feat(booking): add safe retries and operational metrics

2026-08-08 16:25:17 +08:00
1 parent c826b6e574
commit 7330ac853b
@@ -0,0 +1,23 @@
+package cn.nianxx.thhotel.workflows.reservation.booking.common.dto;
+
+/** 项目 schema 内的聚合运行指标；不包含邮件、附件、团号或任何个人信息。 */
+public record BookingOperationalMetricsSnapshot(
+        int windowHours,
+        long totalRuns,
+        long failedRuns,
+        long agentFallbackRuns,
+        long riskRuns,
+        long backlogRuns,
+        long retriedRuns,
+        long retryAttempts) {
+
+    public BookingOperationalMetricsSnapshot {
+        if (windowHours <= 0) {
+            throw new IllegalArgumentException("windowHours 必须大于 0");
+        }
+        if (totalRuns < 0 || failedRuns < 0 || agentFallbackRuns < 0 || riskRuns < 0
+                || backlogRuns < 0 || retriedRuns < 0 || retryAttempts < 0) {
+            throw new IllegalArgumentException("Booking 运行指标不能为负数");
+        }
+    }
+}
@@ -0,0 +1,29 @@
+package cn.nianxx.thhotel.workflows.reservation.booking.common.result;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/** 面向系统管理员的安全 Booking 运行指标视图。 */
+public record BookingOperationalMetricsView(
+        @JsonProperty("window_hours")
+        int windowHours,
+        @JsonProperty("total_runs")
+        long totalRuns,
+        @JsonProperty("failed_runs")
+        long failedRuns,
+        @JsonProperty("failure_rate_percent")
+        double failureRatePercent,
+        @JsonProperty("agent_fallback_runs")
+        long agentFallbackRuns,
+        @JsonProperty("agent_fallback_rate_percent")
+        double agentFallbackRatePercent,
+        @JsonProperty("risk_runs")
+        long riskRuns,
+        @JsonProperty("risk_rate_percent")
+        double riskRatePercent,
+        @JsonProperty("backlog_runs")
+        long backlogRuns,
+        @JsonProperty("retried_runs")
+        long retriedRuns,
+        @JsonProperty("retry_attempts")
+        long retryAttempts) {
+}
@@ -0,0 +1,9 @@
+package cn.nianxx.thhotel.workflows.reservation.booking.repository;
+
+import cn.nianxx.thhotel.workflows.reservation.booking.common.dto.BookingOperationalMetricsSnapshot;
+
+/** 只读聚合项目 schema 的运行指标。 */
+public interface BookingOperationalMetricsRepository {
+
+    BookingOperationalMetricsSnapshot snapshot(int windowHours);
+}
@@ -0,0 +1,59 @@
+package cn.nianxx.thhotel.workflows.reservation.booking.repository;
+
+import cn.nianxx.thhotel.workflows.reservation.booking.common.dto.BookingOperationalMetricsSnapshot;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Repository;
+import org.springframework.transaction.annotation.Transactional;
+
+/** PostgreSQL 项目 schema 的聚合监控查询；不会扫描旧 MySQL 或返回单条邮件数据。 */
+@Repository
+@ConditionalOnProperty(prefix = "booking.postgres", name = "enabled", havingValue = "true")
+public class BookingPostgresOperationalMetricsRepository implements BookingOperationalMetricsRepository {
+
+    private static final String SNAPSHOT_SQL = "SELECT "
+            + "COUNT(*) AS total_runs, "
+            + "COUNT(*) FILTER (WHERE run.run_status = 'FAILED') AS failed_runs, "
+            + "COUNT(*) FILTER (WHERE run.result_disposition = 'RISK_NOTIFICATION') AS risk_runs, "
+            + "COUNT(*) FILTER (WHERE run.run_status IN ("
+            + "'RECEIVED', 'MATERIAL_READY', 'PARSER_COMPLETE', 'PARSER_NOT_APPLICABLE', "
+            + "'PARSER_FAILED', 'CONTEXT_READY', 'AGENT_COMPLETE')) AS backlog_runs, "
+            + "COUNT(*) FILTER (WHERE run.retry_count > 0) AS retried_runs, "
+            + "COALESCE(SUM(run.retry_count), 0) AS retry_attempts, "
+            + "COUNT(*) FILTER (WHERE EXISTS ("
+            + "SELECT 1 FROM th_hotel_booking.booking_contract_artifact artifact "
+            + "WHERE artifact.processing_run_id = run.id "
+            + "AND artifact.artifact_type = 'CANDIDATE_DECISION' "
+            + "AND artifact.payload ->> 'decision_origin' IN ('AGENT', 'MIXED')"
+            + ")) AS agent_fallback_runs "
+            + "FROM th_hotel_booking.booking_contract_processing_run run "
+            + "WHERE run.started_at >= CURRENT_TIMESTAMP - (? * INTERVAL '1 hour')";
+
+    private final JdbcTemplate jdbcTemplate;
+
+    public BookingPostgresOperationalMetricsRepository(
+            @Qualifier("bookingPostgresJdbcTemplate") JdbcTemplate jdbcTemplate) {
+        this.jdbcTemplate = jdbcTemplate;
+    }
+
+    @Override
+    @Transactional(readOnly = true, transactionManager = "bookingPostgresTransactionManager")
+    public BookingOperationalMetricsSnapshot snapshot(int windowHours) {
+        if (windowHours <= 0 || windowHours > 24 * 31) {
+            throw new IllegalArgumentException("Booking 监控窗口必须在 1 到 744 小时之间。");
+        }
+        return jdbcTemplate.queryForObject(
+                SNAPSHOT_SQL,
+                (resultSet, rowNumber) -> new BookingOperationalMetricsSnapshot(
+                        windowHours,
+                        resultSet.getLong("total_runs"),
+                        resultSet.getLong("failed_runs"),
+                        resultSet.getLong("agent_fallback_runs"),
+                        resultSet.getLong("risk_runs"),
+                        resultSet.getLong("backlog_runs"),
+                        resultSet.getLong("retried_runs"),
+                        resultSet.getLong("retry_attempts")),
+                windowHours);
+    }
+}
@@ -404,6 +404,40 @@ public class BookingPostgresProcessingRunRepository implements BookingProcessing
        }
    }

