huangting/th-hotel-simple
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修正SuperAgent任务结果来源消息定位

Browse Source
This commit is contained in:
andy
2026-07-12 14:53:54 +08:00
parent bf796946f9
commit eee37315a0
18 changed files with 789 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
server/src/main/java/cn/nianxx/thhotel/integrations/mcp/superagent/service/impl/SuperAgentMcpServiceImpl.java
View File

@@ -358,8 +358,8 @@ public class SuperAgentMcpServiceImpl implements SuperAgentMcpService {
private Map<String, Object> submitTaskResultsSchema() {
Map<String, Object> propertiesMap = new LinkedHashMap<>();
propertiesMap.put("hotel_id", stringField("可选酒店上下文 ID；缺省由 TH Hotel 后端解析系统酒店"));
propertiesMap.put("source_provider", nullableStringField("来源提供方，默认 AGENTBUS"));
propertiesMap.put("source_channel", nullableStringField("来源渠道，默认 EMAIL"));
propertiesMap.put("source_provider", nullableStringField("兼容字段；写入工具通常不需要传，后端写入定位不使用该字段"));
propertiesMap.put("source_channel", nullableStringField("兼容字段；写入工具通常不需要传，后端写入定位不使用该字段，AgentBus 实际入库渠道可能是 OUTLOOK"));
propertiesMap.put("source_message_id", stringField("外部来源消息 ID，对应 AgentBus source.external_message_id"));
propertiesMap.put("ai_task_results", Map.of(
"type", "array",

18
server/src/main/java/cn/nianxx/thhotel/platform/message/repository/MybatisSourceMessageInboxRepository.java
View File

@@ -157,6 +157,24 @@ public class MybatisSourceMessageInboxRepository implements SourceMessageInboxRe
return Optional.ofNullable(entity).map(this::toSnapshot);
}
/**
* 根据酒店和外部邮件 ID 查询所有候选 Inbox 记录，调用方负责处理零条或多条的业务语义。
*/
@Override
public List<SourceMessageInboxSnapshot> findByExternalMessageId(String hotelId, String externalMessageId) {
if (!hasText(hotelId) || !hasText(externalMessageId)) {
return List.of();
}
return inboxMapper.selectList(Wrappers.<SourceMessageInboxEntity>lambdaQuery()
.eq(SourceMessageInboxEntity::getHotelId, trim(hotelId))
.eq(SourceMessageInboxEntity::getExternalMessageId, trim(externalMessageId))
.orderByDesc(SourceMessageInboxEntity::getReceivedAt)
.orderByDesc(SourceMessageInboxEntity::getId))
.stream()
.map(this::toSnapshot)
.toList();
}
/**
* 执行 SourceMessage 安全摘要分页查询，只读取列表和详情需要的索引字段。
*/

5
server/src/main/java/cn/nianxx/thhotel/platform/message/repository/SourceMessageInboxRepository.java
View File

@@ -50,6 +50,11 @@ public interface SourceMessageInboxRepository {
String channel,
String externalMessageId);
/**
* 按酒店和外部邮件 ID 查询 Inbox 快照，用于外部回调只携带 AgentBus external_message_id 的场景。
*/
List<SourceMessageInboxSnapshot> findByExternalMessageId(String hotelId, String externalMessageId);
/**
* 按查询条件分页读取 Inbox 快照，分页参数已经由 Service 层归一化。
*/

212
server/src/main/java/cn/nianxx/thhotel/workflows/reservation/service/impl/ReservationAiTaskIntakeServiceImpl.java
View File

@@ -39,6 +39,7 @@ import java.util.HexFormat;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
@@ -151,17 +152,21 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
private final ObjectMapper objectMapper;
private final SourceMessageInboxRepository sourceMessageInboxRepository;
private final ReservationAiWorkflowRepository workflowRepository;
private final boolean allowLegacyInternalSourceMessageId;
/**
* 注入 JSON 解析器、SourceMessage 查询边界和 Reservation 持久化边界。
* 注入 JSON 解析器、SourceMessage 查询边界、Reservation 持久化边界和本地旧夹具兼容开关。
*/
public ReservationAiTaskIntakeServiceImpl(
ObjectMapper objectMapper,
SourceMessageInboxRepository sourceMessageInboxRepository,
ReservationAiWorkflowRepository workflowRepository) {
ReservationAiWorkflowRepository workflowRepository,
@Value("${superagent.task-result.allow-legacy-internal-source-message-id:false}")
boolean allowLegacyInternalSourceMessageId) {
this.objectMapper = objectMapper;
this.sourceMessageInboxRepository = sourceMessageInboxRepository;
this.workflowRepository = workflowRepository;
this.allowLegacyInternalSourceMessageId = allowLegacyInternalSourceMessageId;
}
/**
@@ -370,7 +375,8 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
List<SuperAgentTaskResultItemResponse> responseItems = new ArrayList<>(itemCount);
Set<String> acceptedParentSplitCodes = new LinkedHashSet<>();
for (int index = 0; index < messageEvents.size(); index++) {
JsonNode event = messageEvents.get(index);
JsonNode rawEvent = messageEvents.get(index);
JsonNode event = normalizeV3ParentSplitEvent(rawEvent);
int arrayIndex = index + 1;
V3EventContractIssue contractIssue = inspectV3EventContractIssue(event, messageEvents);
if (contractIssue == null) {
@@ -381,9 +387,9 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
hotelId,
sourceMessageId,
batchId,
event,
rawEvent,
arrayIndex,
sourceEventIndex(event, arrayIndex),
sourceEventIndex(rawEvent, arrayIndex),
null,
contractIssue.code(),
contractIssue.message(),
@@ -396,16 +402,16 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
hotelId,
sourceMessageId,
batchId,
event,
rawEvent,
arrayIndex,
sourceEventIndex(event, arrayIndex),
sourceEventIndex(rawEvent, arrayIndex),
null,
"EVENT_ROUTE_UNSUPPORTED",
"message_events[" + index + "] 无法派生稳定路由。",
now));
continue;
}
ObjectNode item = normalizeV3EventToTaskResultItem(event, route, arrayIndex);
ObjectNode item = normalizeV3EventToTaskResultItem(rawEvent, event, route, arrayIndex);
responseItems.add(createItem(hotelId, sourceMessageId, batchId, item, arrayIndex, now));
}
for (int index = 0; index < unhandledIntents.size(); index++) {
@@ -636,31 +642,32 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
/**
* 将 V3 业务 event 转为旧任务结果 item 形态，复用现有订单、任务和任务卡创建逻辑。
* 将 V3 业务 event 转为旧任务结果 item 形态，业务列使用 adapter 派生 event，原始 event 保留在 v3_message_event。
*/
private ObjectNode normalizeV3EventToTaskResultItem(
JsonNode event,
JsonNode rawEvent,
JsonNode derivedEvent,
ReservationAiRouteDefinition route,
int arrayIndex) {
ObjectNode item = objectMapper.createObjectNode();
item.put("source_event_index", sourceEventIndex(event, arrayIndex));
item.put("source_event_index", sourceEventIndex(rawEvent, arrayIndex));
item.put("catalog_code", V3_CATALOG_CODE);
item.put("skill_id", V3_SKILL_ID);
item.put("result_type", route.resultType());
item.put("task_type", route.taskType());
item.put("task_subtype", route.taskSubtype());
copyIfPresent(event, item, "current_or_history");
copyIfPresent(event, item, "relevant_message_excerpt");
copyIfPresent(event, item, "case_keys");
copyIfPresent(event, item, "extracted_fields");
copyIfPresent(event, item, "manual_review");
copyIfPresent(event, item, "attachments");
copyIfPresent(event, item, "file_references");
copyIfPresent(event, item, "context_used");
copyIfPresent(event, item, "parent_source_event_index");
copyIfPresent(event, item, "linked_task_group_id");
copyIfPresent(event, item, "blocked_until_parent_completed");
item.set("v3_message_event", event);
copyIfPresent(rawEvent, item, "current_or_history");
copyIfPresent(rawEvent, item, "relevant_message_excerpt");
copyIfPresent(derivedEvent, item, "case_keys");
copyIfPresent(rawEvent, item, "extracted_fields");
copyIfPresent(rawEvent, item, "manual_review");
copyIfPresent(rawEvent, item, "attachments");
copyIfPresent(rawEvent, item, "file_references");
copyIfPresent(rawEvent, item, "context_used");
copyIfPresent(rawEvent, item, "parent_source_event_index");
copyIfPresent(rawEvent, item, "linked_task_group_id");
copyIfPresent(rawEvent, item, "blocked_until_parent_completed");
item.set("v3_message_event", rawEvent);
item.put("route_code", route.routeCode());
item.put("system_process_category", route.systemProcessCategory().name());
return item;
@@ -804,6 +811,36 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
return "linked_parent_release_after_child_split".equals(firstText(extractedFields, event, "relationship_type"));
}
/**
* 归一 Parent split 的 Parent Group key；只补齐 adapter 派生用副本，不修改请求原文。
*/
private JsonNode normalizeV3ParentSplitEvent(JsonNode event) {
if (event == null || !event.isObject() || !isLinkedParentReleaseCandidate(event)) {
return event;
}
ObjectNode normalizedEvent = event.deepCopy();
JsonNode caseKeysNode = normalizedEvent.path("case_keys");
if (!caseKeysNode.isObject()) {
return normalizedEvent;
}
ObjectNode caseKeys = (ObjectNode) caseKeysNode;
String groupCode = trimToNull(textAt(caseKeys, "group_code"));
String blockCode = trimToNull(textAt(caseKeys, "block_code"));
if (groupCode == null && blockCode != null) {
caseKeys.put("group_code", blockCode);
return normalizedEvent;
}
if (groupCode != null && blockCode == null) {
caseKeys.put("block_code", groupCode);
return normalizedEvent;
}
if (groupCode != null && blockCode != null && !groupCode.equals(blockCode)) {
caseKeys.putNull("group_code");
caseKeys.putNull("block_code");
}
return normalizedEvent;
}
/**
* 判断当前 producer 是否仍在输出旧 Parent Cancel Booking 组合。
*/
@@ -824,8 +861,8 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
if (!isLinkedParentReleaseCandidate(event)) {
return null;
}
String parentGroupCode = trimToNull(textAt(event.path("case_keys"), "group_code"));
if (parentGroupCode != null && !acceptedParentSplitCodes.add(parentGroupCode)) {
ParentSplitIdentity identity = parentSplitIdentity(event);
if (identity.normalizedParentCode() != null && !acceptedParentSplitCodes.add(identity.normalizedParentCode())) {
return new V3EventContractIssue(
"LINKED_PARENT_RELEASE_DUPLICATE_PARENT",
"同一个 Parent split cluster 只能有一个 Parent Cancel Allotment 候选。");
@@ -838,25 +875,25 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
*/
private boolean validLinkedParentReleaseCandidate(JsonNode event, JsonNode messageEvents) {
JsonNode extractedFields = event == null ? null : event.path("extracted_fields");
JsonNode caseKeys = event == null ? null : event.path("case_keys");
String parentGroupCode = trimToNull(textAt(caseKeys, "group_code"));
String parentBlockCode = trimToNull(textAt(caseKeys, "block_code"));
ParentSplitIdentity identity = parentSplitIdentity(event);
if (!("Cancel Allotment".equals(trimToNull(textAt(event, "event_type")))
&& "entire_allotment_control_block".equals(firstText(extractedFields, event, "cancel_scope"))
&& isBooleanTrue(extractedFields.path("parent_release_or_cancel_candidate"))
&& "parent_to_child_allocation_split".equals(firstText(extractedFields, event, "release_reason"))
&& parentGroupCode != null
&& parentGroupCode.equals(firstText(extractedFields, event, "parent_group_code"))
&& isBooleanTrue(extractedFields.path("allocation_split_from_parent"))
&& "New Booking".equals(trimToNull(textAt(event, "related_event_type")))
&& event != null
&& isBooleanTrue(event.path("requires_downstream_hard_validation"))
&& parentGroupCode != null
&& parentGroupCode.equals(parentBlockCode)
&& nonEmptyArray(extractedFields, "child_group_codes")
&& nonEmptyArray(event, "related_source_event_indices"))) {
return false;
}
if (!identity.targetObjectUnclearReview() && !validConcreteParentIdentity(identity, extractedFields)) {
return false;
}
if (identity.targetObjectUnclearReview() && !nonEmptyArray(event.path("context_used"), "parent_identity_candidates")) {
return false;
}
List<String> childGroupCodes = stringArrayValues(extractedFields.path("child_group_codes"));
List<String> relatedSourceEventIndices = stringArrayValues(event.path("related_source_event_indices"));
if (childGroupCodes.isEmpty()
@@ -888,6 +925,30 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
return true;
}
/**
* 校验普通 Parent split 事件是否已经具备明确 Parent Group identity。
*/
private boolean validConcreteParentIdentity(ParentSplitIdentity identity, JsonNode extractedFields) {
return identity.normalizedParentCode() != null
&& identity.normalizedParentCode().equals(identity.groupCode())
&& identity.normalizedParentCode().equals(identity.blockCode())
&& identity.normalizedParentCode().equals(trimToNull(textAt(extractedFields, "parent_group_code")));
}
/**
* 提取 Parent split identity，封装正常归一与 target_object_unclear 复核两种形态。
*/
private ParentSplitIdentity parentSplitIdentity(JsonNode event) {
JsonNode caseKeys = event == null ? null : event.path("case_keys");
String groupCode = trimToNull(textAt(caseKeys, "group_code"));
String blockCode = trimToNull(textAt(caseKeys, "block_code"));
String normalizedParentCode = groupCode != null && groupCode.equals(blockCode) ? groupCode : null;
boolean targetObjectUnclearReview = groupCode == null
&& blockCode == null
&& "target_object_unclear".equals(textAt(event == null ? null : event.path("manual_review"), "reason_code"));
return new ParentSplitIdentity(groupCode, blockCode, normalizedParentCode, targetObjectUnclearReview);
}
/**
* 接收 S000/S999 文本入口结果，创建只读来源消息任务，并隐藏其技术订单。
*/
@@ -914,13 +975,7 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
? requireText(defaultHotelId, "default_hotel_id", LENGTH_64)
: requireText(resolvedSourceMessage.hotelId(), "source_message.hotel_id", LENGTH_64);
SourceMessageInboxSnapshot sourceMessage = resolvedSourceMessage == null
? sourceMessageInboxRepository
.findByIdempotencyKey(
hotelId,
DEFAULT_SOURCE_PROVIDER,
DEFAULT_SOURCE_CHANNEL,
entryResult.externalSourceMessageId())
.orElseThrow(() -> error(HttpStatus.NOT_FOUND, "SOURCE_MESSAGE_NOT_FOUND", "SourceMessage 不存在。"))
? resolveExternalSourceMessageByHotel(hotelId, entryResult.externalSourceMessageId())
: resolvedSourceMessage;
String requestPayloadSha256 = sha256(rawBody == null ? "" : rawBody);
String batchIdempotencyKey = sha256(BATCH_KEY_PREFIX + "|" + sourceMessage.id() + "|" + requestPayloadSha256);
@@ -1575,22 +1630,21 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
if (requestHotelId != null && systemHotelId != null && !requestHotelId.equals(systemHotelId)) {
throw error(HttpStatus.BAD_REQUEST, "HOTEL_ID_MISMATCH", "请求 hotel_id 与系统酒店不一致。");
}
if (requestHotelId == null && isLongText(sourceMessageReference)) {
return resolveLegacyInternalSourceMessage(sourceMessageReference, systemHotelId);
}
String hotelId = requestHotelId == null ? systemHotelId : requestHotelId;
if (hotelId == null) {
return resolveLegacyInternalSourceMessage(sourceMessageReference, null);
if (allowLegacyInternalSourceMessageId) {
return resolveLegacyInternalSourceMessage(sourceMessageReference, null);
}
throw error(HttpStatus.BAD_REQUEST, "HOTEL_ID_REQUIRED", "使用外部 source_message_id 时 hotel_id 不能为空。");
}
validateLength(hotelId, "hotel_id", LENGTH_64);
String sourceProvider = optionalText(textAt(root, "source_provider"), "source_provider", LENGTH_32);
String sourceChannel = optionalText(textAt(root, "source_channel"), "source_channel", LENGTH_32);
String provider = sourceProvider == null ? DEFAULT_SOURCE_PROVIDER : sourceProvider;
String channel = sourceChannel == null ? DEFAULT_SOURCE_CHANNEL : sourceChannel;
SourceMessageInboxSnapshot sourceMessage = sourceMessageInboxRepository
.findByIdempotencyKey(hotelId, provider, channel, sourceMessageReference)
.orElseThrow(() -> error(HttpStatus.NOT_FOUND, "SOURCE_MESSAGE_NOT_FOUND", "SourceMessage 不存在。"));
return new ResolvedSourceMessage(sourceMessage, sourceMessageReference);
// 旧 V2 字段只做格式兼容校验；写入定位统一按系统酒店 + external_message_id 匹配真实 Inbox。
optionalText(textAt(root, "source_provider"), "source_provider", LENGTH_32);
optionalText(textAt(root, "source_channel"), "source_channel", LENGTH_32);
return resolveExternalSourceMessageReferenceByHotel(
hotelId,
sourceMessageReference,
requestHotelId == null && allowLegacyInternalSourceMessageId);
}
/**
@@ -1614,6 +1668,42 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
return new ResolvedSourceMessage(sourceMessage, responseSourceMessageId);
}
/**
* 按系统酒店和 AgentBus 外部邮件 ID 解析真实入库 SourceMessage，不要求 SuperAgent 了解数据库 provider/channel。
*/
private SourceMessageInboxSnapshot resolveExternalSourceMessageByHotel(String hotelId, String externalSourceMessageId) {
List<SourceMessageInboxSnapshot> candidates =
sourceMessageInboxRepository.findByExternalMessageId(hotelId, externalSourceMessageId);
if (candidates.isEmpty()) {
throw error(HttpStatus.NOT_FOUND, "SOURCE_MESSAGE_NOT_FOUND", "SourceMessage 不存在。");
}
if (candidates.size() > 1) {
throw error(HttpStatus.CONFLICT, "SOURCE_MESSAGE_AMBIGUOUS", "同一酒店下存在多个匹配的 SourceMessage。");
}
return candidates.get(0);
}
/**
* V2 写入优先按外部消息 ID 解析；仅在无显式 hotel_id 的旧本地夹具场景下兜底内部 SourceMessage ID。
*/
private ResolvedSourceMessage resolveExternalSourceMessageReferenceByHotel(
String hotelId,
String sourceMessageReference,
boolean allowLegacyInternalFallback) {
List<SourceMessageInboxSnapshot> candidates =
sourceMessageInboxRepository.findByExternalMessageId(hotelId, sourceMessageReference);
if (candidates.isEmpty()) {
if (allowLegacyInternalFallback && isLongText(sourceMessageReference)) {
return resolveLegacyInternalSourceMessage(sourceMessageReference, hotelId);
}
throw error(HttpStatus.NOT_FOUND, "SOURCE_MESSAGE_NOT_FOUND", "SourceMessage 不存在。");
}
if (candidates.size() > 1) {
throw error(HttpStatus.CONFLICT, "SOURCE_MESSAGE_AMBIGUOUS", "同一酒店下存在多个匹配的 SourceMessage。");
}
return new ResolvedSourceMessage(candidates.get(0), sourceMessageReference);
}
/**
* 判断来源消息引用是否为历史本地兼容的内部 SourceMessage ID。
*/
@@ -1823,13 +1913,8 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
}
validateLength(externalSourceMessageId, "source_message.source_message_id", LENGTH_256);
String hotelId = requireText(defaultHotelId, "default_hotel_id", LENGTH_64);
SourceMessageInboxSnapshot sourceMessage = sourceMessageInboxRepository
.findByIdempotencyKey(
hotelId,
DEFAULT_SOURCE_PROVIDER,
DEFAULT_SOURCE_CHANNEL,
externalSourceMessageId)
.orElseThrow(() -> error(HttpStatus.NOT_FOUND, "SOURCE_MESSAGE_NOT_FOUND", "SourceMessage 不存在。"));
SourceMessageInboxSnapshot sourceMessage =
resolveExternalSourceMessageByHotel(hotelId, externalSourceMessageId);
return new ResolvedSourceMessage(sourceMessage, externalSourceMessageId);
}
@@ -2221,6 +2306,17 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
) {
}
/**
* Parent split 父事件身份归一结果。target_object_unclear 代表已知卡型但目标对象需人工确认。
*/
private record ParentSplitIdentity(
String groupCode,
String blockCode,
String normalizedParentCode,
boolean targetObjectUnclearReview
) {
}
/**
* 将空白字符串转换为 null。
*/