From 55cf53b9b41f0670c7043e3ac52e2de358e2b6da Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=80=AA=E7=A8=8B=E4=BC=9F?= Date: Tue, 11 Aug 2026 15:42:43 +0800 Subject: [PATCH] fix(webchat): detach SSE subscribers and reclaim orphan runs Use detach rather than complete for WebChat SSE disconnect callbacks, add an orphan-run grace policy for subscriberless active runs, and emergency-save partial assistant output when reclaimed. --- .../mate/channel/web/ChatStreamTracker.java | 78 +++++++++ .../channel/webchat/WebChatController.java | 81 +++++++++- .../ChatStreamTrackerDetachSemanticsTest.java | 111 +++++++++++++ .../ChatStreamTrackerOrphanPolicyTest.java | 149 ++++++++++++++++++ 4 files changed, 412 insertions(+), 7 deletions(-) create mode 100644 mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerDetachSemanticsTest.java create mode 100644 mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerOrphanPolicyTest.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 5ffb6476..94edb68c 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 @@ -226,6 +226,18 @@ public class ChatStreamTracker { */ volatile long lastEventAt = System.currentTimeMillis(); + /** + * Wall-clock millis at which the subscriber list last became empty + * while the run was still alive (not done). Null when there is at + * least one subscriber, or when the run already finished via the + * normal {@code done} path. Drives the orphan-grace eviction in + * {@link ChatStreamTracker#cleanupStaleRuns()} (issue #587): a run + * whose only subscriber disconnected is invisible to its owner and + * unreachable (webchat has no re-attach endpoint), so it is torn down + * after a grace period instead of burning tokens until the idle sweep. + */ + volatile Long subscribersZeroSince; + /** Bound agent identifier; null while not yet resolved. */ volatile Long agentId; @@ -973,6 +985,10 @@ public class ChatStreamTracker { // Without this, async_task_completed fired after `done` would be silently // dropped, leaving the chat UI stuck on the "正在生成中" placeholder. state.subscribers.add(emitter); + // A (re-)attached subscriber clears the orphan clock — the run is + // visible to its owner again, so the grace-period eviction in + // cleanupStaleRuns() should not fire (issue #587). + state.subscribersZeroSince = null; // Deliver MCP progress snapshots on reconnect (progress events skip buffer replay) sendProgressSnapshots(conversationId, emitter); @@ -1096,6 +1112,14 @@ public class ChatStreamTracker { } synchronized (state.lock) { state.subscribers.remove(emitter); + // When the last subscriber leaves and the run is still alive, arm + // the orphan clock — see RunState.subscribersZeroSince. The run + // is now invisible to its owner and (for webchat) unreachable, so + // cleanupStaleRuns will reclaim it after the grace window unless a + // fresh subscriber re-attaches (which clears the clock in attach()). + if (state.subscribers.isEmpty() && !state.done && state.subscribersZeroSince == null) { + state.subscribersZeroSince = System.currentTimeMillis(); + } } log.debug("Emitter detached from stream: {} (remaining={})", conversationId, state.subscribers.size()); @@ -1538,6 +1562,24 @@ public class ChatStreamTracker { @org.springframework.beans.factory.annotation.Value("${mateclaw.sse.idle-timeout-minutes:30}") private int idleTimeoutMinutes = 30; + /** + * Grace period (seconds) before an orphaned run is reclaimed. A run is + * "orphaned" when its subscriber list has been empty since some instant + * (the only SSE client disconnected) while the agent Flux is still + * running — invisible to its owner and, for the WebChat channel, + * unreachable (no re-attach endpoint). The default 2 minutes tolerates a + * network blip + a client-side regenerate retry; once it elapses with no + * subscriber returning, the run is disposed and its partial assistant + * content is flushed via {@code emergencySaveCallback} (issue #587). + *

+ * Note: a run that keeps producing events but has no subscribers is NOT + * considered stuck — {@code lastEventAt} keeps it out of the idle bucket. + * The orphan bucket specifically catches "alive but nobody's watching", + * which the idle watchdog cannot see. + */ + @org.springframework.beans.factory.annotation.Value("${mateclaw.webchat.orphan-grace-sec:120}") + private int orphanGraceSeconds = 120; + /** * Test hook — backdates the {@code lastEventAt} timestamp on an * existing RunState so {@link #cleanupStaleRuns()} can be exercised @@ -1567,16 +1609,32 @@ public class ChatStreamTracker { this.idleTimeoutMinutes = minutes; } + /** Test hook — backdate the orphan clock on an existing run. */ + void backdateOrphanForTesting(String conversationId, long subscribersZeroSince) { + RunState state = runs.get(conversationId); + if (state != null) { + state.subscribersZeroSince = subscribersZeroSince; + } + } + + /** Test hook — override the orphan grace in pure-unit tests that bypass Spring. */ + void setOrphanGraceSecondsForTesting(int seconds) { + this.orphanGraceSeconds = seconds; + } + /** * 定期清理过期的 RunState，防止内存泄漏。 * - 已完成超过 {@link #DONE_RETENTION_MS} 的 → 移除 * - 自 {@link RunState#lastEventAt} 算起静默超过 * {@link #idleTimeoutMinutes} 分钟的 → 强制移除（视为卡死） + * - 订阅者清零超过 {@link #orphanGraceSeconds} 且仍在运行的孤儿 → + * 移除（webchat 无重连端点，运行对调用方不可见不可达，见 #587） */ @org.springframework.scheduling.annotation.Scheduled(fixedRate = 600_000) public void cleanupStaleRuns() { long now = System.currentTimeMillis(); long idleThresholdMs = (long) idleTimeoutMinutes * 60_000L; + long orphanGraceMs = (long) orphanGraceSeconds * 1000L; int evicted = 0; var iterator = runs.entrySet().iterator(); @@ -1585,6 +1643,16 @@ public class ChatStreamTracker { RunState state = entry.getValue(); long age = now - state.createdAt; long idleMs = now - state.lastEventAt; + Long orphanSince = state.subscribersZeroSince; + long orphanMs = orphanSince != null ? now - orphanSince : -1L; + // Subscriber count under the lock so the orphan decision is + // consistent with the subscriber list (subscribersZeroSince is + // normally null whenever a subscriber is present, but a concurrent + // attach/detach could race the read — guard against that here). + int subCount; + synchronized (state.lock) { + subCount = state.subscribers.size(); + } boolean shouldEvict = false; String reason = null; @@ -1592,6 +1660,16 @@ public class ChatStreamTracker { if (state.done && age > DONE_RETENTION_MS) { shouldEvict = true; reason = "completed and expired"; + } else if (!state.done && subCount == 0 && orphanSince != null && orphanMs > orphanGraceMs) { + // Orphan: subscriber list empty longer than the grace window + // while the agent Flux is still running. Invisible + (for + // webchat) unreachable, so reclaim it instead of letting it + // burn tokens until the idle sweep (issue #587). A run that's + // actively producing events is NOT exempt — the whole point is + // nobody is watching those events. + shouldEvict = true; + reason = "orphaned: no subscribers for " + (orphanMs / 1000) + + "s (grace " + orphanGraceSeconds + "s); run still active"; } else if (idleMs > idleThresholdMs) { shouldEvict = true; reason = "idle for " + (idleMs / 1000) + "s (threshold " diff --git a/mateclaw-server/src/main/java/vip/mate/channel/webchat/WebChatController.java b/mateclaw-server/src/main/java/vip/mate/channel/webchat/WebChatController.java index c27a37d7..7c402edd 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/webchat/WebChatController.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/webchat/WebChatController.java @@ -186,15 +186,25 @@ public class WebChatController { log.info("[WebChat] Stream: agentId={}, conversationId={}, visitor={}", agentId, conversationId, visitorId); - // 注册 emitter 回调 - emitter.onCompletion(() -> log.debug("[WebChat] SSE completed: {}", conversationId)); + // Register emitter callbacks. An SSE disconnect means this subscriber + // left, not that the agent run finished. Use detach() instead of + // complete(); complete() would prematurely mark the RunState done, + // drop later content deltas from the replay buffer, and double-count + // completion when the agent Flux actually finishes. + emitter.onCompletion(() -> { + log.debug("[WebChat] SSE completed: {}", conversationId); + streamTracker.detach(conversationId, emitter); + }); emitter.onTimeout(() -> { log.debug("[WebChat] SSE timeout: {}", conversationId); - streamTracker.complete(conversationId); + streamTracker.detach(conversationId, emitter); + // Explicitly complete after timeout so the servlet container does + // not rethrow AsyncRequestTimeoutException. + emitter.complete(); }); emitter.onError(e -> { log.debug("[WebChat] SSE error: {} - {}", conversationId, e.getMessage()); - streamTracker.complete(conversationId); + streamTracker.detach(conversationId, emitter); }); sseExecutor.execute(() -> { @@ -320,6 +330,11 @@ public class WebChatController { // HTTP call keeps running — token burn + side-effect tools still fire. // Mirrors ChatController#chatStream line 495. streamTracker.setDisposable(conversationId, disposable); + // Wire the emergency save so an orphaned run (only subscriber + // gone) is flushed as an "interrupted" assistant message when + // the grace-period eviction reclaims it — otherwise the visitor + // would see only their own user message (issue #587). + registerEmergencySave(conversationId, assistantReply, usage, modelInfo); } catch (Exception e) { log.error("[WebChat] Error: {}", e.getMessage(), e); @@ -1242,14 +1257,18 @@ public class WebChatController { return emitter; } - emitter.onCompletion(() -> log.debug("[WebChat] approve SSE completed: {}", conversationId)); + emitter.onCompletion(() -> { + log.debug("[WebChat] approve SSE completed: {}", conversationId); + streamTracker.detach(conversationId, emitter); + }); emitter.onTimeout(() -> { log.debug("[WebChat] approve SSE timeout: {}", conversationId); - streamTracker.complete(conversationId); + streamTracker.detach(conversationId, emitter); + emitter.complete(); }); emitter.onError(e -> { log.debug("[WebChat] approve SSE error: {} - {}", conversationId, e.getMessage()); - streamTracker.complete(conversationId); + streamTracker.detach(conversationId, emitter); }); String actor = webchatUsername(visitorId); @@ -1372,6 +1391,7 @@ public class WebChatController { }) .subscribe(); streamTracker.setDisposable(conversationId, disposable); + registerEmergencySave(conversationId, assistantReply, usage, modelInfo); } catch (Exception e) { log.error("[WebChat] approve failed for {}: {}", conversationId, e.getMessage()); try { @@ -1638,6 +1658,53 @@ public class WebChatController { return parts; } + /** + * Register an emergency-save callback that flushes the partial assistant + * reply accumulated so far as an {@code interrupted} message. Wired on + * both {@code /stream} and {@code /sessions/approve} so that when a run is + * reclaimed while its only subscriber is gone (orphan-grace eviction, + * shutdown, admin force-recycle), the visitor can still retrieve the + * partial answer via {@code /sessions/messages} instead of seeing only the + * user message (issue #587). Mirrors ChatController#emergencySaveAccumulator. + * + * @param conversationId target conversation + * @param assistantReply live accumulator appended to in doOnNext + * @param usage [prompt, completion, cacheRead, cacheWrite, reasoning] + * @param modelInfo [runtimeModel, runtimeProvider] + */ + private void registerEmergencySave(String conversationId, StringBuilder assistantReply, + int[] usage, String[] modelInfo) { + streamTracker.setEmergencySaveCallback(conversationId, () -> { + try { + String reply = assistantReply.toString(); + if (reply.isBlank()) { + log.debug("[WebChat] Emergency save skipped (empty reply): {}", conversationId); + return; + } + // usage length varies by call site (chatStream = 5 tokens, + // approve replay = 2); read defensively so the save never + // throws ArrayIndexOutOfBoundsException. + int prompt = usage.length > 0 ? usage[0] : 0; + int completion = usage.length > 1 ? usage[1] : 0; + int cacheRead = usage.length > 2 ? usage[2] : 0; + int cacheWrite = usage.length > 3 ? usage[3] : 0; + int reasoning = usage.length > 4 ? usage[4] : 0; + String runtimeModel = modelInfo.length > 0 ? modelInfo[0] : null; + String runtimeProvider = modelInfo.length > 1 ? modelInfo[1] : null; + conversationService.saveMessage( + conversationId, "assistant", reply, List.of(), + "interrupted", + prompt, completion, cacheRead, cacheWrite, reasoning, + runtimeModel, runtimeProvider, null); + log.info("[WebChat] Emergency-saved partial assistant reply: " + + "conversationId={}, textLen={}", + conversationId, reply.length()); + } catch (Exception e) { + log.warn("[WebChat] Emergency save failed for {}: {}", conversationId, e.getMessage()); + } + }); + } + // ==================== 内部方法 ==================== private static final Pattern SESSION_ID_PATTERN = Pattern.compile("[A-Za-z0-9_-]{1,64}"); diff --git a/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerDetachSemanticsTest.java b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerDetachSemanticsTest.java new file mode 100644 index 00000000..52f0ef95 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerDetachSemanticsTest.java @@ -0,0 +1,111 @@ +package vip.mate.channel.web; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Pins the {@link ChatStreamTracker#detach(String, SseEmitter)} contract: a + * subscriber going away (SSE timeout / error / client close) must NOT mark the + * run as done. This is the tracker-level root of WebChatController issue #587 + * defect 1 — the controller previously called {@code complete()} from its + * {@code onTimeout}/{@code onError} callbacks, which polluted RunState ahead + * of the agent finishing and dropped subsequent content deltas from the replay + * buffer. + * + *

These tests assert the tracker-side invariant the controller now relies on: + * detach only removes the subscriber, the run keeps running, and events still + * reach the buffer for any re-attaching subscriber. + */ +class ChatStreamTrackerDetachSemanticsTest { + + private ChatStreamTracker newTracker() { + return new ChatStreamTracker(new ObjectMapper()); + } + + @Test + @DisplayName("detach() leaves the run running — isRunning() stays true") + void detachKeepsRunRunning() { + ChatStreamTracker tracker = newTracker(); + String cid = "detach-running"; + tracker.register(cid); + tracker.incrementFlux(cid); + + SseEmitter emitter = new SseEmitter(); + tracker.attach(cid, emitter); + // Simulate the SSE onTimeout path: this is what WebChatController now calls. + tracker.detach(cid, emitter); + + assertTrue(tracker.isRunning(cid), + "detach only removes the subscriber; the run must stay running"); + } + + @Test + @DisplayName("detach() does NOT mark done — subsequent events still buffer for replay") + void detachDoesNotMarkDone() { + ChatStreamTracker tracker = newTracker(); + String cid = "detach-buffer"; + tracker.register(cid); + tracker.incrementFlux(cid); + + SseEmitter gone = new SseEmitter(); + tracker.attach(cid, gone); + tracker.detach(cid, gone); + + // After the subscriber left, the agent keeps producing. These events must + // land in the buffer so a re-attaching subscriber can replay them — the + // whole point of not prematurely calling complete(). + tracker.broadcast(cid, "content_delta", "{\"text\":\"still-alive\"}"); + + // A fresh subscriber attaching should be able to see the buffered event + // (proving done was NOT set, which would have dropped the broadcast). + SseEmitter late = new SseEmitter(); + AtomicInteger received = new AtomicInteger(); + late.onCompletion(() -> { + }); + // attach replays the buffer synchronously; we can't easily count sends on a + // raw SseEmitter, but the key assertion is that attach returns true (state + // exists and is not in a terminal window that drops events). + assertTrue(tracker.attach(cid, late), "attach must succeed — run is still alive"); + assertTrue(tracker.isRunning(cid)); + } + + @Test + @DisplayName("contrast: complete() DOES mark done (the old, buggy behavior)") + void completeMarksDone() { + ChatStreamTracker tracker = newTracker(); + String cid = "complete-done"; + tracker.register(cid); + tracker.incrementFlux(cid); + + // complete() is the agent-finished path — it should mark the run done. + tracker.complete(cid); + assertFalse(tracker.isRunning(cid), + "complete() is the real finish signal; detach() must NOT be"); + } + + @Test + @DisplayName("detach is idempotent and safe when no run exists") + void detachSafeWhenAbsent() { + ChatStreamTracker tracker = newTracker(); + SseEmitter emitter = new SseEmitter(); + // No run registered — detach must not throw. + tracker.detach("never-registered", emitter); + + tracker.register("present"); + tracker.attach("present", emitter); + // Detaching twice must be a no-op the second time. + tracker.detach("present", emitter); + tracker.detach("present", emitter); + // run still alive + assertTrue(tracker.isRunning("present")); + assertEquals(0, tracker.getAllSnapshot().getFirst().subscriberCount()); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerOrphanPolicyTest.java b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerOrphanPolicyTest.java new file mode 100644 index 00000000..dd4f78a0 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerOrphanPolicyTest.java @@ -0,0 +1,149 @@ +package vip.mate.channel.web; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Pins the orphan-run policy added for issue #587: when a run's only + * subscriber disconnects (webchat SSE timeout/error) while the agent Flux is + * still running, the run becomes invisible + unreachable and must be reclaimed + * after a grace window instead of burning tokens until the 30-min idle sweep. + * + *

Composes with the detach() fix from #587 defect 1: detach() arms the + * orphan clock when the subscriber list empties; attach()/closeSubscribers() + * clear it. + */ +class ChatStreamTrackerOrphanPolicyTest { + + private ChatStreamTracker newTracker() { + ChatStreamTracker t = new ChatStreamTracker(new ObjectMapper()); + t.setIdleTimeoutMinutesForTesting(30); // keep the idle bucket out of the way + t.setOrphanGraceSecondsForTesting(2); // tight grace for unit-test speed + return t; + } + + @Test + @DisplayName("Orphan run (no subscribers past grace) is evicted") + void orphanRunEvictedAfterGrace() { + ChatStreamTracker tracker = newTracker(); + String cid = "orphan-evict"; + tracker.register(cid); + tracker.incrementFlux(cid); + + // No subscriber ever attached — simulate the clock by backdating, which + // is exactly what detach() would have set when the last subscriber left. + tracker.backdateOrphanForTesting(cid, System.currentTimeMillis() - 3_000L); // 3s > 2s grace + + tracker.cleanupStaleRuns(); + + assertFalse(tracker.hasRunStateForTesting(cid), + "orphan run past the grace window must be evicted"); + } + + @Test + @DisplayName("Orphan run within grace survives") + void orphanRunWithinGraceSurvives() { + ChatStreamTracker tracker = newTracker(); + String cid = "orphan-fresh"; + tracker.register(cid); + tracker.incrementFlux(cid); + + // Just became orphaned — 1s, within the 2s grace. + tracker.backdateOrphanForTesting(cid, System.currentTimeMillis() - 1_000L); + tracker.cleanupStaleRuns(); + + assertTrue(tracker.hasRunStateForTesting(cid), + "orphan run inside the grace window must survive — tolerates a brief disconnect"); + } + + @Test + @DisplayName("Orphan eviction fires emergencySaveCallback before dispose") + void orphanEvictionFiresEmergencySave() { + ChatStreamTracker tracker = newTracker(); + String cid = "orphan-save"; + tracker.register(cid); + tracker.incrementFlux(cid); + + AtomicInteger saveCount = new AtomicInteger(); + tracker.setEmergencySaveCallback(cid, saveCount::incrementAndGet); + + tracker.backdateOrphanForTesting(cid, System.currentTimeMillis() - 3_000L); + tracker.cleanupStaleRuns(); + + assertEquals(1, saveCount.get(), + "partial assistant content must be flushed via the emergency " + + "save before the orphan run is disposed"); + assertFalse(tracker.hasRunStateForTesting(cid)); + } + + @Test + @DisplayName("A done run is never treated as an orphan") + void doneRunNotOrphan() { + ChatStreamTracker tracker = newTracker(); + String cid = "done-not-orphan"; + tracker.register(cid); + tracker.incrementFlux(cid); + tracker.complete(cid); // mark done + + // Even with an orphan clock backdated past grace, a done run must not + // hit the orphan branch (it's finalized via the done path / retention). + tracker.backdateOrphanForTesting(cid, System.currentTimeMillis() - 3_000L); + tracker.cleanupStaleRuns(); + + // done run is kept for DONE_RETENTION_MS (5 min) — still here right after. + assertTrue(tracker.hasRunStateForTesting(cid), + "a done run must not be evicted as an orphan — it's already finalized"); + } + + @Test + @DisplayName("A run with a live subscriber is never an orphan") + void runWithSubscriberNotOrphan() { + ChatStreamTracker tracker = newTracker(); + String cid = "has-sub"; + tracker.register(cid); + tracker.incrementFlux(cid); + + SseEmitter em = new SseEmitter(); + tracker.attach(cid, em); // attaches a subscriber -> clears orphan clock + + // Backdate would-be orphan clock — but a subscriber is present, so the + // orphan branch must not fire regardless. + tracker.backdateOrphanForTesting(cid, System.currentTimeMillis() - 3_000L); + tracker.cleanupStaleRuns(); + + assertTrue(tracker.hasRunStateForTesting(cid), + "a run with a live subscriber must never be evicted as an orphan"); + } + + @Test + @DisplayName("detach() arms the orphan clock; re-attach clears it") + void detachArmsClockAttachClears() { + ChatStreamTracker tracker = newTracker(); + String cid = "detach-rearm"; + tracker.register(cid); + tracker.incrementFlux(cid); + + SseEmitter em = new SseEmitter(); + tracker.attach(cid, em); + // Detach the only subscriber -> clock armed, run still running. + tracker.detach(cid, em); + assertTrue(tracker.isRunning(cid), + "run must still be running after the only subscriber detaches"); + + // A fresh subscriber re-attaches within grace -> clock cleared, survives. + SseEmitter reattached = new SseEmitter(); + assertTrue(tracker.attach(cid, reattached)); + tracker.backdateOrphanForTesting(cid, System.currentTimeMillis() - 3_000L); // would-be past grace + tracker.cleanupStaleRuns(); + assertTrue(tracker.hasRunStateForTesting(cid), + "re-attach clears the orphan clock even if it was backdated"); + } +}