停止V4入站双写旧任务

This commit is contained in:
andy
2026-07-20 17:18:28 +07:00
parent d45ff5b4ed
commit 626e87abdb
16 changed files with 438 additions and 69 deletions

View File

@@ -11,7 +11,7 @@ public record ReservationV4AcceptedEventDraft(
Integer sourceEventIndex,
/** 当前 event 对应的 AI transition ID。 */
Long aiTransitionId,
/** 旧过渡链路已经解析出的本地订单投影 IDV4 第一版复用该绑定。 */
/** 入站阶段解析出的本地订单投影 IDV4 订单任务复用该绑定。 */
Long orderId,
/** 目标订单业务类型,例如 GROUP、FIT。 */
String targetBookingType,

View File

@@ -8,16 +8,16 @@ import com.fasterxml.jackson.annotation.JsonProperty;
* @param sourceEventIndex AI 原始事件序号
* @param arrayIndex 当前批次中的处理顺序
* @param aiTransitionId AI 过渡记录 ID
* @param routeCode M002 V3 稳定路由码S10/S99 直接使用外部路由码
* @param routeCode M002 V3 / V4 稳定路由码S10/S99 直接使用外部路由码
* @param systemProcessCategory 系统处理分类,用于区分业务任务、来源邮件通知和契约错误
* @param adapterErrorCode Adapter 契约错误代码,正常创建任务时为空
* @param orderId 关联订单 ID仅落契约错误时为空
* @param taskId 关联任务 ID仅落契约错误为空
* @param taskId 任务 IDV4 普通业务和契约错误场景为空
* @param systemTaskType 系统主任务类型
* @param taskCardType 任务卡类型
* @param taskStatus 任务状态;仅落契约错误为空
* @param taskStatus 任务状态;V4 普通业务和契约错误场景为空
* @param orderStatus 订单状态;仅落契约错误时为空
* @param executionOrder 同订单任务执行顺序;仅落契约错误为空
* @param executionOrder 同订单任务执行顺序;V4 普通业务和契约错误场景为空
*/
public record SuperAgentTaskResultItemResponse(
@JsonProperty("source_event_index")

View File

@@ -184,7 +184,7 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
}
/**
* 接收已通过 HMAC 鉴权的 SuperAgent 任务结果。先处理旧文本结果,再分流 V3 JSON 和 V2 JSON。
* 接收已通过 HMAC 鉴权的 SuperAgent 任务结果。先处理旧文本结果,再按 V4 / V3 / V2 JSON 契约分流
*/
@Override
@Transactional
@@ -540,7 +540,7 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
}
/**
* 接收 V4 业务根。第一版按 message_events[] 顺序派生现有任务卡event 契约错误只落 transition
* 接收 V4 业务根。V4 新主线只落 AI transition 和 V4 订单任务 / 多卡模型,不再双写旧任务表
*/
private SuperAgentTaskResultResponse acceptV4BusinessRoot(
JsonNode root,
@@ -640,9 +640,22 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
continue;
}
ObjectNode item = v4CallbackRouter.normalizeEventToTaskResultItem(root, event, route, arrayIndex);
SuperAgentTaskResultItemResponse responseItem = createItem(hotelId, sourceMessageId, batchId, item, arrayIndex, now);
responseItems.add(responseItem);
acceptedEvents.add(toV4AcceptedEvent(event, route, responseItem, arrayIndex));
V4BusinessTransitionCreation transitionCreation = createV4BusinessTransition(
hotelId,
sourceMessageId,
batchId,
item,
event,
route,
arrayIndex,
now);
responseItems.add(transitionCreation.responseItem());
acceptedEvents.add(toV4AcceptedEvent(
event,
route,
transitionCreation.transitionId(),
transitionCreation.orderId(),
arrayIndex));
}
v4TaskIntakeService.createBusinessPackageTasks(sourceMessage, batchId, root, acceptedEvents, now);
@@ -1097,19 +1110,20 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
}
/**
* 将已创建 transition 的 V4 event 转成新模型写入草稿,避免 V4 多卡服务反向依赖旧任务响应细节
* 将已创建 transition 的 V4 event 转成新模型写入草稿,避免 V4 多卡服务反向依赖旧任务模型
*/
private ReservationV4AcceptedEventDraft toV4AcceptedEvent(
JsonNode event,
ReservationAiRouteDefinition route,
SuperAgentTaskResultItemResponse responseItem,
Long aiTransitionId,
Long orderId,
int arrayIndex) {
JsonNode targetOrder = event.path("target_order");
return new ReservationV4AcceptedEventDraft(
requireText(textAt(event, "order_ref"), "order_ref", LENGTH_128),
arrayIndex,
longValue(responseItem.aiTransitionId()),
longValue(responseItem.orderId()),
aiTransitionId,
orderId,
optionalText(textAt(targetOrder, "booking_type"), "target_order.booking_type", LENGTH_32),
optionalText(textAt(targetOrder, "locator_type"), "target_order.locator_type", LENGTH_64),
optionalText(textAt(targetOrder, "locator_value"), "target_order.locator_value", LENGTH_128),
@@ -1120,6 +1134,140 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
);
}
/**
* 创建 V4 业务 event 的 AI transition并解析本地订单投影不创建旧 workflow_reservation_task。
*/
private V4BusinessTransitionCreation createV4BusinessTransition(
String hotelId,
Long sourceMessageId,
Long batchId,
JsonNode item,
JsonNode event,
ReservationAiRouteDefinition route,
int arrayIndex,
LocalDateTime now) {
Integer sourceEventIndex = requirePositiveInt(item, "source_event_index");
String catalogCode = requireText(textAt(item, "catalog_code"), "catalog_code", LENGTH_32);
String skillId = requireText(textAt(item, "skill_id"), "skill_id", LENGTH_128);
String resultType = requireText(textAt(item, "result_type"), "result_type", LENGTH_64);
String aiTaskType = requireText(textAt(item, "task_type"), "task_type", LENGTH_64);
String taskSubtype = optionalText(textAt(item, "task_subtype"), "task_subtype", LENGTH_128);
String routeCode = optionalText(textAt(item, "route_code"), "route_code", LENGTH_64);
String systemProcessCategory = optionalText(
textAt(item, "system_process_category"),
"system_process_category",
LENGTH_64);
if (systemProcessCategory == null) {
systemProcessCategory = route.systemProcessCategory().name();
}
JsonNode caseKeys = item.path("case_keys");
String currentOrHistory = optionalText(textAt(item, "current_or_history"), "current_or_history", LENGTH_32);
String groupCode = optionalText(textAt(caseKeys, "group_code"), "group_code", LENGTH_128);
String confirmationNumber = optionalText(textAt(caseKeys, "confirmation_number"), "confirmation_number", LENGTH_128);
String manualReasonCode = optionalText(textAt(item.path("manual_review"), "reason_code"), "manual_review.reason_code", LENGTH_128);
String itemPayloadJson = nodeJson(item);
String itemPayloadSha256 = sha256(itemPayloadJson);
String itemIdempotencyKey = sha256(ITEM_KEY_PREFIX
+ "|" + sourceMessageId
+ "|" + sourceEventIndex
+ "|" + arrayIndex
+ "|" + catalogCode
+ "|" + skillId
+ "|" + resultType
+ "|" + aiTaskType
+ "|" + nullToEmpty(taskSubtype)
+ "|" + itemPayloadSha256);
Long transitionId = workflowRepository.insertTransition(new ReservationAiTransitionDraft(
hotelId,
batchId,
sourceMessageId,
sourceEventIndex,
arrayIndex,
arrayIndex,
catalogCode,
skillId,
resultType,
aiTaskType,
routeCode,
systemProcessCategory,
route.systemTaskType().name(),
route.taskCardType().name(),
taskSubtype,
currentOrHistory,
groupCode,
confirmationNumber,
itemPayloadSha256,
itemIdempotencyKey,
manualReasonCode,
null,
null,
null,
itemPayloadJson,
nodeJson(item.get("case_keys")),
nodeJson(item.get("extracted_fields")),
nodeJson(item.get("manual_review")),
null,
nodeJson(item.get("attachments")),
null,
null,
null,
now
));
ReservationOrderSnapshot order = resolveV4LocalOrder(hotelId, sourceMessageId, event, arrayIndex, now);
SuperAgentTaskResultItemResponse responseItem = new SuperAgentTaskResultItemResponse(
sourceEventIndex,
arrayIndex,
transitionId.toString(),
routeCode,
systemProcessCategory,
null,
order == null ? null : order.id().toString(),
null,
route.systemTaskType().name(),
route.taskCardType().name(),
null,
order == null ? null : order.orderStatus(),
null
);
return new V4BusinessTransitionCreation(
transitionId,
order == null ? null : order.id(),
responseItem);
}
/**
* V4 业务包只在当前能可靠定位 Group Code / Confirmation Number 时绑定本地订单。
*/
private ReservationOrderSnapshot resolveV4LocalOrder(
String hotelId,
Long sourceMessageId,
JsonNode event,
int arrayIndex,
LocalDateTime now) {
JsonNode targetOrder = event == null ? null : event.path("target_order");
String locatorType = trimToNull(textAt(targetOrder, "locator_type"));
String locatorValue = optionalText(textAt(targetOrder, "locator_value"), "target_order.locator_value", LENGTH_128);
BusinessKey businessKey = selectV4BusinessKey(locatorType, locatorValue);
if (businessKey == null) {
return null;
}
return workflowRepository.findActiveOrderByBusinessKey(hotelId, businessKey.keyType().name(), businessKey.value())
.orElseGet(() -> createActiveOrder(hotelId, sourceMessageId, businessKey, arrayIndex, now));
}
/**
* V4 第一版只把 Group Code 和 Confirmation Number 作为可自动落订单的业务号。
*/
private BusinessKey selectV4BusinessKey(String locatorType, String locatorValue) {
if ("GROUP_CODE".equals(locatorType) && hasText(locatorValue)) {
return new BusinessKey(ReservationOrderKeyType.GROUP_CODE, locatorValue);
}
if ("CONFIRMATION_NUMBER".equals(locatorType) && hasText(locatorValue)) {
return new BusinessKey(ReservationOrderKeyType.CONFIRMATION_NUMBER, locatorValue);
}
return null;
}
/**
* 生成 V4 业务卡 AI 原始片段,保留 event 原文和系统派生路由,不复制整封邮件正文。
*/
@@ -1623,7 +1771,7 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
true,
existingBatch.itemCount() == null ? 0 : existingBatch.itemCount(),
List.of(),
List.of(new SuperAgentTaskResultWarning("IDEMPOTENT_REPLAY", "相同请求已经处理,本次未重复创建任务"))
List.of(new SuperAgentTaskResultWarning("IDEMPOTENT_REPLAY", "相同请求已经处理,本次未重复创建业务对象"))
);
}
@@ -3011,6 +3159,16 @@ public class ReservationAiTaskIntakeServiceImpl implements ReservationAiTaskInta
) {
}
/**
* V4 业务 event 的 transition 写入结果V4 新主线不再携带旧 task ID。
*/
private record V4BusinessTransitionCreation(
Long transitionId,
Long orderId,
SuperAgentTaskResultItemResponse responseItem
) {
}
/**
* 任务创建结果,包含任务 ID 和最终写入的同订单执行序号。
*/

