test(ci): 门禁选择器纳入 MQ 定向用例（5 类 38 例）——zcode 审查轮补齐，远端 CI 覆盖本地 97/97 的 FC 半边

ci/test-gate.sh

@@ -1,7 +1,7 @@
 #!/bin/sh
 set -eu
 cd "$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd)"
-TESTS=FcAnswerDeliveryTest,FcDiscoveryAccessTest,FormConsumerGrantPushGateTest,FcReconcileTaskTest,OapiFormControllerTest,FcEventPublisherTest,AnonReplayGuardTest,AppCodeReconcileRoutingTest,OapiSchemaPublishedVersionTest
+TESTS=FcAnswerDeliveryTest,FcDiscoveryAccessTest,FormConsumerGrantPushGateTest,FcReconcileTaskTest,OapiFormControllerTest,FcEventPublisherTest,AnonReplayGuardTest,AppCodeReconcileRoutingTest,OapiSchemaPublishedVersionTest,AnswerDeliveryBrokerRaceTest,AnswerDeliverySourceIdentityTest,AnswerDeliveryMqContractTest,FcEventConverterContractTest,FcEventJsonPublishTest
 # Surefire 的选择器部分匹配也可能成功；每个指定类必须随源码存在。
 oldIFS=$IFS
 IFS=,

docs/mq-local-repair-contract.md

