Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import io.agentscope.dataagent.web.share.AgentAclService.Tier;
import io.agentscope.dataagent.web.toolbus.ToolEventBus;
import io.agentscope.dataagent.web.usage.UsageStore;
import io.agentscope.harness.agent.HarnessAgent;
import io.agentscope.harness.agent.gateway.MsgContext;
import io.agentscope.harness.agent.gateway.channel.InboundMessage;
import io.agentscope.harness.agent.gateway.channel.Peer;
Expand Down Expand Up @@ -364,19 +365,44 @@ private static String normalizedConversationId(String key) {
*/
private String resolveGateKey(String userId, String agentId, String conversationId) {
if (agentId == null || agentId.isBlank()) return null;
if (conversationId == null || conversationId.isBlank()) return null;

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Warning] This new guard makes resolveGateKey return null for a blank conversation id, but executeChat (line 581) still dispatches for a blank id by falling back to the pre-thread DM shape. The two sides now disagree about which session a request belongs to.

Consequence on that path: the message is dispatched and a session is created with a |r:<user>-style gate key, but nothing can resolve that key any more — /reset answers "No active session to reset.", the currentSession probe reports exists=false, and recordRunSession's dedupeKey is null so the RUN_SESSION activity event is silently dropped.

Before this change the two sides agreed, because resolveGateKey built a Peer.direct + accountId(conversationId) probe and produced exactly the same DM-shaped key that InboundMessage.dm(...) does.

The HTTP entry points mint a UUID (lines 164-166, 277-279), so I could not confirm a reachable caller that arrives here with a blank id — but the comment on the fallback branch says it exists for back-compat, which implies it is expected to be live. Please pick one:

  1. mirror the fallback here — build the DM probe when the conversation id is blank instead of returning null; or
  2. if a blank id really cannot reach executeChat, delete the branch at line 581 and make the invariant explicit, so the two paths cannot drift again.

ChatControllerConversationRoutingTest.probeAndDispatchInboundShareCanonicalKey asserts probe/dispatch agreement only for a non-blank conv-a; a blank-id case would pin whichever option you choose.

try {
String gatewayAgentId = catalogService.resolveGatewayAgentId(userId, agentId);
InboundMessage probe =
InboundMessage.builder(ChatUiChannel.CHANNEL_ID, Peer.direct(userId), List.of())
.preferredAgentId(gatewayAgentId)
.accountId(conversationId)
.build();
buildConversationInbound(userId, gatewayAgentId, conversationId, List.of());
return chatUiChannel.previewRoute(probe).context().canonicalKey();
} catch (Exception e) {
return null;
}
}

/**
* Builds an inbound message whose routing key carries {@code conversationId} in the {@code |t:}
* segment via a thread peer. {@code senderId} must be set because thread peers are not DM peers
* and the router would otherwise lose the authenticated user id.
*
* <p>{@code conversationId} must be non-blank — HTTP stream/send mint a UUID before dispatch;
* callers must not fall back to a DM-shaped key.
*/
static InboundMessage buildConversationInbound(
String userId, String gatewayAgentId, String conversationId, List<Msg> messages) {
Objects.requireNonNull(userId, "userId");
if (conversationId == null || conversationId.isBlank()) {
throw new IllegalArgumentException(
"conversationId is required; stream/send must mint a UUID before dispatch");
}
List<Msg> payload = messages != null ? List.copyOf(messages) : List.of();
InboundMessage.Builder builder =
InboundMessage.builder(
ChatUiChannel.CHANNEL_ID, Peer.thread(conversationId), payload)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Peer.thread(conversationId) + senderId(userId) + parentPeer(Peer.direct(userId)) is the kind of triple that breaks quietly if a router rule later keys off parentPeer. Routing both branches through buildConversationInbound is the right shape, but the test only asserts the key format, not that the dispatched inbound in executeChat yields the same canonicalKey as the resolveGateKey probe. One assertion tying probe and dispatch together would prevent exactly the drift that caused this bug.

.senderId(userId)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Info] Setting senderId here because thread peers are not DM peers is a genuinely subtle routing constraint, and the javadoc explains it well.

It is only protected by the canonical-key assertions, though: if a future edit dropped .senderId(userId), gateKey would likely stay stable and these tests would keep passing while the gateway silently lost the authenticated user id on the thread path. An assertion on the built InboundMessage itself (getSenderId(), getParentPeer()) would pin the constraint the comment describes.

