This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 4a1020b08 feat(ai): include bounded prose history in HTTP
conversations (#7051)
4a1020b08 is described below
commit 4a1020b08fbff3e05c20305ddfcf8242b8574205
Author: aias00 <[email protected]>
AuthorDate: Mon Oct 26 17:09:22 2026 +0800
feat(ai): include bounded prose history in HTTP conversations (#7051)
* feat(ai): include bounded prose history in HTTP conversations
* refactor(ai): fix the HTTP history flag at admission and answer budget
breaches with 400
RunContext kept every input final except httpHistoryEnabled, which
AiRunService set
through a Lombok setter immediately after construction; make it the last
constructor
argument so a run's shape cannot change mid-flight.
The two history budget guards threw IllegalArgumentException where the
neighbouring
AiPayloadGuard checks answer with BusinessException(400), so an over-budget
history
reached the operator as a 500.
Also document the HTTP side of `resume` in api-spec.md: for the http engine
the flag
decides whether the turn carries bounded history (200 events scanned, at
most 16
turns and 64 KiB), history never gains the system prompt's authority, an
over-budget
history is a 400, and resume=false applies to that request only.
---
docs/ai-agent-architecture.md | 13 +++
docs/api-spec.md | 6 +-
.../studio/ops/ai/OpenAiCompatibleLlmClient.java | 35 +++++++
.../ops/ai/conversation/AiEventRepository.java | 3 +
.../studio/ops/ai/conversation/AiRunExecutor.java | 21 +++-
.../studio/ops/ai/conversation/AiRunService.java | 5 +-
.../ai/conversation/HttpConversationContext.java | 80 +++++++++++++++
.../conversation/MybatisPlusAiEventRepository.java | 14 +++
.../ops/ai/OpenAiCompatibleLlmClientTest.java | 32 ++++++
.../AiConversationPersistenceIntegrationTest.java | 12 +++
.../ops/ai/conversation/AiRunExecutorTest.java | 34 +++++-
.../ops/ai/conversation/AiRunServiceTest.java | 21 +++-
.../ai/conversation/AiTimelineRepositoryTest.java | 26 +++++
.../conversation/HttpConversationContextTest.java | 114 +++++++++++++++++++++
14 files changed, 400 insertions(+), 16 deletions(-)
diff --git a/docs/ai-agent-architecture.md b/docs/ai-agent-architecture.md
index 3d42f78a2..fb64c8897 100644
--- a/docs/ai-agent-architecture.md
+++ b/docs/ai-agent-architecture.md
@@ -157,6 +157,19 @@ Agent 循环 ⊃ 一条时间线条目。不带多租户 `tenant_id`、`shareSco
`server/src/test/resources/ai/ai-event-contract.json`：Java 的
`AiEventContractTest` 与 TS 的
`aiEvents.contract.test.ts` 读同一个文件，各自断言类型集合与示例。任一侧漂移，另一侧 CI 变红。
+### HTTP 多轮文本上下文
+
+HTTP 正式回答会读取当前 run 的 `start_seq` 之前、同一 conversation 最近的 200 条事件
+（额外一条仅用于探测裁剪）。先按索引选出有界 seq，再读取这些事件的 payload；不对大文本排序。
+只把完整的用户问题和助手文本组成独立的 `user`/`assistant` 消息，忽略 reasoning、工具和诊断。
+同一轮连续文本按原顺序合并，最多保留最近 16 轮、64 KiB UTF-8 历史；超预算时丢弃整轮，
+不切断 Unicode 或拼出半轮回答。当前问题单独追加且始终保留，总输入仍经既有 payload 校验。
+历史被裁剪或存在不完整/不可读轮次时，使用已有 Notice 告知上下文有限。
+
+`resume=false` 让该 HTTP 请求不携带历史；它不是删除数据库历史或为后续请求建立持久重置点。
+旧的单 prompt HTTP 调用和 CLI 会话保持原行为。HTTP 仍没有 CLI 的 runtime session、工具执行或
+完整无限上下文；未出现在有界窗口中的内容不会自动总结，限制是应用文本预算而非精确 token 预算。
+
## 7. 生成与 HTTP 连接解耦
SSE 流是**观察者，不是所有者**。今天（改造前）`LlmSseSession` 把 `emitter.onCompletion/onTimeout/
diff --git a/docs/api-spec.md b/docs/api-spec.md
index 798cb4d18..9066bfba0 100644
--- a/docs/api-spec.md
+++ b/docs/api-spec.md
@@ -3026,7 +3026,11 @@ POST /api/ai/conversations/:id/messages
| `engine` | `string` | 否 | 覆盖本轮引擎（`http` / `claude-code` / `qoder`），最长 16 字符 |
| `mode` | `string` | 否 | 覆盖本轮模式，最长 16 字符 |
| `enhance` | `boolean` | 否 | 是否先做 prompt 增强改写；改写过程以 `source: "enhance"` 的
`thinking` 帧流回，默认 `false` |
-| `resume` | `boolean` | 否 | 是否复用上游 Agent 会话（`claude --resume`）。**省略 =
沿用会话自身状态**，`false` = 强制开新会话，`true` = 强制复用 |
+| `resume` | `boolean` | 否 | 是否复用上游会话。**省略 = 沿用会话自身状态**，`false` =
强制开新会话，`true` = 强制复用。CLI 引擎（`claude-code` / `qoder`）表现为是否带 `--resume`；**`http`
引擎表现为是否随本轮携带此前对话历史**——`false` 只发本轮提问，省略或 `true` 时附带**有界**历史（读当前 run `start_seq`
之前最近 200 条事件，最多 16 轮 / 64 KiB UTF-8，只取完整的用户问题与助手文本，超预算整轮丢弃） |
+
+> `http` 引擎的历史只作为普通 `user` / `assistant` 消息附带，不取得系统提示词的权限；历史超过 64 条、
+> 或累计超过出站上下文预算时按 **400** 拒绝（与 `AiPayloadGuard` 同一口径）。`resume=false` 只影响
+> 本次请求，不删除库里的历史、也不为后续请求建立持久重置点。CLI 引擎与旧的单 prompt HTTP 调用行为不变。
**Response:** `text/event-stream`
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClient.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClient.java
index 63966eb36..42173aa6a 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClient.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClient.java
@@ -19,6 +19,7 @@ package org.apache.rocketmq.studio.ops.ai;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
@@ -141,9 +142,43 @@ public class OpenAiCompatibleLlmClient {
}
public void stream(LlmConfigVO config, String prompt, String
modelOverride, Consumer<String> tokenConsumer) {
+ streamConversation(config, List.of(), prompt, modelOverride,
tokenConsumer);
+ }
+
+ /** Prior conversational content never acquires the authority of the
formatting system message. */
+ public record ChatMessage(String role, String content) {
+ public ChatMessage {
+ if (!("user".equals(role) || "assistant".equals(role)) || content
== null) {
+ throw new IllegalArgumentException("History messages require a
user/assistant role and content");
+ }
+ }
+ }
+
+ public void streamConversation(LlmConfigVO config, List<ChatMessage>
history, String prompt,
+ String modelOverride, Consumer<String>
tokenConsumer) {
validate(config);
AiPayloadGuard.validateOutboundPrompt(prompt, effectiveModel(config,
modelOverride));
Map<String, Object> requestBody = requestBody(config, prompt,
modelOverride, true);
+ if (history.size() > 64) {
+ throw new BusinessException(400, "LLM history must not exceed 64
messages");
+ }
+ List<Object> messages = new ArrayList<>();
+ messages.add(Map.of("role", "system", "content",
MARKDOWN_SYSTEM_PROMPT));
+ int historyBytes = 0;
+ StringBuilder combined = new StringBuilder(prompt == null ? "" :
prompt);
+ for (ChatMessage message : history) {
+ AiPayloadGuard.validateOutboundPrompt(message.content(),
effectiveModel(config, modelOverride));
+ historyBytes +=
message.content().getBytes(StandardCharsets.UTF_8).length;
+ if (historyBytes > AiPayloadGuard.MAX_CONTEXT_BYTES) {
+ throw new BusinessException(400, "LLM history must not exceed "
+ + AiPayloadGuard.MAX_CONTEXT_BYTES + " UTF-8 bytes");
+ }
+ messages.add(message);
+ combined.append(message.content());
+ }
+ AiPayloadGuard.validateOutboundPrompt(combined.toString(),
effectiveModel(config, modelOverride));
+ messages.add(Map.of("role", "user", "content",
StringUtils.hasText(prompt) ? prompt.trim() : ""));
+ requestBody.put("messages", messages);
HttpRequest request = request(config, "text/event-stream",
requestBody);
try {
HttpResponse<StreamBody> response = httpClient.send(request,
streamBodyHandler());
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiEventRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiEventRepository.java
index f6868e8e1..0b18044f6 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiEventRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiEventRepository.java
@@ -44,6 +44,9 @@ public interface AiEventRepository {
*/
List<RmqAiEvent> findByConversationIdAfterSeq(Long conversationId, int
afterSeq, int limit);
+ /** Latest bounded rows before an admitted run, returned in ascending seq
order. */
+ List<RmqAiEvent> findRecentBeforeSeq(Long conversationId, int beforeSeq,
int limit);
+
int deleteByConversationId(Long conversationId);
int deleteByConversationIds(List<Long> conversationIds);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutor.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutor.java
index ac2206b3c..c99549af6 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutor.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutor.java
@@ -381,10 +381,8 @@ public class AiRunExecutor {
}
/**
- * The {@code http} engine has no agent CLI, so it stays a single-turn
completion streamed as text
- * deltas: no reasoning, no tools, no resumable session id. That is a real
limitation of the engine
- * and not something to paper over — the run row keeps {@code
runtime_session_id} null and the next
- * turn starts from scratch.
+ * HTTP carries bounded persisted prose history, not a CLI session,
reasoning or tool execution.
+ * The run row still has no provider runtime_session_id; resume=false
omits prior messages.
*/
private void streamHttp(RunContext context, String prompt,
Consumer<AgentEvent> events) {
LlmConfigVO config = context.getConfig();
@@ -393,7 +391,15 @@ public class AiRunExecutor {
"LLM provider is not supported by the OpenAI-compatible
gateway",
"Use one of: openai, deepseek, tongyi, ollama.");
}
- llmClient.stream(config, prompt, context.getRun().getModel(),
+ HttpConversationContext history = context.isHttpHistoryEnabled()
+ ? HttpConversationContext.read(eventRepository, objectMapper,
+ context.getRun().getConversationId(),
context.getRun().getStartSeq())
+ : new HttpConversationContext(java.util.List.of(), false);
+ if (history.truncated()) {
+ events.accept(new
AgentEvent.ProviderNotice(AgentEventProjector.LEVEL_WARN,
+ "Only bounded recent question/answer history is included
in this HTTP turn."));
+ }
+ llmClient.streamConversation(config, history.messages(), prompt,
context.getRun().getModel(),
token -> events.accept(new AgentEvent.TextDelta(token)));
}
@@ -723,6 +729,11 @@ public class AiRunExecutor {
private final Duration timeout;
/** Null when the conversation runs without RocketMQ tools. */
private final RmqctlWorkspace.Preparation preparation;
+ /**
+ * False when the caller asked not to resume, so an HTTP run carries
only this turn. Final
+ * like every other input: a run's shape is fixed when it is admitted,
not mutated mid-flight.
+ */
+ private final boolean httpHistoryEnabled;
private final AtomicReference<State> state = new
AtomicReference<>(State.ACTIVE);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunService.java
index 4f279ab0a..426bd7566 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunService.java
@@ -236,7 +236,8 @@ public class AiRunService {
AgentStreamSession session = runExecutor.newSession(run.getId(),
timeoutMillis);
AiRunExecutor.RunContext context = new
AiRunExecutor.RunContext(conversation, run, sink, handle,
config, engine, message, resumeSessionId, request.enhance(),
- Duration.ofMillis(timeoutMillis), preparation);
+ Duration.ofMillis(timeoutMillis), preparation,
+ !Boolean.FALSE.equals(request.resume()));
registry.register(run.getId(), handle);
try {
// enhancedPrompt stays null on this row: the rewrite happens on
the worker, after the turn
@@ -596,7 +597,7 @@ public class AiRunService {
AiEventSink sink = runExecutor.newSink(run.getConversationId(),
run.getId(),
run.getTurn() == null ? 0 : run.getTurn(), seedSeq);
return new AiRunExecutor.RunContext(conversation, run, sink, null,
null, run.getEngine(), null,
- null, false, null, null);
+ null, false, null, null, true);
}
/** The notice a conversation gets when it has to do without RocketMQ
tools. */
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/HttpConversationContext.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/HttpConversationContext.java
new file mode 100644
index 000000000..84d1d7e3a
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/HttpConversationContext.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.ops.ai.conversation;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.studio.ops.ai.OpenAiCompatibleLlmClient.ChatMessage;
+import org.apache.rocketmq.studio.ops.ai.conversation.event.TimelineEvent;
+import org.apache.rocketmq.studio.persistence.entity.RmqAiEvent;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.TreeMap;
+
+/** A bounded suffix of complete prose turns; reasoning and tools never become
model instructions. */
+record HttpConversationContext(List<ChatMessage> messages, boolean truncated) {
+ private static final int MAX_ROWS = 200;
+ private static final int MAX_TURNS = 16;
+ private static final int MAX_BYTES = 64 * 1024;
+
+ static HttpConversationContext read(AiEventRepository repository,
ObjectMapper mapper,
+ Long conversationId, Integer
beforeSeq) {
+ if (beforeSeq == null || beforeSeq <= 1) return new
HttpConversationContext(List.of(), false);
+ List<RmqAiEvent> rows = repository.findRecentBeforeSeq(conversationId,
beforeSeq, MAX_ROWS + 1);
+ boolean truncated = rows.size() > MAX_ROWS;
+ TreeMap<Integer, Turn> turns = new TreeMap<>();
+ for (RmqAiEvent row : rows.subList(Math.max(0, rows.size() -
MAX_ROWS), rows.size())) {
+ Turn turn = turns.computeIfAbsent(row.getTurn(), ignored -> new
Turn());
+ TimelineEvent event = AiEventCodec.read(mapper, row).orElse(null);
+ if (event == null) {
+ turn.invalid = true;
+ } else if (event instanceof TimelineEvent.User user) {
+ turn.question = user.text();
+ } else if (event instanceof TimelineEvent.Text text) {
+ if (text.text() == null) turn.invalid = true;
+ else turn.answer.append(text.text());
+ }
+ }
+ List<List<ChatMessage>> selected = new ArrayList<>();
+ int bytes = 0;
+ for (Turn turn : turns.descendingMap().values()) {
+ if (turn.invalid || turn.question == null ||
turn.answer.isEmpty()) {
+ truncated = true;
+ continue;
+ }
+ String answer = turn.answer.toString();
+ int size = turn.question.getBytes(StandardCharsets.UTF_8).length
+ + answer.getBytes(StandardCharsets.UTF_8).length;
+ if (selected.size() >= MAX_TURNS || size > MAX_BYTES - bytes) {
+ truncated = true;
+ break;
+ }
+ bytes += size;
+ selected.add(List.of(new ChatMessage("user", turn.question), new
ChatMessage("assistant", answer)));
+ }
+ List<ChatMessage> messages = new ArrayList<>();
+ for (int i = selected.size() - 1; i >= 0; i--)
messages.addAll(selected.get(i));
+ return new HttpConversationContext(List.copyOf(messages), truncated);
+ }
+
+ private static final class Turn {
+ private String question;
+ private final StringBuilder answer = new StringBuilder();
+ private boolean invalid;
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/MybatisPlusAiEventRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/MybatisPlusAiEventRepository.java
index c9ef64f03..c977971b3 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/MybatisPlusAiEventRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/MybatisPlusAiEventRepository.java
@@ -73,6 +73,20 @@ public class MybatisPlusAiEventRepository implements
AiEventRepository {
return sorted;
}
+ @Override
+ public List<RmqAiEvent> findRecentBeforeSeq(Long conversationId, int
beforeSeq, int limit) {
+ if (conversationId == null || beforeSeq <= 0 || limit <= 0) return
List.of();
+ // Sort only seq keys on the unique index; fetch payloads by those
bounded keys, never filesort text.
+ List<Object> keys = eventMapper.selectObjs(new
QueryWrapper<RmqAiEvent>()
+ .select("seq").eq("conversation_id", conversationId).lt("seq",
beforeSeq)
+ .orderByDesc("seq").last("LIMIT " + Math.min(limit,
MAX_LIMIT)));
+ if (keys.isEmpty()) return List.of();
+ List<RmqAiEvent> rows = new ArrayList<>(eventMapper.selectList(new
QueryWrapper<RmqAiEvent>()
+ .eq("conversation_id", conversationId).in("seq", keys)));
+ rows.sort(Comparator.comparing(RmqAiEvent::getSeq));
+ return rows;
+ }
+
@Override
public int deleteByConversationId(Long conversationId) {
if (conversationId == null) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClientTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClientTest.java
index 86583f40d..1348f480a 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClientTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/OpenAiCompatibleLlmClientTest.java
@@ -134,6 +134,38 @@ class OpenAiCompatibleLlmClientTest {
});
}
+ @Test
+ void conversationStreamShouldSendHistoryAsSeparateRolesTest() {
+ AtomicReference<JsonNode> body = new AtomicReference<>();
+ server.createContext("/v1/chat/completions", exchange -> {
+ body.set(objectMapper.readTree(exchange.getRequestBody()));
+ respond(exchange, 200, "data: [DONE]\n\n", "text/event-stream");
+ });
+ client.streamConversation(config("openai", "sk-test"), List.of(
+ new OpenAiCompatibleLlmClient.ChatMessage("user", "previous
question"),
+ new OpenAiCompatibleLlmClient.ChatMessage("assistant",
"previous answer")),
+ "follow-up", "selected-model", ignored -> { });
+
+
assertThat(body.get().path("model").asText()).isEqualTo("selected-model");
+ JsonNode messages = body.get().path("messages");
+ assertThat(messages.size()).isEqualTo(4);
+ assertThat(messages.get(0).path("role").asText()).isEqualTo("system");
+
assertThat(messages.get(1).path("content").asText()).isEqualTo("previous
question");
+
assertThat(messages.get(2).path("role").asText()).isEqualTo("assistant");
+
assertThat(messages.get(3).path("content").asText()).isEqualTo("follow-up");
+ assertThatThrownBy(() -> new
OpenAiCompatibleLlmClient.ChatMessage("system", "untrusted"))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void conversationContextMustFitTheCombinedOutboundBudgetTest() {
+ assertThatThrownBy(() -> client.streamConversation(config("openai",
"sk-test"),
+ List.of(new OpenAiCompatibleLlmClient.ChatMessage("user",
"x".repeat(200_000))),
+ "y".repeat(200_000), null, ignored -> { }))
+ .isInstanceOf(LlmGatewayException.class)
+ .hasMessageContaining("UTF-8 bytes");
+ }
+
@Test
void streamShouldParseOpenAiCompatibleSseDeltas() {
AtomicReference<JsonNode> requestBody = new AtomicReference<>();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationPersistenceIntegrationTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationPersistenceIntegrationTest.java
index d5cad3686..c761526c0 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationPersistenceIntegrationTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationPersistenceIntegrationTest.java
@@ -307,6 +307,18 @@ class AiConversationPersistenceIntegrationTest {
markCompleted(seeded.secondRunId);
}
+ @Test
+ void
recentHistoryShouldStayBeforeTheCurrentRunAndInsideItsConversationTest() {
+ Seeded seeded = seed();
+ Seeded other = seed();
+ assertThat(eventRepository.findRecentBeforeSeq(seeded.conversationId,
101, 3))
+ .extracting(RmqAiEvent::getSeq).containsExactly(98, 99, 100);
+ assertThat(eventRepository.findRecentBeforeSeq(seeded.conversationId,
101, 3))
+ .allSatisfy(row -> assertThat(row.getConversationId())
+
.isEqualTo(seeded.conversationId).isNotEqualTo(other.conversationId));
+ assertThat(eventRepository.findRecentBeforeSeq(seeded.conversationId,
1, 3)).isEmpty();
+ }
+
@Test
void deletingAConversationShouldTakeItsChildrenWithItTest() {
Seeded seeded = seed();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutorTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutorTest.java
index 56004a90f..f3b5fa477 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutorTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunExecutorTest.java
@@ -60,6 +60,7 @@ import java.util.function.Consumer;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
@@ -205,7 +206,7 @@ class AiRunExecutorTest {
AiRunExecutor.RunContext base = context(ENGINE, true);
AiRunExecutor.RunContext shortRun = new
AiRunExecutor.RunContext(conversation, run,
base.getSink(), base.getHandle(), base.getConfig(),
ENGINE, "hello", null,
- true, Duration.ofMillis(500), null);
+ true, Duration.ofMillis(500), null, true);
executor.submit(shortRun);
assertThat(enhancing.await(3, TimeUnit.SECONDS)).isTrue();
assertThat(terminal.await(3, TimeUnit.SECONDS)).isTrue();
@@ -338,7 +339,7 @@ class AiRunExecutorTest {
session.finishReplay();
lateExecutor.submit(new AiRunExecutor.RunContext(conversation, run,
sink, handle,
LlmConfigVO.builder().engine(ENGINE).model("qwen3.8-max").enabled(true).build(),
- ENGINE, "hello", null, false, Duration.ofSeconds(300), null));
+ ENGINE, "hello", null, false, Duration.ofSeconds(300), null,
true));
assertThat(runRow().getStatus()).isEqualTo(RunStatus.COMPLETED.name());
assertThat(runRow().getStopReason()).isNull();
@@ -580,14 +581,37 @@ class AiRunExecutorTest {
assertThat(payloads().get(2)).contains("checkpoint");
}
+ @Test
+ void httpTurnShouldReceiveOnlyItsPriorConversationMessagesTest() {
+ when(llmClient.supports(any())).thenReturn(true);
+ RmqAiEvent question = new RmqAiEvent();
+ question.setSeq(1);
+ question.setTurn(1);
+ question.setPayload("{\"type\":\"user\",\"text\":\"previous
question\"}");
+ RmqAiEvent answer = new RmqAiEvent();
+ answer.setSeq(2);
+ answer.setTurn(1);
+ answer.setPayload("{\"type\":\"text\",\"text\":\"previous answer\"}");
+ when(eventRepository.findRecentBeforeSeq(eq(CONVERSATION_ID), eq(3),
anyInt()))
+ .thenReturn(List.of(question, answer));
+ run.setStartSeq(3);
+
+ executor.submit(context(LlmConfigVO.ENGINE_HTTP));
+
+ verify(llmClient).streamConversation(any(), eq(List.of(
+ new OpenAiCompatibleLlmClient.ChatMessage("user", "previous
question"),
+ new OpenAiCompatibleLlmClient.ChatMessage("assistant",
"previous answer"))),
+ eq("hello"), eq("qwen3.8-max"), any());
+ }
+
@Test
void httpEngineShouldStreamTokensAsTextDeltasWithoutAnAgentProviderTest() {
doAnswer(invocation -> {
- Consumer<String> tokens = invocation.getArgument(3);
+ Consumer<String> tokens = invocation.getArgument(4);
tokens.accept("hello ");
tokens.accept("world");
return null;
- }).when(llmClient).stream(any(), any(), any(), any());
+ }).when(llmClient).streamConversation(any(), any(), any(), any(),
any());
when(llmClient.supports(any())).thenReturn(true);
AiRunExecutor.RunContext context = context(LlmConfigVO.ENGINE_HTTP);
@@ -725,7 +749,7 @@ class AiRunExecutorTest {
session.finishReplay();
return new AiRunExecutor.RunContext(conversation, run, sink, handle,
LlmConfigVO.builder().engine(engine).model("qwen3.8-max").enabled(true).build(),
- engine, "hello", resumeSessionId, enhance,
Duration.ofSeconds(300), null);
+ engine, "hello", resumeSessionId, enhance,
Duration.ofSeconds(300), null, true);
}
private List<String> types() {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunServiceTest.java
index bd35285ad..bfb734997 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiRunServiceTest.java
@@ -61,6 +61,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -250,14 +251,28 @@ class AiRunServiceTest {
assertThat(inserted.get(1).getPayload()).contains(RmqctlWorkspace.UNBOUND_NOTICE_MESSAGE);
}
+ @Test
+ void resumeFalseShouldKeepAnHttpTurnIndependentOfPriorHistoryTest() {
+ when(eventRepository.maxSeq(CONVERSATION_ID)).thenReturn(10);
+ when(llmClient.supports(any())).thenReturn(true);
+ service.sendMessage(CONVERSATION_ID, new AiRunService.RunRequest(
+ "fresh question", null, LlmConfigVO.ENGINE_HTTP, null, false,
false));
+
+ verify(eventRepository, never()).findRecentBeforeSeq(anyLong(),
anyInt(), anyInt());
+ verify(llmClient).streamConversation(any(),
org.mockito.ArgumentMatchers.eq(List.of()),
+ org.mockito.ArgumentMatchers.eq("fresh question"), any(),
any());
+ assertThat(runUpdates.get(runUpdates.size() - 1).getStatus())
+ .isEqualTo(RunStatus.COMPLETED.name());
+ }
+
@Test
void anHttpTurnShouldNotPrepareAnAgentWorkspaceItCannotUseTest() {
when(llmClient.supports(any())).thenReturn(true);
doAnswer(invocation -> {
- Consumer<String> tokens = invocation.getArgument(3);
+ Consumer<String> tokens = invocation.getArgument(4);
tokens.accept("pong");
return null;
- }).when(llmClient).stream(any(), any(), any(), any());
+ }).when(llmClient).streamConversation(any(), any(), any(), any(),
any());
when(workspace.prepare(anyLong(), any()))
.thenThrow(new BusinessException(500, "the workspace directory
is unusable"));
@@ -550,7 +565,7 @@ class AiRunServiceTest {
AiRunExecutor.RunContext context = new
AiRunExecutor.RunContext(conversation,
AiRunTestSupport.run(RUN_ID, CONVERSATION_ID, 1,
RunStatus.RUNNING), sink,
new AgentRunHandle(RUN_ID, Duration.ofMillis(1)), null,
provider.engine(), "hello",
- null, false, Duration.ofSeconds(300), null);
+ null, false, Duration.ofSeconds(300), null, true);
// A stop, a provider failure, the worker's own safety net and a
shutdown drain can all reach
// finalisation for the same run; the compareAndSet is what makes
exactly one of them write.
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiTimelineRepositoryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiTimelineRepositoryTest.java
index 67362d056..c45c38030 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiTimelineRepositoryTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/AiTimelineRepositoryTest.java
@@ -171,6 +171,32 @@ class AiTimelineRepositoryTest {
// --- the cursor the caller derives
-------------------------------------------
+ @Test
+ void recentHistoryShouldSortOnlyBoundedIndexKeysNotPayloadsTest() {
+ when(eventMapper.<Object>selectObjs(any())).thenReturn(List.of(9, 8));
+ when(eventMapper.selectList(any())).thenReturn(List.of(row(909L, 9),
row(908L, 8)));
+
+ assertThat(repository.findRecentBeforeSeq(CONVERSATION_ID, 10, 9999))
+ .extracting(RmqAiEvent::getSeq).containsExactly(8, 9);
+ QueryWrapper<RmqAiEvent> keys = capturedEventAggregate();
+ assertThat(keys.getSqlSelect()).isEqualTo("seq");
+ assertThat(keys.getTargetSql()).contains("conversation_id = ? AND seq
< ?")
+ .contains("ORDER BY seq DESC LIMIT 500");
+ assertThat(keys.getParamNameValuePairs().values())
+ .containsExactlyInAnyOrder(CONVERSATION_ID, 10);
+ assertThat(capturedQuery().getTargetSql()).contains("conversation_id =
? AND seq IN (?,?)")
+ .doesNotContain("ORDER BY");
+ }
+
+ @Test
+ void recentHistoryShouldSkipSqlWithoutAConversationRangeTest() {
+ assertThat(repository.findRecentBeforeSeq(null, 10, 10)).isEmpty();
+ assertThat(repository.findRecentBeforeSeq(CONVERSATION_ID, 0,
10)).isEmpty();
+ assertThat(repository.findRecentBeforeSeq(CONVERSATION_ID, 10,
0)).isEmpty();
+ verify(eventMapper, never()).selectObjs(any());
+ verify(eventMapper, never()).selectList(any());
+ }
+
@Test
void theCursorShouldBeTheLastRowOfThePageWithTheProbeRowDroppedTest() {
AiConversationService service = serviceOverTheRealRepository();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/HttpConversationContextTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/HttpConversationContextTest.java
new file mode 100644
index 000000000..485718ef5
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/conversation/HttpConversationContextTest.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.ops.ai.conversation;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.studio.ops.ai.OpenAiCompatibleLlmClient.ChatMessage;
+import org.apache.rocketmq.studio.persistence.entity.RmqAiEvent;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class HttpConversationContextTest {
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final AiEventRepository repository = mock(AiEventRepository.class);
+
+ @Test
+ void onlyCompleteQuestionAnswerTurnsShouldBecomeRoleMessagesTest() {
+ when(repository.findRecentBeforeSeq(eq(7L), eq(20),
anyInt())).thenReturn(List.of(
+ row(1, 1, "user", "question"), row(2, 1, "text", "answer "),
+ row(3, 1, "thinking", "private reasoning"), row(4, 1, "text",
"tail"),
+ row(5, 2, "user", "unanswered"), row(6, 3, "text", "orphan
fragment")));
+ HttpConversationContext context =
HttpConversationContext.read(repository, mapper, 7L, 20);
+ assertThat(context.messages()).containsExactly(new ChatMessage("user",
"question"),
+ new ChatMessage("assistant", "answer tail"));
+ assertThat(context.truncated()).isTrue();
+ }
+
+ @Test
+ void recentTurnsShouldWinTheMessageBudgetTest() {
+ List<RmqAiEvent> rows = new ArrayList<>();
+ for (int i = 1; i <= 18; i++) {
+ rows.add(row(2 * i - 1, i, "user", "q" + i));
+ rows.add(row(2 * i, i, "text", "a" + i));
+ }
+ when(repository.findRecentBeforeSeq(eq(7L), eq(100),
anyInt())).thenReturn(rows);
+ HttpConversationContext context =
HttpConversationContext.read(repository, mapper, 7L, 100);
+ assertThat(context.messages()).hasSize(32);
+ assertThat(context.messages().get(0).content()).isEqualTo("q3");
+ assertThat(context.messages().get(31).content()).isEqualTo("a18");
+ assertThat(context.truncated()).isTrue();
+ }
+
+ @Test
+ void eventWindowShouldNotIncludeAnOrphanedLeadingAnswerTest() {
+ List<RmqAiEvent> rows = new ArrayList<>();
+ for (int i = 1; i <= 101; i++) {
+ rows.add(row(2 * i - 1, i, "user", "q" + i));
+ if (i <= 100) rows.add(row(2 * i, i, "text", "a" + i));
+ }
+ when(repository.findRecentBeforeSeq(eq(7L), eq(300),
anyInt())).thenReturn(rows);
+ HttpConversationContext context =
HttpConversationContext.read(repository, mapper, 7L, 300);
+ verify(repository).findRecentBeforeSeq(7L, 300, 201);
+ assertThat(context.messages()).hasSize(32);
+ assertThat(context.messages().get(0).content()).isEqualTo("q85");
+ assertThat(context.messages().get(31).content()).isEqualTo("a100");
+ assertThat(context.truncated()).isTrue();
+ }
+
+ @Test
+ void byteBudgetShouldDropWholeTurnsRatherThanSplitUnicodeTest() {
+ String large = new String(Character.toChars(0x1F600)).repeat(17_000);
+ when(repository.findRecentBeforeSeq(eq(7L), eq(100),
anyInt())).thenReturn(List.of(
+ row(1, 1, "user", "older"), row(2, 1, "text", "old answer"),
+ row(3, 2, "user", "newer"), row(4, 2, "text", large)));
+ HttpConversationContext context =
HttpConversationContext.read(repository, mapper, 7L, 100);
+ assertThat(context.messages()).isEmpty();
+ assertThat(context.truncated()).isTrue();
+ }
+
+ @Test
+ void unreadableTurnShouldNotBecomeAnUnpairedAnswerTest() {
+ RmqAiEvent bad = row(1, 1, "user", "bad");
+ bad.setPayload("{broken");
+ when(repository.findRecentBeforeSeq(eq(7L), eq(100),
anyInt())).thenReturn(List.of(
+ bad, row(2, 1, "text", "do not replay"),
+ row(3, 2, "user", "valid"), row(4, 2, "text", "valid
answer")));
+ HttpConversationContext context =
HttpConversationContext.read(repository, mapper, 7L, 100);
+ assertThat(context.messages()).containsExactly(new ChatMessage("user",
"valid"),
+ new ChatMessage("assistant", "valid answer"));
+ assertThat(context.truncated()).isTrue();
+ }
+
+ private RmqAiEvent row(int seq, int turn, String type, String text) {
+ RmqAiEvent row = new RmqAiEvent();
+ row.setConversationId(7L);
+ row.setSeq(seq);
+ row.setTurn(turn);
+ String source = "thinking".equals(type) ? ",\"source\":\"model\"" : "";
+ row.setPayload("{\"type\":\"" + type + "\",\"text\":\"" + text + "\""
+ source + "}");
+ return row;
+ }
+}
