fix(server): 账号被顶下线/踢下线后释放其全部案件数据源

兜底回收器原先按「账号是否还有有效 token」判活：被顶下线的用户通常马上在新设备重新登录，导致监听器因在途任务跳过释放的案件永远等不到回收。

- CaseDataSourceReaper 改为判断「打开案件的 token」是否仍有效（被顶/被踢/登出/过期都会失效），任务结束后下一轮扫描即回收，多节点部署下同样成立
- CaseDataSourceRegistry 幂等开案时刷新归属 token，避免用户重新登录后重开同一案件时，回收器按旧 token 把新会话正在用的数据源关掉
- 补充 CaseDataSourceReaperTest / CaseDataSourceRegistryTest

ai-server/src/main/java/com/zsjz/ai/common/datasource/CaseDataSourceReaper.java

@@ -16,8 +16,9 @@ import java.util.List;
  * 30 天后 token 过期，其打开的案件数据源就会永久留在内存里，最终把
  * {@code zsjz.duckdb.max-open-cases} 的配额占满，导致其他人无法开案。
  *
- * <p>本类定期扫描已打开的案件数据源，检查其归属用户是否还有有效 token，
- * 没有则释放。
+ * <p>本类定期扫描已打开的案件数据源，检查<b>打开它的那个 token</b> 是否仍然有效，
+ * 失效则释放。按 token 而不是按用户判断：账号被挤下线后往往很快在新设备重新登录，
+ * 按用户判断会因为「该账号还有有效 token」而永远不回收旧 token 打开的案件。
  *
  * <p><b>保守策略</b>：查询会话状态失败时一律不释放（宁可多占一点内存，
  * 也不要把正在使用的案件数据源关掉）。
@@ -53,17 +54,45 @@ public class CaseDataSourceReaper {
                 // 无归属记录（理论上不会出现），交给开案方主动关闭，不在这里处理
                 continue;
             }
-            if (!hasActiveSession(ownerUserId)) {
-                if (hasRunningWork(caseId)) {
-                    log.info("用户会话已过期，但该案件仍有在途任务，本轮跳过回收: caseId={}", caseId);
-                    continue;
-                }
-                log.info("用户会话已过期，回收案件数据源: userId={}, caseId={}", ownerUserId, caseId);
-                caseDataSourceRegistry.close(caseId);
-                CaseDataCache.cleanCache(caseId);
-                CaseDataCache.cleanGovernConfig(caseId);
+            boolean ownerSessionAlive = isOwnerSessionAlive(ownerUserId, caseId);
+            // 排查「案件为什么没被回收」时先看这一行：会话还活着 / 有在途任务 / 已回收
+            log.debug("兜底扫描: caseId={}, userId={}, 归属会话存活={}", caseId, ownerUserId, ownerSessionAlive);
+            if (ownerSessionAlive) {
+                continue;
+            }
+            if (hasRunningWork(caseId)) {
+                log.info("归属会话已结束，但该案件仍有在途任务，本轮跳过回收: caseId={}", caseId);
+                continue;
+            }
+            log.info("归属会话已结束，回收案件数据源: userId={}, caseId={}", ownerUserId, caseId);
+            caseDataSourceRegistry.close(caseId);
+            CaseDataCache.cleanCache(caseId);
+            CaseDataCache.cleanGovernConfig(caseId);
+        }
+    }
+
+    /**
+     * 打开该案件的会话是否仍然有效。
+     *
+     * <p>被新登录顶下线 / 被踢下线 / 登出 / 过期后，token 立即失效（sa-token 会在
+     * token→loginId 映射里写入异常标记，{@code getLoginIdByToken} 对这类 token 返回 null），
+     * 所以只看 ownerToken 就够。<b>不能</b>退化成「账号是否还有有效 token」：
+     * 被顶下线的用户通常马上在新设备重新登录，那会让旧案件永远等不到兜底回收 ——
+     * 尤其是监听器因在途任务跳过的那一轮。
+     *
+     * <p>老数据没记录 ownerToken，退回按用户维度判断（保守，宁可晚点回收）。
+     */
+    boolean isOwnerSessionAlive(Long ownerUserId, Long caseId) {
+        String ownerToken = caseDataSourceRegistry.ownerTokenOf(caseId);
+        if (ownerToken != null && !ownerToken.isBlank()) {
+            try {
+                return StpUtil.getLoginIdByToken(ownerToken) != null;
+            } catch (Exception e) {
+                log.error("检查案件归属 token 状态失败，本次跳过回收: caseId={}", caseId, e);
+                return true;
             }
         }
+        return hasActiveSession(ownerUserId);
     }
 
     /** 该案件是否有业务模块声明的在途任务（治理/清洗批次等） */

