From d074b3a5aec5954565a5ffcd76dee7aab6cadaaa Mon Sep 17 00:00:00 2001 From: xy-ygz <779323365@qq.com> Date: Thu, 10 Sep 2026 22:19:09 +0800 Subject: [PATCH] fix(example): fix/data-agent-multi-session-inbox-routing --- .../dataagent/web/api/ChatController.java | 70 +++++++-- ...ChatControllerConversationRoutingTest.java | 134 ++++++++++++++++++ 2 files changed, 190 insertions(+), 14 deletions(-) create mode 100644 agentscope-examples/agents/agentscope-dataagent/src/test/java/io/agentscope/dataagent/web/api/ChatControllerConversationRoutingTest.java diff --git a/agentscope-examples/agents/agentscope-dataagent/src/main/java/io/agentscope/dataagent/web/api/ChatController.java b/agentscope-examples/agents/agentscope-dataagent/src/main/java/io/agentscope/dataagent/web/api/ChatController.java index c697c1fcfa..f1b2330f68 100644 --- a/agentscope-examples/agents/agentscope-dataagent/src/main/java/io/agentscope/dataagent/web/api/ChatController.java +++ b/agentscope-examples/agents/agentscope-dataagent/src/main/java/io/agentscope/dataagent/web/api/ChatController.java @@ -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; @@ -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; 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. + * + *

{@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 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 payload = messages != null ? List.copyOf(messages) : List.of(); + InboundMessage.Builder builder = + InboundMessage.builder( + ChatUiChannel.CHANNEL_ID, Peer.thread(conversationId), payload) + .senderId(userId) + .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} @@ -537,28 +563,44 @@ static List shapeInboundMessages( * *

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. + * + *

{@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 executeChat( String userId, String agentId, String message, String conversationId) { long startMs = System.currentTimeMillis(); + if (agentId != null && !agentId.isBlank()) { + HarnessAgent ha = catalogService.getRunningAgent(userId, agentId); + 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 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 { 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 call = chatUiChannel.dispatch(inbound); diff --git a/agentscope-examples/agents/agentscope-dataagent/src/test/java/io/agentscope/dataagent/web/api/ChatControllerConversationRoutingTest.java b/agentscope-examples/agents/agentscope-dataagent/src/test/java/io/agentscope/dataagent/web/api/ChatControllerConversationRoutingTest.java new file mode 100644 index 0000000000..1ad6968872 --- /dev/null +++ b/agentscope-examples/agents/agentscope-dataagent/src/test/java/io/agentscope/dataagent/web/api/ChatControllerConversationRoutingTest.java @@ -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( + 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); + } +}