docs: README 已知限制补 trim 默认开启的升级醒目提示

配置表行内 ⚠️ 之外，在已知限制一节开头再立一条：升级后订阅 stream 按
trim.maxlen 封顶、堆积超限删最老未消费消息，附调大/关闭与容量估算指引。

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

CLAUDE.md

@@ -32,12 +32,12 @@ mvn -pl mq-starter-core test -Dtest=MqConsumeDispatcherTest#someMethod # 单个
 ```
 mq-starter-core             门面 + 注解扫描 + 消费链 + SPI 契约（中立，不依赖 Spring/Redis 的核心编排逻辑，可纯单测）
 mq-starter-idempotent-redis 消费去重的 Redis 存储（driver 中立，不按 mq.type 门控；任何 driver 有 Redis 就能用）
-mq-starter-redis-stream     Redis Stream driver（去重/批量/落库/reclaim 重试全接入；去重 store 依赖 idempotent-redis 模块）
+mq-starter-redis-stream     Redis Stream driver（去重/批量/落库/reclaim 重试/tag 消费端过滤/订阅 stream 定时裁剪全接入；去重 store 依赖 idempotent-redis 模块。裁剪复用底层 redis-stream.trim.* 配置、默认开启，maxlen<=0 关，只裁订阅过的 stream——中间件自己的 trim 任务只认它的 handler，对 mq-starter 无效）
 mq-starter-rocketmq         RocketMQ driver（收发 + 原生重试/死信 + RETRY_EXHAUSTED 落库 + 去重信号处理；去重 store 经聚合坐标间接获得）
 mq-spring-boot-starter      聚合 jar，业务方只引这一个（恒含 idempotent-redis）
 ```
 
-> ⚠️ **两个 driver 仍有差异**（README 有对照表）：去重、失败落库两端已对齐（2026-07 起 rocketmq `idempotent=true` 生效，需 Redis 可达）；剩余差距是死信模型（redis 只落库 / rocketmq 原生 `%DLQ%`）、批量攒批语义、`concurrency` 两端都未兑现。改跨 driver 行为前先对照 README 能力表。
+> ⚠️ **两个 driver 仍有差异**（README 有对照表）：去重、失败落库、tag 过滤三者两端已对齐（2026-07 起 rocketmq `idempotent=true` 生效、需 Redis 可达；tag：rocketmq broker 原生 / redis-stream 消费端模拟，语义子集见 core 的 `MqTagMatcher`，dispatch 顶部与 reclaim 耗尽分支前两处调用）；剩余差距是死信模型（redis 只落库 / rocketmq 原生 `%DLQ%`）、批量**整批失败**的处理（逐条 ACK 后无法精确重投整批，故落库+吞掉）与**多环境隔离**（redis-stream 经底层 starter 的 `redis-stream.env-isolation`/`env` 拼 `topic_env` 后缀；rocketmq 无对应机制，切换后隔离静默消失）；真正两端都未兑现的只有 `concurrency`。批量攒批本身（含 idle flush 超时/停机强刷，2026-07 已落地）两端已对齐。改跨 driver 行为前先对照 README 能力表。
 >
 > **单独只引 `mq-starter-rocketmq`（不走聚合坐标）时去重需自行加 `mq-starter-idempotent-redis` 依赖**——rocketmq 模块本身不依赖它（刻意零耦合）。
 

README.md

@@ -128,7 +128,8 @@ public void onFile(FileData msg) { }          // 只收 tag=node-01 的消息，
 
 - 消费端不写 `tag` 默认 `*`，整个 topic 全收；发送端用两参 `send` 则消息不带 tag——**对端若按 tag 过滤，不带标签的消息它收不到**，跨系统对接时这是个经典坑。
 - rocketmq 模式：过滤发生在 **broker 服务端**，不匹配的消息根本不会投到消费者；tag 表达式直接透传 RocketMQ 原生语法，`"A || B"` 多选一也支持。
-- redis-stream 模式：**当前不过滤、全收**（见能力表），tag 只是随消息带到消费端，可从 `MqContext.getTag()` 读到，需要分流的话业务自己判断——这也是切换 `mq.type` 时的行为翻转点之一。
+- redis-stream 模式：**消费端模拟过滤**——driver 收到消息后按订阅 tag 比对，不匹配的直接 ACK 丢弃、不进消费链（不碰去重、不算重试、不落库），支持与 rocketmq 相同的写法子集（`*` 全收、单 tag 精确、`A || B` 多选一）。与 rocketmq 的差别只剩过滤位置与成本：消息会先走一趟网络到消费端才被丢弃。
+- **同一 group 的所有订阅必须用同一个 tag 表达式**（含滚动发布期间的新旧实例）——要按 tag 分流，就各用各的 group。同组配不同 tag 时，消息只会投给组内某一个消费者，拿到"不归自己"消息的那个会直接 ACK 吞掉，对方永远收不到（两端皆然，rocketmq 生态称之为「订阅关系不一致」）。
 
 ## 进阶用法
 
