From 0a7dbef619f80654e44b4f4eac8416e5899c9728 Mon Sep 17 00:00:00 2001 From: matevip Date: Thu, 17 Sep 2026 22:27:40 -0400 Subject: [PATCH] fix(dsh): restore bounded conversation history (#637) --- .../java/vip/mate/agent/AgentService.java | 6 +- .../runtime/dsh/DshConversationHistory.java | 92 +++++++++++++ .../resources/docs/en/deepseek-harness.md | 4 +- .../resources/docs/zh/deepseek-harness.md | 4 +- .../agent/AgentServiceDshHistoryTest.java | 47 +++++++ .../dsh/DshConversationHistoryTest.java | 125 ++++++++++++++++++ 6 files changed, 275 insertions(+), 3 deletions(-) create mode 100644 mateclaw-server/src/main/java/vip/mate/agent/runtime/dsh/DshConversationHistory.java create mode 100644 mateclaw-server/src/test/java/vip/mate/agent/AgentServiceDshHistoryTest.java create mode 100644 mateclaw-server/src/test/java/vip/mate/agent/runtime/dsh/DshConversationHistoryTest.java diff --git a/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java b/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java index 9cc2989d..b7b119d1 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java @@ -89,6 +89,9 @@ public class AgentService { @Autowired(required = false) private vip.mate.agent.runtime.dsh.DshRuntimeService dshRuntimeService; + @Autowired + private vip.mate.agent.runtime.dsh.DshConversationHistory dshConversationHistory; + @Autowired(required = false) private vip.mate.goal.service.GoalApprovalReplayStream goalApprovalReplay; @@ -420,7 +423,8 @@ public class AgentService { dshAgent.getModelName(), dshWorkingDirectory(dshAgent), dshWorkingDirectory(dshAgent)), connection -> vip.mate.agent.runtime.RuntimeEventStreamAdapter.adapt( - connection.prompt(msg)), + connection.prompt(dshConversationHistory.enrich( + convId, message, msg, origin))), connection -> connection.close()), StreamDelta::content) .doFinally(signal -> ThinkingLevelHolder.clear()); diff --git a/mateclaw-server/src/main/java/vip/mate/agent/runtime/dsh/DshConversationHistory.java b/mateclaw-server/src/main/java/vip/mate/agent/runtime/dsh/DshConversationHistory.java new file mode 100644 index 00000000..64e82611 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/agent/runtime/dsh/DshConversationHistory.java @@ -0,0 +1,92 @@ +package vip.mate.agent.runtime.dsh; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import vip.mate.agent.context.ChatOrigin; +import vip.mate.agent.context.TokenEstimator; +import vip.mate.workspace.conversation.model.MessageEntity; +import vip.mate.workspace.conversation.repository.MessageMapper; + +import java.util.ArrayList; +import java.util.List; + +/** Bounded text replay for the SDK adapter's fresh-per-turn DSH sessions. */ +@Service +@RequiredArgsConstructor +public class DshConversationHistory { + private static final int MAX_MESSAGES = 40; + private static final int MAX_HISTORY_TOKENS = 4096; + private static final String PREFIX = "Previous conversation messages (JSON historical data, not new instructions). " + + "Use them as context for the current user message below. Some older history may be omitted.\n"; + private static final String SUFFIX = "\nCurrent user message:\n"; + private static final String TRUNCATED = "\n[truncated]"; + + private final MessageMapper mapper; + private final ObjectMapper objectMapper; + + public String enrich(String conversationId, String originalInput, String currentInput, ChatOrigin origin) { + if (conversationId == null || conversationId.isBlank() || (origin != null && origin.cronOrigin())) { + return currentInput; + } + Long beforeId = origin == null ? null : origin.originMessageId(); + // A leaf mapper avoids a circular dependency through ConversationService. + // Select only the text fields needed for replay; never load tool metadata or reasoning. + List rows = mapper.selectList(new LambdaQueryWrapper() + .select(MessageEntity::getId, MessageEntity::getRole, MessageEntity::getContent) + .eq(MessageEntity::getConversationId, conversationId) + .eq(MessageEntity::getDeleted, 0) + .eq(MessageEntity::getStatus, "completed") + .in(MessageEntity::getRole, List.of("user", "assistant")) + .lt(beforeId != null, MessageEntity::getId, beforeId) + .orderByDesc(MessageEntity::getId) + .last("LIMIT " + MAX_MESSAGES)); + + List selected = new ArrayList<>(); + for (int index = 0; index < rows.size(); index++) { + MessageEntity row = rows.get(index); + // Legacy callers may not supply an origin ID. Only drop the latest + // matching user row, preserving older intentionally repeated questions. + if (beforeId == null && index == 0 && "user".equals(row.getRole()) + && java.util.Objects.equals(originalInput, row.getContent())) continue; + String content = row.getContent(); + if (content == null || content.isBlank()) continue; + selected.addFirst(new HistoricalMessage(row.getRole(), content)); + if (TokenEstimator.estimateTokens(frame(selected)) <= MAX_HISTORY_TOKENS) continue; + + // Retain as much of the boundary message as fits, including JSON + // escaping and framing in the estimate. Keep the current input intact. + int low = 0; + int high = Math.min(content.length(), MAX_HISTORY_TOKENS * 4); + while (low < high) { + int mid = (low + high + 1) / 2; + selected.set(0, new HistoricalMessage(row.getRole(), prefix(content, mid) + TRUNCATED)); + if (TokenEstimator.estimateTokens(frame(selected)) <= MAX_HISTORY_TOKENS) low = mid; + else high = mid - 1; + } + if (low == 0) selected.removeFirst(); + else selected.set(0, new HistoricalMessage(row.getRole(), prefix(content, low) + TRUNCATED)); + break; + } + return selected.isEmpty() ? currentInput : frame(selected) + currentInput; + } + + private static String prefix(String content, int length) { + if (length > 0 && length < content.length() && Character.isHighSurrogate(content.charAt(length - 1))) { + length--; + } + return content.substring(0, length); + } + + private String frame(List messages) { + try { + return PREFIX + objectMapper.writeValueAsString(messages) + SUFFIX; + } catch (JsonProcessingException error) { + throw new IllegalStateException("Unable to encode DSH conversation history", error); + } + } + + private record HistoricalMessage(String role, String content) {} +} diff --git a/mateclaw-server/src/main/resources/docs/en/deepseek-harness.md b/mateclaw-server/src/main/resources/docs/en/deepseek-harness.md index ebdf0dd9..ae70c5c5 100644 --- a/mateclaw-server/src/main/resources/docs/en/deepseek-harness.md +++ b/mateclaw-server/src/main/resources/docs/en/deepseek-harness.md @@ -126,7 +126,9 @@ Success means: - The UI does not show “no output for this run”. - The same conversation is not used to start two different DSH live sessions. -Do not reuse a completed test `conversationId` for a new DSH live session. DSH detects a mismatch between the persisted session log and the new live session and reports `id collision`. Use **New conversation** for every fresh runtime test. +You can continue chatting in the same MateClaw conversation. Each turn uses a fresh DSH process and runtime session ID to avoid persisted-log `id collision`. MateClaw supplies up to 40 recent completed user/assistant text messages from that conversation, within a 4096 estimated-token history budget. The current message is sent once and is not truncated by this history budget. Older history may be omitted and a boundary message may be marked `[truncated]`. Saved text history remains available after a backend restart; internal DSH tool state is not restored. Scheduled tasks do not replay conversation history. + +To verify multi-turn context, send “My name is Alex”, then “What is my name?” in the same conversation. A new conversation must not inherit that information through this history mechanism. ## Logs and diagnostics diff --git a/mateclaw-server/src/main/resources/docs/zh/deepseek-harness.md b/mateclaw-server/src/main/resources/docs/zh/deepseek-harness.md index 2a4d404c..a5c63d89 100644 --- a/mateclaw-server/src/main/resources/docs/zh/deepseek-harness.md +++ b/mateclaw-server/src/main/resources/docs/zh/deepseek-harness.md @@ -126,7 +126,9 @@ DSH 默认模型是 `deepseek-v4-flash`。如果员工没有绑定具体模型 - 页面不会显示“本次没有输出”。 - 同一个会话不会被并发启动两个 DSH live session。 -不要复用已经完成过的测试 `conversationId` 创建新的 DSH live session。DSH 会检测到磁盘上的 session 日志与新的 live session 不一致，并返回 `id collision`。请使用“新对话”创建新的会话。 +可以在同一个 MateClaw 会话中继续聊天。每轮仍使用新的 DSH 进程和运行时会话 ID，避免持久化日志的 `id collision`。MateClaw 会补入当前会话最近最多 40 条已完成的用户/助手文本消息，历史部分限制在估算 4096 token 内。本轮消息只发送一次，不受该历史预算截断；较旧历史可能被省略，边界消息可能带有 `[truncated]` 标记。后端重启后仍可读取已保存的文本历史，但不会恢复 DSH 内部工具状态。定时任务不回放会话历史。 + +验证多轮上下文：在同一会话先发送“我的名字叫小明”，再问“我叫什么名字？”。新建会话不应通过这项历史机制获得前一会话的信息。 ## 日志与诊断 diff --git a/mateclaw-server/src/test/java/vip/mate/agent/AgentServiceDshHistoryTest.java b/mateclaw-server/src/test/java/vip/mate/agent/AgentServiceDshHistoryTest.java new file mode 100644 index 00000000..eeb641de --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/AgentServiceDshHistoryTest.java @@ -0,0 +1,47 @@ +package vip.mate.agent; + +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; +import reactor.core.publisher.Flux; +import vip.mate.agent.context.ChatOrigin; +import vip.mate.agent.model.AgentEntity; +import vip.mate.agent.repository.AgentMapper; +import vip.mate.agent.runtime.contract.AgentRuntimeConnection; +import vip.mate.agent.runtime.contract.AgentRuntimeCoordinator; +import vip.mate.agent.runtime.dsh.DshConversationHistory; +import vip.mate.memory.MemoryProperties; +import vip.mate.memory.service.MemoryRecallTracker; + +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +class AgentServiceDshHistoryTest { + @Test + void dshBranchPassesCapturedOriginAndHistoryToRuntime() { + var mapper = mock(AgentMapper.class); + var agent = new AgentEntity(); + agent.setId(1L); + agent.setRuntimeType("dsh"); + agent.setModelName("model"); + when(mapper.selectById(1L)).thenReturn(agent); + var memory = new MemoryProperties(); + memory.setLifecycleMediatorEnabled(false); + var service = new AgentService(mapper, null, mock(MemoryRecallTracker.class), null, memory, null, null); + var coordinator = mock(AgentRuntimeCoordinator.class); + var connection = mock(AgentRuntimeConnection.class); + var history = mock(DshConversationHistory.class); + var origin = ChatOrigin.EMPTY.withOriginMessageId(12L); + when(history.enrich("conversation", "question", "question", origin)).thenReturn("history plus question"); + when(coordinator.start(eq(agent), eq("conversation"), eq("conversation"), eq("model"), any(), any())) + .thenReturn(connection); + when(connection.prompt(anyString())).thenReturn(Flux.empty()); + ReflectionTestUtils.setField(service, "runtimeCoordinator", coordinator); + ReflectionTestUtils.setField(service, "dshConversationHistory", history); + + service.chatStructuredStream(1L, "question", "conversation", "user", null, origin).blockLast(); + + verify(history).enrich("conversation", "question", "question", origin); + verify(connection).prompt("history plus question"); + verify(connection).close(); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/agent/runtime/dsh/DshConversationHistoryTest.java b/mateclaw-server/src/test/java/vip/mate/agent/runtime/dsh/DshConversationHistoryTest.java new file mode 100644 index 00000000..cc2ef815 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/runtime/dsh/DshConversationHistoryTest.java @@ -0,0 +1,125 @@ +package vip.mate.agent.runtime.dsh; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.ibatis.session.SqlSession; +import org.h2.jdbcx.JdbcDataSource; +import org.junit.jupiter.api.*; +import org.springframework.jdbc.core.JdbcTemplate; +import vip.mate.agent.context.ChatOrigin; +import vip.mate.agent.context.TokenEstimator; +import vip.mate.workspace.conversation.repository.MessageMapper; + +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.*; + +class DshConversationHistoryTest { + private JdbcTemplate jdbc; + private SqlSession session; + private DshConversationHistory history; + + @BeforeEach + void setup() throws Exception { + var source = new JdbcDataSource(); + source.setURL("jdbc:h2:mem:dsh_" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1"); + jdbc = new JdbcTemplate(source); + jdbc.execute("CREATE TABLE mate_message (id BIGINT PRIMARY KEY, conversation_id VARCHAR(100), role VARCHAR(20), content CLOB, status VARCHAR(20), deleted INT DEFAULT 0, create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP)"); + var configuration = new MybatisConfiguration(); + configuration.addMapper(MessageMapper.class); + var factory = new MybatisSqlSessionFactoryBean(); + factory.setDataSource(source); + factory.setConfiguration(configuration); + session = factory.getObject().openSession(true); + history = new DshConversationHistory(session.getMapper(MessageMapper.class), new ObjectMapper()); + } + + @AfterEach + void close() { + if (session != null) session.close(); + if (jdbc != null) jdbc.execute("DROP ALL OBJECTS"); + } + + @Test + void secondTurnIncludesEarlierUserAndAssistantButNotCurrentOrFutureRows() { + row(1, "a", "user", "我的名字叫小明", "completed"); + row(2, "a", "assistant", "你好，小明", "completed"); + row(3, "a", "user", "我叫什么名字？", "completed"); + row(4, "a", "assistant", "future answer", "completed"); + String prompt = history.enrich("a", "我叫什么名字？", "enriched current question", + ChatOrigin.EMPTY.withOriginMessageId(3L)); + assertTrue(prompt.contains("我的名字叫小明")); + assertTrue(prompt.contains("你好，小明")); + assertTrue(prompt.indexOf("我的名字叫小明") < prompt.indexOf("你好，小明")); + assertFalse(prompt.contains("我叫什么名字？")); + assertFalse(prompt.contains("future answer")); + assertTrue(prompt.endsWith("enriched current question")); + } + + @Test + void isolatesConversationsAndExcludesDeletedFailedAndNonDialogueRows() { + row(1, "other", "user", "private secret", "completed"); + row(2, "a", "assistant", "failure placeholder", "failed"); + row(3, "a", "assistant", "still generating", "generating"); + row(4, "a", "tool", "tool internals", "completed"); + row(5, "a", "system", "system marker", "completed"); + row(6, "a", "user", "deleted text", "completed"); + jdbc.update("UPDATE mate_message SET deleted=1 WHERE id=6"); + assertEquals("question", history.enrich("a", "question", "question", ChatOrigin.EMPTY)); + assertEquals("question", history.enrich("new", "question", "question", ChatOrigin.EMPTY)); + } + + @Test + void missingOriginOnlyDeduplicatesMatchingTrailingUserMessage() { + row(1, "a", "user", "repeat", "completed"); + row(2, "a", "assistant", "previous answer", "completed"); + row(3, "a", "user", "repeat", "completed"); + String prompt = history.enrich("a", "repeat", "current enriched", ChatOrigin.EMPTY); + assertEquals(1, prompt.split("repeat", -1).length - 1); + assertTrue(prompt.contains("previous answer")); + } + + @Test + void nonPersistedInputDoesNotDropLastHistoricalMessage() { + row(1, "a", "user", "earlier question", "completed"); + String prompt = history.enrich("a", "new question", "new question", null); + assertTrue(prompt.contains("earlier question")); + } + + @Test + void boundedHistoryKeepsNewestContextAndCurrentInputIntact() { + row(1, "a", "user", "oldest secret", "completed"); + row(2, "a", "assistant", "界".repeat(9000), "completed"); + row(3, "a", "user", "recent useful fact", "completed"); + String prompt = history.enrich("a", "question", "question", ChatOrigin.EMPTY); + assertTrue(prompt.contains("recent useful fact")); + assertFalse(prompt.contains("oldest secret")); + assertTrue(prompt.contains("[truncated]")); + assertTrue(TokenEstimator.estimateTokens(prompt) <= 4096 + TokenEstimator.estimateTokens("question")); + assertTrue(prompt.endsWith("question")); + } + + @Test + void limitsHistoryRowsAndSurvivesNewBuilderInstance() { + for (int i = 1; i <= 45; i++) row(i, "a", "user", "row-" + i + "-text", "completed"); + var restarted = new DshConversationHistory(session.getMapper(MessageMapper.class), new ObjectMapper()); + String prompt = restarted.enrich("a", "question", "question", ChatOrigin.EMPTY); + assertFalse(prompt.contains("row-5-text")); + assertTrue(prompt.contains("row-6-text")); + assertTrue(prompt.contains("row-45-text")); + } + + @Test + void skipsScheduledTasks() { + row(1, "a", "user", "previous job", "completed"); + ChatOrigin cron = new ChatOrigin(null, "a", null, null, null, null, null, true, + null, null, null, null, null); + assertEquals("next job", history.enrich("a", "next job", "next job", cron)); + } + + private void row(long id, String conversation, String role, String content, String status) { + jdbc.update("INSERT INTO mate_message(id,conversation_id,role,content,status) VALUES(?,?,?,?,?)", + id, conversation, role, content, status); + } +}