opensource/mateclaw
2
0
Fork 0
mirror of https://gitee.com/mateos/mateclaw.git synced 2026-09-15 20:08:18 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(channel): preserve queued chat inputs across turns

Browse Source
This commit is contained in:
matevip 2026-05-24 22:57:08 +08:00
parent a9c2d45790
commit 481cece733
2 changed files with 64 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java
View File

@ -458,7 +458,18 @@ public class ChatStreamTracker {
RunState state = runs.get(conversationId); RunState state = runs.get(conversationId);
if (state != null && state.done) { if (state != null && state.done) {
stopHeartbeat(conversationId); stopHeartbeat(conversationId);
runs.put(conversationId, new RunState(conversationId)); RunState nextState = new RunState(conversationId);
int carried = 0;
QueuedInput queued;
while ((queued = state.messageQueue.poll()) != null) {
nextState.messageQueue.offer(queued);
carried++;
}
runs.put(conversationId, nextState);
if (carried > 0) {
log.info("[ChatStreamTracker] Carried {} queued message(s) into next run: {}",
carried, conversationId);
}
} else if (state != null) { } else if (state != null) {
// Reuse path: when complete() early-returns due to activeFluxCount > 0 // Reuse path: when complete() early-returns due to activeFluxCount > 0
// (approval replay / interrupt / any leaked flux increment), the RunState // (approval replay / interrupt / any leaked flux increment), the RunState

52
mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerQueueDrainTest.java Normal file
View File

@ -0,0 +1,52 @@
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.*;
class ChatStreamTrackerQueueDrainTest {
private ChatStreamTracker newTracker() {
return new ChatStreamTracker(new ObjectMapper());
}
@Test
@DisplayName("Queued inputs survive RunState replacement and drain FIFO")
void queuedInputsSurviveRunStateReplacementAndDrainFifo() {
ChatStreamTracker tracker = newTracker();
String conversationId = "queue-drain";
tracker.register(conversationId);
tracker.incrementFlux(conversationId);
assertTrue(tracker.enqueueMessage(conversationId, "q1", 101L, false));
assertTrue(tracker.enqueueMessage(conversationId, "q2", 101L, false));
assertTrue(tracker.enqueueMessage(conversationId, "q3", 101L, false));
ChatStreamTracker.CompletionResult first = tracker.completeAndConsumeIfLast(conversationId);
assertTrue(first.allDone());
assertNotNull(first.queuedInput());
assertEquals("q1", first.queuedInput().message());
tracker.register(conversationId);
tracker.incrementFlux(conversationId);
ChatStreamTracker.CompletionResult second = tracker.completeAndConsumeIfLast(conversationId);
assertTrue(second.allDone());
assertNotNull(second.queuedInput());
assertEquals("q2", second.queuedInput().message());
tracker.register(conversationId);
tracker.incrementFlux(conversationId);
ChatStreamTracker.CompletionResult third = tracker.completeAndConsumeIfLast(conversationId);
assertTrue(third.allDone());
assertNotNull(third.queuedInput());
assertEquals("q3", third.queuedInput().message());
tracker.register(conversationId);
tracker.incrementFlux(conversationId);
ChatStreamTracker.CompletionResult empty = tracker.completeAndConsumeIfLast(conversationId);
assertTrue(empty.allDone());
assertNull(empty.queuedInput());
}
}