+    /** 失败 run 的显式重试起点；不新建来源邮件、不改变 revision，也不删除前一 attempt 的审计。 */
+    @Override
+    @Transactional(transactionManager = "bookingPostgresTransactionManager")
+    public BookingProcessingRunSnapshot beginRetry(String runId) {
+        requireText(runId, "runId");
+        Long processingRunDatabaseId = jdbcTemplate.query(
+                        "UPDATE th_hotel_booking.booking_contract_processing_run SET "
+                                + "retry_count = retry_count + 1, run_status = 'CONTEXT_READY', "
+                                + "result_disposition = NULL, error_code = NULL, error_summary = NULL, "
+                                + "completed_at = NULL, updated_at = CURRENT_TIMESTAMP "
+                                + "WHERE run_id = ? AND run_status = 'FAILED' RETURNING id",
+                        (resultSet, rowNumber) -> resultSet.getLong("id"),
+                        runId)
+                .stream()
+                .findFirst()
+                .orElse(null);
+        if (processingRunDatabaseId == null) {
+            throw new IllegalStateException("只有 FAILED 状态的 Booking processing run 可以重试。");
+        }
+        Integer nextAttemptNumber = jdbcTemplate.queryForObject(
+                "SELECT COALESCE(MAX(attempt_number), 0) + 1 "
+                        + "FROM th_hotel_booking.booking_contract_processing_attempt WHERE processing_run_id = ?",
+                Integer.class,
+                processingRunDatabaseId);
+        jdbcTemplate.update(
+                "INSERT INTO th_hotel_booking.booking_contract_processing_attempt ("
+                        + "processing_run_id, attempt_number, attempt_status"
+                        + ") VALUES (?, ?, 'STARTED')",
+                processingRunDatabaseId,
+                nextAttemptNumber == null ? 1 : nextAttemptNumber);
+        return findByRunId(runId)
+                .orElseThrow(() -> new IllegalStateException("Booking processing run 重试后无法读取。"));
+    }
+
    /** 按 run ID 查询安全摘要。 */
    @Override
    @Transactional(readOnly = true, transactionManager = "bookingPostgresTransactionManager")
@@ -52,6 +52,12 @@ public interface BookingProcessingRunRepository {
            String errorCode,
            String errorSummary);

+    /**
+     * 仅将失败 run 置回 Context-ready 并新增一条 processing attempt；调用方随后必须只从已持久化的安全
+     * contracts artifact 重放，不能重新下载原始邮件或附件。
+     */
+    BookingProcessingRunSnapshot beginRetry(String runId);
+
    /** 读取某个安全 contracts artifact 的 JSON；只用于后端确认/查询，不返回原始邮件材料。 */
    Optional<String> findArtifactPayload(String runId, String artifactType);