View File

@@ -7,7 +7,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
/**
* V4 SuperAgent 回调路由器。负责把 V4 event 映射到当前系统稳定 route 和旧任务创建链路 item。
* V4 SuperAgent 回调路由器。负责把 V4 event 映射到当前系统稳定 route 和入站 transition item。
*/
final class ReservationAiV4CallbackRouter {
@@ -41,7 +41,7 @@ final class ReservationAiV4CallbackRouter {
}
/**
* 将 V4 event 归一为当前任务创建链路可消费的 item并保留 V4 原始结构。
* 将 V4 event 归一为当前入站 transition 可消费的 item并保留 V4 原始结构。
*/
ObjectNode normalizeEventToTaskResultItem(
JsonNode root,
@@ -70,7 +70,7 @@ final class ReservationAiV4CallbackRouter {
}
/**
* 将 V4 target_order 映射为旧任务链路可识别的 case_keys同时保留 Booking Code。
* 将 V4 target_order 映射为 transition 查询摘要可识别的 case_keys同时保留 Booking Code。
*/
ObjectNode buildCaseKeys(JsonNode targetOrder) {
ObjectNode caseKeys = objectMapper.createObjectNode();

View File

@@ -502,7 +502,7 @@ public class ReservationDemoDataServiceImpl implements ReservationDemoDataServic
}
/**
* 调用 AI 入站服务创建订单、任务和任务卡
* 调用历史 AI 入站服务创建 V2/V3 演示订单、任务和任务卡V4 smoke 不使用该入口造数
*/
private SuperAgentTaskResultResponse intake(String rawBody, String demoRunId, String scenarioCode) {
SuperAgentTaskResultResponse response = intakeService.accept(

View File

@@ -877,6 +877,8 @@ class ReservationFrontendQueryControllerTest {
.value(contains(0)))
.andExpect(jsonPath("$.items[?(@.order_id=='" + completedOnlyOrderId + "')].next_v4_order_task_id")
.value(contains(nullValue())))
.andExpect(jsonPath("$.items[?(@.order_id=='" + completedOnlyOrderId + "')].next_processable_task_id")
.value(contains(nullValue())))
.andExpect(jsonPath("$.items[?(@.order_id=='" + completedOnlyOrderId + "')].next_v4_action_type")
.value(contains("NONE")))
.andExpect(jsonPath("$.items[?(@.order_id=='" + completedOnlyOrderId + "')].next_v4_action_status")

View File

@@ -990,7 +990,7 @@ class SuperAgentTaskResultControllerTest {
}
@Test
void shouldCreateBusinessTasksFromV4OrderContextsAndMessageEventsInOrder() throws Exception {
void shouldCreateV4OrderTaskAndCardsWithoutLegacyTaskDualWrite() throws Exception {
SourceMessageCaptureResult source = captureSourceMessage("mail-v4-business-root-001");
String body = v4BusinessRootBody("mail-v4-business-root-001");
@@ -1003,13 +1003,17 @@ class SuperAgentTaskResultControllerTest {
.andExpect(jsonPath("$.items[0].route_code").value("R02_NEW_GROUP_BLOCK_NORMAL"))
.andExpect(jsonPath("$.items[0].system_task_type").value("NEW_BOOKING"))
.andExpect(jsonPath("$.items[0].task_card_type").value("NEW_BOOKING"))
.andExpect(jsonPath("$.items[0].execution_order").value(1))
.andExpect(jsonPath("$.items[0].task_id").doesNotExist())
.andExpect(jsonPath("$.items[0].task_status").doesNotExist())
.andExpect(jsonPath("$.items[0].execution_order").doesNotExist())
.andExpect(jsonPath("$.items[1].source_event_index").value(2))
.andExpect(jsonPath("$.items[1].array_index").value(2))
.andExpect(jsonPath("$.items[1].route_code").value("R10_PAYMENT_EVIDENCE_NORMAL"))
.andExpect(jsonPath("$.items[1].system_task_type").value("UPDATE_BOOKING"))
.andExpect(jsonPath("$.items[1].task_card_type").value("PAYMENT_EVIDENCE"))
.andExpect(jsonPath("$.items[1].execution_order").value(2))
.andExpect(jsonPath("$.items[1].task_id").doesNotExist())
.andExpect(jsonPath("$.items[1].task_status").doesNotExist())
.andExpect(jsonPath("$.items[1].execution_order").doesNotExist())
.andReturn();
String firstOrderId = com.jayway.jsonpath.JsonPath.read(
@@ -1069,6 +1073,12 @@ class SuperAgentTaskResultControllerTest {
)
""", Long.class, source.inboxId());
assertThat(v4CardStatusCount).isEqualTo(4L);
Long legacyTaskCount = jdbcTemplate.queryForObject("""
SELECT COUNT(*)
FROM workflow_reservation_task
WHERE source_message_id = ?
""", Long.class, source.inboxId());
assertThat(legacyTaskCount).isZero();
String sourceDisplayPayload = jdbcTemplate.queryForObject("""
SELECT display_payload_json
@@ -1115,8 +1125,8 @@ class SuperAgentTaskResultControllerTest {
mockMvc.perform(signedPost(body, "nonce-v4-two-order-refs-001"))
.andExpect(status().isCreated())
.andExpect(jsonPath("$.accepted_count").value(2))
.andExpect(jsonPath("$.items[0].execution_order").value(1))
.andExpect(jsonPath("$.items[1].execution_order").value(1));
.andExpect(jsonPath("$.items[0].execution_order").doesNotExist())
.andExpect(jsonPath("$.items[1].execution_order").doesNotExist());
List<String> orderRefs = jdbcTemplate.queryForList("""
SELECT order_ref
@@ -1148,7 +1158,8 @@ class SuperAgentTaskResultControllerTest {
mockMvc.perform(signedPost(body, "nonce-v4-fit-booking-code-001"))
.andExpect(status().isCreated())
.andExpect(jsonPath("$.accepted_count").value(1))
.andExpect(jsonPath("$.items[0].order_status").value("TEMPORARY"));
.andExpect(jsonPath("$.items[0].order_id").doesNotExist())
.andExpect(jsonPath("$.items[0].order_status").doesNotExist());
Long unresolvedCount = jdbcTemplate.queryForObject("""
SELECT COUNT(*)
@@ -1320,15 +1331,15 @@ class SuperAgentTaskResultControllerTest {
.andExpect(jsonPath("$.items[0].route_code").value("R06_CANCEL_GROUP_BLOCK_NORMAL"))
.andExpect(jsonPath("$.items[0].system_task_type").value("CANCEL_BOOKING"))
.andExpect(jsonPath("$.items[0].task_card_type").value("CANCEL_BOOKING"))
.andExpect(jsonPath("$.items[0].execution_order").value(1))
.andExpect(jsonPath("$.items[0].execution_order").doesNotExist())
.andExpect(jsonPath("$.items[1].route_code").value("R14_GENERAL_REQUEST_NORMAL"))
.andExpect(jsonPath("$.items[1].system_task_type").value("UPDATE_BOOKING"))
.andExpect(jsonPath("$.items[1].task_card_type").value("TRACE_RESERVATION_NOTES"))
.andExpect(jsonPath("$.items[1].execution_order").value(2))
.andExpect(jsonPath("$.items[1].execution_order").doesNotExist())
.andExpect(jsonPath("$.items[2].route_code").value("R11_ROOMING_LIST_NORMAL"))
.andExpect(jsonPath("$.items[2].system_task_type").value("UPDATE_BOOKING"))
.andExpect(jsonPath("$.items[2].task_card_type").value("ROOMING_LIST"))
.andExpect(jsonPath("$.items[2].execution_order").value(3))
.andExpect(jsonPath("$.items[2].execution_order").doesNotExist())
.andReturn();
String firstOrderId = com.jayway.jsonpath.JsonPath.read(
@@ -1554,8 +1565,8 @@ class SuperAgentTaskResultControllerTest {
.andExpect(jsonPath("$.items[0].adapter_error_code").value("PAYMENT_ATTACHMENT_ID_NOT_FOUND"))
.andExpect(jsonPath("$.items[0].task_id").doesNotExist())
.andExpect(jsonPath("$.items[1].route_code").value("R06_CANCEL_GROUP_BLOCK_NORMAL"))
.andExpect(jsonPath("$.items[1].task_id").exists())
.andExpect(jsonPath("$.items[1].execution_order").value(1));
.andExpect(jsonPath("$.items[1].task_id").doesNotExist())
.andExpect(jsonPath("$.items[1].execution_order").doesNotExist());
Long taskCount = jdbcTemplate.queryForObject("""
SELECT COUNT(*)
@@ -1569,7 +1580,7 @@ class SuperAgentTaskResultControllerTest {
AND system_process_category = 'ADAPTER_CONTRACT_ERROR'
AND adapter_error_code = 'PAYMENT_ATTACHMENT_ID_NOT_FOUND'
""", Long.class, source.inboxId());
assertThat(taskCount).isEqualTo(1L);
assertThat(taskCount).isZero();
assertThat(adapterErrorCount).isEqualTo(1L);
List<String> v4CardTypes = jdbcTemplate.queryForList("""