@@ -0,0 +1,46 @@
+# MQ 独立修复契约（2026-10-23，本地候选）
+
+主链已经部署并由用户验收。本补丁只处理 MQ，不改变 UserBase 身份模型、HTTP 收据主链、FL、B5 或对账调度。
+代码基线为 FC 核心 SHA 1717f4c47e26918e7f582d0c00ddd23b732ab98a；旧 MQ WIP 保留在原 worktree。
+
+## 身份与路由
+
+- `application_info.id` 是业务应用主身份；`appKey/appSecret` 是签名凭据；`appCode` 只用于部署/寻址以及现有注册查询。
+- 兼容现有 v2 协议：`sourceAppKey/consumerAppKey` 仍携带凭据键，不改名、不把业务 id 填进去、不迁移已有业务绑定。
+- source 默认引用最终 `yh.app-auth.app-key`；显式 source 与签名键不一致时拒绝。真正发布前还按本应用的部署 code 调已有 `getAppByCode`，核验注册行的正整数 id、appCode、appKey。允许 id != appKey。不可达/错配进入有界失败退避，不记 broker 成功。
+- 不使用答卷的资源 `snapshot.appCode` 推断发布者身份。该字段是资源所属应用技术标识，原样保留。
+- v2 routing key 固定为 `fc.snapshot.v2.<consumer credential key>.<schoolId>`，不能把应用主键或 appCode 代入。拒绝能扩展 topic 段的通配符/额外分隔符与非正学校号。
+- SH 的 binding、信封消费身份与最终签名凭据必须同源；不硬编码为 100008 或 100018。
+
+## 投递状态裁决
+
+| 输入 | 条件 | 结果 |
+|---|---|---|
+| ACK | 当前代且没有已返回事实 | BROKER_ACCEPTED；不写 consumer_applied |
+| 重复 ACK | 当前代已接受 | 时间戳/计数不再改写 |
+| NACK / mandatory return | 当前代，包括先前已收到 ACK | FAILED 或预算耗尽 DEAD；清 broker_accepted_at，保留原因并退避 |
+| ACK 带 CorrelationData.returned | 独立 return 回调未持久化也一样 | 先按 NO_ROUTE 失败处理，不能写成功 |
+| 重复负回调/迟到旧代 ACK | 负回调已推进 retry_count 或授权重放操作已变化 | 幂等忽略，不清错误、不重复扣重试预算 |
+| DEAD/GRANT_REVOKED | 任意 broker 回调 | 不复活 |
+| 序列化/快照损坏/注册不可达 | 未实际发送 | 失败退避；同 eventId 与不可变快照恢复，不伪造消费成功 |
+
+回调以现有 retry_count 与重放操作键摘要标识代际，读取同一投递行时使用写锁；没有新增库列。
+这是至少一次投递，不能声称 broker/数据库的真实并发恰好一次已验证。
+发送通道必须配置 correlated confirm、publisher-returns 和 mandatory；缺少返回事实观察能力时不启用扫描发送。
+`push_allowed=0` 保留 read/subscribe 数据权，既不建新 v2 outbox 行，也不继续发失去 push 权的旧行。
+
+## N-1 元数据事件
+
+`fc.events.v1` 现在显式序列化为 UTF-8 JSON Message，再调用 RabbitTemplate.send；不更换共享模板的转换器，不影响 v2。
+信封仍为十字段闭集；不含 values/附件字节/凭据。messageId 与 event_id 一致，提交后使用冻结的坐标发送。
+`fc.events.enabled` 默认 false，且仍须 broker host；仅平台注入 host 不会打开已修好的事件通道。
+该元数据通知与 v2 push 授权是两个协议；没有引入新的订阅授权模型。
+保持原 best-effort 语义：转换/声明/发送失败不反向破坏答卷提交，不增加 v1 outbox 或自动重放；后续事件可以恢复。v2 才有有界可靠重试台账。
+
+## B-03 及验证边界
+
+B-03 原定义是共享 `/` vhost 与同一 admin 账号，不是转换器问题。独立 vhost、分权账号与 topic 前缀约束仍需环境交付。
+FC 会主动 declare 自己的 exchange；SH RabbitAdmin 会声明队列/exchange/DLX/binding。因此权限方案须包含这些实际操作，不能只写“发布方 write”就认为配置完毕。
+普通资源权限与 topic binding key 权限须分开核验；依据 [RabbitMQ 官方访问控制表](https://www.rabbitmq.com/docs/access-control)，不是本轮 broker 实测结论。
+本轮仅本地规则、真实 JSON 转换、Spring 条件装配测试。真实路由、NO_ROUTE、ACL、HMAC身份、DB并发与生产验收均 NOT RUN。
+不得开启生产 MQ、修改 push_allowed 或 B-08。请使用独立测试 vhost 完成后续环境验证。

src/main/java/com/yunhai/formcenter/repository/AnswerDeliveryRepository.java

@@ -4,6 +4,8 @@ import com.yunhai.formcenter.entity.AnswerDelivery;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Lock;
+import javax.persistence.LockModeType;
 import org.springframework.data.jpa.repository.Query;
 import org.springframework.data.repository.query.Param;
 
@@ -20,13 +22,20 @@ import java.util.Optional;
  */
 public interface AnswerDeliveryRepository extends JpaRepository<AnswerDelivery, Long> {
 
+    /** 回调/发送失败按同一行串行裁决，避免 return 与 ACK 各读旧状态后互相覆盖。 */
+    @Override
+    @Lock(LockModeType.PESSIMISTIC_WRITE)
+    Optional<AnswerDelivery> findById(Long id);
+
     /** 回执端点定位（eventId 全局唯一，另三键做逐字复核） */
+    @Lock(LockModeType.PESSIMISTIC_WRITE)
     Optional<AnswerDelivery> findByEventId(String eventId);
 
     /** 同 revision 同消费方是否已有行（入队幂等；正常由 uk_delivery_revision 兜底） */
     Optional<AnswerDelivery> findByAnswerUuidAndAnswerRevisionAndConsumerAppKeyAndConsumerSchoolId(
             String answerUuid, Long answerRevision, String consumerAppKey, Long consumerSchoolId);
 
+    @Lock(LockModeType.PESSIMISTIC_WRITE)
     List<AnswerDelivery> findByAnswerUuidAndConsumerAppKeyOrderByIdAsc(String answerUuid, String consumerAppKey);
 
     /** 到点待投（PENDING 首投与 FAILED 退避重试同一条队列） */

src/main/java/com/yunhai/formcenter/service/AnswerDeliveryService.java

@@ -1,6 +1,7 @@
 package com.yunhai.formcenter.service;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.yunhai.formcenter.entity.Answer;
 import com.yunhai.formcenter.entity.AnswerDelivery;
 import com.yunhai.formcenter.model.AnswerDeliveryModels;
@@ -22,6 +23,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.core.ReturnedMessage;
 import org.springframework.amqp.rabbit.connection.CorrelationData;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.ObjectProvider;
@@ -38,10 +40,13 @@ import org.springframework.transaction.support.TransactionTemplate;
 
 import javax.annotation.PostConstruct;
 import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Base64;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.ExecutorService;
@@ -74,6 +79,7 @@ public class AnswerDeliveryService {
     private static final DateTimeFormatter OCCURRED_AT = DateTimeFormatter.ISO_OFFSET_DATE_TIME;
     private static final DateTimeFormatter TS = DateTimeFormatter.ISO_LOCAL_DATE_TIME;
     private static final String HEADER_DELIVERY_ID = "fc-delivery-id";
+    private static final String HEADER_DELIVERY_ATTEMPT = "fc-delivery-attempt";
     /** 退避序列：base×2^n，封顶 cap（broker 故障时不许热循环） */
     private static final List<String> SCANNABLE_STATES =
             Arrays.asList(AnswerDelivery.STATE_PENDING, AnswerDelivery.STATE_FAILED);
@@ -87,9 +93,18 @@ public class AnswerDeliveryService {
     private final ObjectProvider<RabbitTemplate> templates;
 
     /** 发布方身份（服务端配置；不接受消费方申报，也不从资源 appCode 推） */
-    @Value("${fc.answer.delivery.source-app-key:${spring.application.name:form-center}}")
+    @Value("${fc.answer.delivery.source-app-key:${yh.app-auth.app-key:}}")
     private String sourceAppKey;
 
+    @Value("${yh.app-auth.app-key:}")
+    private String signingAppKey;
+
+    @Autowired
+    private com.yunhai.platform.openapi.base.OpenApiAppClient appRegistryClient;
+
+    @Value("${spring.application.name:form-center}")
+    private String applicationCode;
+
     @Value("${spring.rabbitmq.host:}")
     private String host;
 
@@ -97,6 +112,12 @@ public class AnswerDeliveryService {
     @Value("${spring.rabbitmq.publisher-confirm-type:none}")
     private String confirmType;
 
+    @Value("${spring.rabbitmq.publisher-returns:false}")
+    private boolean publisherReturns = true;
+
+    @Value("${spring.rabbitmq.template.mandatory:false}")
+    private boolean mandatory = true;
+
     @Value("${fc.answer.delivery.batch-size:50}")
     private int batchSize;
 
@@ -138,11 +159,15 @@ public class AnswerDeliveryService {
 
     @PostConstruct
     public void logState() {
+        if (isBlank(signingAppKey) || isBlank(sourceAppKey) || !signingAppKey.trim().equals(sourceAppKey.trim())) {
+            throw new IllegalStateException("fc.answer.delivery.source-app-key 必须等于 yh.app-auth.app-key 签名身份");
+        }
+        sourceAppKey = sourceAppKey.trim();
         if (!hostConfigured()) {
             log.info("[v2投递] 未配置 RABBITMQ_HOST：outbox 仍照常落库并留在 PENDING（不丢事件，B.4）");
             return;
         }
-        if (!"correlated".equalsIgnoreCase(confirmType)) {
+        if (!"correlated".equalsIgnoreCase(confirmType) || !publisherReturns || !mandatory) {
             log.error("[v2投递] 已配置 broker 但 spring.rabbitmq.publisher-confirm-type={} —— "
                     + "拿不到逐条 confirm 就不发送（宁可积压也不按未确认投递），请置 correlated", confirmType);
             return;
@@ -159,6 +184,7 @@ public class AnswerDeliveryService {
     /** 投递通道是否可用（broker + confirm 双条件；不满足=只入队不发送） */
     public boolean sendEnabled() {
         return hostConfigured() && "correlated".equalsIgnoreCase(confirmType)
+                && publisherReturns && mandatory
                 && templates.getIfAvailable() != null;
     }
 
@@ -340,21 +366,24 @@ public class AnswerDeliveryService {
             markFailed(row, "容器内无 RabbitTemplate");
             return false;
         }
+        final String publishKey = callbackKey(row);
         try {
+            requireSourceApplication(row);
             declareExchangeOnce(template);
             byte[] body = objectMapper.writeValueAsString(envelopeOf(row)).getBytes(StandardCharsets.UTF_8);
             MessageProperties props = new MessageProperties();
             props.setMessageId(row.getEventId());
-            props.setCorrelationId(String.valueOf(row.getId()));
+            props.setCorrelationId(publishKey);
             props.setHeader(HEADER_DELIVERY_ID, String.valueOf(row.getId()));
+            props.setHeader(HEADER_DELIVERY_ATTEMPT, publishKey);
             props.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
             props.setContentType(MessageProperties.CONTENT_TYPE_JSON);
             props.setContentEncoding(StandardCharsets.UTF_8.name());
             template.send(AnswerDeliveryModels.EXCHANGE, routingKey(row), new Message(body, props),
-                    new CorrelationData(String.valueOf(row.getId())));
+                    new CorrelationData(publishKey));
             return true;
         } catch (Exception e) {
-            markFailed(row, e.toString());
+            markFailed(row, publishKey, e.toString());
             log.warn("[v2投递] 发送异常，退避重试：eventId={} cause={}", row.getEventId(), e.toString());
             return false;
         }
@@ -366,9 +395,28 @@ public class AnswerDeliveryService {
     }
 
     public static String routingKey(String consumerAppKey, Long consumerSchoolId) {
+        if (consumerAppKey == null || !consumerAppKey.matches("[A-Za-z0-9_-]+")
+                || consumerSchoolId == null || consumerSchoolId <= 0) {
+            throw new IllegalArgumentException("INVALID_MQ_ROUTE_COORDINATE");
+        }
         return "fc.snapshot.v2." + consumerAppKey + "." + consumerSchoolId;
     }
 
+    /** Wire *AppKey fields contain credentials' keys, never application_info.id or resource appCode. */
+    private void requireSourceApplication(AnswerDelivery row) {
+        if (isBlank(sourceAppKey) || !sourceAppKey.equals(signingAppKey == null ? null : signingAppKey.trim())
+                || !sourceAppKey.equals(row.getSourceAppKey())) {
+            throw new IllegalStateException("SOURCE_CREDENTIAL_MISMATCH");
+        }
+        JsonNode application = appRegistryClient == null ? null : appRegistryClient.getAppByCode(applicationCode);
+        if (application == null || !application.path("id").isIntegralNumber()
+                || !application.path("id").canConvertToLong() || application.path("id").asLong() <= 0
+                || !sourceAppKey.equals(application.path("appKey").asText())
+                || !applicationCode.equals(application.path("appCode").asText())) {
+            throw new IllegalStateException("SOURCE_APPLICATION_MISMATCH");
+        }
+    }
+
     private SnapshotEnvelope envelopeOf(AnswerDelivery row) {
         SnapshotEnvelope envelope = new SnapshotEnvelope();
         envelope.setProtocolVersion(row.getProtocolVersion());
@@ -441,7 +489,16 @@ public class AnswerDeliveryService {
                 return; // v1 广播通道不带我们的相关键
             }
             try {
-                onBrokerConfirm(Long.valueOf(correlationData.getId()), ack, cause);
+                String key = correlationData.getId();
+                // CorrelationData retains the return even if its separate callback could not persist it.
+                if (correlationData.getReturned() != null) {
+                    ReturnedMessage returned = correlationData.getReturned();
+                    onUnroutable(callbackDeliveryId(key), key, returned.getExchange() + "/"
+                            + returned.getRoutingKey() + " reply=" + returned.getReplyCode()
+                            + " " + returned.getReplyText());
+                    return;
+                }
+                onBrokerConfirm(callbackDeliveryId(key), key, ack, cause);
             } catch (Exception e) {
                 log.warn("[v2投递] confirm 回写失败 id={}：{}", correlationData.getId(), e.toString());
             }
@@ -451,7 +508,10 @@ public class AnswerDeliveryService {
             if (id == null) {
                 return;
             }
-            onUnroutable(id, returned.getExchange() + "/" + returned.getRoutingKey()
+            Object generation = returned.getMessage().getMessageProperties()
+                    .getHeaders().get(HEADER_DELIVERY_ATTEMPT);
+            String key = generation == null ? legacyCallbackKey(id) : String.valueOf(generation);
+            onUnroutable(id, key, returned.getExchange() + "/" + returned.getRoutingKey()
                     + " reply=" + returned.getReplyCode() + " " + returned.getReplyText());
         });
     }
@@ -471,44 +531,94 @@ public class AnswerDeliveryService {
         }
     }
 
-    /** confirm 只落 brokerAccepted；nack 走退避（B.4"confirm 只记 brokerAccepted"） */
+    /** 代际来自现有 retry_count + 授权重放操作键，无新列；失败/新重放后旧回调不能改变新轮。 */
+    static String callbackKey(AnswerDelivery row) {
+        String replay = row.getLastReplayOperationId();
+        String digest = "";
+        if (replay != null && !replay.isEmpty()) {
+            try {
+                digest = Base64.getUrlEncoder().withoutPadding().encodeToString(
+                        MessageDigest.getInstance("SHA-256").digest(replay.getBytes(StandardCharsets.UTF_8)));
+            } catch (NoSuchAlgorithmException e) {
+                throw new IllegalStateException("SHA-256 unavailable", e);
+            }
+        }
+        return row.getId() + ":" + (row.getRetryCount() == null ? 0 : row.getRetryCount()) + ":" + digest;
+    }
+
+    private static String legacyCallbackKey(Long id) {
+        return id + ":0:"; // 旧无代际回调只可能确认原始首投，不能猜测后续重试/重放。
+    }
+
+    private static Long callbackDeliveryId(String key) {
+        int separator = key.indexOf(':');
+        return Long.valueOf(separator < 0 ? key : key.substring(0, separator));
+    }
+
+    /** 兼容首投的旧包内调用；生产新投递始终携带完整代际。 */
     void onBrokerConfirm(Long deliveryId, boolean ack, String cause) {
+        onBrokerConfirm(deliveryId, legacyCallbackKey(deliveryId), ack, cause);
+    }
+
+    /** confirm 不代表可路由；同代 return 的失败事实优先，旧代 ACK/nack 不能抹掉失败或改新轮。 */
+    void onBrokerConfirm(Long deliveryId, String key, boolean ack, String cause) {
         transactionTemplate.executeWithoutResult(tx -> {
             AnswerDelivery row = deliveryRepository.findById(deliveryId).orElse(null);
-            if (row == null) {
+            if (row == null || !callbackKey(row).equals(normalizeCallbackKey(key, deliveryId))
+                    || terminalDelivery(row)
+                    || (ack && AnswerDelivery.STATE_BROKER_ACCEPTED.equals(row.getState()))) {
                 return;
             }
-            if (AnswerDelivery.STATE_BROKER_ACCEPTED.equals(row.getState())) {
-                return; // 重复 confirm（broker 重投同一相关键）幂等吸收
-            }
             if (ack) {
                 row.setState(AnswerDelivery.STATE_BROKER_ACCEPTED);
                 row.setBrokerAcceptedAt(LocalDateTime.now());
                 row.setLastError(null);
+                row.setNextRetryAt(null);
             } else {
+                row.setBrokerAcceptedAt(null);
                 bumpRetry(row, "broker nack: " + cause);
             }
             deliveryRepository.save(row);
         });
     }
 
-    /** mandatory 不可路由=目标私有队列没绑上（部署契约缺隔离配置），按失败退避而不是静默丢弃 */
     void onUnroutable(Long deliveryId, String detail) {
+        onUnroutable(deliveryId, legacyCallbackKey(deliveryId), detail);
+    }
+
+    /** 同代 return 即使迟于 ACK 也要保留 NO_ROUTE；重复 return 代际已失效，不能重复增加退避。 */
+    void onUnroutable(Long deliveryId, String key, String detail) {
         transactionTemplate.executeWithoutResult(tx -> {
             AnswerDelivery row = deliveryRepository.findById(deliveryId).orElse(null);
-            if (row == null || AnswerDelivery.STATE_BROKER_ACCEPTED.equals(row.getState())) {
+            if (row == null || !callbackKey(row).equals(normalizeCallbackKey(key, deliveryId))
+                    || terminalDelivery(row)) {
                 return;
             }
-            markFailed(row, "不可路由（mandatory return）：" + detail);
+            row.setBrokerAcceptedAt(null);
+            bumpRetry(row, "NO_ROUTE: 不可路由（mandatory return）：" + detail);
             deliveryRepository.save(row);
         });
         log.warn("[v2投递] 消息不可路由：id={} {}", deliveryId, detail);
     }
 
+    private static String normalizeCallbackKey(String key, Long id) {
+        return String.valueOf(id).equals(key) ? legacyCallbackKey(id) : key;
+    }
+
+    private static boolean terminalDelivery(AnswerDelivery row) {
+        return AnswerDelivery.STATE_DEAD.equals(row.getState())
+                || AnswerDelivery.STATE_GRANT_REVOKED.equals(row.getState());
+    }
+
     private void markFailed(AnswerDelivery row, String error) {
+        markFailed(row, callbackKey(row), error);
+    }
+
+    private void markFailed(AnswerDelivery row, String key, String error) {
         transactionTemplate.executeWithoutResult(tx -> {
             AnswerDelivery fresh = deliveryRepository.findById(row.getId()).orElse(null);
-            if (fresh == null) {
+            if (fresh == null || !callbackKey(fresh).equals(key) || terminalDelivery(fresh)
+                    || AnswerDelivery.STATE_BROKER_ACCEPTED.equals(fresh.getState())) {
                 return;
             }
             bumpRetry(fresh, error);

src/main/java/com/yunhai/formcenter/service/FcEventPublisher.java

@@ -1,12 +1,16 @@
 package com.yunhai.formcenter.service;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.yunhai.formcenter.entity.Answer;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
@@ -58,7 +62,13 @@ public class FcEventPublisher {
      */
     private final ObjectProvider<RabbitTemplate> templates;
 
-    /** 是否启用的唯一判据=宿主是否注入了 RABBITMQ_HOST（yml 把它映射到 spring.rabbitmq.host） */
+    // Local serialization leaves the shared RabbitTemplate converter and v2 channel unchanged.
+    private final ObjectMapper eventMapper = new ObjectMapper();
+
+    @Value("${fc.events.enabled:false}")
+    private boolean eventsEnabled;
+
+    /** Explicit events switch plus broker host; independent of the v2 snapshot outbox. */
     @Value("${spring.rabbitmq.host:}")
     private String host;
 
@@ -69,12 +79,12 @@ public class FcEventPublisher {
         if (enabled()) {
             log.info("[MQ] 事件发布启用（exchange={}，host={}）", EXCHANGE, host);
         } else {
-            log.info("[MQ] 未配置 RABBITMQ_HOST，事件发布禁用——拉取端点即事实源，服务照常（契约 v2 §6.2 裁决 3）");
+            log.info("[MQ] events.enabled=false 或未配置 broker，事件发布禁用；拉取仍是事实源");
         }
     }
 
     public boolean enabled() {
-        return host != null && !host.trim().isEmpty();
+        return eventsEnabled && host != null && !host.trim().isEmpty();
     }
 
     // ==================== 三个发布点 ====================
@@ -123,13 +133,19 @@ public class FcEventPublisher {
 
     private void send(Envelope envelope, String action) {
         try {
+            byte[] body = eventMapper.writeValueAsBytes(envelope);
             RabbitTemplate template = templates.getIfAvailable();
             if (template == null) {
                 log.warn("[MQ] 已配置 RABBITMQ_HOST 但容器里没有 RabbitTemplate，跳过发布（不影响主流程）");
                 return;
             }
             declareExchangeOnce(template);
-            template.convertAndSend(EXCHANGE, routingKey(envelope.getAppCode(), action), envelope);
+            MessageProperties properties = new MessageProperties();
+            properties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
+            properties.setContentEncoding("UTF-8");
+            properties.setMessageId(envelope.getEventId());
+            properties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
+            template.send(EXCHANGE, routingKey(envelope.getAppCode(), action), new Message(body, properties));
         } catch (Exception e) {
             // 裁决 2：broker 故障绝不阻断提交。消费方以游标拉取为事实源，这条事件丢了由下一轮增量补回
             log.warn("[MQ] 事件发布失败（忽略，拉取兜底）：type={} answer={} cause={}",

src/main/resources/application.yml

@@ -148,9 +148,12 @@ form-center:
 # source-app-key 默认取本应用签名身份（yh.app-auth.app-key），信封里的 sourceAppKey 由服务端给，
 # 不接受消费方申报。未配 broker 时行留在 PENDING、不丢事件（B.4）。
 fc:
+  events:
+    # Metadata notification is opt-in too; fixing conversion must not enable a production channel.
+    enabled: ${FC_EVENTS_ENABLED:false}
   answer:
     delivery:
-      source-app-key: ${YH_APP_AUTH_APP_KEY:100021}
+      source-app-key: ${yh.app-auth.app-key}
       scan-interval-ms: ${FC_DELIVERY_SCAN_INTERVAL_MS:10000}
       scan-initial-delay-ms: ${FC_DELIVERY_SCAN_INITIAL_DELAY_MS:30000}
       batch-size: ${FC_DELIVERY_BATCH_SIZE:50}

src/test/java/com/yunhai/formcenter/service/AnswerDeliveryBrokerRaceTest.java

@@ -0,0 +1,229 @@
+package com.yunhai.formcenter.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.yunhai.formcenter.entity.AnswerDelivery;
+import com.yunhai.formcenter.repository.AnswerDeliveryRepository;
+import com.yunhai.formcenter.repository.AnswerRepository;
+import com.yunhai.formcenter.model.FormAccessModels.SubscriptionGrant;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.ReturnedMessage;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.mockito.ArgumentCaptor;
+import java.util.Collections;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.data.jpa.repository.Lock;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import javax.persistence.LockModeType;
+import java.time.LocalDateTime;
+import java.util.Optional;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.verify;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+
+/** B-09 独立规则回归；不冒充真实 RabbitMQ 或真实数据库并发验收。 */
+class AnswerDeliveryBrokerRaceTest {
+    private AnswerDelivery row;
+    private AnswerDeliveryService service;
+
+    @BeforeEach
+    void setUp() {
+        row = new AnswerDelivery();
+        row.setId(7L);
+        row.setEventId("event");
+        row.setConsumerAppKey("consumer");
+        row.setState(AnswerDelivery.STATE_PENDING);
+        row.setRetryCount(0);
+        AnswerDeliveryRepository rows = mock(AnswerDeliveryRepository.class);
+        when(rows.findById(7L)).thenReturn(Optional.of(row));
+        TransactionTemplate tx = mock(TransactionTemplate.class);
+        doAnswer(call -> {
+            Consumer<TransactionStatus> action = call.getArgument(0);
+            action.accept(null);
+            return null;
+        }).when(tx).executeWithoutResult(any());
+        @SuppressWarnings("unchecked")
+        ObjectProvider<RabbitTemplate> templates = mock(ObjectProvider.class);
+        service = new AnswerDeliveryService(rows, mock(AnswerRepository.class),
+                mock(AnswerSnapshotService.class), mock(FormConsumerAccessService.class),
+                new ObjectMapper(), tx, templates);
+        ReflectionTestUtils.setField(service, "maxRetries", 3);
+        ReflectionTestUtils.setField(service, "backoffBaseSeconds", 30L);
+        ReflectionTestUtils.setField(service, "backoffCapSeconds", 3600L);
+    }
+
+    @Test
+    void noRouteIsNotOverwrittenByFollowingAck() {
+        String sent = AnswerDeliveryService.callbackKey(row);
+        service.onUnroutable(7L, sent, "312 NO_ROUTE");
+        String error = row.getLastError();
+        LocalDateTime retryAt = row.getNextRetryAt();
+        service.onBrokerConfirm(7L, sent, true, null);
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED);
+        assertThat(row.getLastError()).isEqualTo(error).contains("NO_ROUTE");
+        assertThat(row.getRetryCount()).isEqualTo(1);
+        assertThat(row.getNextRetryAt()).isEqualTo(retryAt).isAfter(LocalDateTime.now());
+        assertThat(row.getBrokerAcceptedAt()).isNull();
+    }
+
+    @Test
+    void returnAlsoWinsIfAckCallbackAcquiredTheRowFirst() {
+        String sent = AnswerDeliveryService.callbackKey(row);
+        service.onBrokerConfirm(7L, sent, true, null);
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_BROKER_ACCEPTED);
+        service.onUnroutable(7L, sent, "312 NO_ROUTE");
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED);
+        assertThat(row.getBrokerAcceptedAt()).isNull();
+        assertThat(row.getLastError()).contains("NO_ROUTE");
+        assertThat(row.getRetryCount()).isEqualTo(1);
+    }
+
+    @Test
+    void duplicateReturnAndNackDoNotDoubleChargeSameFailedAttempt() {
+        String sent = AnswerDeliveryService.callbackKey(row);
+        service.onUnroutable(7L, sent, "312 NO_ROUTE");
+        service.onUnroutable(7L, sent, "duplicate");
+        service.onBrokerConfirm(7L, sent, false, "late nack");
+        assertThat(row.getRetryCount()).isEqualTo(1);
+        assertThat(row.getLastError()).contains("NO_ROUTE").doesNotContain("duplicate", "late nack");
+    }
+
+    @Test
+    void onlyCurrentRetryCanClearFailureAfterSuccessfulRoute() {
+        String first = AnswerDeliveryService.callbackKey(row);
+        service.onUnroutable(7L, first, "312 NO_ROUTE");
+        String retry = AnswerDeliveryService.callbackKey(row);
+        assertThat(retry).isNotEqualTo(first);
+        service.onBrokerConfirm(7L, retry, true, null);
+        service.onUnroutable(7L, first, "late old return");
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_BROKER_ACCEPTED);
+        assertThat(row.getLastError()).isNull();
+        assertThat(row.getNextRetryAt()).isNull();
+        assertThat(row.getBrokerAcceptedAt()).isNotNull();
+        assertThat(row.getConsumerAppliedAt()).isNull();
+    }
+
+    @Test
+    void unroutableAttemptsBackOffThenStopAtExistingBudget() {
+        service.onUnroutable(7L, AnswerDeliveryService.callbackKey(row), "NO_ROUTE #1");
+        LocalDateTime firstRetry = row.getNextRetryAt();
+        service.onUnroutable(7L, AnswerDeliveryService.callbackKey(row), "NO_ROUTE #2");
+        assertThat(row.getNextRetryAt()).isAfter(firstRetry);
+        String last = AnswerDeliveryService.callbackKey(row);
+        service.onUnroutable(7L, last, "NO_ROUTE #3");
+        service.onBrokerConfirm(7L, last, true, null);
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_DEAD);
+        assertThat(row.getRetryCount()).isEqualTo(3);
+        assertThat(row.getNextRetryAt()).isNull();
+        assertThat(row.getLastError()).contains("NO_ROUTE #3");
+    }
+
+    @Test
+    void oldReplayCannotConfirmOrFailNewAuthorizedOperation() {
+        row.setLastReplayOperationId("old:operation");
+        String old = AnswerDeliveryService.callbackKey(row);
+        row.setLastReplayOperationId("new:operation");
+        service.onBrokerConfirm(7L, old, true, null);
+        service.onUnroutable(7L, old, "old operation NO_ROUTE");
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_PENDING);
+        assertThat(row.getRetryCount()).isZero();
+    }
+
+    @Test
+    void multibyteReplayOperationKeepsCorrelationWithinAmqpShortStringLimit() {
+        StringBuilder operation = new StringBuilder();
+        for (int i = 0; i < 64; i++) operation.append("重");
+        row.setLastReplayOperationId(operation.toString());
+        String key = AnswerDeliveryService.callbackKey(row);
+        assertThat(key.getBytes(java.nio.charset.StandardCharsets.UTF_8).length).isLessThan(255);
+        assertThat(key).startsWith("7:0:");
+    }
+
+    @Test
+    void revokedGrantCannotBeReopenedByBrokerCallbacks() {
+        String sent = AnswerDeliveryService.callbackKey(row);
+        row.setState(AnswerDelivery.STATE_GRANT_REVOKED);
+        row.setLastError("revoked");
+        service.onBrokerConfirm(7L, sent, true, null);
+        service.onUnroutable(7L, sent, "NO_ROUTE");
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_GRANT_REVOKED);
+        assertThat(row.getLastError()).isEqualTo("revoked");
+    }
+
+    @Test
+    void productionSendAndCallbacksCarryTheCapturedRetryGeneration() {
+        AnswerDeliveryRepository rows = mock(AnswerDeliveryRepository.class);
+        when(rows.findById(7L)).thenReturn(Optional.of(row));
+        row.setFormUuid("form");
+        row.setConsumerSchoolId(27L);
+        row.setGrantVersion(3);
+        row.setSnapshotJson("{}");
+        when(rows.findDue(any(), any())).thenReturn(Collections.singletonList(row));
+        FormConsumerAccessService access = mock(FormConsumerAccessService.class);
+        SubscriptionGrant grant = new SubscriptionGrant();
+        grant.setConsumerAppKey("consumer");
+        grant.setGrantVersion(3);
+        grant.setPushAllowed(true);
+        when(access.listEnabledSubscriptions("form", 27L)).thenReturn(Collections.singletonList(grant));
+        TransactionTemplate tx = mock(TransactionTemplate.class);
+        doAnswer(call -> {
+            Consumer<TransactionStatus> action = call.getArgument(0);
+            action.accept(null);
+            return null;
+        }).when(tx).executeWithoutResult(any());
+        RabbitTemplate template = mock(RabbitTemplate.class);
+        @SuppressWarnings("unchecked")
+        ObjectProvider<RabbitTemplate> templates = mock(ObjectProvider.class);
+        when(templates.getIfAvailable()).thenReturn(template);
+        AnswerDeliveryService sender = new AnswerDeliveryService(rows, null, null, access,
+                new ObjectMapper(), tx, templates);
+        ReflectionTestUtils.setField(sender, "sourceAppKey", "100021");
+        row.setSourceAppKey("100021");
+        ReflectionTestUtils.setField(sender, "signingAppKey", "100021");
+        ReflectionTestUtils.setField(sender, "applicationCode", "form-center");
+        com.yunhai.platform.openapi.base.OpenApiAppClient registry = mock(com.yunhai.platform.openapi.base.OpenApiAppClient.class);
+        when(registry.getAppByCode("form-center")).thenReturn(new ObjectMapper().createObjectNode()
+                .put("id", 700021).put("appKey", 100021).put("appCode", "form-center"));
+        ReflectionTestUtils.setField(sender, "appRegistryClient", registry);
+        ReflectionTestUtils.setField(sender, "host", "broker");
+        ReflectionTestUtils.setField(sender, "confirmType", "correlated");
+        ReflectionTestUtils.setField(sender, "maxRetries", 3);
+        ReflectionTestUtils.setField(sender, "backoffBaseSeconds", 30L);
+        ReflectionTestUtils.setField(sender, "backoffCapSeconds", 3600L);
+        ReflectionTestUtils.invokeMethod(sender, "registerCallbacks");
+        assertThat(sender.sendDueBatch()).isEqualTo(1);
+        ArgumentCaptor<Message> message = ArgumentCaptor.forClass(Message.class);
+        ArgumentCaptor<CorrelationData> correlation = ArgumentCaptor.forClass(CorrelationData.class);
+        verify(template).send(anyString(), anyString(), message.capture(), correlation.capture());
+        assertThat(correlation.getValue().getId()).isEqualTo("7:0:");
+        assertThat(message.getValue().getMessageProperties().getHeaders().get("fc-delivery-attempt"))
+                .isEqualTo(correlation.getValue().getId());
+        ArgumentCaptor<RabbitTemplate.ReturnsCallback> returns = ArgumentCaptor.forClass(RabbitTemplate.ReturnsCallback.class);
+        ArgumentCaptor<RabbitTemplate.ConfirmCallback> confirms = ArgumentCaptor.forClass(RabbitTemplate.ConfirmCallback.class);
+        verify(template).setReturnsCallback(returns.capture());
+        verify(template).setConfirmCallback(confirms.capture());
+        returns.getValue().returnedMessage(new ReturnedMessage(message.getValue(), 312, "NO_ROUTE", "exchange", "route"));
+        confirms.getValue().confirm(correlation.getValue(), true, null);
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED);
+        assertThat(row.getLastError()).contains("NO_ROUTE");
+    }
+
+    @Test
+    void callbackReadsUseTheSameDatabaseWriteLock() throws Exception {
+        Lock lock = AnswerDeliveryRepository.class.getMethod("findById", Long.class).getAnnotation(Lock.class);
+        assertThat(lock).isNotNull();
+        assertThat(lock.value()).isEqualTo(LockModeType.PESSIMISTIC_WRITE);
+    }
+}

src/test/java/com/yunhai/formcenter/service/AnswerDeliveryMqContractTest.java

@@ -0,0 +1,239 @@
+package com.yunhai.formcenter.service;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.yunhai.formcenter.entity.AnswerDelivery;
+import com.yunhai.formcenter.model.AnswerDeliveryModels;
+import com.yunhai.formcenter.model.FormAccessModels.SubscriptionGrant;
+import com.yunhai.formcenter.repository.AnswerDeliveryRepository;
+import com.yunhai.platform.openapi.base.OpenApiAppClient;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.ReturnedMessage;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.time.LocalDateTime;
+import java.util.Collections;
+import java.util.Optional;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+/** Local state/serialization/wiring tests; no broker and no real database concurrency. */
+class AnswerDeliveryMqContractTest {
+    private final ObjectMapper mapper = new ObjectMapper();
+    private AnswerDelivery row;
+    private AnswerDeliveryService service;
+    private RabbitTemplate template;
+    private OpenApiAppClient registry;
+    private ObjectNode application;
+    private RabbitTemplate.ConfirmCallback confirm;
+    private RabbitTemplate.ReturnsCallback returns;
+
+    @BeforeEach
+    void setUp() {
+        row = new AnswerDelivery();
+        row.setId(9L); row.setEventId("stable-event");
+        row.setProtocolVersion(AnswerDeliveryModels.PROTOCOL_VERSION); row.setEventType("updated");
+        row.setSourceAppKey("100025"); row.setConsumerAppKey("100018"); row.setConsumerSchoolId(27L);
+        row.setGrantVersion(4); row.setFormUuid("form-uuid"); row.setFormVersion(1);
+        row.setAnswerUuid("answer-uuid"); row.setAnswerRevision(8L); row.setContentHash("hash");
+        row.setOccurredAt(LocalDateTime.of(2026, 10, 23, 10, 0));
+        row.setSnapshotJson("{\"schId\":27,\"appCode\":\"business-owner\",\"code\":\"leave\","
+                + "\"submitterId\":41,\"subjectId\":42,\"status\":\"processed\",\"values\":{\"reason\":\"leave\"},"
+                + "\"answerNo\":\"LEAVE-8\",\"createTime\":\"2026-10-23T10:00:00\","
+                + "\"updateTime\":\"2026-10-23T10:01:00\",\"intent\":{\"intentId\":\"intent-1\"}}");
+        AnswerDeliveryRepository rows = mock(AnswerDeliveryRepository.class);
+        when(rows.findById(9L)).thenReturn(Optional.of(row));
+        when(rows.findDue(any(), any())).thenReturn(Collections.singletonList(row));
+        FormConsumerAccessService access = mock(FormConsumerAccessService.class);
+        SubscriptionGrant grant = new SubscriptionGrant();
+        grant.setConsumerAppKey("100018"); grant.setGrantVersion(4); grant.setPushAllowed(true);
+        when(access.listEnabledSubscriptions("form-uuid", 27L)).thenReturn(Collections.singletonList(grant));
+        TransactionTemplate tx = mock(TransactionTemplate.class);
+        doAnswer(call -> { Consumer<TransactionStatus> action = call.getArgument(0); action.accept(null); return null; })
+                .when(tx).executeWithoutResult(any());
+        template = mock(RabbitTemplate.class);
+        @SuppressWarnings("unchecked") ObjectProvider<RabbitTemplate> provider = mock(ObjectProvider.class);
+        when(provider.getIfAvailable()).thenReturn(template);
+        service = new AnswerDeliveryService(rows, null, null, access, mapper, tx, provider);
+        registry = mock(OpenApiAppClient.class);
+        application = mapper.createObjectNode().put("id", 700025).put("appKey", 100025).put("appCode", "form-center");
+        when(registry.getAppByCode("form-center")).thenReturn(application);
+        ReflectionTestUtils.setField(service, "sourceAppKey", "100025");
+        ReflectionTestUtils.setField(service, "signingAppKey", "100025");
+        ReflectionTestUtils.setField(service, "applicationCode", "form-center");
+        ReflectionTestUtils.setField(service, "appRegistryClient", registry);
+        ReflectionTestUtils.setField(service, "host", "local-test-only");
+        ReflectionTestUtils.setField(service, "confirmType", "correlated");
+        ReflectionTestUtils.setField(service, "batchSize", 10);
+        ReflectionTestUtils.setField(service, "maxRetries", 3);
+        ReflectionTestUtils.setField(service, "backoffBaseSeconds", 30L);
+        ReflectionTestUtils.setField(service, "backoffCapSeconds", 3600L);
+        service.logState();
+        ArgumentCaptor<RabbitTemplate.ConfirmCallback> confirms = ArgumentCaptor.forClass(RabbitTemplate.ConfirmCallback.class);
+        ArgumentCaptor<RabbitTemplate.ReturnsCallback> returned = ArgumentCaptor.forClass(RabbitTemplate.ReturnsCallback.class);
+        verify(template).setConfirmCallback(confirms.capture()); verify(template).setReturnsCallback(returned.capture());
+        confirm = confirms.getValue(); returns = returned.getValue();
+    }
+
+    @Test void routingUsesConsumerCredentialAndSchoolNotBusinessIdOrAppCode() {
+        assertThat(AnswerDeliveryService.routingKey("100018", 27L)).isEqualTo("fc.snapshot.v2.100018.27");
+        assertThat(AnswerDeliveryService.routingKey("100008", 27L)).isEqualTo("fc.snapshot.v2.100008.27")
+                .isNotEqualTo("fc.snapshot.v2.100018.27");
+        assertThat(AnswerDeliveryService.routingKey("700025", 27L)).isNotEqualTo(AnswerDeliveryService.routingKey(row));
+        assertThatThrownBy(() -> AnswerDeliveryService.routingKey("100018.*", 27L)).isInstanceOf(IllegalArgumentException.class);
+        assertThatThrownBy(() -> AnswerDeliveryService.routingKey("100018", 0L)).isInstanceOf(IllegalArgumentException.class);
+    }
+
+    @Test void successfulWirePreservesAllCoordinatesAndAnswerData() throws Exception {
+        assertThat(service.sendDueBatch()).isEqualTo(1);
+        ArgumentCaptor<Message> message = ArgumentCaptor.forClass(Message.class);
+        verify(template).send(eq(AnswerDeliveryModels.EXCHANGE), eq("fc.snapshot.v2.100018.27"), message.capture(), any(CorrelationData.class));
+        JsonNode json = mapper.readTree(message.getValue().getBody());
+        assertThat(json.size()).isEqualTo(15);
+        assertThat(json.get("sourceAppKey").asText()).isEqualTo("100025");
+        assertThat(json.get("consumerAppKey").asText()).isEqualTo("100018");
+        assertThat(json.get("consumerSchoolId").asLong()).isEqualTo(27);
+        assertThat(json.get("eventId").asText()).isEqualTo("stable-event");
+        assertThat(json.get("formUuid").asText()).isEqualTo("form-uuid");
+        assertThat(json.get("formVersion").asInt()).isEqualTo(1);
+        assertThat(json.get("answerUuid").asText()).isEqualTo("answer-uuid");
+        assertThat(json.get("answerRevision").asLong()).isEqualTo(8);
+        assertThat(json.get("contentHash").asText()).isEqualTo("hash");
+        assertThat(json.get("snapshot")).isEqualTo(mapper.readTree(row.getSnapshotJson()));
+        assertThat(message.getValue().getMessageProperties().getMessageId()).isEqualTo("stable-event");
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_PENDING); // send is not a confirm
+        verify(registry).getAppByCode("form-center"); verify(registry, never()).getAppById(anyString());
+    }
+
+    @Test void applicationIdMayDifferFromSigningKey() {
+        assertThat(application.get("id").asText()).isNotEqualTo(application.get("appKey").asText());
+        assertThat(service.sendDueBatch()).isEqualTo(1);
+    }
+
+    @Test void sourceKeyIsNotTheApplicationPrimaryId() {
+        row.setSourceAppKey("700025");
+        assertThat(service.sendDueBatch()).isZero();
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED);
+        assertThat(row.getLastError()).contains("SOURCE_CREDENTIAL_MISMATCH");
+        verify(template, never()).send(anyString(), anyString(), any(Message.class), any(CorrelationData.class));
+    }
+
+    @Test void signingCredentialBelongingToAnotherApplicationIsRejected() {
+        application.put("appKey", 100099);
+        assertThat(service.sendDueBatch()).isZero();
+        assertThat(row.getLastError()).contains("SOURCE_APPLICATION_MISMATCH");
+    }
+
+    @Test void wrongApplicationCodeOrMissingBusinessIdCannotMasqueradeAsSource() {
+        application.put("appCode", "svc-hall");
+        assertThat(service.sendDueBatch()).isZero();
+        application.put("appCode", "form-center").remove("id");
+        assertThat(service.sendDueBatch()).isZero();
+        assertThat(row.getLastError()).contains("SOURCE_APPLICATION_MISMATCH");
+    }
+
+    @Test void registryFailureIsRetryableAndCannotBeConfirmedByTheFailedGeneration() {
+        String failed = AnswerDeliveryService.callbackKey(row);
+        when(registry.getAppByCode("form-center")).thenThrow(new IllegalStateException("registry unavailable"));
+        assertThat(service.sendDueBatch()).isZero();
+        confirm.confirm(new CorrelationData(failed), true, null);
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED);
+        assertThat(row.getNextRetryAt()).isNotNull();
+        assertThat(row.getLastError()).contains("registry unavailable");
+    }
+
+    @Test void returnedFactOnCorrelationPreventsSuccessEvenWithoutTheSeparateReturnCallback() {
+        assertThat(service.sendDueBatch()).isEqualTo(1);
+        ArgumentCaptor<Message> message = ArgumentCaptor.forClass(Message.class);
+        ArgumentCaptor<CorrelationData> key = ArgumentCaptor.forClass(CorrelationData.class);
+        verify(template).send(anyString(), anyString(), message.capture(), key.capture());
+        key.getValue().setReturned(new ReturnedMessage(message.getValue(), 312, "NO_ROUTE", AnswerDeliveryModels.EXCHANGE, "wrong.route"));
+        confirm.confirm(key.getValue(), true, null);
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED);
+        assertThat(row.getLastError()).contains("NO_ROUTE");
+        assertThat(row.getBrokerAcceptedAt()).isNull();
+    }
+
+    @Test void bindingMismatchReturnThenAckKeepsFailureAndRetry() {
+        assertThat(service.sendDueBatch()).isEqualTo(1);
+        ArgumentCaptor<Message> message = ArgumentCaptor.forClass(Message.class);
+        ArgumentCaptor<CorrelationData> key = ArgumentCaptor.forClass(CorrelationData.class);
+        verify(template).send(anyString(), anyString(), message.capture(), key.capture());
+        returns.returnedMessage(new ReturnedMessage(message.getValue(), 312, "NO_ROUTE", AnswerDeliveryModels.EXCHANGE, "fc.snapshot.v2.100018.27"));
+        confirm.confirm(key.getValue(), true, null);
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED);
+        assertThat(row.getLastError()).contains("NO_ROUTE");
+        assertThat(row.getRetryCount()).isEqualTo(1);
+        assertThat(row.getNextRetryAt()).isNotNull();
+    }
+
+    @Test void nackBeforeAckAndDuplicateNackCannotBecomeSuccess() {
+        CorrelationData key = new CorrelationData(AnswerDeliveryService.callbackKey(row));
+        confirm.confirm(key, false, "broker rejected"); confirm.confirm(key, true, null); confirm.confirm(key, false, "duplicate");
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED);
+        assertThat(row.getRetryCount()).isEqualTo(1);
+        assertThat(row.getLastError()).contains("broker rejected").doesNotContain("duplicate");
+    }
+
+    @Test void sameGenerationNackAfterAckOverridesOnlyTheBrokerState() {
+        CorrelationData key = new CorrelationData(AnswerDeliveryService.callbackKey(row));
+        confirm.confirm(key, true, null); confirm.confirm(key, false, "negative wins"); confirm.confirm(key, true, null);
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED);
+        assertThat(row.getBrokerAcceptedAt()).isNull();
+        assertThat(row.getConsumerAppliedAt()).isNull();
+        assertThat(row.getLastError()).contains("negative wins");
+    }
+
+    @Test void duplicateAckDoesNotRewriteTheAcceptedTimestamp() {
+        CorrelationData key = new CorrelationData(AnswerDeliveryService.callbackKey(row));
+        confirm.confirm(key, true, null); LocalDateTime first = row.getBrokerAcceptedAt(); confirm.confirm(key, true, null);
+        assertThat(row.getBrokerAcceptedAt()).isEqualTo(first); assertThat(row.getRetryCount()).isZero();
+    }
+
+    @Test void conversionFailureRetriesWithSameImmutableEventAndNewGeneration() throws Exception {
+        ObjectMapper failing = spy(new ObjectMapper());
+        doThrow(new JsonProcessingException("encode failure") {}).when(failing).writeValueAsString(any());
+        ReflectionTestUtils.setField(service, "objectMapper", failing);
+        String original = AnswerDeliveryService.callbackKey(row);
+        assertThat(service.sendDueBatch()).isZero();
+        assertThat(row.getLastError()).contains("encode failure");
+        confirm.confirm(new CorrelationData(original), true, null);
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED);
+        ReflectionTestUtils.setField(service, "objectMapper", mapper);
+        row.setNextRetryAt(LocalDateTime.now().minusSeconds(1));
+        assertThat(service.sendDueBatch()).isEqualTo(1);
+        ArgumentCaptor<CorrelationData> retry = ArgumentCaptor.forClass(CorrelationData.class);
+        verify(template).send(anyString(), anyString(), any(Message.class), retry.capture());
+        assertThat(retry.getValue().getId()).isNotEqualTo(original);
+        confirm.confirm(retry.getValue(), true, null);
+        assertThat(row.getEventId()).isEqualTo("stable-event");
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_BROKER_ACCEPTED);
+        assertThat(row.getLastError()).isNull(); assertThat(row.getConsumerAppliedAt()).isNull();
+    }
+
+    @Test void corruptSnapshotCannotReachBroker() {
+        row.setSnapshotJson("{broken"); assertThat(service.sendDueBatch()).isZero();
+        assertThat(row.getState()).isEqualTo(AnswerDelivery.STATE_FAILED);
+        verify(template, never()).send(anyString(), anyString(), any(Message.class), any(CorrelationData.class));
+    }
+
+    @Test void returnsAndMandatoryAreRequiredBeforeTheScannerCanSend() {
+        assertThat(service.sendEnabled()).isTrue();
+        ReflectionTestUtils.setField(service, "publisherReturns", false); assertThat(service.sendEnabled()).isFalse();
+        ReflectionTestUtils.setField(service, "publisherReturns", true);
+        ReflectionTestUtils.setField(service, "mandatory", false); assertThat(service.sendEnabled()).isFalse();
+    }
+}

src/test/java/com/yunhai/formcenter/service/AnswerDeliverySourceIdentityTest.java

@@ -0,0 +1,53 @@
+package com.yunhai.formcenter.service;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.env.YamlPropertySourceLoader;
+import org.springframework.core.env.MapPropertySource;
+import org.springframework.core.env.StandardEnvironment;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+
+/** B-05 只证明最终配置解析与错配拒绝；没有核实 160 的实际签名配置。 */
+class AnswerDeliverySourceIdentityTest {
+    @Test
+    void deliverySourceTracksFinalSigningPropertyInsteadOfSeparateEnvironmentFallback() throws Exception {
+        StandardEnvironment env = new StandardEnvironment();
+        new YamlPropertySourceLoader().load("application", new ClassPathResource("application.yml"))
+                .forEach(env.getPropertySources()::addLast);
+        env.getPropertySources().addFirst(new MapPropertySource("platform-final-signing-property",
+                Collections.singletonMap("yh.app-auth.app-key", "100008")));
+        assertThat(env.getProperty("fc.answer.delivery.source-app-key")).isEqualTo("100008");
+    }
+
+    @Test
+    void mismatchedExplicitSourceFailsEvenWithoutBroker() {
+        AnswerDeliveryService service = service("100021", "100008");
+        assertThatThrownBy(service::logState).isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining("source-app-key", "yh.app-auth.app-key");
+    }
+
+    @Test
+    void matchingFinalSigningIdentityDoesNotNeedBroker() {
+        assertThatCode(service("100008", "100008")::logState).doesNotThrowAnyException();
+    }
+
+    @Test
+    void missingSigningIdentityFailsClosed() {
+        assertThatThrownBy(service("100008", "")::logState).isInstanceOf(IllegalStateException.class);
+    }
+
+    private AnswerDeliveryService service(String source, String signing) {
+        AnswerDeliveryService service = new AnswerDeliveryService(null, null, null, null, null, null, null);
+        ReflectionTestUtils.setField(service, "sourceAppKey", source);
+        ReflectionTestUtils.setField(service, "signingAppKey", signing);
+        ReflectionTestUtils.setField(service, "host", "");
+        return service;
+    }
+}

src/test/java/com/yunhai/formcenter/service/FcEventConverterContractTest.java

@@ -0,0 +1,37 @@
+package com.yunhai.formcenter.service;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
+import org.springframework.amqp.support.converter.SimpleMessageConverter;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** N-1 legacy-converter evidence; production now sends explicit JSON bytes. No broker acceptance claim. */
+class FcEventConverterContractTest {
+    @Test
+    void defaultRabbitTemplateCannotEncodeTheCurrentNonSerializableEnvelope() {
+        RabbitTemplate template = new RabbitTemplate();
+        assertThat(template.getMessageConverter()).isInstanceOf(SimpleMessageConverter.class);
+        assertThatThrownBy(() -> template.getMessageConverter()
+                .toMessage(new FcEventPublisher.Envelope(), new MessageProperties()))
+                .isInstanceOf(IllegalArgumentException.class).hasMessageContaining("Serializable");
+    }
+
+    @Test
+    void jsonConverterIsASeparatePossibleFixAndKeepsClosedTenFields() throws Exception {
+        ObjectMapper mapper = new ObjectMapper();
+        FcEventPublisher.Envelope envelope = new FcEventPublisher.Envelope();
+        envelope.setType("fc.answer.created");
+        Message encoded = new Jackson2JsonMessageConverter(mapper).toMessage(envelope, new MessageProperties());
+        JsonNode json = mapper.readTree(encoded.getBody());
+        assertThat(json.size()).isEqualTo(10);
+        assertThat(json.get("type").asText()).isEqualTo("fc.answer.created");
+        assertThat(encoded.getMessageProperties().getContentType()).isEqualTo(MessageProperties.CONTENT_TYPE_JSON);
+    }
+}

src/test/java/com/yunhai/formcenter/service/FcEventJsonPublishTest.java

@@ -0,0 +1,127 @@
+package com.yunhai.formcenter.service;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.yunhai.formcenter.entity.Answer;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.support.converter.SimpleMessageConverter;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+/** Real JSON encoding and template boundary, no RabbitMQ connection. v1 remains best effort. */
+class FcEventJsonPublishTest {
+    private FcEventPublisher publisher;
+    private RabbitTemplate template;
+    private ObjectProvider<RabbitTemplate> provider;
+    private Answer answer;
+    private final ObjectMapper mapper = new ObjectMapper();
+
+    @BeforeEach
+    @SuppressWarnings("unchecked")
+    void setUp() {
+        template = spy(new RabbitTemplate());
+        doReturn(null).when(template).execute(any());
+        doNothing().when(template).send(anyString(), anyString(), any(Message.class));
+        provider = mock(ObjectProvider.class); when(provider.getIfAvailable()).thenReturn(template);
+        publisher = new FcEventPublisher(provider);
+        ReflectionTestUtils.setField(publisher, "host", "local-test-only");
+        ReflectionTestUtils.setField(publisher, "eventsEnabled", true);
+        answer = new Answer(); answer.setUuid("answer-uuid"); answer.setFormUuid("form-uuid");
+        answer.setSchId(27L); answer.setAppCode("svc-hall"); answer.setCode("leave");
+        answer.setVersion(1); answer.setAnswerNo("LEAVE-1"); answer.setValuesJson("{\"secret\":\"must-not-broadcast\"}");
+    }
+
+    @Test void hostInjectionDoesNotEnableTheFixedNotificationChannel() {
+        FcEventPublisher disabled = new FcEventPublisher(provider);
+        ReflectionTestUtils.setField(disabled, "host", "platform-host");
+        assertThat(disabled.enabled()).isFalse(); disabled.answerCreated(answer);
+        verifyNoInteractions(provider);
+    }
+
+    @Test void actualSendWorksWithDefaultConverterAndPreservesTheTenFieldContract() throws Exception {
+        publisher.answerCreated(answer);
+        ArgumentCaptor<Message> wire = ArgumentCaptor.forClass(Message.class);
+        verify(template).send(eq(FcEventPublisher.EXCHANGE), eq("fc.answer.svc-hall.created"), wire.capture());
+        assertThat(template.getMessageConverter()).isInstanceOf(SimpleMessageConverter.class);
+        JsonNode json = mapper.readTree(wire.getValue().getBody());
+        List<String> keys = new ArrayList<>(); json.fieldNames().forEachRemaining(keys::add);
+        assertThat(keys).containsExactlyInAnyOrder("event_id", "type", "occurred_at", "app_code", "sch_id",
+                "form_uuid", "form_code", "answer_uuid", "version", "answer_no");
+        assertThat(json.get("event_id").asText()).isNotEmpty();
+        assertThat(json.get("type").asText()).isEqualTo("fc.answer.created");
+        assertThat(json.get("occurred_at").asText()).contains("+08:00");
+        assertThat(json.get("app_code").asText()).isEqualTo("svc-hall");
+        assertThat(json.get("sch_id").asLong()).isEqualTo(27);
+        assertThat(json.get("form_uuid").asText()).isEqualTo("form-uuid");
+        assertThat(json.get("form_code").asText()).isEqualTo("leave");
+        assertThat(json.get("answer_uuid").asText()).isEqualTo("answer-uuid");
+        assertThat(json.get("version").asInt()).isEqualTo(1);
+        assertThat(json.get("answer_no").asText()).isEqualTo("LEAVE-1");
+        assertThat(json.toString()).doesNotContain("must-not-broadcast", "values", "appSecret");
+        assertThat(wire.getValue().getMessageProperties().getMessageId()).isEqualTo(json.get("event_id").asText());
+        assertThat(wire.getValue().getMessageProperties().getContentType()).isEqualTo(MessageProperties.CONTENT_TYPE_JSON);
+        assertThat(wire.getValue().getMessageProperties().getContentEncoding()).isEqualTo("UTF-8");
+    }
+
+    @Test void conversionFailureCannotReachTheTemplateOrBreakTheAnswerWrite() throws Exception {
+        ObjectMapper broken = spy(new ObjectMapper());
+        doThrow(new JsonProcessingException("encode failure") {}).when(broken).writeValueAsBytes(any());
+        ReflectionTestUtils.setField(publisher, "eventMapper", broken);
+        assertThatCode(() -> publisher.answerCreated(answer)).doesNotThrowAnyException();
+        verifyNoInteractions(provider); assertThat(answer.getUuid()).isEqualTo("answer-uuid");
+        assertThat(answer.getVersion()).isEqualTo(1);
+    }
+
+    @Test void aFailedEncodeDoesNotPermanentlyDisableLaterNotifications() throws Exception {
+        ObjectMapper recovering = spy(new ObjectMapper());
+        doThrow(new JsonProcessingException("first encode failed") {}).doCallRealMethod()
+                .when(recovering).writeValueAsBytes(any());
+        ReflectionTestUtils.setField(publisher, "eventMapper", recovering);
+        publisher.answerUpdated(answer); publisher.answerUpdated(answer);
+        verify(template, times(1)).send(eq(FcEventPublisher.EXCHANGE), eq("fc.answer.svc-hall.updated"), any(Message.class));
+    }
+
+    @Test void afterCommitUsesTheFrozenAnswerCoordinates() throws Exception {
+        TransactionSynchronizationManager.initSynchronization();
+        try {
+            publisher.answerCreated(answer);
+            verify(template, never()).send(anyString(), anyString(), any(Message.class));
+            answer.setUuid("later-answer"); answer.setVersion(99); answer.setAnswerNo("later-number");
+            TransactionSynchronizationManager.getSynchronizations().forEach(s -> s.afterCommit());
+            ArgumentCaptor<Message> wire = ArgumentCaptor.forClass(Message.class);
+            verify(template).send(anyString(), anyString(), wire.capture());
+            JsonNode json = mapper.readTree(wire.getValue().getBody());
+            assertThat(json.get("answer_uuid").asText()).isEqualTo("answer-uuid");
+            assertThat(json.get("version").asInt()).isEqualTo(1);
+            assertThat(json.get("answer_no").asText()).isEqualTo("LEAVE-1");
+        } finally { TransactionSynchronizationManager.clearSynchronization(); }
+    }
+
+    @Test void failedDeclarationIsRetriedByTheNextNotification() {
+        doThrow(new IllegalStateException("declare failed")).doReturn(null).when(template).execute(any());
+        assertThatCode(() -> publisher.answerCreated(answer)).doesNotThrowAnyException();
+        publisher.answerCreated(answer);
+        verify(template, times(2)).execute(any()); verify(template).send(anyString(), anyString(), any(Message.class));
+    }
+
+    @Test void sendExceptionRemainsBestEffortAndNeverClaimsDelivery() {
+        doThrow(new IllegalStateException("send failed")).when(template).send(anyString(), anyString(), any(Message.class));
+        assertThatCode(() -> publisher.answerTombstoned(answer)).doesNotThrowAnyException();
+        verify(template, times(1)).send(anyString(), eq("fc.answer.svc-hall.tombstoned"), any(Message.class));
+        assertThat(answer.getUuid()).isEqualTo("answer-uuid");
+    }
+}

src/test/java/com/yunhai/formcenter/service/FcEventPublisherTest.java

@@ -7,6 +7,7 @@ import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentMatchers;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.core.Message;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.test.util.ReflectionTestUtils;
 import org.springframework.transaction.support.TransactionSynchronization;
@@ -71,6 +72,7 @@ class FcEventPublisherTest {
         @SuppressWarnings("unchecked")
         ObjectProvider<RabbitTemplate> provider = mock(ObjectProvider.class);
         FcEventPublisher publisher = new FcEventPublisher(provider);
+        ReflectionTestUtils.setField(publisher, "eventsEnabled", true);
         ReflectionTestUtils.setField(publisher, "host", "");
 
         assertThat(publisher.enabled()).isFalse();
@@ -88,21 +90,22 @@ class FcEventPublisherTest {
         ObjectProvider<RabbitTemplate> provider = mock(ObjectProvider.class);
         when(provider.getIfAvailable()).thenReturn(template);
         FcEventPublisher publisher = new FcEventPublisher(provider);
+        ReflectionTestUtils.setField(publisher, "eventsEnabled", true);
         ReflectionTestUtils.setField(publisher, "host", "localhost");
 
         TransactionSynchronizationManager.initSynchronization();
         try {
             publisher.answerCreated(answer());
-            verify(template, never()).convertAndSend(ArgumentMatchers.anyString(), ArgumentMatchers.anyString(),
-                    ArgumentMatchers.any(Object.class));
+            verify(template, never()).send(ArgumentMatchers.anyString(), ArgumentMatchers.anyString(),
+                    ArgumentMatchers.any(Message.class));
 
             for (TransactionSynchronization sync
                     : TransactionSynchronizationManager.getSynchronizations()) {
                 sync.afterCommit();
             }
-            verify(template).convertAndSend(ArgumentMatchers.eq(FcEventPublisher.EXCHANGE),
+            verify(template).send(ArgumentMatchers.eq(FcEventPublisher.EXCHANGE),
                     ArgumentMatchers.eq("fc.answer.demo.created"),
-                    ArgumentMatchers.any(FcEventPublisher.Envelope.class));
+                    ArgumentMatchers.any(Message.class));
         } finally {
             TransactionSynchronizationManager.clearSynchronization();
         }
@@ -113,12 +116,13 @@ class FcEventPublisherTest {
     void sendFailureIsSwallowed() {
         RabbitTemplate template = mock(RabbitTemplate.class);
         doThrow(new IllegalStateException("broker down")).when(template)
-                .convertAndSend(ArgumentMatchers.anyString(), ArgumentMatchers.anyString(),
-                        ArgumentMatchers.any(Object.class));
+                .send(ArgumentMatchers.anyString(), ArgumentMatchers.anyString(),
+                        ArgumentMatchers.any(Message.class));
         @SuppressWarnings("unchecked")
         ObjectProvider<RabbitTemplate> provider = mock(ObjectProvider.class);
         when(provider.getIfAvailable()).thenReturn(template);
         FcEventPublisher publisher = new FcEventPublisher(provider);
+        ReflectionTestUtils.setField(publisher, "eventsEnabled", true);
         ReflectionTestUtils.setField(publisher, "host", "localhost");
 
         // 无事务上下文时就地发布（同步器未激活的那条分支）