@@ -0,0 +1,9 @@
+package cn.nianxx.thhotel.workflows.reservation.booking.service;
+
+import cn.nianxx.thhotel.workflows.reservation.booking.common.result.BookingOperationalMetricsView;
+
+/** Booking 运行状态的安全聚合监控服务。 */
+public interface BookingOperationalMetricsService {
+
+    BookingOperationalMetricsView last24Hours();
+}
@@ -9,5 +9,8 @@ public interface BookingProcessingRunService {

    BookingProcessingRunView get(String runId);

+    /** 对失败 run 仅使用已持久化的安全 contracts artifact 发起异步重试。 */
+    BookingProcessingRunView retry(String runId);
+
    BookingConfirmationResult confirm(String runId, BookingConfirmationRequest request, String actorId);
 }
@@ -0,0 +1,44 @@
+package cn.nianxx.thhotel.workflows.reservation.booking.service.impl;
+
+import cn.nianxx.thhotel.workflows.reservation.common.dto.booking.BookingContracts;
+import java.util.List;
+import java.util.Locale;
+
+/** 统一判断 current message 是否具有 Booking 业务信号；不读取 quoted history 或附件。 */
+final class BookingMessageSignalPolicy {
+
+    private static final List<String> BUSINESS_SIGNAL_KEYWORDS = List.of(
+            "BOOKING",
+            "AMEND",
+            "UPDATE",
+            "CANCEL",
+            "CXL",
+            "ROOMING",
+            "PAYMENT",
+            "VOUCHER",
+            "TRACE",
+            "ALLOTMENT",
+            "TOUR CODE",
+            "ยกเลิก",
+            "จอง",
+            "ห้อง");
+
+    private BookingMessageSignalPolicy() {
+    }
+
+    static boolean containsBusinessSignal(String safeSubject, String currentBody) {
+        String source = String.join("\n",
+                safeSubject == null ? "" : safeSubject,
+                currentBody == null ? "" : currentBody)
+                .toUpperCase(Locale.ROOT);
+        return BUSINESS_SIGNAL_KEYWORDS.stream()
+                .map(keyword -> keyword.toUpperCase(Locale.ROOT))
+                .anyMatch(source::contains);
+    }
+
+    /** retry 只可使用已持久化的受控 current-message 摘要重建业务信号。 */
+    static boolean containsBusinessSignal(BookingContracts.AgentInputMaterial agentInputMaterial) {
+        return agentInputMaterial != null
+                && containsBusinessSignal(null, agentInputMaterial.currentMessageText());
+    }
+}
@@ -0,0 +1,46 @@
+package cn.nianxx.thhotel.workflows.reservation.booking.service.impl;
+
+import cn.nianxx.thhotel.workflows.reservation.booking.common.dto.BookingOperationalMetricsSnapshot;
+import cn.nianxx.thhotel.workflows.reservation.booking.common.result.BookingOperationalMetricsView;
+import cn.nianxx.thhotel.workflows.reservation.booking.repository.BookingOperationalMetricsRepository;
+import cn.nianxx.thhotel.workflows.reservation.booking.service.BookingOperationalMetricsService;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Service;
+
+/** 将项目 schema 的计数聚合为可比较的百分比；不读取单封邮件内容。 */
+@Service
+@ConditionalOnProperty(prefix = "booking.postgres", name = "enabled", havingValue = "true")
+public class BookingOperationalMetricsServiceImpl implements BookingOperationalMetricsService {
+
+    private static final int DEFAULT_WINDOW_HOURS = 24;
+
+    private final BookingOperationalMetricsRepository repository;
+
+    public BookingOperationalMetricsServiceImpl(BookingOperationalMetricsRepository repository) {
+        this.repository = repository;
+    }
+
+    @Override
+    public BookingOperationalMetricsView last24Hours() {
+        BookingOperationalMetricsSnapshot snapshot = repository.snapshot(DEFAULT_WINDOW_HOURS);
+        return new BookingOperationalMetricsView(
+                snapshot.windowHours(),
+                snapshot.totalRuns(),
+                snapshot.failedRuns(),
+                percentage(snapshot.failedRuns(), snapshot.totalRuns()),
+                snapshot.agentFallbackRuns(),
+                percentage(snapshot.agentFallbackRuns(), snapshot.totalRuns()),
+                snapshot.riskRuns(),
+                percentage(snapshot.riskRuns(), snapshot.totalRuns()),
+                snapshot.backlogRuns(),
+                snapshot.retriedRuns(),
+                snapshot.retryAttempts());
+    }
+
+    private double percentage(long numerator, long denominator) {
+        if (denominator <= 0) {
+            return 0D;
+        }
+        return Math.round((numerator * 10_000D / denominator)) / 100D;
+    }
+}
@@ -19,7 +19,6 @@ import java.security.NoSuchAlgorithmException;
 import java.util.HexFormat;
 import java.util.LinkedHashMap;
 import java.util.List;