ai-server/src/main/java/com/zsjz/ai/common/datasource/CaseDataSourceRegistry.java

@@ -84,6 +84,14 @@ public class CaseDataSourceRegistry {
                 // 业务上不该发生（案件不共享），属防御性检查
                 throw new ServerException(403, "该案件已被其他用户打开");
             }
+            // 同一用户换会话重新打开（被顶下线后重新登录，且旧案件因在途任务尚未释放）：
+            // 归属 token 必须刷新成新会话。否则兜底回收器仍按旧 token 判断归属，
+            // 会在旧任务跑完后把新会话正在使用的数据源回收掉。
+            if (ownerToken != null && !ownerToken.equals(existing.ownerToken())) {
+                opened.put(caseId, new OpenedCase(existing.dataSource(),
+                        ownerUserId != null ? ownerUserId : owner, ownerToken, existing.openedAt()));
+                log.info("案件归属会话已刷新: caseId={}, userId={}", caseId, ownerUserId);
+            }
             return;
         }
         if (opened.size() >= budget.getMaxOpenCases()) {

ai-server/src/test/java/com/zsjz/ai/common/datasource/CaseDataSourceReaperTest.java

@@ -0,0 +1,118 @@
+package com.zsjz.ai.common.datasource;
+
+import cn.dev33.satoken.stp.StpUtil;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * 兜底回收器的判定契约。
+ *
+ * <p>守两件事：
+ * <ol>
+ *   <li><b>按「打开案件的 token」判活</b>：被顶下线后用户往往马上重新登录，
+ *       按用户维度判断会把旧案件永远留在内存里；</li>
+ *   <li><b>在途任务优先</b>：会话结束了但清洗/治理还在跑，本轮不关数据源。
+ *       任务结束后下一轮必须能接着关（这正是监听器跳过时承诺的「交给 Reaper 兜底」）。</li>
+ * </ol>
+ */
+class CaseDataSourceReaperTest {
+
+    private static final Long CASE_ID = 27L;
+    private static final Long USER_ID = 1L;
+    private static final String OWNER_TOKEN = "owner-token";
+
+    private final CaseDataSourceRegistry registry = mock(CaseDataSourceRegistry.class);
+
+    private void stubOpenedCase() {
+        when(registry.openedCount()).thenReturn(1);
+        when(registry.openedCaseIds()).thenReturn(Set.of(CASE_ID));
+        when(registry.ownerOf(CASE_ID)).thenReturn(USER_ID);
+        when(registry.ownerTokenOf(CASE_ID)).thenReturn(OWNER_TOKEN);
+    }
+
+    /** 用可变的「会话是否有效 / 是否有在途任务」驱动 reap，避免依赖真实 sa-token 上下文 */
+    private CaseDataSourceReaper reaper(AtomicBoolean sessionAlive, AtomicBoolean workRunning) {
+        stubOpenedCase();
+        CaseActivityProbe probe = caseId -> workRunning.get();
+        return new CaseDataSourceReaper(registry, List.of(probe)) {
+            @Override
+            boolean isOwnerSessionAlive(Long ownerUserId, Long caseId) {
+                return sessionAlive.get();
+            }
+        };
+    }
+
+    @Test
+    @DisplayName("归属会话有效：保留案件数据源")
+    void keepsCaseWhenOwnerSessionAlive() {
+        CaseDataSourceReaper reaper = reaper(new AtomicBoolean(true), new AtomicBoolean(false));
+
+        reaper.reap();
+
+        verify(registry, never()).close(CASE_ID);
+    }
+
+    @Test
+    @DisplayName("归属会话已结束：回收案件数据源")
+    void reapsCaseWhenOwnerSessionEnded() {
+        CaseDataSourceReaper reaper = reaper(new AtomicBoolean(false), new AtomicBoolean(false));
+
+        reaper.reap();
+
+        verify(registry).close(CASE_ID);
+    }
+
+    @Test
+    @DisplayName("会话结束但在途任务未完成：本轮跳过，任务结束后下一轮回收")
+    void skipsCaseWithRunningWorkThenReaps() {
+        AtomicBoolean workRunning = new AtomicBoolean(true);
+        CaseDataSourceReaper reaper = reaper(new AtomicBoolean(false), workRunning);
+
+        reaper.reap();
+        verify(registry, never()).close(CASE_ID);
+
+        workRunning.set(false);
+        reaper.reap();
+        verify(registry).close(CASE_ID);
+    }
+
+    @Test
+    @DisplayName("ownerToken 已被顶下线：判定会话已结束（即使账号还有别的有效 token）")
+    void replacedOwnerTokenIsNotAlive() {
+        stubOpenedCase();
+        CaseDataSourceReaper reaper = new CaseDataSourceReaper(registry, List.of());
+
+        try (MockedStatic<StpUtil> stp = mockStatic(StpUtil.class)) {
+            // sa-token 对「被顶下线/被踢/过期」的 token 返回 null（映射里写的是异常标记）
+            stp.when(() -> StpUtil.getLoginIdByToken(OWNER_TOKEN)).thenReturn(null);
+
+            assertFalse(reaper.isOwnerSessionAlive(USER_ID, CASE_ID));
+        }
+    }
+
+    @Test
+    @DisplayName("ownerToken 仍有效：保留案件数据源")
+    void validOwnerTokenIsAlive() {
+        stubOpenedCase();
+        CaseDataSourceReaper reaper = new CaseDataSourceReaper(registry, List.of());
+
+        try (MockedStatic<StpUtil> stp = mockStatic(StpUtil.class)) {
+            stp.when(() -> StpUtil.getLoginIdByToken(OWNER_TOKEN)).thenReturn(String.valueOf(USER_ID));
+
+            assertTrue(reaper.isOwnerSessionAlive(USER_ID, CASE_ID));
+        }
+    }
+}

ai-server/src/test/java/com/zsjz/ai/common/datasource/CaseDataSourceRegistryTest.java

@@ -0,0 +1,66 @@
+package com.zsjz.ai.common.datasource;
+
+import com.zsjz.ai.common.config.DuckdbBudgetProperties;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.Statement;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+
+/**
+ * 案件数据源注册表的归属记账契约。
+ *
+ * <p>幂等开案<b>不能只更新案件快照、不更新归属 token</b>：被顶下线后重新登录的用户
+ * 可能再次打开同一个案件（旧案件因在途任务还没被释放），此时注册表里的 ownerToken
+ * 仍是已失效的旧会话 —— 兜底回收器会在任务结束后把新会话正在用的数据源回收掉。
+ */
+class CaseDataSourceRegistryTest {
+
+    private static final Long CASE_ID = 27L;
+    private static final Long USER_ID = 1L;
+
+    @TempDir
+    Path dir;
+
+    private CaseDataSourceRegistry registry;
+
+    @BeforeEach
+    void setUp() throws Exception {
+        Path dbFile = dir.resolve("case27.duckdb");
+        try (Connection con = DriverManager.getConnection("jdbc:duckdb:" + dbFile);
+             Statement st = con.createStatement()) {
+            st.execute("CREATE TABLE demo (id INTEGER)");
+        }
+        DuckdbBudgetProperties budget = new DuckdbBudgetProperties();
+        budget.setMaxOpenCases(2);
+        registry = new CaseDataSourceRegistry(mock(CaseRoutingDataSource.class), budget);
+    }
+
+    @AfterEach
+    void tearDown() {
+        registry.close(CASE_ID);
+    }
+
+    @Test
+    @DisplayName("同一用户换会话重新开案：归属 token 刷新为新会话")
+    void refreshesOwnerTokenWhenSameUserReopens() {
+        Path dbFile = dir.resolve("case27.duckdb");
+
+        registry.open(CASE_ID, dbFile.toString(), USER_ID, "token-old");
+        assertEquals("token-old", registry.ownerTokenOf(CASE_ID));
+
+        registry.open(CASE_ID, dbFile.toString(), USER_ID, "token-new");
+
+        assertEquals("token-new", registry.ownerTokenOf(CASE_ID));
+        assertEquals(USER_ID, registry.ownerOf(CASE_ID));
+        assertEquals(1, registry.openedCount());
+    }
+}