@@ -172,26 +173,28 @@ public class MyErrorHandler implements MqConsumeErrorHandler {
 
 ## 两个 driver 不太一样
 
-两套实现的能力大头已经对齐：去重、失败落库两端都接上了。剩下的差异主要在死信模型和批量语义：
+两套实现的能力大头已经对齐：去重、失败落库、批量攒批（含 idle flush）、tag 过滤两端都接上了。剩下的差异主要在死信模型，以及批量的**整批失败处理**（攒批本身两端已对齐，糙点在整批失败被吞、无法精确重投）：
 
 | 能力 | Redis Stream | RocketMQ |
 |---|:---:|:---:|
 | 收发 | ✅ | ✅ |
+| tag 过滤 | ✅ 消费端模拟（不匹配 ACK 丢弃，写法子集同右） | ✅ broker 服务端原生 |
+| 多环境隔离（topic 拼环境后缀） | ✅ 底层 starter 能力（非 mq-starter 实现）：配 `redis-stream.env-isolation=true` + `redis-stream.env=dev` 后实际 stream 名为 `topic_dev`，各环境共用一套 Redis 互不串流 | ❌ topic 原样使用，多环境共用集群需自行隔离（独立集群 / topic 命名约定） |
 | 并发消费 `concurrency` | ⚠️ 单线程 poll，未兑现 | ⚠️ 原生线程池，不受 concurrency 控制 |
 | 去重 `idempotent`（默认开启） | ✅（需 Redis） | ✅（需 Redis） |
 | 失败落库 `RETRY_EXHAUSTED`（需配 `errorHandler`，否则仅 ACK 丢弃） | ✅ | ✅ 落库+ACK |
 | 失败落库 `SINGLE` | ⚠️ 单条：落库+重投；批量：失败被吞（两端一致） | ⚠️ 单条：落库+重投；批量：失败被吞（两端一致） |
-| 批量 `batchSize` | ⚠️ core 内存攒批（driver 逐条喂）；超时/停机强制刷出，仅 kill -9 丢未满批（两端一致） | ⚠️ core 内存攒批（driver 逐条喂）；超时/停机强制刷出，仅 kill -9 丢未满批（两端一致） |
+| 批量 `batchSize` | ✅ core 内存攒批（driver 逐条喂）；超时/停机强制刷出，仅 kill -9 丢未满批（两端一致） | ✅ core 内存攒批（driver 逐条喂）；超时/停机强制刷出，仅 kill -9 丢未满批（两端一致） |
 | 重试 | ✅ PEL reclaim | ✅ 原生重投 |
 | 死信队列 | ❌ 只落库 | ⚠️ 原生 `%DLQ%` 通常被耗尽 ACK 旁路（InProgress 白跑烧光计数时仍会进） |
 | 顺序消息 | ❌ | ❌ 用的 Concurrently |
 | 事务 / 延迟消息 | ❌ | ❌ 没暴露 |
 | 序列化 | fastjson | fastjson |
 
-> ⚠️ **切换 `mq.type` 业务代码零改动，但三处运行时行为会跟着翻转**，切换前按这三条过一遍消费方假设：
+> ⚠️ **切换 `mq.type` 业务代码零改动，但两处运行时行为会跟着翻转**，切换前按这两条过一遍消费方假设：
-> 1. **tag 过滤**：redis-stream 消费端不过滤（全收）、rocketmq 走 broker 服务端按 tag 过滤——依赖 tag 分流的业务切 rocketmq 会"只收匹配的"、切回 redis 会"收到全部"。
+> 1. **并发模型**：redis-stream 单线程串行、rocketmq 原生线程池并发——消费逻辑若有非线程安全假设，切到 rocketmq 可能暴露。
-> 2. **并发模型**：redis-stream 单线程串行、rocketmq 原生线程池并发——消费逻辑若有非线程安全假设，切到 rocketmq 可能暴露。
+> 2. **顺序**：两端都不保证严格有序（rocketmq 用的 Concurrently 而非 Orderly）。
-> 3. **顺序**：两端都不保证严格有序（rocketmq 用的 Concurrently 而非 Orderly）。
+> 3. **多环境隔离**：redis-stream 的环境后缀是底层 starter 拼的，rocketmq 没有对应机制——切过去后隔离**静默消失**，多环境共用一套 RocketMQ 时同名 topic+group 会被当成同一消费组互相抢消息，切换前先规划好隔离方式。
 
 去重按**消息 key** 判重：经 `MqTemplate` 发送时自动生成雪花 ID、天然唯一，业务无感；**不经 `MqTemplate` 的异构发送方必须自己保证 key 逐条唯一**（redis-stream = 信封 `key` field，rocketmq = 报文头 KEYS 属性；为空或固定值会把不同消息误判成重复而静默丢弃，详见「Redis Stream 消息结构」一节及其末尾的 RocketMQ 补注）。去重的状态存在 Redis（`processing / success / fail` 三态 + TTL），由独立模块 `mq-starter-idempotent-redis` 提供，**不看 `mq.type`**——所以 RocketMQ 部署只要能连上 Redis（标准 `spring.redis.*` 配置），去重一样生效。引聚合坐标的话这个模块已经打在里面；单独只引 `mq-starter-rocketmq` 的话需要自己再加一个 `mq-starter-idempotent-redis` 依赖。**去重默认开启**，消费链路会读写 Redis，务必保证 Redis 可达——连不上时消费抛连接异常走重投（日志可见、不是静默丢，但业务会一直消费不进去）。确实不需要去重的监听器显式写 `idempotent = false`；或配 `mq.idempotent.enabled=false` 全局关掉 store（关掉后去重不生效、启动打 WARN 提示，完全不碰 Redis、不影响启动）。
 
@@ -202,7 +205,7 @@ redis-stream 模式下，消息在 Redis 里**不是裸 JSON**，而是固定五
 | field | 内容 |
 |---|---|
 | `key` | 消息业务 key，**每条必须唯一**——消费端去重按它判重（`idempotent` 默认开启），key 重复或恒定会把不同消息误判成重复而丢弃。经 `MqTemplate` 发送时自动生成雪花 ID，无需关心 |
-| `tag` | 路由 / 过滤标记，可为空串（注意 redis-stream 消费端当前不按 tag 过滤，全收） |
+| `tag` | 路由 / 过滤标记，可为空串——消费端按订阅 tag 过滤，不带 tag 的消息只有订 `*`（缺省）的监听器收得到 |
 | `enc` | **body 的编码方式，压缩标识就放这里**：`plain`（明文 JSON）或 `gzip`（压缩，见下） |
 | `body` | 消息体：`enc=plain` 时为 JSON 明文；`enc=gzip` 时为 `base64(gzip(json))`——先标准 gzip（RFC1952，带 gzip 头），再 base64（基础编码、不带换行） |
 | `ts` | 发送时间戳（毫秒） |
@@ -266,7 +269,7 @@ core 是中立的，编排逻辑不依赖 Spring 和 Redis，可以直接单测
 | `batchSize` | 1 | 攒批大小，大于 1 要用 `List` 参数；与首参是否 `List` 错配时启动打 WARN、仍按首参跑 |
 | `batchTimeoutMs` | 60000 | 批量攒批最长等待毫秒数（从本批第一条算起，到点未满也强制刷出）；0/负数关闭；仅批量监听器生效 |
 | `maxRetry` | 3 | 最大重试次数（redis-stream 按 PEL delivery 计数判耗尽 / rocketmq 写入 broker `maxReconsumeTimes`；被去重标记挡回的 InProgress 白跑**两端都计入**） |
-| `tag` | `*` | 只收打了指定 tag 的消息（发送端三参 `send` 的第二参打 tag）。rocketmq 为 broker 服务端过滤、支持原生表达式（如 `A \|\| B`）；redis-stream 当前不过滤、全收。详见「消费」一节的 tag 小节 |
+| `tag` | `*` | 只收打了指定 tag 的消息（发送端三参 `send` 的第二参打 tag）。两端都生效：rocketmq 为 broker 服务端过滤（原生表达式透传）；redis-stream 为消费端模拟过滤，支持相同写法子集（`*`、单 tag、`A \|\| B`）。详见「消费」一节的 tag 小节 |
 | `idempotent` | true | 是否去重（默认开启，需 Redis 可达；单个监听器关闭写 `idempotent = false`，全局关闭配 `mq.idempotent.enabled=false`） |
 | `errorHandler` | `""` | 失败落库的 bean 名 |
 
@@ -288,7 +291,18 @@ Redis Stream 这边的参数（前缀 `mq.redis-stream`）：
 |---|---|---|
 | `reclaim-min-idle` | 30s | 消息投出去多久没 ack 才 reclaim 重投，要大于单条处理耗时；**不要调到比 `processing-ttl` 还大**（需保持 `processing-ttl ≥ reclaim-min-idle`，否则存活的慢消费者会被 reclaim 重投副本并行双消费）；也别调得太小——每次 reclaim（xClaim）都使 PEL delivery +1，被 processing 挡回的白跑同样计入，需保持 `maxRetry × reclaim-min-idle > processing-ttl`，否则崩溃残留会在业务没重跑的情况下烧光重试直接判耗尽 |
 
-Redis 连接走标准 `spring.redis.*`；更底层的 `redis-stream.*`（readCount / blockMs / consumerName 等）来自内部 stream 中间件（`redis-stream-springboot-starter`），一般不用动。
+Redis 连接走标准 `spring.redis.*`。另有一组更底层的参数（前缀 `redis-stream`，来自内部 stream 中间件 `redis-stream-springboot-starter`，mq-starter 的 driver 复用其中一部分）：
+
+| 配置 | 默认 | 说明 |
+|---|---|---|
+| `env-isolation` | false | 多环境隔离开关：开启且 `env` 非空时，实际 stream 名 = `topic_环境值`（如 `order_dev`），各环境共用一套 Redis 互不串流。发送、订阅两侧都生效；跨应用收发时两边的 `env` 配置必须一致，否则流名对不上收不到 |
+| `env` | 空 | 环境标识（`dev` / `test` / `prod`…），配合上面的开关用 |
+| `read-count` | 10 | 单次 XREADGROUP 拉取条数。大堆积追赶想快些可调大（如 100），代价是单批占内存多、批内逐条处理时间拉长 |
+| `block-ms` | 2000 | 拉取无消息时的阻塞等待毫秒数，到点空手返回进下一轮 |
+| `consumer-name` | 空 | 组内消费者名。留空则每次启动生成随机名（推荐，重启遗留的 PEL 由 reclaim 拿回）；写死会让多实例同名、PEL 归属混在一起 |
+| `max-retry` | 3 | 兜底重试次数，仅 `@MqListener` 的 `maxRetry` 配成 0/负数时才用到 |
+| `trim.maxlen` / `trim.interval-ms` | 100000 / 60s | stream 封顶裁剪，**默认开启**：driver 对**订阅过**的 stream 每周期执行一次 `XTRIM MAXLEN ~ maxlen`（`~` 近似裁剪，实际长度可能略超 maxlen）。⚠️ 裁剪只按条数砍最老的、**不看消费进度**——堆积超过 maxlen 时未消费的老消息也会被删，容量按「可容忍的最大停机时长 × 生产速率」估算；`maxlen` 配 0/负数关闭；纯发送、不订阅的 topic 不裁（裁剪节奏归消费方） |
+| `autoclaim.*` | — | 对 mq-starter **不生效**：中间件的 autoclaim 只认自己注册的 handler 的 stream（无 handler 任务不启动），mq-starter 不走那套。mq-starter 有自己的 reclaim，配上表的 `mq.redis-stream.reclaim-min-idle` |
 
 RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置（`name-server`、`producer.group` 等）。
 
@@ -319,10 +333,12 @@ RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置（`nam
 多半是「处理中」标记没过期挡住了重投：消费者崩溃残留标记，或单条处理耗时超过 `mq.idempotent.processing-ttl`。等 TTL 过期会自动恢复；调 `processing-ttl` 时注意配置表里的上下界约束。rocketmq 下持续重投烧光 `maxReconsumeTimes` 会进原生 `%DLQ%`（进死信，不是丢消息）。
 
 **从 redis-stream 切到 rocketmq 后行为变了（多收 / 漏收、并发不一样）？**
-正常，见「两个 driver 不太一样」里的三条运行时翻转：tag 过滤、并发模型、顺序保证。
+正常，见「两个 driver 不太一样」里的两条运行时翻转：并发模型、顺序保证。
 
 ## 已知限制 / TODO
 
+> ⚠️ **升级注意（redis-stream）**：订阅 stream 的定时裁剪**默认开启**（2026-07 起）。升级后每个订阅过的 stream 按 `redis-stream.trim.maxlen`（未配则 10 万条）封顶，**堆积超过封顶时最老的未消费消息会被直接删掉**——裁剪只按条数、不看消费进度。堆积可能超限的场景，升级前把 `trim.maxlen` 调大或配 0/负数关闭；容量按「可容忍的最大停机时长 × 生产速率」估算，详见「配置项」一节的 `redis-stream.*` 表。
+
 按优先级列，详细的在 [`docs/mq-starter-改进优化清单.md`](./docs/mq-starter-改进优化清单.md)。
 
 P0，放量到生产前想补的：

mq-starter-core/src/main/java/com/njcn/mq/spi/MqTagMatcher.java

@@ -0,0 +1,42 @@
+package com.njcn.mq.spi;
+
+/**
+ * 订阅 tag 表达式匹配，语义对齐 RocketMQ 原生 tag 表达式**子集**：
+ * {@code *}/空 = 全收、单 tag 精确匹配、{@code A || B} 多选一。不支持 SQL92 / 否定 / 通配。
+ *
+ * <p>无服务端过滤能力的 driver（如 redis-stream）在消费端调用本类模拟过滤；
+ * 语义保持单一来源，勿在 driver 内自行实现比对逻辑。rocketmq driver 无需本类（broker 原生过滤）。
+ */
+public final class MqTagMatcher {
+
+    private MqTagMatcher() {
+    }
+
+    /**
+     * @param subscriptionTag 订阅方声明的 tag 表达式（{@code @MqListener.tag}，缺省 {@code *}）
+     * @param messageTag      消息携带的 tag（null/空 = 未打标签，只被全收表达式命中）
+     * @return true = 该消息归本订阅消费
+     */
+    public static boolean matches(String subscriptionTag, String messageTag) {
+        if (subscriptionTag == null) {
+            return true;
+        }
+        String expr = subscriptionTag.trim();
+        if (expr.isEmpty() || "*".equals(expr)) {
+            return true;
+        }
+        boolean hasValidPart = false;
+        for (String part : expr.split("\\|\\|")) {
+            String p = part.trim();
+            if (p.isEmpty()) {
+                continue; // 空段忽略（如 "A ||" 的尾段）
+            }
+            hasValidPart = true;
+            if (p.equals(messageTag)) {
+                return true;
+            }
+        }
+        // 全空段（如 "||"）= 没声明任何有效 tag = 视同全收；有有效段但都没命中 = 不匹配
+        return !hasValidPart;
+    }
+}

mq-starter-core/src/test/java/com/njcn/mq/spi/MqTagMatcherTest.java

@@ -0,0 +1,45 @@
+package com.njcn.mq.spi;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class MqTagMatcherTest {
+
+    @Test
+    void catchAllExpressions_alwaysMatch() {
+        assertTrue(MqTagMatcher.matches(null, "A"));
+        assertTrue(MqTagMatcher.matches("", "A"));
+        assertTrue(MqTagMatcher.matches("   ", "A"));
+        assertTrue(MqTagMatcher.matches("*", "A"));
+        assertTrue(MqTagMatcher.matches("||", "A"));   // 全空段 = 空表达式 = 全收
+        assertTrue(MqTagMatcher.matches("*", null));   // 全收对无 tag 消息同样放行
+        assertTrue(MqTagMatcher.matches("*", ""));
+    }
+
+    @Test
+    void singleTag_exactCaseSensitiveMatch() {
+        assertTrue(MqTagMatcher.matches("A", "A"));
+        assertFalse(MqTagMatcher.matches("A", "B"));
+        assertFalse(MqTagMatcher.matches("A", "a"));   // 大小写敏感
+        assertTrue(MqTagMatcher.matches(" A ", "A"));  // 订阅侧逐段 trim
+        assertFalse(MqTagMatcher.matches("A", "A "));  // 消息侧不 trim
+    }
+
+    @Test
+    void orExpression_matchesAnyValidPart() {
+        assertTrue(MqTagMatcher.matches("A || B", "A"));
+        assertTrue(MqTagMatcher.matches("A || B", "B"));
+        assertFalse(MqTagMatcher.matches("A || B", "C"));
+        assertTrue(MqTagMatcher.matches("A ||", "A")); // 空段忽略，不影响有效段
+        assertFalse(MqTagMatcher.matches("A ||", "B"));
+    }
+
+    @Test
+    void untaggedMessage_onlyMatchedByCatchAll() {
+        assertFalse(MqTagMatcher.matches("A", null));
+        assertFalse(MqTagMatcher.matches("A", ""));
+        assertFalse(MqTagMatcher.matches("A || B", null));
+    }
+}

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java

@@ -8,6 +8,7 @@ import com.njcn.mq.core.MqMessage;
 import com.njcn.mq.spi.MqDriver;
 import com.njcn.mq.spi.MqMessageInProgressException;
 import com.njcn.mq.spi.MqSubscription;
+import com.njcn.mq.spi.MqTagMatcher;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.connection.stream.*;
 import org.springframework.data.domain.Range;
@@ -26,6 +27,7 @@ public class RedisStreamMqDriver implements MqDriver {
     private final RedisStreamProperties props;
     private final long reclaimMinIdleMs;
     private final List<Thread> workers = new ArrayList<>();
+    private final RedisStreamTrimScheduler trimScheduler;
     private volatile boolean running = true;
 
     public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props) {
@@ -37,6 +39,7 @@ public class RedisStreamMqDriver implements MqDriver {
     public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props, long reclaimMinIdleMs) {
         this.redis = redis; this.keyBuilder = keyBuilder; this.props = props;
         this.reclaimMinIdleMs = reclaimMinIdleMs;
+        this.trimScheduler = new RedisStreamTrimScheduler(redis, props);
     }
 
     @Override public void send(String topic, MqMessage message) {
@@ -56,6 +59,7 @@ public class RedisStreamMqDriver implements MqDriver {
         String stream = keyBuilder.buildStream(sub.getTopic());
         String group = sub.getGroup();
         ensureGroup(stream, group);
+        trimScheduler.register(stream);
         String consumer = (props.getConsumerName() == null || props.getConsumerName().isEmpty())
                 ? "mq-" + UUID.randomUUID() : props.getConsumerName();
         Thread t = new Thread(() -> loop(sub, stream, group, consumer), "mq-redis-" + stream + "-" + group);
@@ -87,12 +91,23 @@ public class RedisStreamMqDriver implements MqDriver {
 
     /** @return true=ACK；false=留 PEL（重试）。 */
     private boolean dispatch(MqSubscription sub, Map<String, String> fields, int maxRetry) {
+        String msgTag = fields.get(StreamMessageConstant.F_TAG);
+        if (!MqTagMatcher.matches(sub.getTag(), msgTag)) {
+            // tag 不匹配：不归本订阅消费，直接 ACK 丢弃（对齐 rocketmq broker 过滤语义，只是位置在消费端）。
+            // 置于反序列化之前：不碰去重标记、不进攒批、不走重试，连 JSON 都不用解。
+            log.debug("[mq-redis] tag 不匹配，ACK 跳过 key={} msgTag={} subTag={}",
+                    fields.get(StreamMessageConstant.F_KEY), msgTag, sub.getTag());
+            return true;
+        }
         Object payload;
         try {
             String json = MessageCodec.decodeBody(fields.get(StreamMessageConstant.F_ENC), fields.get(StreamMessageConstant.F_BODY));
             payload = JSON.parseObject(json, sub.getPayloadType());
         } catch (Exception poison) {
-            log.error("[mq-redis] 消息反序列化失败，按毒消息 ACK 丢弃 key={}", fields.get(StreamMessageConstant.F_KEY), poison);
+            String key = fields.get(StreamMessageConstant.F_KEY);
+            log.error("[mq-redis] 消息反序列化失败，按毒消息 ACK 丢弃 key={} type={} message={}",
+                    key, poison.getClass().getSimpleName(), poison.getMessage());
+            log.debug("[mq-redis] 消息反序列化失败堆栈 key={}", key, poison);
             return true; // 毒消息直接 ACK，不卡队列
         }
         MqMessage msg = MqMessage.of(fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), payload);
@@ -137,6 +152,13 @@ public class RedisStreamMqDriver implements MqDriver {
             Map<String, String> fields = new HashMap<>();
             rec.getValue().forEach((k, v) -> fields.put(
                     new String(k, StandardCharsets.UTF_8), new String(v, StandardCharsets.UTF_8)));
+            if (!MqTagMatcher.matches(sub.getTag(), fields.get(StreamMessageConstant.F_TAG))) {
+                // tag 不匹配：不归本组消费，直接 ACK 丢弃——不得进耗尽分支以本组名义落库，也无须重投。
+                redis.opsForStream().acknowledge(stream, group, rec.getId());
+                log.debug("[mq-redis] tag 不匹配（reclaim），ACK 丢弃 key={} delivery={}",
+                        fields.get(StreamMessageConstant.F_KEY), deliveryCount);
+                continue;
+            }
             if (deliveryCount > maxRetry) {
                 if (sub.getRetryExhaustedHandler() != null) {
                     try {
@@ -186,5 +208,5 @@ public class RedisStreamMqDriver implements MqDriver {
     }
     private static void sleep(long ms) { try { Thread.sleep(ms); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); } }
 
-    @Override public void stop() { running = false; for (Thread t : workers) t.interrupt(); workers.clear(); }
+    @Override public void stop() { running = false; trimScheduler.stop(); for (Thread t : workers) t.interrupt(); workers.clear(); }
 }

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamTrimScheduler.java

@@ -0,0 +1,76 @@
+package com.njcn.mq.driver.redis;
+
+import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 订阅 stream 的定时裁剪（{@code XTRIM <stream> MAXLEN ~ <maxlen>}，近似裁剪）。
+ *
+ * <p>底层中间件的 StreamTrimScheduler 只裁自己注册的 handler 的 stream，而 mq-starter 的消费
+ * 不走 handler 体系——driver 不自己裁的话，只经 mq-starter 收发的 topic 无界增长。配置复用中间件的
+ * {@code redis-stream.trim.*}（默认开启：maxlen 10 万 / 周期 60s）。
+ *
+ * <p>两条语义约束：{@code maxlen<=0} 视为关闭——XTRIM 0 会清空整条 stream，绝不能发出；
+ * 只裁「订阅过」的 stream——与中间件“handler 即消费者”的语义对齐，纯发送方不裁
+ * （裁剪节奏该由消费方掌握，发送方不知道还有谁没消费到哪）。
+ */
+@Slf4j
+class RedisStreamTrimScheduler {
+    private final StringRedisTemplate redis;
+    private final RedisStreamProperties props;
+    private final Set<String> streams = ConcurrentHashMap.newKeySet();
+    private ScheduledExecutorService scheduler;
+
+    RedisStreamTrimScheduler(StringRedisTemplate redis, RedisStreamProperties props) {
+        this.redis = redis;
+        this.props = props;
+    }
+
+    /** 订阅时登记 stream；首个登记且 maxlen&gt;0 时拉起调度线程（重复登记幂等）。 */
+    synchronized void register(String stream) {
+        streams.add(stream);
+        long maxlen = props.getTrim().getMaxlen();
+        if (scheduler != null || maxlen <= 0) return;
+        long interval = props.getTrim().getIntervalMs();
+        scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+            Thread t = new Thread(r, "mq-redis-trim");
+            t.setDaemon(true);
+            return t;
+        });
+        scheduler.scheduleWithFixedDelay(this::trimAll, interval, interval, TimeUnit.MILLISECONDS);
+        log.info("[mq-redis] stream 裁剪调度启动 interval={}ms maxlen={}（近似裁剪，只裁订阅过的 stream）", interval, maxlen);
+    }
+
+    void trimAll() {
+        long maxlen = props.getTrim().getMaxlen();
+        if (maxlen <= 0) return;
+        for (String stream : streams) {
+            try {
+                redis.execute((RedisCallback<Object>) c -> c.execute("XTRIM",
+                        stream.getBytes(StandardCharsets.UTF_8),
+                        "MAXLEN".getBytes(StandardCharsets.UTF_8),
+                        "~".getBytes(StandardCharsets.UTF_8),
+                        String.valueOf(maxlen).getBytes(StandardCharsets.UTF_8)));
+            } catch (Exception e) {
+                // 单条失败不得中断其余 stream 的裁剪
+                log.warn("[mq-redis] stream 裁剪失败 stream={}", stream, e);
+            }
+        }
+    }
+
+    synchronized void stop() {
+        if (scheduler != null) {
+            scheduler.shutdownNow();
+            scheduler = null;
+        }
+    }
+}

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverPoisonLogTest.java

@@ -0,0 +1,82 @@
+package com.njcn.mq.driver.redis;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import com.njcn.middle.stream.constant.StreamMessageConstant;
+import com.njcn.mq.spi.MqSubscription;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class RedisStreamMqDriverPoisonLogTest {
+
+    private final ListAppender<ILoggingEvent> logAppender = new ListAppender<>();
+    private Logger logger;
+    private Level originalLevel;
+
+    @BeforeEach
+    void attachAppender() {
+        logger = (Logger) LoggerFactory.getLogger(RedisStreamMqDriver.class);
+        originalLevel = logger.getLevel();
+        logger.setLevel(Level.DEBUG);
+        logAppender.start();
+        logger.addAppender(logAppender);
+    }
+
+    @AfterEach
+    void detachAppender() {
+        logger.detachAppender(logAppender);
+        logger.setLevel(originalLevel);
+    }
+
+    @Test
+    void poisonMessage_logsErrorSummaryAndDebugStackTrace() throws Exception {
+        RedisStreamMqDriver driver = new RedisStreamMqDriver(null, null, null);
+        MqSubscription sub = MqSubscription.builder()
+                .tag("*")
+                .payloadType(String.class)
+                .listener(msg -> fail("poison message must not reach listener"))
+                .build();
+
+        Map<String, String> fields = new HashMap<>();
+        fields.put(StreamMessageConstant.F_KEY, "poison-1");
+        fields.put(StreamMessageConstant.F_TAG, "");
+        fields.put(StreamMessageConstant.F_ENC, "plain");
+        fields.put(StreamMessageConstant.F_BODY, "new");
+
+        Method dispatch = RedisStreamMqDriver.class.getDeclaredMethod("dispatch", MqSubscription.class, Map.class, int.class);
+        dispatch.setAccessible(true);
+
+        Object ack = dispatch.invoke(driver, sub, fields, 3);
+
+        assertEquals(Boolean.TRUE, ack, "poison message should still be ACKed");
+        List<ILoggingEvent> errors = events(Level.ERROR);
+        assertEquals(1, errors.size());
+        ILoggingEvent error = errors.get(0);
+        assertNull(error.getThrowableProxy(), "ERROR summary should not print a stack trace by default");
+        assertTrue(error.getFormattedMessage().contains("key=poison-1"));
+        assertTrue(error.getFormattedMessage().contains("type=JSONException"));
+        assertTrue(error.getFormattedMessage().contains("message="));
+
+        List<ILoggingEvent> debugs = events(Level.DEBUG);
+        assertEquals(1, debugs.size());
+        assertNotNull(debugs.get(0).getThrowableProxy(), "DEBUG log should retain the stack trace for deep diagnosis");
+    }
+
+    private List<ILoggingEvent> events(Level level) {
+        return logAppender.list.stream()
+                .filter(e -> e.getLevel() == level)
+                .collect(Collectors.toList());
+    }
+}

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverTagFilterIT.java

@@ -0,0 +1,123 @@
+package com.njcn.mq.driver.redis;
+
+import com.njcn.mq.core.MqMessage;
+import com.njcn.mq.spi.MqSubscription;
+import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
+import com.njcn.middle.stream.support.StreamKeyBuilder;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.connection.stream.*;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+class RedisStreamMqDriverTagFilterIT {
+
+    @Test
+    void tagFilter_onlyMatchingDelivered_othersAckedCleanly() throws Exception {
+        StringRedisTemplate redis = newRedisOrSkip();
+        StreamKeyBuilder kb = new StreamKeyBuilder(false, "");
+        RedisStreamProperties props = new RedisStreamProperties();
+        props.setConsumerName("it-tag");
+        props.setReadCount(10);
+        props.setBlockMs(300);
+        props.setMaxRetry(3);
+        RedisStreamMqDriver driver = new RedisStreamMqDriver(redis, kb, props, 200L);
+
+        String topic = "MQ_TAG_IT_" + System.nanoTime();
+        BlockingQueue<String> got = new LinkedBlockingQueue<>();
+        driver.subscribe(MqSubscription.builder()
+                .topic(topic).group("g1").tag("A").payloadType(String.class)
+                .concurrency(1).batchSize(1).maxRetry(3)
+                .listener(m -> got.add((String) m.getPayload()))
+                .build());
+
+        driver.send(topic, MqMessage.of("k-a", "A", "match"));
+        driver.send(topic, MqMessage.of("k-b", "B", "mismatch"));
+        driver.send(topic, MqMessage.of("k-none", "", "untagged"));
+
+        assertEquals("match", got.poll(5, TimeUnit.SECONDS), "tag=A 的消息应正常消费");
+        assertNull(got.poll(1500, TimeUnit.MILLISECONDS), "tag 不匹配/未打 tag 的消息不得进业务方法");
+
+        Thread.sleep(500); // 留时间让过滤 ACK 落地
+        PendingMessagesSummary summary = redis.opsForStream().pending(kb.buildStream(topic), "g1");
+        assertEquals(0L, summary.getTotalPendingMessages(), "被过滤的消息应 ACK 干净，PEL 无残留");
+        driver.stop();
+    }
+
+    @Test
+    void reclaimExhausted_mismatchedTag_ackedWithoutExhaustedCallback() throws Exception {
+        StringRedisTemplate redis = newRedisOrSkip();
+        StreamKeyBuilder kb = new StreamKeyBuilder(false, "");
+        RedisStreamProperties props = new RedisStreamProperties();
+        props.setConsumerName("it-tag-reclaim");
+        props.setReadCount(10);
+        props.setBlockMs(300);
+        props.setMaxRetry(1);
+        RedisStreamMqDriver driver = new RedisStreamMqDriver(redis, kb, props, 200L);
+
+        String topic = "MQ_TAG_RECLAIM_IT_" + System.nanoTime();
+        String stream = kb.buildStream(topic);
+
+        // 1. 手工建组（driver 尚未 subscribe，模拟历史遗留 PEL 场景）
+        redis.execute((RedisCallback<Object>) c -> c.execute("XGROUP",
+                "CREATE".getBytes(StandardCharsets.UTF_8), stream.getBytes(StandardCharsets.UTF_8),
+                "g1".getBytes(StandardCharsets.UTF_8), "$".getBytes(StandardCharsets.UTF_8),
+                "MKSTREAM".getBytes(StandardCharsets.UTF_8)));
+
+        // 2. 发一条 tag=B 的消息（send 不依赖 subscribe）
+        driver.send(topic, MqMessage.of("k-stale", "B", "stale"));
+
+        // 3. 手工 XREADGROUP 不 ACK：delivery=1，模拟消费者读走后崩溃
+        List<MapRecord<String, Object, Object>> recs = redis.opsForStream().read(
+                Consumer.from("g1", "seed"),
+                StreamReadOptions.empty().count(1),
+                StreamOffset.create(stream, ReadOffset.lastConsumed()));
+        assertNotNull(recs);
+        assertEquals(1, recs.size(), "种子消息应已进 PEL");
+        RecordId staleId = recs.get(0).getId();
+
+        // 4. 手工 XCLAIM 一次：delivery=2 > maxRetry=1，reclaim 将走耗尽分支
+        redis.execute((RedisCallback<Object>) conn -> conn.streamCommands()
+                .xClaim(stream.getBytes(StandardCharsets.UTF_8), "g1", "seed2", Duration.ZERO, staleId));
+
+        // 5. 启动订阅 tag=A 的 driver，等 reclaim 处理这条 tag=B 的残留
+        BlockingQueue<MqMessage> exhausted = new LinkedBlockingQueue<>();
+        BlockingQueue<Object> delivered = new LinkedBlockingQueue<>();
+        driver.subscribe(MqSubscription.builder()
+                .topic(topic).group("g1").tag("A").payloadType(String.class)
+                .concurrency(1).batchSize(1).maxRetry(1)
+                .listener(m -> delivered.add(m.getPayload()))
+                .retryExhaustedHandler(exhausted::add)
+                .build());
+
+        assertNull(exhausted.poll(3, TimeUnit.SECONDS), "tag 不匹配的消息不得以本组名义耗尽落库");
+        assertTrue(delivered.isEmpty(), "tag 不匹配的消息不得进业务方法");
+        PendingMessagesSummary summary = redis.opsForStream().pending(stream, "g1");
+        assertEquals(0L, summary.getTotalPendingMessages(), "应被 ACK 丢弃，PEL 清空");
+        driver.stop();
+    }
+
+    private static StringRedisTemplate newRedisOrSkip() {
+        try {
+            LettuceConnectionFactory cf = new LettuceConnectionFactory("localhost", 6379);
+            cf.afterPropertiesSet();
+            StringRedisTemplate t = new StringRedisTemplate(cf);
+            t.afterPropertiesSet();
+            t.hasKey("ping"); // 触发真实连接
+            return t;
+        } catch (Exception e) { assumeTrue(false, "no local redis, skip"); return null; }
+    }
+}

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverTrimTest.java

@@ -0,0 +1,181 @@
+package com.njcn.mq.driver.redis;
+
+import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
+import com.njcn.middle.stream.support.StreamKeyBuilder;
+import com.njcn.mq.spi.MqSubscription;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.StreamOperations;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * 订阅 stream 的定时裁剪（XTRIM MAXLEN ~）。
+ *
+ * <p>底层中间件的 trim 任务只认自己注册的 handler，mq-starter 不走那套——
+ * 所以 driver 必须自己裁订阅过的 stream，否则只经 mq-starter 收发的 topic 无界增长。
+ * 配置复用中间件的 {@code redis-stream.trim.*}（maxlen 默认 10 万 / 周期默认 60s），默认开启。
+ */
+class RedisStreamMqDriverTrimTest {
+
+    /** 记录 connection.execute 收到的每次调用：[command, arg1, arg2, ...]（args 按 UTF-8 还原成字符串）。 */
+    private final List<List<String>> executed = Collections.synchronizedList(new ArrayList<>());
+    private StringRedisTemplate redis;
+
+    @BeforeEach
+    void mockRedis() {
+        redis = mock(StringRedisTemplate.class);
+        RedisConnection conn = recordingConnection(null);
+        when(redis.execute(any(RedisCallback.class))).thenAnswer(inv ->
+                ((RedisCallback<?>) inv.getArgument(0)).doInRedis(conn));
+    }
+
+    /**
+     * 记录 execute 调用的 RedisConnection——用默认 Answer 而非参数匹配，
+     * 因为 execute(String, byte[]...) 是 varargs，这版 Mockito 的 any(byte[][].class) 匹配不上。
+     */
+    private RedisConnection recordingConnection(String failStream) {
+        return mock(RedisConnection.class, inv -> {
+            if ("toString".equals(inv.getMethod().getName())) return "mock-conn";
+            if (!"execute".equals(inv.getMethod().getName())) return null;
+            List<String> call = new ArrayList<>();
+            for (Object arg : inv.getArguments()) {
+                if (arg instanceof byte[]) call.add(new String((byte[]) arg, StandardCharsets.UTF_8));
+                else if (arg instanceof byte[][]) {
+                    for (byte[] b : (byte[][]) arg) call.add(new String(b, StandardCharsets.UTF_8));
+                } else call.add(String.valueOf(arg));
+            }
+            if (failStream != null && call.size() > 1 && failStream.equals(call.get(1))) {
+                throw new IllegalStateException("boom");
+            }
+            executed.add(call);
+            return null;
+        });
+    }
+
+    private static RedisStreamProperties propsWithTrim(long maxlen, long intervalMs) {
+        RedisStreamProperties props = new RedisStreamProperties();
+        props.getTrim().setMaxlen(maxlen);
+        props.getTrim().setIntervalMs(intervalMs);
+        props.setConsumerName("test-consumer");
+        return props;
+    }
+
+    private List<List<String>> xtrims() {
+        synchronized (executed) {
+            return executed.stream().filter(c -> "XTRIM".equals(c.get(0))).collect(Collectors.toList());
+        }
+    }
+
+    // ---- RedisStreamTrimScheduler 单元 ----
+
+    @Test
+    void trimAll_issuesApproximateXtrimForEachRegisteredStream() {
+        RedisStreamTrimScheduler trim = new RedisStreamTrimScheduler(redis, propsWithTrim(50000, 60000));
+        trim.register("order_dev");
+        trim.register("pay_dev");
+
+        trim.trimAll();
+
+        List<List<String>> calls = xtrims();
+        assertThat(calls).hasSize(2);
+        assertThat(calls).allSatisfy(c -> {
+            assertThat(c.subList(2, 5)).as("XTRIM <stream> MAXLEN ~ <maxlen>")
+                    .containsExactly("MAXLEN", "~", "50000");
+        });
+        assertThat(calls.stream().map(c -> c.get(1)))
+                .containsExactlyInAnyOrder("order_dev", "pay_dev");
+        trim.stop();
+    }
+
+    @Test
+    void register_sameStreamTwice_trimsOnlyOnce() {
+        RedisStreamTrimScheduler trim = new RedisStreamTrimScheduler(redis, propsWithTrim(100, 60000));
+        trim.register("order_dev");
+        trim.register("order_dev");
+
+        trim.trimAll();
+
+        assertThat(xtrims()).hasSize(1);
+        trim.stop();
+    }
+
+    @Test
+    void maxlenNotPositive_treatedAsDisabled_neverTrims() {
+        RedisStreamTrimScheduler trim = new RedisStreamTrimScheduler(redis, propsWithTrim(0, 60000));
+        trim.register("order_dev");
+
+        trim.trimAll();
+
+        assertThat(xtrims()).as("maxlen<=0 视为关闭，XTRIM 0 会清空整条 stream，绝不能发出").isEmpty();
+        trim.stop();
+    }
+
+    @Test
+    void oneStreamFails_othersStillTrimmed() {
+        RedisConnection conn = recordingConnection("bad_dev");
+        when(redis.execute(any(RedisCallback.class))).thenAnswer(inv ->
+                ((RedisCallback<?>) inv.getArgument(0)).doInRedis(conn));
+
+        RedisStreamTrimScheduler trim = new RedisStreamTrimScheduler(redis, propsWithTrim(100, 60000));
+        trim.register("bad_dev");
+        trim.register("good_dev");
+
+        trim.trimAll(); // bad_dev 抛异常不得中断 good_dev
+
+        List<List<String>> calls = xtrims();
+        assertThat(calls).hasSize(1);
+        assertThat(calls.get(0).get(1)).isEqualTo("good_dev");
+        trim.stop();
+    }
+
+    // ---- driver 接线：订阅登记 + 周期触发 + 停机停止 ----
+
+    @Test
+    void subscribe_startsPeriodicTrimOfSubscribedStream_andStopHaltsIt() throws Exception {
+        RedisStreamProperties props = propsWithTrim(1234, 50); // 50ms 一裁，便于观察
+        StreamOperations<String, Object, Object> streamOps = mock(StreamOperations.class);
+        when(redis.opsForStream()).thenReturn(streamOps);
+        when(streamOps.read(any(org.springframework.data.redis.connection.stream.Consumer.class),
+                any(), any())).thenAnswer(inv -> { Thread.sleep(20); return Collections.emptyList(); });
+
+        RedisStreamMqDriver driver = new RedisStreamMqDriver(
+                redis, new StreamKeyBuilder(true, "dev"), props);
+        driver.subscribe(MqSubscription.builder()
+                .topic("t-order").group("g1").tag("*").maxRetry(3)
+                .payloadType(String.class).listener(msg -> { }).build());
+
+        List<List<String>> calls = awaitXtrim(2000);
+        assertThat(calls).as("订阅后应周期性裁剪").isNotEmpty();
+        assertThat(calls.get(0).get(1)).as("裁的是拼了环境后缀的实际 stream 名").isEqualTo("t-order_dev");
+        assertThat(calls.get(0).subList(2, 5)).containsExactly("MAXLEN", "~", "1234");
+
+        driver.stop();
+        Thread.sleep(120); // 让在途 tick 收尾
+        int settled = xtrims().size();
+        Thread.sleep(250);
+        assertThat(xtrims()).as("stop 后不得再裁").hasSize(settled);
+    }
+
+    private List<List<String>> awaitXtrim(long timeoutMs) throws InterruptedException {
+        long deadline = System.currentTimeMillis() + timeoutMs;
+        while (System.currentTimeMillis() < deadline) {
+            List<List<String>> calls = xtrims();
+            if (!calls.isEmpty()) return calls;
+            Thread.sleep(20);
+        }
+        return xtrims();
+    }
+}