-import java.util.Locale;
 import java.util.Map;
 import java.util.UUID;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -36,22 +35,6 @@ import org.springframework.stereotype.Service;
@ConditionalOnProperty(prefix = "booking.postgres", name = "enabled", havingValue = "true")
 public class BookingPostgresMessageOrchestrator implements BookingMessageOrchestrator {

-    private static final List<String> BUSINESS_SIGNAL_KEYWORDS = List.of(
-            "BOOKING",
-            "AMEND",
-            "UPDATE",
-            "CANCEL",
-            "CXL",
-            "ROOMING",
-            "PAYMENT",
-            "VOUCHER",
-            "TRACE",
-            "ALLOTMENT",
-            "TOUR CODE",
-            "ยกเลิก",
-            "จอง",
-            "ห้อง");
-
    private final BookingContractAssembler contractAssembler;
    private final BookingProcessingRunRepository processingRunRepository;
    private final BookingContextAssembler contextAssembler;
@@ -160,7 +143,8 @@ public class BookingPostgresMessageOrchestrator implements BookingMessageOrchest
                        mergeWarningCodes(masterData.warningCodes(), completion.warningCodes()));
            }

-            boolean businessSignalPresent = containsBusinessSignal(input);
+            boolean businessSignalPresent = BookingMessageSignalPolicy.containsBusinessSignal(
+                    input.safeSubject(), input.currentBody());
            boolean attachmentPresent = !input.attachments().isEmpty();
            if (shouldDispatchAgentAsynchronously(parsedFactSet)) {
                agentContinuationDispatcher.completeAsync(
@@ -277,16 +261,6 @@ public class BookingPostgresMessageOrchestrator implements BookingMessageOrchest
        }
    }

