You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2026/09/21 13:09:09 UTC
(rocketmq-dashboard) branch rocketmq-studio updated: fix(ai): retry a turn whose remembered agent session is gone (#4693)
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 02d81ec28 fix(ai): retry a turn whose remembered agent session is gone (#4693)
02d81ec28 is described below
commit 02d81ec286b068f1ec42c9827d970f52dcc63c55
Author: Apulupie <16...@users.noreply.github.com>
AuthorDate: Mon Sep 21 21:09:03 2026 +0800
fix(ai): retry a turn whose remembered agent session is gone (#4693)
`ResumeRecovery` documented the lost-resume case in detail but had no production caller. `RmqctlWorkspace` keeps the agent workspace under `/tmp`, so a container restart wipes it while `conversation.runtime_session_id` survives in MySQL; the CLI then exits non-zero with "No conversation found with session ID" and emits only a result frame whose subtype is `error_during_execution`, echoing the requested id back. `AiRunExecutor.record` trusted `ResultMeta.runtimeSessionId()` uncondition [...]
`ClaudeCodeStreamParser` now captures the result subtype and `ClaudeCodeAgentProvider.streamEvents` raises an `LlmGatewayException` carrying `ResumeRecovery.RESUME_LOST_CODE` when a resumed run reports it. `AiRunExecutor.streamWithLostResumeRetry` catches that code, forgets the session through the new `clearRuntimeSessionId` — needed because `updateById`'s NOT_NULL field strategy cannot express a null — discards the first attempt's outcome fields, emits a warn `ProviderNotice` and re- [...]
Fixes #4694
---
.../studio/ops/ai/ClaudeCodeAgentProvider.java | 25 +++++-
.../studio/ops/ai/ClaudeCodeStreamParser.java | 15 +++-
.../ai/conversation/AiConversationRepository.java | 6 ++
.../studio/ops/ai/conversation/AiRunExecutor.java | 98 +++++++++++++++++++---
.../MybatisPlusAiConversationRepository.java | 12 +++
.../ops/ai/conversation/agent/ResumeRecovery.java | 12 +++
.../studio/ops/ai/ClaudeCodeAgentProviderTest.java | 41 +++++++++
.../ops/ai/conversation/AiRunExecutorTest.java | 86 ++++++++++++++++++-
.../ai/conversation/AiTimelineRepositoryTest.java | 34 ++++++++
9 files changed, 315 insertions(+), 14 deletions(-)
diff --git a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeAgentProvider.java b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeAgentProvider.java
index 606983824..08c3d1a3d 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeAgentProvider.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeAgentProvider.java
@@ -18,6 +18,7 @@ package org.apache.rocketmq.studio.ops.ai;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.studio.ops.ai.conversation.agent.AgentStreamOptions;
+import org.apache.rocketmq.studio.ops.ai.conversation.agent.ResumeRecovery;
import org.apache.rocketmq.studio.ops.ai.conversation.event.AgentEvent;
import org.apache.rocketmq.studio.ops.ai.conversation.event.AgentEventProjector;
import org.springframework.stereotype.Component;
@@ -180,7 +181,8 @@ public class ClaudeCodeAgentProvider extends CliAgentProvider {
* <p>A non-zero exit is only an error when the CLI never produced its terminal {@code result}
* frame. When it did, the frame already said what happened — {@code error_max_turns}, an
* {@code api_error_status}, a permission denial — and the parser turned it into events, so
- * throwing here would replace a precise diagnosis with a generic one.
+ * throwing here would replace a precise diagnosis with a generic one. The single exception is a
+ * {@code --resume} session that no longer exists: see {@link #throwIfTheResumeSessionWasLost}.
*/
@Override
public void streamEvents(LlmConfigVO config, AgentStreamOptions options, Consumer<AgentEvent> sink) {
@@ -199,6 +201,7 @@ public class ClaudeCodeAgentProvider extends CliAgentProvider {
SpawnResult spawn = spawn(command, childEnv(config, options), timeoutSeconds,
options.getWorkspaceDir(), line -> parser.parseLine(line).forEach(sink),
options.getProcessSink());
+ throwIfTheResumeSessionWasLost(options, parser, spawn);
if (spawn.exitCode() != 0 && !parser.resultFrameSeen()) {
throw new LlmGatewayException(502, "llm.provider.cli_error",
BINARY + " CLI failed: " + describeStderr(spawn.stderr()),
@@ -206,6 +209,26 @@ public class ClaudeCodeAgentProvider extends CliAgentProvider {
}
}
+ /**
+ * Reports the one failure a caller can repair by retrying: the conversation's {@code --resume}
+ * session is gone. The frames have already said so, so this adds no diagnosis — what the caller
+ * cannot know on its own is that the retry has to drop {@code --resume}, and the command is built
+ * here. {@link ResumeRecovery} holds the contract.
+ *
+ * <p>Both halves of the signal are needed: the stderr line is a human-readable string a CLI
+ * upgrade may reword, and {@code error_during_execution} also reports failures no retry can fix.
+ */
+ private void throwIfTheResumeSessionWasLost(AgentStreamOptions options,
+ ClaudeCodeStreamParser parser, SpawnResult spawn) {
+ if (!ResumeRecovery.shouldRetryWithoutResume(StringUtils.hasText(options.getResumeSessionId()),
+ spawn.exitCode(), parser.resultSubtype(), spawn.stderr())) {
+ return;
+ }
+ throw new LlmGatewayException(502, ResumeRecovery.RESUME_LOST_CODE,
+ "the session conversation resumed no longer exists: " + describeStderr(spawn.stderr()),
+ "The turn is retried once without --resume; the earlier turns' context is lost.");
+ }
+
/**
* Streams completion tokens as plain text.
*
diff --git a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeStreamParser.java b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeStreamParser.java
index 552bb9da4..fa40f9bf6 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeStreamParser.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeStreamParser.java
@@ -221,6 +221,9 @@ final class ClaudeCodeStreamParser {
/** True once the terminal {@code result} frame arrived. */
private boolean resultFrameSeen;
+ /** Subtype of that frame, resolved by {@link #resolveSubtype}; null while none has arrived. */
+ private String resultSubtype;
+
/** Session id from the most recent frame that carried one; the init frame is the first. */
private String runtimeSessionId;
@@ -288,6 +291,15 @@ final class ClaudeCodeStreamParser {
return resultFrameSeen;
}
+ /**
+ * The subtype of that frame, or null when none arrived. One half of the lost-resume signal: the
+ * caller pairs it with the exit code and the stderr, because a subtype alone cannot tell a
+ * vanished {@code --resume} session from a failure no retry can fix.
+ */
+ String resultSubtype() {
+ return resultSubtype;
+ }
+
/**
* Every frame of a run carries {@code session_id}, including {@code system/init} and
* {@code stream_event}, so the id is known before anything interesting happens.
@@ -558,10 +570,11 @@ final class ClaudeCodeStreamParser {
}
}
resultFrameSeen = true;
+ resultSubtype = resolveSubtype(node);
JsonNode usage = node.path("usage");
events.add(new AgentEvent.ResultMeta(runtimeSessionId, longOrNull(node, "duration_ms"),
intOrNull(usage, "input_tokens"), intOrNull(usage, "output_tokens"),
- resolveSubtype(node)));
+ resultSubtype));
return events;
}
diff --git a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationRepository.java b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationRepository.java
index 1fa3dd477..98580081f 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationRepository.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/AiConversationRepository.java
@@ -45,6 +45,12 @@ public interface AiConversationRepository {
void update(RmqAiConversation conversation);
+ /**
+ * Forgets the remembered {@code --resume} session id. A method of its own because {@link #update}
+ * writes non-null fields only, so it cannot express "forget this value".
+ */
+ void clearRuntimeSessionId(Long id);
+
int deleteById(Long id);
/** Ids created strictly before the cutoff, oldest first, capped at limit. */
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 e2d0c4d86..756d51895 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
@@ -27,6 +27,7 @@ import org.apache.rocketmq.studio.ops.ai.LlmGatewayException;
import org.apache.rocketmq.studio.ops.ai.OpenAiCompatibleLlmClient;
import org.apache.rocketmq.studio.ops.ai.conversation.agent.AgentStreamOptions;
import org.apache.rocketmq.studio.ops.ai.conversation.agent.PromptEnhancer;
+import org.apache.rocketmq.studio.ops.ai.conversation.agent.ResumeRecovery;
import org.apache.rocketmq.studio.ops.ai.conversation.agent.RmqctlWorkspace;
import org.apache.rocketmq.studio.ops.ai.conversation.event.AgentEvent;
import org.apache.rocketmq.studio.ops.ai.conversation.event.AgentEventProjector;
@@ -275,7 +276,7 @@ public class AiRunExecutor {
Outcome outcome = new Outcome();
try {
markRunning(context);
- stream(context, outcome);
+ streamWithLostResumeRetry(context, outcome);
} catch (LlmGatewayException exception) {
log.warn("agent run {} failed: {} - {}", run.getId(), exception.getCode(), exception.getMessage());
outcome.gatewayFailure = exception;
@@ -304,20 +305,57 @@ public class AiRunExecutor {
run.getId(), run.getConversationId(), run.getTurn(), run.getEngine());
}
- private void stream(RunContext context, Outcome outcome) {
- String prompt = context.getPrompt();
- if (context.isEnhance()) {
- prompt = promptEnhancer.enhance(context.getConfig(), context.getEngine(), prompt,
- chunk -> onAgentEvent(context, outcome,
- new AgentEvent.ThinkingDelta(chunk, ThinkingSource.ENHANCE)));
+ /**
+ * Streams the turn, and repairs the one provider failure a retry can repair: the {@code --resume}
+ * session the conversation remembers no longer exists on disk. The provider reports it as
+ * {@link ResumeRecovery#RESUME_LOST_CODE}, because only the provider knows the retry has to drop
+ * {@code --resume} from the command.
+ *
+ * <p>Exactly one retry, and the dead id is forgotten <em>before</em> it: were it kept, the retry
+ * would hit the same wall and so would every turn after it. What the conversation loses is the
+ * earlier turns' context, which is the price {@link ResumeRecovery} documents.
+ */
+ private void streamWithLostResumeRetry(RunContext context, Outcome outcome) {
+ // Prepared once, outside the retry: re-running the enhancer would pay for the rewrite twice
+ // and put its reasoning in the timeline a second time.
+ String prompt = preparePrompt(context, outcome);
+ boolean resumeRequested = StringUtils.hasText(context.getResumeSessionId());
+ try {
+ stream(context, prompt, outcome, resumeRequested);
+ } catch (LlmGatewayException exception) {
+ if (!resumeRequested || !ResumeRecovery.RESUME_LOST_CODE.equals(exception.getCode())) {
+ throw exception;
+ }
+ log.warn("agent run {} could not resume the session conversation {} remembers; retrying"
+ + " without --resume",
+ context.getRun().getId(), context.getConversation().getId());
+ forgetRuntimeSession(context);
+ outcome.forgetFirstAttempt();
+ context.getSink().emit(new AgentEvent.ProviderNotice(AgentEventProjector.LEVEL_WARN,
+ "The agent session this conversation was resuming no longer exists; the turn was"
+ + " retried without the earlier turns' context."));
+ stream(context, prompt, outcome, false);
}
+ }
+
+ /** The prompt as the provider receives it, after the optional rewrite. */
+ private String preparePrompt(RunContext context, Outcome outcome) {
+ if (!context.isEnhance()) {
+ return context.getPrompt();
+ }
+ return promptEnhancer.enhance(context.getConfig(), context.getEngine(), context.getPrompt(),
+ chunk -> onAgentEvent(context, outcome,
+ new AgentEvent.ThinkingDelta(chunk, ThinkingSource.ENHANCE)));
+ }
+
+ private void stream(RunContext context, String prompt, Outcome outcome, boolean allowResume) {
Consumer<AgentEvent> events = event -> onAgentEvent(context, outcome, event);
if (isHttpEngine(context.getEngine())) {
streamHttp(context, prompt, events);
return;
}
agentProviders.forEngine(context.getEngine())
- .streamEvents(context.getConfig(), options(context, prompt), events);
+ .streamEvents(context.getConfig(), options(context, prompt, allowResume), events);
}
/**
@@ -337,12 +375,13 @@ public class AiRunExecutor {
token -> events.accept(new AgentEvent.TextDelta(token)));
}
- private AgentStreamOptions options(RunContext context, String prompt) {
+ private AgentStreamOptions options(RunContext context, String prompt, boolean allowResume) {
RmqctlWorkspace.Preparation preparation = context.getPreparation();
AgentStreamOptions.AgentStreamOptionsBuilder builder = AgentStreamOptions.builder()
.prompt(prompt)
.model(context.getRun().getModel())
- .resumeSessionId(context.getResumeSessionId())
+ // Dropped for the retry: that is the whole repair.
+ .resumeSessionId(allowResume ? context.getResumeSessionId() : null)
.instanceId(context.getConversation().getInstanceId())
// Registered so a stop kills the real process tree instead of relying on an interrupt.
.processSink(context.getHandle())
@@ -397,7 +436,11 @@ public class AiRunExecutor {
return;
}
if (event instanceof AgentEvent.ResultMeta meta) {
- if (StringUtils.hasText(meta.runtimeSessionId())) {
+ // Only from a successful frame: a failed one echoes the *requested* session id back, so
+ // persisting it would point the next turn at a session that does not exist. The init
+ // frame's id, taken above, is a real one and survives a later failure.
+ if (AgentEventProjector.SUCCESS_SUBTYPE.equals(meta.subtype())
+ && StringUtils.hasText(meta.runtimeSessionId())) {
outcome.runtimeSessionId = meta.runtimeSessionId().trim();
}
if (meta.inputTokens() != null) {
@@ -528,6 +571,24 @@ public class AiRunExecutor {
}
}
+ /**
+ * Drops the {@code --resume} id the conversation remembers, so the retry — and every turn after it
+ * — starts a fresh CLI session instead of failing on the same dead one again.
+ */
+ private void forgetRuntimeSession(RunContext context) {
+ RmqAiConversation conversation = context.getConversation();
+ if (!StringUtils.hasText(conversation.getRuntimeSessionId())) {
+ return;
+ }
+ try {
+ conversationRepository.clearRuntimeSessionId(conversation.getId());
+ conversation.setRuntimeSessionId(null);
+ } catch (RuntimeException exception) {
+ log.warn("could not forget the runtime session id of conversation {}: {}",
+ conversation.getId(), exception.toString());
+ }
+ }
+
private long durationMs(RmqAiRun run, LocalDateTime finishedAt) {
LocalDateTime startedAt = run.getStartedAt() != null ? run.getStartedAt() : run.getGmtCreate();
if (startedAt == null) {
@@ -664,6 +725,21 @@ public class AiRunExecutor {
private boolean stopRacedSuccess;
private LlmGatewayException gatewayFailure;
private RuntimeException unexpected;
+
+ /**
+ * Drops everything the attempt that asked for the missing session reported, so the retry's own
+ * frames decide the terminal state. Its session id in particular must not survive: it is the
+ * dead one, echoed back.
+ */
+ void forgetFirstAttempt() {
+ runtimeSessionId = null;
+ subtype = null;
+ inputTokens = null;
+ outputTokens = null;
+ providerDurationMs = null;
+ successTerminalProjected = false;
+ stopRacedSuccess = false;
+ }
}
/** The terminal state to write, and whether the projector already wrote one. */
diff --git a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/MybatisPlusAiConversationRepository.java b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/MybatisPlusAiConversationRepository.java
index be88197d6..3448dcf46 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/MybatisPlusAiConversationRepository.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/MybatisPlusAiConversationRepository.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.studio.ops.ai.conversation;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.studio.common.domain.PageResult;
@@ -79,6 +80,17 @@ public class MybatisPlusAiConversationRepository implements AiConversationReposi
conversationMapper.updateById(conversation);
}
+ @Override
+ public void clearRuntimeSessionId(Long id) {
+ if (id == null) {
+ return;
+ }
+ // updateById skips null fields, so "forget this value" has to name the column itself.
+ conversationMapper.update(null, new UpdateWrapper<RmqAiConversation>()
+ .eq("id", id)
+ .set("runtime_session_id", null));
+ }
+
@Override
public int deleteById(Long id) {
return id == null ? 0 : conversationMapper.deleteById(id);
diff --git a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/ResumeRecovery.java b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/ResumeRecovery.java
index 7481f916e..d2a3d0704 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/ResumeRecovery.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/conversation/agent/ResumeRecovery.java
@@ -46,9 +46,21 @@ import org.springframework.util.StringUtils;
* </ol>
* Losing the earlier turns' context is the price; without the retry the conversation is permanently
* broken, because every subsequent turn would resume the same missing id.
+ *
+ * <p>The signal is detected where the exit code and the stderr are still in scope — the provider that
+ * spawned the CLI — and travels to the caller as {@link #RESUME_LOST_CODE} on a
+ * {@code LlmGatewayException}, because the frames alone cannot express it: the command that has to
+ * drop {@code --resume} is the provider's to build.
*/
public final class ResumeRecovery {
+ /**
+ * The error code a provider reports when a run could not resume the session its conversation
+ * remembers. The caller that sees it owns the recovery: retry the turn once without
+ * {@code --resume}, after clearing {@code conversation.runtime_session_id}.
+ */
+ public static final String RESUME_LOST_CODE = "llm.provider.resume_lost";
+
/** The CLI's stderr line, matched as a prefix of the first non-blank content. */
static final String SESSION_NOT_FOUND_PREFIX = "No conversation found with session ID:";
diff --git a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeAgentProviderTest.java b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeAgentProviderTest.java
index be0755bd8..826eeaf44 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeAgentProviderTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/ClaudeCodeAgentProviderTest.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.studio.ops.ai;
import org.apache.rocketmq.studio.ops.ai.conversation.agent.AgentStreamOptions;
+import org.apache.rocketmq.studio.ops.ai.conversation.agent.ResumeRecovery;
import org.apache.rocketmq.studio.ops.ai.conversation.event.AgentEvent;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -353,6 +354,46 @@ class ClaudeCodeAgentProviderTest {
new AgentEvent.ResultMeta("s-1", 12L, 3, 4, "error_max_turns"));
}
+ @Test
+ void streamEventsShouldReportALostResumeSessionTest() {
+ // The measured shape of a stale --resume: exit 1, the session-not-found line on stderr, and a
+ // result frame whose subtype is error_during_execution with the dead id echoed back.
+ StreamingTestProvider provider = new StreamingTestProvider(
+ List.of("sh", "-c", "printf '%s' "
+ + "'{\"type\":\"result\",\"subtype\":\"error_during_execution\",\"is_error\":true,"
+ + "\"num_turns\":0,\"session_id\":\"gone-session\"}'; "
+ + "echo 'No conversation found with session ID: gone-session' >&2; exit 1"), 30);
+
+ // The caller can only retry correctly if it knows the retry has to drop --resume, and the
+ // command is built here. See ResumeRecovery.
+ assertThatThrownBy(() -> provider.streamEvents(LlmConfigVO.builder().build(),
+ AgentStreamOptions.builder().prompt("hi").model("qwen3.8-max")
+ .resumeSessionId("gone-session").build(), event -> { }))
+ .isInstanceOfSatisfying(LlmGatewayException.class, exception -> {
+ assertThat(exception.getStatusCode()).isEqualTo(502);
+ assertThat(exception.getCode()).isEqualTo(ResumeRecovery.RESUME_LOST_CODE);
+ assertThat(exception.getMessage()).contains("gone-session");
+ });
+ }
+
+ @Test
+ void streamEventsShouldNotReportALostResumeWhenTheRunNeverResumedTest() {
+ StreamingTestProvider provider = new StreamingTestProvider(
+ List.of("sh", "-c", "printf '%s' "
+ + "'{\"type\":\"result\",\"subtype\":\"error_during_execution\",\"is_error\":true,"
+ + "\"num_turns\":0,\"session_id\":\"s-1\"}'; "
+ + "echo 'No conversation found with session ID: s-1' >&2; exit 1"), 30);
+ List<AgentEvent> events = new ArrayList<>();
+
+ // Nothing was resumed, so the signal is not the one a retry repairs: the frames explain the
+ // failure and retrying the same command would only repeat it.
+ provider.streamEvents(LlmConfigVO.builder().build(),
+ AgentStreamOptions.builder().prompt("hi").model("qwen3.8-max").build(), events::add);
+
+ assertThat(events).containsExactly(
+ new AgentEvent.ResultMeta("s-1", null, null, null, "error_during_execution"));
+ }
+
private static int count(List<AgentEvent> events, Class<?> type) {
return (int) events.stream().filter(type::isInstance).count();
}
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 2c3fec1f7..31e1f87c0 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
@@ -22,6 +22,7 @@ import org.apache.rocketmq.studio.ops.ai.LlmConfigVO;
import org.apache.rocketmq.studio.ops.ai.LlmGatewayException;
import org.apache.rocketmq.studio.ops.ai.OpenAiCompatibleLlmClient;
import org.apache.rocketmq.studio.ops.ai.conversation.agent.PromptEnhancer;
+import org.apache.rocketmq.studio.ops.ai.conversation.agent.ResumeRecovery;
import org.apache.rocketmq.studio.ops.ai.conversation.event.AgentEvent;
import org.apache.rocketmq.studio.ops.ai.conversation.event.AgentEventProjector;
import org.apache.rocketmq.studio.ops.ai.conversation.event.RunStatus;
@@ -272,6 +273,81 @@ class AiRunExecutorTest {
assertThat(runRow().getErrorCode()).isEqualTo("llm.provider.error_max_turns");
}
+ @Test
+ void aFailedResultFrameShouldNotPersistTheSessionIdItEchoesBackTest() {
+ // A failed frame carries the session id that was *requested*, not a live one: the parser says so
+ // in as many words. Persisting it would point the next turn at a session that does not exist.
+ provider.emit(new AgentEvent.ResultMeta("echoed-session", 12L, 1, 2, "error_max_turns"));
+
+ startAndRun();
+
+ assertThat(runRow().getStatus()).isEqualTo(RunStatus.FAILED.name());
+ assertThat(runRow().getRuntimeSessionId()).isNull();
+ assertThat(conversation.getRuntimeSessionId()).isNull();
+ }
+
+ @Test
+ void aLostResumeSessionShouldBeForgottenAndTheTurnRetriedWithoutResumeTest() {
+ List<String> resumedSessions = new CopyOnWriteArrayList<>();
+ conversation.setRuntimeSessionId("gone-session");
+ // The measured shape of a stale --resume: exit 1, one error_during_execution result frame with
+ // the dead id echoed back, and the stderr line naming the session.
+ provider.emit(new AgentEvent.ResultMeta("gone-session", null, null, null, "error_during_execution"));
+ provider.failure = new LlmGatewayException(502, ResumeRecovery.RESUME_LOST_CODE,
+ "the session conversation resumed no longer exists", "Retried without --resume.");
+ provider.beforeStream = options -> {
+ resumedSessions.add(options.getResumeSessionId());
+ if (provider.calls > 1) {
+ provider.scripted.clear();
+ provider.failure = null;
+ provider.emit(new AgentEvent.TextDelta("fresh answer"));
+ provider.emit(new AgentEvent.ResultMeta("new-session", 30L, 5, 6,
+ AgentEventProjector.SUCCESS_SUBTYPE));
+ }
+ };
+
+ startAndRun("gone-session");
+
+ // The repair is a second attempt at the same turn, and it is the whole repair: no --resume on it.
+ assertThat(provider.calls).isEqualTo(2);
+ assertThat(resumedSessions).containsExactly("gone-session", null);
+ assertThat(provider.lastPrompt).isEqualTo("hello");
+ verify(conversationRepository).clearRuntimeSessionId(CONVERSATION_ID);
+ // The retry's session replaces the dead one, and the user is told the context is gone.
+ assertThat(runRow().getStatus()).isEqualTo(RunStatus.COMPLETED.name());
+ assertThat(runRow().getRuntimeSessionId()).isEqualTo("new-session");
+ assertThat(conversation.getRuntimeSessionId()).isEqualTo("new-session");
+ assertThat(types()).containsExactly("user", "error", "notice", "text", "run_status");
+ // Exactly one terminal reached the wire: the retry's, not one per attempt.
+ assertThat(emitters.get(0).eventCount("\"type\":\"run_finished\"")).isEqualTo(1);
+ }
+
+ @Test
+ void aRetryThatFailsTooShouldStillLeaveTheDeadSessionIdForgottenTest() {
+ conversation.setRuntimeSessionId("gone-session");
+ provider.emit(new AgentEvent.ResultMeta("gone-session", null, null, null, "error_during_execution"));
+ provider.failure = new LlmGatewayException(502, ResumeRecovery.RESUME_LOST_CODE,
+ "the session conversation resumed no longer exists", "Retried without --resume.");
+ provider.beforeStream = options -> {
+ if (provider.calls > 1) {
+ provider.scripted.clear();
+ provider.failure = new LlmGatewayException(504, "llm.provider.timeout",
+ "claude CLI stream timed out after 300s", "Retry with a shorter prompt.");
+ provider.emit(new AgentEvent.TextDelta("partial"));
+ }
+ };
+
+ startAndRun("gone-session");
+
+ // A retry that fails must not put the dead id back: every turn after it would otherwise resume a
+ // session that is still gone. Null is the honest state, and the next turn starts fresh.
+ assertThat(provider.calls).isEqualTo(2);
+ verify(conversationRepository).clearRuntimeSessionId(CONVERSATION_ID);
+ assertThat(runRow().getStatus()).isEqualTo(RunStatus.FAILED.name());
+ assertThat(runRow().getRuntimeSessionId()).isNull();
+ assertThat(conversation.getRuntimeSessionId()).isNull();
+ }
+
@Test
void anUnexpectedProviderFailureShouldNotLeaveTheRunNonTerminalTest() {
provider.failure = new IllegalStateException("provider exploded");
@@ -469,11 +545,19 @@ class AiRunExecutorTest {
executor.submit(context(ENGINE, enhance));
}
+ private void startAndRun(String resumeSessionId) {
+ executor.submit(context(ENGINE, false, resumeSessionId));
+ }
+
private AiRunExecutor.RunContext context(String engine) {
return context(engine, false);
}
private AiRunExecutor.RunContext context(String engine, boolean enhance) {
+ return context(engine, enhance, null);
+ }
+
+ private AiRunExecutor.RunContext context(String engine, boolean enhance, String resumeSessionId) {
AgentRunHandle handle = executor.newHandle(RUN_ID);
AiEventSink sink = executor.newSink(CONVERSATION_ID, RUN_ID, 1, 0);
session = executor.newSession(RUN_ID, executor.streamTimeoutMillis(engine));
@@ -483,7 +567,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", null, enhance, Duration.ofSeconds(300), null);
+ engine, "hello", resumeSessionId, enhance, Duration.ofSeconds(300), null);
}
private List<String> types() {
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 8a82cd186..b305858e4 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
@@ -18,6 +18,7 @@ package org.apache.rocketmq.studio.ops.ai.conversation;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -49,6 +50,7 @@ import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -475,6 +477,31 @@ class AiTimelineRepositoryTest {
verify(conversationMapper, never()).selectObjs(any());
}
+ /**
+ * Forgetting the remembered {@code --resume} session is the recovery for a stale one, so this write
+ * has to actually null the column. {@code updateById} skips null fields, which is why the port has a
+ * method of its own — a fake pass through the ordinary update would be a silent no-op that leaves the
+ * conversation pointing at a session that no longer exists.
+ */
+ @Test
+ void forgettingTheRuntimeSessionShouldNullTheColumnByNameTest() {
+ conversations.clearRuntimeSessionId(CONVERSATION_ID);
+
+ UpdateWrapper<RmqAiConversation> update = capturedConversationUpdate();
+ assertThat(update.getSqlSet()).contains("runtime_session_id");
+ // The bound value is what makes it a clear rather than a no-op.
+ assertThat(update.getParamNameValuePairs()).containsValue(null);
+ assertThat(update.getTargetSql()).contains("id = ?");
+ assertThat(update.getParamNameValuePairs().values()).contains(CONVERSATION_ID);
+ }
+
+ @Test
+ void forgettingTheRuntimeSessionShouldNeedAnIdTest() {
+ conversations.clearRuntimeSessionId(null);
+
+ verify(conversationMapper, never()).update(any(), any());
+ }
+
// --- the run rows -----------------------------------------------------------------
/**
@@ -714,6 +741,13 @@ class AiTimelineRepositoryTest {
return queryOf(captor.getValue());
}
+ @SuppressWarnings("unchecked")
+ private UpdateWrapper<RmqAiConversation> capturedConversationUpdate() {
+ ArgumentCaptor<Wrapper<RmqAiConversation>> captor = wrapperCaptor();
+ verify(conversationMapper).update(isNull(), captor.capture());
+ return (UpdateWrapper<RmqAiConversation>) captor.getValue();
+ }
+
private QueryWrapper<RmqAiRun> capturedRunSelectList() {
ArgumentCaptor<Wrapper<RmqAiRun>> captor = wrapperCaptor();
verify(runMapper).selectList(captor.capture());