.parentPeer(Peer.direct(userId));
if (gatewayAgentId != null && !gatewayAgentId.isBlank()) {
builder.preferredAgentId(gatewayAgentId);
}
return builder.build();
}

/**
* Translates a gateway routing key into the real {@code SessionEntry.sessionKey()}, by
* scanning registered MAIN sessions for the matching {@code gateKey}. Returns {@code null}
Expand Down Expand Up @@ -537,28 +563,44 @@ static List<Msg> shapeInboundMessages(
*
* <p>When {@code agentId} is blank (defensive — controller always supplies one), falls back to
* pure binding-driven routing: the chatui channel's default agent or matching binding wins.
*
* <p>{@code conversationId} must already be pinned (stream/send mint a UUID when the client
* omits {@code sessionKey}). A blank id is rejected so probe ({@link #resolveGateKey}) and
* dispatch never diverge on session identity.
*/
private Mono<Msg> executeChat(
String userId, String agentId, String message, String conversationId) {
long startMs = System.currentTimeMillis();

if (agentId != null && !agentId.isBlank()) {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The new ha.getModel() == null pre-check hardcodes provider-specific guidance (DASHSCOPE_API_KEY, dataagent.dashscope.api-key) into a generic controller, while HarnessAgent is model-agnostic across the OpenAI/Anthropic/Gemini/Ollama extensions — this message will mislead users on another provider. Consider a provider-neutral wording, and/or surfacing it where the model is actually resolved so all entry points benefit, not only executeChat.

HarnessAgent ha = catalogService.getRunningAgent(userId, agentId);

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Info] Good UX improvement turning the missing-model condition into an actionable error instead of a downstream failure. Minor: this is a check-then-act on catalogService.getRunningAgent — the agent can be stopped or reconfigured between here and chatUiChannel.dispatch, so the guard is best-effort only (dispatch still needs its own error path, which it has). Also consider emitting the same IllegalStateException text from the stream() SSE error frame so the frontend shows the hint rather than a generic stream error — quick check whether the existing error mapping already covers it.

if (ha != null && ha.getModel() == null) {
return Mono.error(
new IllegalStateException(
"No LLM model configured for agent '"
+ agentId
+ "'. Provide a Model Spring bean or configure model"
+ " options under dataagent.* in application.yml."));
}
}

String pinnedConversationId = normalizedConversationId(conversationId);
if (pinnedConversationId == null) {
return Mono.error(
new IllegalArgumentException(
"conversationId is required; stream/send must mint a UUID before"
+ " executeChat"));
}

List<Msg> msgs =
shapeInboundMessages(userBindings.list(userId), ChatUiChannel.CHANNEL_ID, message);

InboundMessage inbound;
if (agentId == null || agentId.isBlank()) {
// No agent override and no conversation scoping — pure binding-driven routing.
inbound = InboundMessage.dm(ChatUiChannel.CHANNEL_ID, userId, List.copyOf(msgs));
inbound = buildConversationInbound(userId, null, pinnedConversationId, msgs);
} else {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Warning] The dispatch shape change from Peer.direct(userId) + accountId(conversationId) to Peer.thread(conversationId) + parentPeer(direct) changes the canonical gateKey (|t: segment replaces the DM-shaped key). For the example's own new-session flow this is exactly the fix (distinct conversations no longer collapse), but for a deployed DataAgent with pre-existing MAIN sessions keyed under the old DM-shaped gateKey, the first message after upgrade will resolve to a new gateKey — history/listen-by-gate lookups (findSessionKeyByGate, /reset, tool-event bus subscription at the first turn) will miss the old session until it ages out. If that migration blind-spot matters for users running the example against persisted state, consider either (a) a fallback lookup on the legacy key shape during resolveGateKey, or (b) an explicit note in the PR description / release note that existing chat sessions start fresh after upgrade.