-    private boolean containsBusinessSignal(BookingMessageInput input) {
-        String source = String.join("\n",
-                input.safeSubject() == null ? "" : input.safeSubject(),
-                input.currentBody() == null ? "" : input.currentBody())
-                .toUpperCase(Locale.ROOT);
-        return BUSINESS_SIGNAL_KEYWORDS.stream()
-                .map(keyword -> keyword.toUpperCase(Locale.ROOT))
-                .anyMatch(source::contains);
-    }
-
    private String sha256(String source) {
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
@@ -35,16 +35,19 @@ public class BookingProcessingRunServiceImpl implements BookingProcessingRunServ
    private final BookingProcessingRunRepository processingRunRepository;
    private final BookingDecisionValidator decisionValidator;
    private final BookingConfirmationProjectionFactory projectionFactory;
+    private final BookingAgentContinuationDispatcher agentContinuationDispatcher;
    private final ObjectMapper objectMapper;

    public BookingProcessingRunServiceImpl(
            BookingProcessingRunRepository processingRunRepository,
            BookingDecisionValidator decisionValidator,
            BookingConfirmationProjectionFactory projectionFactory,
+            BookingAgentContinuationDispatcher agentContinuationDispatcher,
            ObjectMapper objectMapper) {
        this.processingRunRepository = processingRunRepository;
        this.decisionValidator = decisionValidator;
        this.projectionFactory = projectionFactory;
+        this.agentContinuationDispatcher = agentContinuationDispatcher;
        this.objectMapper = objectMapper;
    }

@@ -52,7 +55,54 @@ public class BookingProcessingRunServiceImpl implements BookingProcessingRunServ
    @Override
    public BookingProcessingRunView get(String runId) {
        BookingProcessingRunSnapshot snapshot = snapshot(runId);
-        return view(snapshot, confirmationProjection(runId));
+        return view(snapshot, projectionFor(snapshot));
+    }
+
+    /**
+     * 重试只允许 FAILED run，且只会重用已持久化、脱敏的 contracts artifact。为避免用户请求线程等待真实
+     * Agent，所有 retry 都经既有 worker 异步执行并立即返回 Context-ready 状态。
+     */
+    @Override
+    public BookingProcessingRunView retry(String runId) {
+        BookingProcessingRunSnapshot before = snapshot(runId);
+        if (before.status() != BookingContracts.ProcessingStatus.FAILED) {
+            throw workflowError(
+                    HttpStatus.CONFLICT,
+                    "BOOKING_PROCESSING_RUN_NOT_RETRYABLE",
+                    "只有失败的 Booking 邮件可以重试。");
+        }
+        BookingContracts.ParsedFactSet parsedFactSet = artifact(
+                runId,
+                "PARSED_FACT_SET",
+                BookingContracts.ParsedFactSet.class);
+        BookingContracts.ContextPackage contextPackage = artifact(
+                runId,
+                "CONTEXT_PACKAGE",
+                BookingContracts.ContextPackage.class);
+        BookingContracts.MaterialPackage materialPackage = artifact(
+                runId,
+                "MATERIAL_PACKAGE",
+                BookingContracts.MaterialPackage.class);
+        BookingProcessingRunSnapshot retryStarted = processingRunRepository.beginRetry(runId);
+        try {
+            agentContinuationDispatcher.completeAsync(
+                    runId,
+                    parsedFactSet,
+                    contextPackage,
+                    BookingMessageSignalPolicy.containsBusinessSignal(parsedFactSet.agentInputMaterial()),
+                    materialPackage.currentMaterials().stream()
+                            .anyMatch(material -> material.currentMaterial()
+                                    && material.kind() != BookingContracts.EvidenceKind.CURRENT_BODY));
+        } catch (RuntimeException exception) {
+            processingRunRepository.updateRunStatus(
+                    runId,
+                    BookingContracts.ProcessingStatus.FAILED,
+                    BookingContracts.ResultDisposition.RISK_NOTIFICATION,
+                    "BOOKING_RETRY_DISPATCH_FAILED",
+                    "Booking retry dispatch failed before processing.");
+            throw exception;
+        }
+        return view(retryStarted, null);
    }

    /**
@@ -308,6 +358,14 @@ public class BookingProcessingRunServiceImpl implements BookingProcessingRunServ
        return payload == null ? null : read(payload, BookingContracts.ConfirmationProjection.class);
    }

+    private BookingContracts.ConfirmationProjection projectionFor(BookingProcessingRunSnapshot snapshot) {
+        return switch (snapshot.status()) {
+            case VALIDATED, AWAITING_CONFIRMATION, CONFIRMED -> confirmationProjection(snapshot.runId());
+            case RECEIVED, MATERIAL_READY, PARSER_COMPLETE, PARSER_NOT_APPLICABLE,
+                    PARSER_FAILED, CONTEXT_READY, AGENT_COMPLETE, FAILED -> null;
+        };
+    }
+
    private <T> T artifact(String runId, String artifactType, Class<T> type) {
        String payload = processingRunRepository.findArtifactPayload(runId, artifactType)
                .orElseThrow(() -> workflowError(
@@ -0,0 +1,42 @@
+package cn.nianxx.thhotel.workflows.reservation.control;
+
+import cn.nianxx.thhotel.platform.access.common.enums.PlatformPermissionCode;
+import cn.nianxx.thhotel.platform.security.service.FrontendAuthorizationService;
+import cn.nianxx.thhotel.workflows.reservation.booking.common.result.BookingOperationalMetricsView;
+import cn.nianxx.thhotel.workflows.reservation.booking.service.BookingOperationalMetricsService;
+import cn.nianxx.thhotel.workflows.reservation.service.impl.ReservationTaskWorkflowException;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/** 系统管理员可读取的 Booking 聚合监控 API；不返回任一邮件、附件或团号。 */
+@RestController
+@RequestMapping("/api/reservation/booking-processing-metrics")
+public class ReservationBookingOperationalMetricsController {
+
+    private final ObjectProvider<BookingOperationalMetricsService> metricsServiceProvider;
+    private final FrontendAuthorizationService authorizationService;
+
+    public ReservationBookingOperationalMetricsController(
+            ObjectProvider<BookingOperationalMetricsService> metricsServiceProvider,
+            FrontendAuthorizationService authorizationService) {
+        this.metricsServiceProvider = metricsServiceProvider;
+        this.authorizationService = authorizationService;
+    }
+
+    @GetMapping(produces = MediaType.APPLICATION_JSON_VALUE)
+    public BookingOperationalMetricsView getLast24Hours() {
+        authorizationService.requirePermission(PlatformPermissionCode.SYSTEM_ADMIN_CONSOLE_ACCESS.name());
+        BookingOperationalMetricsService service = metricsServiceProvider.getIfAvailable();
+        if (service == null) {
+            throw new ReservationTaskWorkflowException(
+                    HttpStatus.SERVICE_UNAVAILABLE,
+                    "BOOKING_POSTGRES_RUNTIME_DISABLED",
+                    "新 Booking PostgreSQL 运行时尚未启用。");
+        }
+        return service.last24Hours();
+    }
+}
@@ -16,6 +16,7 @@ import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.ResponseStatus;
 import org.springframework.web.bind.annotation.RestController;

 /** contracts-v1 processing run 查询和用户确认 API；没有 PMS/Opera 调用。 */
@@ -40,6 +41,14 @@ public class ReservationBookingProcessingRunController {
        return service().get(runId);
    }

+    /** 仅重放失败 run 的安全 contracts artifact；异步 worker 完成后由前端继续轮询。 */
+    @PostMapping(value = "/{runId}/retry", produces = MediaType.APPLICATION_JSON_VALUE)
+    @ResponseStatus(HttpStatus.ACCEPTED)
+    public BookingProcessingRunView retry(@PathVariable String runId) {
+        authorizationService.requirePermission(PlatformPermissionCode.RESERVATION_TASK_EDIT.name());
+        return service().retry(runId);
+    }
+
    /** 冻结经校验的参数并写确认审计；不会创建 PMS/Opera 操作。 */
    @PostMapping(
            value = "/{runId}/confirmations",
@@ -0,0 +1,29 @@
+package cn.nianxx.thhotel.workflows.reservation.booking.service.impl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import cn.nianxx.thhotel.workflows.reservation.common.dto.booking.BookingContracts;
+import org.junit.jupiter.api.Test;
+
+class BookingMessageSignalPolicyTest {
+
+    @Test
+    void shouldRecognizeCurrentMessageSignalsWithoutUsingHistory() {
+        assertThat(BookingMessageSignalPolicy.containsBusinessSignal("Update booking", "ordinary text")).isTrue();
+        assertThat(BookingMessageSignalPolicy.containsBusinessSignal(null, "ยกเลิกห้องพัก")).isTrue();
+        assertThat(BookingMessageSignalPolicy.containsBusinessSignal("FYI", "ordinary notice")).isFalse();
+    }
+
+    @Test
+    void shouldUseOnlyPersistedSafeAgentMaterialForRetry() {
+        BookingContracts.AgentInputMaterial material = new BookingContracts.AgentInputMaterial(
+                "current-evidence-001",
+                "Subject: booking update\nPlease amend one room.",
+                52,
+                false,
+                "booking-agent-current-message-v1");
+
+        assertThat(BookingMessageSignalPolicy.containsBusinessSignal(material)).isTrue();
+        assertThat(BookingMessageSignalPolicy.containsBusinessSignal((BookingContracts.AgentInputMaterial) null)).isFalse();
+    }
+}
@@ -0,0 +1,42 @@
+package cn.nianxx.thhotel.workflows.reservation.booking.service.impl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import cn.nianxx.thhotel.workflows.reservation.booking.common.dto.BookingOperationalMetricsSnapshot;
+import cn.nianxx.thhotel.workflows.reservation.booking.common.result.BookingOperationalMetricsView;
+import cn.nianxx.thhotel.workflows.reservation.booking.service.BookingOperationalMetricsService;
+import org.junit.jupiter.api.Test;
+
+class BookingOperationalMetricsServiceImplTest {
+
+    @Test
+    void shouldExposeSafeCountsAndRoundedRatesForTheFixedWindow() {
+        BookingOperationalMetricsService service = new BookingOperationalMetricsServiceImpl(hours -> {
+            assertThat(hours).isEqualTo(24);
+            return new BookingOperationalMetricsSnapshot(24, 8, 2, 3, 1, 4, 2, 5);
+        });
+
+        BookingOperationalMetricsView view = service.last24Hours();
+
+        assertThat(view.windowHours()).isEqualTo(24);
+        assertThat(view.totalRuns()).isEqualTo(8);
+        assertThat(view.failureRatePercent()).isEqualTo(25D);
+        assertThat(view.agentFallbackRatePercent()).isEqualTo(37.5D);
+        assertThat(view.riskRatePercent()).isEqualTo(12.5D);
+        assertThat(view.backlogRuns()).isEqualTo(4);
+        assertThat(view.retriedRuns()).isEqualTo(2);
+        assertThat(view.retryAttempts()).isEqualTo(5);
+    }
+
+    @Test
+    void shouldAvoidDivisionByZeroForAnEmptyWindow() {
+        BookingOperationalMetricsService service = new BookingOperationalMetricsServiceImpl(
+                hours -> new BookingOperationalMetricsSnapshot(24, 0, 0, 0, 0, 0, 0, 0));
+
+        BookingOperationalMetricsView view = service.last24Hours();
+
+        assertThat(view.failureRatePercent()).isZero();
+        assertThat(view.agentFallbackRatePercent()).isZero();
+        assertThat(view.riskRatePercent()).isZero();
+    }
+}
@@ -0,0 +1,192 @@
+package cn.nianxx.thhotel.workflows.reservation.booking.service.impl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import cn.nianxx.thhotel.workflows.reservation.booking.common.dto.BookingProcessingRunSnapshot;
+import cn.nianxx.thhotel.workflows.reservation.booking.common.result.BookingProcessingRunView;
+import cn.nianxx.thhotel.workflows.reservation.booking.repository.BookingProcessingRunRepository;
+import cn.nianxx.thhotel.workflows.reservation.booking.service.BookingConfirmationProjectionFactory;
+import cn.nianxx.thhotel.workflows.reservation.booking.service.BookingDecisionValidator;
+import cn.nianxx.thhotel.workflows.reservation.common.dto.booking.BookingContracts;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.time.OffsetDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import org.junit.jupiter.api.Test;
+
+class BookingProcessingRunServiceImplTest {
+
+    @Test
+    void shouldRetryOnlyFromSafePersistedContractsAndReturnAsyncContextReadyView() throws Exception {
+        BookingProcessingRunRepository repository = mock(BookingProcessingRunRepository.class);
+        BookingAgentContinuationDispatcher dispatcher = mock(BookingAgentContinuationDispatcher.class);
+        ObjectMapper objectMapper = new ObjectMapper().findAndRegisterModules();
+        BookingProcessingRunServiceImpl service = new BookingProcessingRunServiceImpl(
+                repository,
+                mock(BookingDecisionValidator.class),
+                mock(BookingConfirmationProjectionFactory.class),
+                dispatcher,
+                objectMapper);
+        BookingContracts.ContractMeta meta = meta();
+        BookingContracts.EvidenceReference evidence = evidence(meta);
+        BookingContracts.ParsedFactSet parsedFactSet = new BookingContracts.ParsedFactSet(
+                meta,
+                List.of(evidence),
+                BookingContracts.ParserApplicability.NOT_APPLICABLE,
+                null,
+                List.of(),
+                List.of(),
+                List.of("NO_FIXED_CHANNEL"),
+                List.of(),
+                new BookingContracts.AgentInputMaterial(
+                        evidence.evidenceId(),
+                        "Subject: Booking update\nPlease amend this reservation.",
+                        54,
+                        false,
+                        "booking-agent-current-message-v1"));
+        BookingContracts.ContextPackage contextPackage = new BookingContracts.ContextPackage(
+                meta,
+                List.of(evidence),
+                "MESSAGE",
+                List.of(),
+                Map.of(),
+                Map.of(),
+                Map.of(),
+                Map.of(),
+                List.of());
+        BookingContracts.MaterialPackage materialPackage = materialPackage(meta, evidence);
+
+        when(repository.findByRunId("run-retry-001")).thenReturn(Optional.of(snapshot(
+                BookingContracts.ProcessingStatus.FAILED,
+                0)));
+        when(repository.findArtifactPayload("run-retry-001", "PARSED_FACT_SET"))
+                .thenReturn(Optional.of(objectMapper.writeValueAsString(parsedFactSet)));
+        when(repository.findArtifactPayload("run-retry-001", "CONTEXT_PACKAGE"))
+                .thenReturn(Optional.of(objectMapper.writeValueAsString(contextPackage)));
+        when(repository.findArtifactPayload("run-retry-001", "MATERIAL_PACKAGE"))
+                .thenReturn(Optional.of(objectMapper.writeValueAsString(materialPackage)));
+        when(repository.beginRetry("run-retry-001")).thenReturn(snapshot(
+                BookingContracts.ProcessingStatus.CONTEXT_READY,
+                1));
+
+        BookingProcessingRunView view = service.retry("run-retry-001");
+
+        assertThat(view.status()).isEqualTo("CONTEXT_READY");
+        assertThat(view.retryCount()).isEqualTo(1);
+        assertThat(view.confirmationProjection()).isNull();
+        verify(repository).beginRetry("run-retry-001");
+        verify(dispatcher).completeAsync(
+                eq("run-retry-001"),
+                eq(parsedFactSet),
+                eq(contextPackage),
+                eq(true),
+                eq(true));
+    }
+
+    @Test
+    void shouldRejectRetryWhenRunHasNotFailed() {
+        BookingProcessingRunRepository repository = mock(BookingProcessingRunRepository.class);
+        BookingProcessingRunServiceImpl service = new BookingProcessingRunServiceImpl(
+                repository,
+                mock(BookingDecisionValidator.class),
+                mock(BookingConfirmationProjectionFactory.class),
+                mock(BookingAgentContinuationDispatcher.class),
+                new ObjectMapper().findAndRegisterModules());
+        when(repository.findByRunId("run-retry-002")).thenReturn(Optional.of(snapshot(
+                BookingContracts.ProcessingStatus.AWAITING_CONFIRMATION,
+                0)));
+
+        assertThatThrownBy(() -> service.retry("run-retry-002"))
+                .hasMessageContaining("只有失败的 Booking 邮件可以重试");
+
+        verify(repository, never()).beginRetry("run-retry-002");
+    }
+
+    private BookingContracts.MaterialPackage materialPackage(
+            BookingContracts.ContractMeta meta,
+            BookingContracts.EvidenceReference evidence) {
+        BookingContracts.SourceMessageEnvelope envelope = new BookingContracts.SourceMessageEnvelope(
+                meta,
+                List.of(evidence),
+                BookingContracts.MessageOrigin.MANUAL_EML,
+                "external-retry-001",
+                "conversation-retry-001",
+                null,
+                "idempotency-retry-001",
+                OffsetDateTime.parse("2026-08-08T00:00:00Z"),
+                "Booking update",
+                evidence.evidenceId(),
+                null,
+                List.of(),
+                null);
+        return new BookingContracts.MaterialPackage(
+                meta,
+                List.of(evidence),
+                envelope,
+                List.of(
+                        new BookingContracts.MaterialReference(
+                                evidence.evidenceId(),
+                                BookingContracts.EvidenceKind.CURRENT_BODY,
+                                true,
+                                Map.of("content_available", true)),
+                        new BookingContracts.MaterialReference(
+                                "attachment-evidence-retry-001",
+                                BookingContracts.EvidenceKind.ATTACHMENT,
+                                true,
+                                Map.of("attachment_id", "attachment-retry-001"))),
+                List.of(),
+                List.of(),
+                List.of(),
+                BookingContracts.MaterialSelectionStatus.READY,
+                List.of());
+    }
+
+    private BookingProcessingRunSnapshot snapshot(
+            BookingContracts.ProcessingStatus status,
+            int retryCount) {
+        return new BookingProcessingRunSnapshot(
+                1L,
+                "run-retry-001",
+                2L,
+                "source-retry-001",
+                1,
+                status,
+                status == BookingContracts.ProcessingStatus.FAILED
+                        ? BookingContracts.ResultDisposition.RISK_NOTIFICATION
+                        : null,
+                false,
+                retryCount,
+                null,
+                null,
+                OffsetDateTime.parse("2026-08-08T00:00:00Z"),
+                null);
+    }
+
+    private BookingContracts.ContractMeta meta() {
+        return new BookingContracts.ContractMeta(
+                BookingContracts.CONTRACT_VERSION,
+                "catalog-v1",
+                "parser-v1",
+                BookingContracts.AGENT_NOT_INVOKED,
+                "run-retry-001",
+                "source-retry-001",
+                0L);
+    }
+
+    private BookingContracts.EvidenceReference evidence(BookingContracts.ContractMeta meta) {
+        return new BookingContracts.EvidenceReference(
+                "current-evidence-retry-001",
+                BookingContracts.EvidenceKind.CURRENT_BODY,
+                meta.sourceMessageId(),
+                "current-body",
+                "a".repeat(64),
+                true);
+    }
+}