From f1d9104422a3d6228e6cc5e215c5284dbe8790c8 Mon Sep 17 00:00:00 2001 From: matevip Date: Sun, 24 May 2026 23:01:04 +0800 Subject: [PATCH] fix(channel): evict SSE RunState by inactivity, not wall-clock age --- .../mate/channel/web/ChatStreamTracker.java | 62 +++++++++++-- .../web/ChatStreamTrackerCleanupTest.java | 86 +++++++++++++++++++ 2 files changed, 142 insertions(+), 6 deletions(-) create mode 100644 mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerCleanupTest.java diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java index 09a68103..63f220c3 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java @@ -1464,17 +1464,64 @@ public class ChatStreamTracker { /** 已完成的 RunState 保留时间（5 分钟） */ private static final long DONE_RETENTION_MS = 5 * 60 * 1000; - /** RunState 最大存活时间（30 分钟，防止挂起的流永远占内存） */ - private static final long MAX_LIFETIME_MS = 30 * 60 * 1000; + + /** + * RunState 最长无活动时间。从 wall-clock {@code MAX_LIFETIME_MS=30min} + * 切换到 inactivity-based 后默认 30 min — 与 hermes-agent 的 + * {@code gateway_timeout=1800s} 同口径：只要 agent 还在持续产事件 + * （tool call / content delta / phase transition / progress_update）， + * 就一直活下去，墙钟跑 1 小时 2 小时都可以。只有真正"完全静默 ≥ N 分钟" + * 才视为卡死并强制清理。 + * + *

修复的背景：round-6 的 10-LLM 横评任务实际跑了 47 min，全程都在 + * 出 tool call，但旧的 wall-clock 30 min 死线在 iter 128 / 8 of 10 + * 就把 RunState 清掉了 — SSE 流死、UI 空白、用户以为任务挂了。换成 + * inactivity 后，那种长任务永远不会被误清，而真正卡死的 agent（无活动 + * 5+ 分钟）会按时清理。可通过 property + * {@code mateclaw.sse.idle-timeout-minutes} 调整。 + */ + @org.springframework.beans.factory.annotation.Value("${mateclaw.sse.idle-timeout-minutes:30}") + private int idleTimeoutMinutes = 30; + + /** + * Test hook — backdates the {@code lastEventAt} timestamp on an + * existing RunState so {@link #cleanupStaleRuns()} can be exercised + * deterministically without sleeping for minutes. Package-private on + * purpose; production callers go through {@link #broadcast} which + * stamps the field forward. + */ + void backdateLastEventForTesting(String conversationId, long lastEventAt) { + RunState state = runs.get(conversationId); + if (state != null) { + state.lastEventAt = lastEventAt; + } + } + + /** Test hook — true when a RunState row exists for the conversation. */ + boolean hasRunStateForTesting(String conversationId) { + return runs.containsKey(conversationId); + } + + /** Test hook — exposes the configurable timeout for assertion. */ + int idleTimeoutMinutesForTesting() { + return idleTimeoutMinutes; + } + + /** Test hook — override the timeout in pure-unit tests that bypass Spring. */ + void setIdleTimeoutMinutesForTesting(int minutes) { + this.idleTimeoutMinutes = minutes; + } /** * 定期清理过期的 RunState，防止内存泄漏。 - * - 已完成超过 5 分钟的 → 移除 - * - 存活超过 30 分钟的（无论是否完成）→ 强制移除 + * - 已完成超过 {@link #DONE_RETENTION_MS} 的 → 移除 + * - 自 {@link RunState#lastEventAt} 算起静默超过 + * {@link #idleTimeoutMinutes} 分钟的 → 强制移除（视为卡死） */ @org.springframework.scheduling.annotation.Scheduled(fixedRate = 600_000) public void cleanupStaleRuns() { long now = System.currentTimeMillis(); + long idleThresholdMs = (long) idleTimeoutMinutes * 60_000L; int evicted = 0; var iterator = runs.entrySet().iterator(); @@ -1482,6 +1529,7 @@ public class ChatStreamTracker { var entry = iterator.next(); RunState state = entry.getValue(); long age = now - state.createdAt; + long idleMs = now - state.lastEventAt; boolean shouldEvict = false; String reason = null; @@ -1489,9 +1537,11 @@ public class ChatStreamTracker { if (state.done && age > DONE_RETENTION_MS) { shouldEvict = true; reason = "completed and expired"; - } else if (age > MAX_LIFETIME_MS) { + } else if (idleMs > idleThresholdMs) { shouldEvict = true; - reason = "exceeded max lifetime (" + (age / 1000) + "s)"; + reason = "idle for " + (idleMs / 1000) + "s (threshold " + + idleTimeoutMinutes + "min); total wall-clock age " + + (age / 1000) + "s"; } if (shouldEvict) { diff --git a/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerCleanupTest.java b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerCleanupTest.java new file mode 100644 index 00000000..98ab6acf --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerCleanupTest.java @@ -0,0 +1,86 @@ +package vip.mate.channel.web; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Pins {@link ChatStreamTracker#cleanupStaleRuns()} — the switch from + * wall-clock {@code MAX_LIFETIME_MS} to inactivity-based eviction. + * + *

Before the fix, a long-running agent that kept producing tool calls + * (47-minute LLM-review smoke test, round 6) was killed at the 30-minute + * wall-clock mark mid-task. The new behaviour mirrors hermes-agent's + * {@code gateway_timeout}: only completely idle runs are evicted, the + * actively-producing ones can run as long as they need to. + */ +class ChatStreamTrackerCleanupTest { + + private static ChatStreamTracker newTracker() { + ChatStreamTracker t = new ChatStreamTracker(new ObjectMapper()); + // Tighten the idle threshold so the test stays at unit-test speed. + t.setIdleTimeoutMinutesForTesting(5); + return t; + } + + @Test + @DisplayName("Active run (recent lastEventAt) survives cleanup regardless of total age.") + void activeRunSurvives() { + ChatStreamTracker tracker = newTracker(); + tracker.register("conv-active"); + // lastEventAt was set to "now" inside the RunState constructor — + // no backdate, so even a "very old createdAt" would be irrelevant. + tracker.cleanupStaleRuns(); + assertTrue(tracker.hasRunStateForTesting("conv-active"), + "actively-producing run must not be evicted"); + } + + @Test + @DisplayName("Idle run beyond threshold is evicted.") + void idleRunEvicted() { + ChatStreamTracker tracker = newTracker(); + tracker.register("conv-idle"); + // 6 minutes ago — past the 5-minute threshold set above. + tracker.backdateLastEventForTesting("conv-idle", System.currentTimeMillis() - 6 * 60_000L); + tracker.cleanupStaleRuns(); + assertFalse(tracker.hasRunStateForTesting("conv-idle"), + "idle run past threshold must be evicted"); + } + + @Test + @DisplayName("Idle run just inside the threshold survives — no premature eviction.") + void idleRunWithinThresholdSurvives() { + ChatStreamTracker tracker = newTracker(); + tracker.register("conv-borderline"); + // 4 minutes idle — within 5-minute window. + tracker.backdateLastEventForTesting("conv-borderline", System.currentTimeMillis() - 4 * 60_000L); + tracker.cleanupStaleRuns(); + assertTrue(tracker.hasRunStateForTesting("conv-borderline"), + "run idle below threshold must survive — would otherwise be a regression of the wall-clock bug"); + } + + @Test + @DisplayName("Mixed: active + idle runs — only the idle one is evicted.") + void mixedRunsSelectiveEviction() { + ChatStreamTracker tracker = newTracker(); + tracker.register("conv-active"); + tracker.register("conv-idle"); + tracker.backdateLastEventForTesting("conv-idle", System.currentTimeMillis() - 10 * 60_000L); + tracker.cleanupStaleRuns(); + assertTrue(tracker.hasRunStateForTesting("conv-active")); + assertFalse(tracker.hasRunStateForTesting("conv-idle")); + } + + @Test + @DisplayName("Default idle timeout from @Value matches the documented 30-minute fallback.") + void defaultIdleTimeoutIs30() { + // Bypass the @Value injection (no Spring context in this unit test) — + // the field initialiser pins the default so a refactor that drops the + // = 30 falls over here. + ChatStreamTracker tracker = new ChatStreamTracker(new ObjectMapper()); + org.junit.jupiter.api.Assertions.assertEquals(30, tracker.idleTimeoutMinutesForTesting()); + } +}