String gatewayAgentId = catalogService.resolveGatewayAgentId(userId, agentId);
inbound =
InboundMessage.builder(
ChatUiChannel.CHANNEL_ID,
Peer.direct(userId),
List.copyOf(msgs))
.preferredAgentId(gatewayAgentId)
.accountId(conversationId)
.build();
inbound = buildConversationInbound(userId, gatewayAgentId, pinnedConversationId, msgs);
}
Mono<Msg> call = chatUiChannel.dispatch(inbound);

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,134 @@
/*
* Copyright 2024-2026 the original author or authors.
*
* Licensed 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 io.agentscope.dataagent.web.api;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import io.agentscope.core.message.Msg;
import io.agentscope.core.message.MsgRole;
import io.agentscope.harness.agent.gateway.channel.ChannelConfig;
import io.agentscope.harness.agent.gateway.channel.DmScope;
import io.agentscope.harness.agent.gateway.channel.InboundMessage;
import io.agentscope.harness.agent.gateway.channel.Peer;
import io.agentscope.harness.agent.gateway.channel.chatui.ChatUiChannel;
import java.util.List;
import org.junit.jupiter.api.Test;

/** Verifies conversation-scoped inbound routing produces distinct gate keys per conversation. */
class ChatControllerConversationRoutingTest {

private static final ChatUiChannel CHANNEL =
ChatUiChannel.create(

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Solid focused test. Two additions would lock the fix down: (1) blank/null conversationId on the no-agent path — asserting the resulting key documents the migration decision above; (2) a regression that two turns with the same conversationId map to the same key while a different one does not, which is the user-visible invariant.

ChannelConfig.builder("chatui")
.defaultAgentId("data-agent")
.dmScope(DmScope.MAIN)
.build());

private static String gateKey(InboundMessage inbound) {
return CHANNEL.previewRoute(inbound).context().canonicalKey();
}

@Test
void differentConversationIdsProduceDistinctGateKeysWithThreadSegment() {
InboundMessage a =
ChatController.buildConversationInbound(
"user-1", "data-agent", "conv-a", List.of());
InboundMessage b =
ChatController.buildConversationInbound(
"user-1", "data-agent", "conv-b", List.of());

String gateKeyA = gateKey(a);
String gateKeyB = gateKey(b);

assertThat(gateKeyA).isNotEqualTo(gateKeyB);
assertThat(gateKeyA).contains("|t:conv-a");
assertThat(gateKeyB).contains("|t:conv-b");
assertThat(SessionController.extractConversationId(gateKeyA)).isEqualTo("conv-a");
assertThat(SessionController.extractConversationId(gateKeyB)).isEqualTo("conv-b");
}

@Test
void sameConversationIdMapsToSameGateKey() {
InboundMessage first =
ChatController.buildConversationInbound(
"user-1", "data-agent", "conv-a", List.of());
InboundMessage second =
ChatController.buildConversationInbound(
"user-1", "data-agent", "conv-a", List.of());

assertThat(gateKey(first)).isEqualTo(gateKey(second));
}

@Test
void probeAndDispatchInboundShareCanonicalKey() {
InboundMessage probe =
ChatController.buildConversationInbound(
"user-1", "data-agent", "conv-a", List.of());
InboundMessage dispatch =
ChatController.buildConversationInbound(
"user-1",
"data-agent",
"conv-a",
List.of(Msg.builder().role(MsgRole.USER).textContent("hello").build()));

assertThat(gateKey(probe)).isEqualTo(gateKey(dispatch));
}

@Test
void blankConversationIdIsRejectedByBuildConversationInbound() {
assertThatThrownBy(
() ->
ChatController.buildConversationInbound(
"user-1", "data-agent", null, List.of()))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("conversationId");
assertThatThrownBy(
() ->
ChatController.buildConversationInbound(
"user-1", "data-agent", " ", List.of()))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("conversationId");
}

@Test
void threadInboundCarriesSenderIdAndParentPeer() {
InboundMessage inbound =
ChatController.buildConversationInbound(
"user-1", "data-agent", "conv-a", List.of());

assertThat(inbound.senderId()).isEqualTo("user-1");
assertThat(inbound.parentPeer()).isEqualTo(Peer.direct("user-1"));
assertThat(inbound.peer()).isEqualTo(Peer.thread("conv-a"));
}

@Test
void sameUserDifferentConversationsShareRoomButNotThread() {
InboundMessage a =
ChatController.buildConversationInbound(
"user-1", "data-agent", "conv-a", List.of());
InboundMessage b =
ChatController.buildConversationInbound(
"user-1", "data-agent", "conv-b", List.of());

String gateKeyA = gateKey(a);
String gateKeyB = gateKey(b);

assertThat(gateKeyA).contains("|r:user-1");
assertThat(gateKeyB).contains("|r:user-1");
assertThat(gateKeyA).isNotEqualTo(gateKeyB);
}
}
Loading