release: FL 2.0 候选（C01 记账边界+P2-2 认领协议+D10 变量契约+D11 资格终审+D09 激活代次+T01/T03/T04/T05）

按《FL最终代码审计与上线阻断清单》194 项已审计工作树形成候选，基座 = origin/main 11e0778。

- 源码/测试/资源：沿用已审计工作树内容（96 项相对远端为新增或更新），含未跟踪实现
  CallbackFactStore、CallbackSourcePolicy、VariableContract、FormalApplyCommandService、
  FlowActorContextGate 等；C01 续验夹具修正一并入内。
- CI：保留远端 11e0778 的门禁强化（CI_SOURCE_SHA 核对、容器回收验证、空 schema 反向门禁、
  main ref 才推镜像）；ci/flow-test-schema.sql 按当前实体重新生成（14→20 表，补 6 新表与
  6 新列，回调日志复合唯一键 uk_flw_log_event_reg），workflow 表数断言同步 14→20。
- 依据：docs/tasks/evidence/FL-final-audit-20261027（194 项哈希/版本核对）。

src/main/java/com/yunhai/flowcenter/callback/CallbackDispatcher.java

@@ -13,6 +13,7 @@ import com.yunhai.flowcenter.repository.FlwDefRepository;
 import com.yunhai.flowcenter.repository.FlwInstanceRepository;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.dao.DuplicateKeyException;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.event.TransactionPhase;
@@ -28,10 +29,22 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
 /**
- * 回调分发器（概览 §8.4 可靠性设计）：引擎 AFTER_COMMIT 事件 → 匹配 flw_callback_reg
- * （flowGuid + 订阅事件）→ 落 flw_callback_log(status=0) → 异步首推。
+ * 回调分发器（概览 §8.4 可靠性设计；2026-10-25 R2 返修）：引擎 AFTER_COMMIT 事件 →
+ * 匹配 flw_callback_reg（flowGuid + 订阅事件）→ 落 flw_callback_log(status=0) → 异步首推。
  * 落库成功即视为主业务完成，推送失败不影响审批（异步解耦是特性）。
  * @Scheduled(fixedDelay=30s) 扫 (status=0, next_retry_time<=now) 重推；死信留管理端手工重推。
+ *
+ * <p>R2（Codex D04）投递可靠性口径：
+ * <ul>
+ *   <li><b>统一冻结信封</b>——事实行在状态事务内冻结完整投递体（含 eventId/eventSeq/flowGuid）；
+ *       正常分发优先复用冻结体，崩溃恢复分发原样复用同一冻结体，两条路径消息体同构；恢复体
+ *       对 R2 前存量事实行按列值幂等补齐 eventId/eventSeq/flowGuid 后投出。</li>
+ *   <li><b>未可靠建行不记账</b>——任一匹配订阅方的投递行建行失败（含持久化异常）时，事实行
+ *       不标记 dispatched：正常链不标记（恢复扫描补建），恢复链保持 dispatched=0 等下轮重扫；
+ *       异步首推的提交失败不回退建行结果（行已在，重试扫描接手）。</li>
+ *   <li><b>并发幂等靠数据库唯一键</b>——(event_id, reg_id) 复合唯一约束兜底 exists→insert 竞态，
+ *       DuplicateKeyException 一律视为"已建行"幂等成功。</li>
+ * </ul>
  */
 @Slf4j
 @Component
@@ -48,6 +61,17 @@ public class CallbackDispatcher {
     private final CallbackPushService pushService;
     private final CallbackProperties properties;
     private final ObjectMapper objectMapper;
+    private final com.yunhai.flowcenter.repository.FlwStateEventRepository stateEventRepository;
+    private final CallbackFactStore factStore;
+
+    @org.springframework.beans.factory.annotation.Autowired
+    private CallbackSourcePolicy sourcePolicy;
+
+    private boolean self(FlwCallbackReg reg, FlwInstance instance) {
+        if (instance.getSourcePrincipalId() == null) return true; // legacy cutover is a separate migration
+        if (sourcePolicy == null) throw new IllegalStateException("Missing formal callback source policy");
+        return sourcePolicy.matches(reg, instance);
+    }
 
     private volatile ExecutorService pushExecutor;
 
@@ -70,14 +94,129 @@ public class CallbackDispatcher {
         }
         try {
             dispatch(event);
+            markFactDispatched(event);
         } catch (Exception e) {
-            // 分发失败只影响回调（对账兜底会追平），绝不能反噬引擎主流程
-            log.error("[flow-center] 回调分发异常：instance={} event={} {}",
+            // 分发/建行失败只影响回调（对账兜底会追平），绝不能反噬引擎主流程；
+            // R2：不标记 dispatched——恢复扫描稍后按事实行补建缺失的投递行
+            log.error("[flow-center] 回调分发异常（事实保持待补分发）：instance={} event={} {}",
                     event.getInstanceId(), event.getType().getCode(), e.toString(), e);
         }
     }
 
-    private void dispatch(FlowEvent event) {
+    /** T05：事实行分发完成记账（崩溃窗口恢复扫描据此区分"已分发"与"待补分发"） */
+    private void markFactDispatched(FlowEvent event) {
+        if (event.getStableEventId() == null) {
+            return; // 无事实行（存量发布路径/记录器缺席），维持旧行为
+        }
+        try {
+            // Cross the separate bean's proxy: the completed business transaction cannot commit this write.
+            factStore.markDispatched(event.getStableEventId());
+        } catch (Exception e) {
+            log.warn("[flow-center] 事件事实分发记账失败（重扫会幂等补）：eventId={} {}",
+                    event.getStableEventId(), e.toString());
+        }
+    }
+
+    /**
+     * T05 崩溃窗口恢复扫描：状态事务已提交但分发未及执行（进程被杀/分发异常）的事实行，
+     * 过缓冲期后按存量事实补分发——幂等（同 eventId 已建投递行则跳过），不重放旧轮身份
+     * （载荷即事实快照）。R2：任一订阅方建行失败则保持 dispatched=0，下轮重扫补齐。
+     */
+    @Scheduled(fixedDelay = 60_000)
+    public void sweepUndispatchedFacts() {
+        try {
+            java.util.List<com.yunhai.flowcenter.entity.FlwStateEvent> due =
+                    stateEventRepository.findTop50ByDispatchedAndCreateTimeLessThanEqualOrderByCreateTimeAsc(
+                            0, java.time.LocalDateTime.now().minusSeconds(60));
+            for (com.yunhai.flowcenter.entity.FlwStateEvent fact : due) {
+                executor().submit(() -> dispatchFact(fact));
+            }
+        } catch (Exception e) {
+            log.warn("[flow-center] 事件事实补分发扫描异常：{}", e.toString());
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    private void dispatchFact(com.yunhai.flowcenter.entity.FlwStateEvent fact) {
+        try {
+            FlwInstance instance = instanceRepository.findById(fact.getInstanceId()).orElse(null);
+            if (instance == null) {
+                return;
+            }
+            FlwDef def = defRepository.findById(instance.getDefId()).orElse(null);
+            if (def == null) {
+                return;
+            }
+            Map<String, Object> stored = objectMapper.readValue(fact.getPayloadJson(), Map.class);
+            java.util.List<FlwCallbackReg> regs = regRepository.findBySchIdAndFlowGuidAndStatus(
+                    instance.getSchId(), def.getFlowGuid(), 0);
+            boolean allDelivered = true;
+            for (FlwCallbackReg reg : regs) {
+                if (!self(reg, instance)) continue;
+                if (!subscribed(reg.getEventTypes(), fact.getEventType())) {
+                    continue;
+                }
+                if (logRepository.existsByEventIdAndRegId(fact.getEventId(), reg.getId())) {
+                    continue; // 幂等：崩溃前已建的投递行不重建
+                }
+                try {
+                    persistFactDelivery(reg, instance, def, fact, stored);
+                } catch (DuplicateKeyException dup) {
+                    // uk(event_id, reg_id) 并发兜底：另一路径已建行，视为已投递
+                } catch (Exception e) {
+                    allDelivered = false;
+                    log.error("[flow-center] 事件事实补分发建行失败（保留待重扫）：eventId={} regId={} {}",
+                            fact.getEventId(), reg.getId(), e.toString());
+                }
+            }
+            if (!allDelivered) {
+                return; // R2：未全部可靠建行，不标记 dispatched
+            }
+            fact.setDispatched(1);
+            stateEventRepository.save(fact);
+        } catch (Exception e) {
+            log.warn("[flow-center] 事件事实补分发失败（下轮重扫）：eventId={} {}", fact.getEventId(), e.toString());
+        }
+    }
+
+    /**
+     * 补分发建行：载荷以事实冻结体为准（R2 冻结信封原样复用，与正常链同构）；
+     * R2 前存量事实行缺 eventId/eventSeq/flowGuid 时按列值幂等补齐。异步首推同正常链。
+     */
+    private void persistFactDelivery(FlwCallbackReg reg, FlwInstance instance, FlwDef def,
+                                     com.yunhai.flowcenter.entity.FlwStateEvent fact,
+                                     Map<String, Object> stored) throws Exception {
+        String payloadJson;
+        if (stored.containsKey("eventId") && stored.containsKey("flowGuid")) {
+            // R2 冻结体：完整信封已含全部投递字段，原样复用（不可变）
+            payloadJson = fact.getPayloadJson();
+        } else {
+            // R2 前存量事实行：信封字段缺失，按事实列值+现查 flowGuid 幂等修复
+            Map<String, Object> repaired = new HashMap<>(stored);
+            repaired.put("eventId", fact.getEventId());
+            if (fact.getEventSeq() != null) {
+                repaired.put("eventSeq", fact.getEventSeq());
+            }
+            repaired.put("flowGuid", def.getFlowGuid());
+            payloadJson = objectMapper.writeValueAsString(repaired);
+        }
+        FlwCallbackLog callbackLog = new FlwCallbackLog();
+        callbackLog.setSchId(instance.getSchId());
+        callbackLog.setRegId(reg.getId());
+        callbackLog.setEventId(fact.getEventId());
+        callbackLog.setEvent(fact.getEventType());
+        callbackLog.setInstanceId(instance.getId());
+        callbackLog.setBizType(instance.getBizType());
+        callbackLog.setBizId(instance.getBizId());
+        callbackLog.setPayloadJson(payloadJson);
+        callbackLog.setStatus(FlwEnums.CallbackLogStatus.PENDING.getCode());
+        callbackLog.setRetryCount(0);
+        callbackLog.setNextRetryTime(LocalDateTime.now());
+        Long logId = logStore.persist(callbackLog);
+        submitPush(logId);
+    }
+
+    private void dispatch(FlowEvent event) throws Exception {
         FlwInstance instance = instanceRepository.findById(event.getInstanceId()).orElse(null);
         if (instance == null) {
             return;
@@ -90,6 +229,7 @@ public class CallbackDispatcher {
         List<FlwCallbackReg> regs = regRepository.findBySchIdAndFlowGuidAndStatus(
                 instance.getSchId(), def.getFlowGuid(), 0);
         for (FlwCallbackReg reg : regs) {
+                if (!self(reg, instance)) continue;
             if (!subscribed(reg.getEventTypes(), eventCode)) {
                 continue;
             }
@@ -97,58 +237,87 @@ public class CallbackDispatcher {
         }
     }
 
-    private void persistAndPush(FlwCallbackReg reg, FlwInstance instance, FlwDef def, FlowEvent event) {
+    private void persistAndPush(FlwCallbackReg reg, FlwInstance instance, FlwDef def, FlowEvent event)
+            throws Exception {
+        // T05：eventId 用事实行稳定身份（全部订阅方/重试复用）；事实行缺席（存量路径）才现场生成
+        String eventId = event.getStableEventId() != null ? event.getStableEventId()
+                : UUID.randomUUID().toString().replace("-", "").toLowerCase();
+        if (logRepository.existsByEventIdAndRegId(eventId, reg.getId())) {
+            return; // 幂等：补分发与正常分发竞争时不重复建投递行
+        }
+        // R2：优先复用事实冻结信封（与崩溃恢复链同构、消息体恒含 eventId/eventSeq）；
+        // 冻结体缺席（存量路径/记录器缺席）才现场构建
+        String payloadJson = event.getStablePayloadJson() != null
+                ? event.getStablePayloadJson()
+                : objectMapper.writeValueAsString(legacyEnvelope(event, instance, def, eventId));
+        FlwCallbackLog callbackLog = new FlwCallbackLog();
+        callbackLog.setSchId(instance.getSchId());
+        callbackLog.setRegId(reg.getId());
+        callbackLog.setEventId(eventId);
+        callbackLog.setEvent(event.getType().getCode());
+        callbackLog.setInstanceId(instance.getId());
+        callbackLog.setBizType(instance.getBizType());
+        callbackLog.setBizId(instance.getBizId());
+        callbackLog.setPayloadJson(payloadJson);
+        callbackLog.setStatus(FlwEnums.CallbackLogStatus.PENDING.getCode());
+        callbackLog.setRetryCount(0);
+        callbackLog.setNextRetryTime(LocalDateTime.now());
+        // REQUIRES_NEW 落库（afterCommit 阶段原事务已完结，REQUIRED 会挂空事务）；
+        // R2：建行失败（含并发撞 uk）向外传播——调用方据此不标记事实已分发/保持待重扫
+        Long logId;
         try {
-            String eventId = UUID.randomUUID().toString().replace("-", "").toLowerCase();
-            Map<String, Object> payload = new HashMap<>();
-            payload.put("eventId", eventId);
-            payload.put("event", event.getType().getCode());
-            payload.put("schId", instance.getSchId());
-            payload.put("flowGuid", def.getFlowGuid());
-            payload.put("instanceId", instance.getId());
-            payload.put("instanceVersion", instance.getVersion());
-            payload.put("bizType", instance.getBizType());
-            payload.put("bizId", instance.getBizId());
-            payload.put("bizTitle", instance.getBizTitle());
-            payload.put("occurredTime", LocalDateTime.now().format(TIME));
-            // 2026-09-20 信封升级：任务定位与操作人（均可空，办件级事件/历史重放体为 null——
-            // 契约见《子应用接入流程中心规范.md》§6.3/§6.4）
-            payload.put("taskId", event.getTaskId());
-            payload.put("operatorUserId", event.getOperatorId());
-            payload.put("operatorName", event.getOperatorName());
-            Map<String, Object> data = new HashMap<>();
-            if (event.getData() != null) {
-                data.putAll(event.getData());
-            }
-            // 身份在事件发布线程中冻结，不能用 AFTER_COMMIT 重读到的新轮给旧事件重新贴标。
-            if (event.getRoundIdentity() != null) {
-                data.putAll(event.getRoundIdentity());
-            }
-            payload.put("data", data);
-            String payloadJson = objectMapper.writeValueAsString(payload);
+            logId = logStore.persist(callbackLog);
+        } catch (DuplicateKeyException dup) {
+            return; // uk(event_id, reg_id) 并发兜底：另一路径已建行，幂等成功
+        }
+        submitPush(logId);
+    }
 
-            FlwCallbackLog callbackLog = new FlwCallbackLog();
-            callbackLog.setSchId(instance.getSchId());
-            callbackLog.setRegId(reg.getId());
-            callbackLog.setEventId(eventId);
-            callbackLog.setEvent(event.getType().getCode());
-            callbackLog.setInstanceId(instance.getId());
-            callbackLog.setBizType(instance.getBizType());
-            callbackLog.setBizId(instance.getBizId());
-            callbackLog.setPayloadJson(payloadJson);
-            callbackLog.setStatus(FlwEnums.CallbackLogStatus.PENDING.getCode());
-            callbackLog.setRetryCount(0);
-            callbackLog.setNextRetryTime(LocalDateTime.now());
-            // REQUIRES_NEW 落库（afterCommit 阶段原事务已完结，REQUIRED 会挂空事务）
-            Long logId = logStore.persist(callbackLog);
-            // 异步首推：失败由退避重试接手（1m 起）
+    /** 异步首推提交：失败不回退建行结果（行已在，重试扫描接手） */
+    private void submitPush(Long logId) {
+        try {
             executor().submit(() -> pushService.pushAsync(logId));
         } catch (Exception e) {
-            log.error("[flow-center] 回调日志落库失败（对账兜底会追平）：instance={} event={} {}",
-                    instance.getId(), event.getType().getCode(), e.toString(), e);
+            log.warn("[flow-center] 异步首推提交失败（重试扫描接手）：logId={} {}", logId, e.toString());
         }
     }
 
+    /**
+     * 存量现场信封（事实冻结体缺席的兜底路径）：字段口径与 FlwStateEventRecorder 冻结体一致。
+     */
+    private Map<String, Object> legacyEnvelope(FlowEvent event, FlwInstance instance, FlwDef def,
+                                               String eventId) {
+        Map<String, Object> payload = new HashMap<>();
+        payload.put("eventId", eventId);
+        if (event.getStableEventSeq() != null) {
+            payload.put("eventSeq", event.getStableEventSeq());
+        }
+        payload.put("event", event.getType().getCode());
+        payload.put("schId", instance.getSchId());
+        payload.put("flowGuid", def.getFlowGuid());
+        payload.put("instanceId", instance.getId());
+        payload.put("instanceVersion", instance.getVersion());
+        payload.put("bizType", instance.getBizType());
+        payload.put("bizId", instance.getBizId());
+        payload.put("bizTitle", instance.getBizTitle());
+        payload.put("occurredTime", LocalDateTime.now().format(TIME));
+        // 2026-09-20 信封升级：任务定位与操作人（均可空，办件级事件/历史重放体为 null——
+        // 契约见《子应用接入流程中心规范.md》§6.3/§6.4）
+        payload.put("taskId", event.getTaskId());
+        payload.put("operatorUserId", event.getOperatorId());
+        payload.put("operatorName", event.getOperatorName());
+        Map<String, Object> data = new HashMap<>();
+        if (event.getData() != null) {
+            data.putAll(event.getData());
+        }
+        // 身份在事件发布线程中冻结，不能用 AFTER_COMMIT 重读到的新轮给旧事件重新贴标。
+        if (event.getRoundIdentity() != null) {
+            data.putAll(event.getRoundIdentity());
+        }
+        payload.put("data", data);
+        return payload;
+    }
+
     /** 重推扫描（fixedDelay 串行防重入；推拉结合里 Webhook 永远只是加速器） */
     @Scheduled(fixedDelay = 30_000)
     public void scanRetry() {

src/main/java/com/yunhai/flowcenter/callback/CallbackFactStore.java

@@ -0,0 +1,24 @@
+package com.yunhai.flowcenter.callback;
+
+import com.yunhai.flowcenter.repository.FlwStateEventRepository;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+
+/** Fact bookkeeping has its own commit; AFTER_COMMIT still has the completed business resources bound. */
+@Service
+@RequiredArgsConstructor
+public class CallbackFactStore {
+
+    private final FlwStateEventRepository stateEventRepository;
+
+    /** Called only after every matching delivery has been persisted (including the zero-subscriber case). */
+    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    public void markDispatched(String eventId) {
+        stateEventRepository.findByEventId(eventId).ifPresent(fact -> {
+            fact.setDispatched(1);
+            stateEventRepository.save(fact);
+        });
+    }
+}

src/main/java/com/yunhai/flowcenter/callback/CallbackLogStore.java

@@ -2,21 +2,45 @@ package com.yunhai.flowcenter.callback;
 
 import com.yunhai.flowcenter.entity.FlwCallbackLog;
 import com.yunhai.flowcenter.repository.FlwCallbackLogRepository;
-import lombok.RequiredArgsConstructor;
+import com.yunhai.platform.apistd.BizException;
+import com.yunhai.platform.utils.exception.ErrorCode;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.jdbc.core.BeanPropertyRowMapper;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
 
+import javax.sql.DataSource;
+import java.sql.*;
+import java.time.LocalDateTime;
+import java.util.*;
+
 /**
  * 回调日志落库（独立 bean 保证 REQUIRES_NEW 生效）：AFTER_COMMIT 监听阶段原事务已提交
  * 但事务资源仍挂在线程上，REQUIRED 会"加入"已完结事务并抛 no transaction is in progress——
  * 必须显式开新事务写日志。落库成功即视为主业务完成，推送失败不影响审批。
  */
 @Service
-@RequiredArgsConstructor
+@Slf4j
 public class CallbackLogStore {
 
     private final FlwCallbackLogRepository logRepository;
+    private final DataSource dataSource;
+    private final int claimIdleSeconds;
+
+    @Autowired
+    public CallbackLogStore(FlwCallbackLogRepository logRepository, DataSource dataSource) {
+        this(logRepository, dataSource, 60);
+    }
+
+    // 窄范围故障测试可缩短会话租期；生产固定 60s，大于正常 HTTP 5s 连接 + 10s 读取超时。
+    CallbackLogStore(FlwCallbackLogRepository logRepository, DataSource dataSource, int claimIdleSeconds) {
+        if (claimIdleSeconds < 1) throw new IllegalArgumentException("claimIdleSeconds must be positive");
+        this.logRepository = logRepository;
+        this.dataSource = dataSource;
+        this.claimIdleSeconds = claimIdleSeconds;
+    }
 
     /** @return 落库后的日志行 id */
     @Transactional(propagation = Propagation.REQUIRES_NEW)
@@ -24,4 +48,198 @@ public class CallbackLogStore {
         FlwCallbackLog saved = logRepository.saveAndFlush(callbackLog);
         return saved.getId();
     }
+
+    /**
+     * MySQL 会话锁不是事务/行锁。专用借出连接始终 autocommit，HTTP 不占用数据库事务。
+     * 连接空闲 60s 后由数据库关闭并释放锁；所有结果必须在原连接上验证锁所有权，禁止换连接补写。
+     */
+    public Claim tryClaim(Long id) {
+        Claim claim = openSession();
+        try {
+            if (!claim.acquire(id)) {
+                claim.close();
+                return null;
+            }
+            claim.row = claim.read(id);
+            if (claim.row == null) {
+                claim.close();
+                return null;
+            }
+            return claim;
+        } catch (RuntimeException | SQLException failure) {
+            claim.close();
+            throw new IllegalStateException("回调认领失败", failure);
+        }
+    }
+
+    private Claim openSession() {
+        Connection connection = null;
+        try {
+            // 不用 DataSourceUtils：绝不借用调用方已绑定的业务事务连接。
+            connection = dataSource.getConnection();
+            if (!connection.getAutoCommit()) throw new SQLException("callback claim requires autocommit");
+            long previous;
+            try (Statement st = connection.createStatement(); ResultSet rs = st.executeQuery("SELECT @@session.wait_timeout")) {
+                rs.next(); previous = rs.getLong(1);
+            }
+            try (Statement st = connection.createStatement()) {
+                st.execute("SET SESSION wait_timeout = " + claimIdleSeconds);
+            }
+            return new Claim(connection, previous);
+        } catch (SQLException failure) {
+            if (connection != null) {
+                try { connection.abort(Runnable::run); } catch (SQLException ignored) { }
+                try { connection.close(); } catch (SQLException ignored) { }
+            }
+            throw new IllegalStateException("回调认领连接不可用", failure);
+        }
+    }
+
+    /** 人工重推先完成整批短事务，再交回推送层发 HTTP；成功行不能被旧管理快照复活。 */
+    public List<Long> resetForReplay(Long school, Collection<Long> ids) {
+        try (Claim claim = openSession()) {
+            List<Long> sorted = new ArrayList<>(new TreeSet<>(ids));
+            for (Long id : sorted) {
+                if (!claim.acquire(id)) throw new BizException(ErrorCode.CUSTOM, "回调正在派发，请稍后重推：" + id);
+                FlwCallbackLog row = claim.read(id);
+                if (row == null || !school.equals(row.getSchId())) throw new BizException(ErrorCode.FORBIDDEN, "回调日志不存在或不属于该校");
+            }
+            claim.connection.setAutoCommit(false);
+            List<Long> reset = new ArrayList<>();
+            try {
+                for (Long id : sorted) {
+                    // 也检查所有权，旧 MySQL 若不支持同会话多锁会 fail closed，而不会无锁更新。
+                    if (!claim.owns(id)) throw new BizException(ErrorCode.CUSTOM, "回调派发权已失效，请重试");
+                    int changed = claim.update(id, "status=0, retry_count=0, next_retry_time=?, update_time=?, update_by=?",
+                            "status in (0,2)", Timestamp.valueOf(LocalDateTime.now()), Timestamp.valueOf(LocalDateTime.now()), actor());
+                    if (changed == 1) reset.add(id);
+                }
+                claim.connection.commit();
+                return reset;
+            } catch (RuntimeException | SQLException failure) {
+                claim.connection.rollback();
+                throw failure;
+            } finally { claim.connection.setAutoCommit(true); }
+        } catch (SQLException failure) {
+            throw new IllegalStateException("回调重推状态提交失败", failure);
+        }
+    }
+
+    private static String actor() {
+        com.yunhai.platform.auth.LoginUser user = com.yunhai.platform.auth.UserContextHolder.get();
+        return user == null ? null : user.getLoginName();
+    }
+
+    /** 令牌即拥有命名锁的原始数据库会话；状态更新只能在该连接上进行。 */
+    public static final class Claim implements AutoCloseable {
+        private final Connection connection;
+        private final long previousWaitTimeout;
+        private final Map<Long, String> keys = new LinkedHashMap<>();
+        private FlwCallbackLog row;
+        private boolean closed;
+        private boolean settled;
+
+        private Claim(Connection connection, long previousWaitTimeout) {
+            this.connection = connection;
+            this.previousWaitTimeout = previousWaitTimeout;
+        }
+
+        public FlwCallbackLog row() { return row; }
+
+        public boolean active() {
+            try { return !closed && row != null && owns(row.getId()); }
+            catch (SQLException expired) { return false; }
+        }
+
+        private boolean acquire(Long id) throws SQLException {
+            String key;
+            // 64 字符锁名：库名隔离，同 (eventId,regId) 的全部执行者用同一身份。
+            try (PreparedStatement st = connection.prepareStatement(
+                    "SELECT SHA2(CONCAT(LOWER(DATABASE()), ':flw_callback:', event_id, ':', reg_id),256) FROM flw_callback_log WHERE id=?")) {
+                st.setLong(1, id);
+                try (ResultSet rs = st.executeQuery()) { if (!rs.next()) return false; key = rs.getString(1); }
+            }
+            try (PreparedStatement st = connection.prepareStatement("SELECT GET_LOCK(?,0)")) {
+                st.setString(1, key);
+                try (ResultSet rs = st.executeQuery()) {
+                    rs.next();
+                    if (rs.getInt(1) != 1 || rs.wasNull()) return false;
+                }
+            }
+            keys.put(id, key);
+            return true;
+        }
+
+        private FlwCallbackLog read(Long id) throws SQLException {
+            try (PreparedStatement st = connection.prepareStatement("SELECT * FROM flw_callback_log WHERE id=?")) {
+                st.setLong(1, id);
+                try (ResultSet rs = st.executeQuery()) {
+                    return rs.next() ? new BeanPropertyRowMapper<>(FlwCallbackLog.class).mapRow(rs, 0) : null;
+                }
+            }
+        }
+
+        private boolean owns(Long id) throws SQLException {
+            try (PreparedStatement st = connection.prepareStatement("SELECT IS_USED_LOCK(?)=CONNECTION_ID()")) {
+                st.setString(1, keys.get(id));
+                try (ResultSet rs = st.executeQuery()) { rs.next(); return rs.getBoolean(1); }
+            }
+        }
+
+        private int update(Long id, String assignments, String condition, Object... values) throws SQLException {
+            try (PreparedStatement st = connection.prepareStatement("UPDATE flw_callback_log SET " + assignments
+                    + " WHERE id=? AND " + condition + " AND IS_USED_LOCK(?)=CONNECTION_ID()")) {
+                int i=1;
+                for (Object value : values) st.setObject(i++, value);
+                st.setLong(i++, id); st.setString(i, keys.get(id));
+                return st.executeUpdate();
+            }
+        }
+
+        public boolean success(int httpStatus) { return finish("status=1, last_http_status=?, last_error=NULL, push_time=?, next_retry_time=NULL",
+                String.valueOf(httpStatus), Timestamp.valueOf(LocalDateTime.now())); }
+
+        public boolean failure(String error, String httpStatus, long backoffMinutes) {
+            return finish("retry_count=COALESCE(retry_count,0)+1, last_http_status=?, last_error=?, next_retry_time=?",
+                    httpStatus, error, Timestamp.valueOf(LocalDateTime.now().plusMinutes(backoffMinutes)));
+        }
+
+        public boolean dead(String error) { return finish("status=2, last_error=?", error); }
+
+        private boolean finish(String assignments, Object... values) {
+            if (closed || settled || row == null) return false;
+            Object[] audited = Arrays.copyOf(values, values.length+2);
+            audited[values.length] = Timestamp.valueOf(LocalDateTime.now()); audited[values.length+1] = actor();
+            try {
+                // count 是附加 CAS；不是把旧实体全量 merge 回库。会话失权/成功/人工处理后均不能覆盖。
+                settled = update(row.getId(), assignments + ", update_time=?, update_by=?",
+                        "status=0 AND retry_count=" + row.getRetryCount(), audited) == 1;
+                if (!settled) log.debug("[flow-center] 回调旧结果已忽略：logId={}", row.getId());
+                return settled;
+            } catch (SQLException failure) {
+                log.warn("[flow-center] 回调结果未记账（会话失效或数据库不可用，留待重试）：logId={} {}", row.getId(), failure.toString());
+                return false;
+            }
+        }
+
+        @Override public void close() {
+            if (closed) return;
+            closed = true;
+            try {
+                for (String key : keys.values()) {
+                    try (PreparedStatement st = connection.prepareStatement("SELECT RELEASE_LOCK(?)")) {
+                        st.setString(1, key); st.execute();
+                    }
+                }
+                try (Statement st = connection.createStatement()) {
+                    st.execute("SET SESSION wait_timeout = " + previousWaitTimeout);
+                }
+            } catch (SQLException failedRelease) {
+                // 不能把可能残留会话锁的物理连接还给池；显式中止，让数据库释放全部锁。
+                try { connection.abort(Runnable::run); } catch (SQLException ignored) { }
+            } finally {
+                try { connection.close(); } catch (SQLException ignored) { }
+            }
+        }
+    }
 }

src/main/java/com/yunhai/flowcenter/callback/CallbackPushService.java

@@ -21,8 +21,8 @@ import javax.crypto.spec.SecretKeySpec;
 import java.nio.charset.StandardCharsets;
 import java.time.LocalDateTime;
 import java.util.Base64;
-import java.util.HashSet;
-import java.util.Set;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.UUID;
 
 /**
@@ -32,7 +32,7 @@ import java.util.UUID;
  * <b>持续重试不终止</b>（T1.1：消费方只承诺"瞬时失败回 5xx"，恢复后自动追平，消费方免对账）；
  * 挂起（retryCount 首达 hang-alert-retries）时 UserMsg 告警恰一次；
  * 死信（status=2）仅两个入口——注册停用、管理端手工标记，重试次数永不自动触发。
- * 推送一律在新事务里落库（AFTER_COMMIT 监听器/调度线程均无外层事务）。
+ * HTTP 在业务事务之外执行，结果在认领会话上以自动提交的条件 UPDATE 落定。
  * 推送本身与落库解耦：payload_json 是完整快照，可重放。
  */
 @Slf4j
@@ -40,43 +40,72 @@ import java.util.UUID;
 @RequiredArgsConstructor
 public class CallbackPushService {
 
-    private static final Set<Long> IN_FLIGHT = new HashSet<>();
+    /** 仅观测，不阻断数据库租期后的恢复；每个执行者有独立本地状态。 */
+    private final Map<Long, Integer> IN_FLIGHT = new HashMap<>();
+
+    /** 行是否有本地执行者在途（仅观测；互斥由数据库认领负责）。 */
+    boolean inFlight(Long logId) {
+        synchronized (IN_FLIGHT) {
+            return IN_FLIGHT.containsKey(logId);
+        }
+    }
 
     private final FlwCallbackLogRepository logRepository;
     private final FlwCallbackRegRepository regRepository;
     private final CallbackProperties properties;
+    private final CallbackLogStore logStore;
     private final UserMsgClient userMsgClient;
     private final org.springframework.beans.factory.ObjectProvider<com.yunhai.platform.appauth.LocalAppIdentityProvider>
             appIdentityProvider;
 
+    @org.springframework.beans.factory.annotation.Autowired
+    private CallbackSourcePolicy sourcePolicy;
+
     /** 推送指定日志行（幂等入口：成功/终态/在途跳过） */
+    @org.springframework.transaction.annotation.Transactional(propagation = org.springframework.transaction.annotation.Propagation.NOT_SUPPORTED)
     public void pushAsync(Long logId) {
         synchronized (IN_FLIGHT) {
-            if (!IN_FLIGHT.add(logId)) {
+            IN_FLIGHT.merge(logId, 1, Integer::sum);
+        }
+        try (CallbackLogStore.Claim claim = logStore.tryClaim(logId)) {
+            if (claim == null) {
+                log.debug("[flow-center] 回调数据库派发权未取得：logId={}", logId);
                 return;
             }
-        }
-        try {
-            doPush(logId);
+            doPush(claim);
         } finally {
             synchronized (IN_FLIGHT) {
-                IN_FLIGHT.remove(logId);
+                IN_FLIGHT.computeIfPresent(logId, (id, count) -> count == 1 ? null : count - 1);
             }
         }
     }
 
-    private void doPush(Long logId) {
-        FlwCallbackLog callbackLog = logRepository.findById(logId).orElse(null);
+    private void doPush(CallbackLogStore.Claim claim) {
+        FlwCallbackLog callbackLog = claim.row();
         if (callbackLog == null
                 || callbackLog.getStatus() != FlwEnums.CallbackLogStatus.PENDING.getCode()) {
+            log.debug("[flow-center] 回调推送早退（行缺席或非待推）：logId={} status={}",
+                    callbackLog == null ? null : callbackLog.getId(), callbackLog == null ? null : callbackLog.getStatus());
             return;
         }
         FlwCallbackReg reg = regRepository.findById(callbackLog.getRegId()).orElse(null);
         if (reg == null || reg.getStatus() != 0) {
             // 注册已停用：保留日志待查，不再推送（置死信便于管理端一次性清理）
-            markDead(callbackLog, "registration disabled");
+            markDead(claim, "registration disabled");
             return;
         }
+        if (sourcePolicy != null) {
+            try {
+                if (!sourcePolicy.permits(reg, callbackLog)) {
+                    markDead(claim, "SELF source or command identity mismatch");
+                    return;
+                }
+            } catch (RuntimeException unavailable) {
+                markRetryOrFail(claim, "source authentication unavailable", null);
+                return;
+            }
+        }
+        if (!claim.active()) return; // 权威查询迟到/会话租期已失效：不再发送。
         String body = callbackLog.getPayloadJson();
         String ts = String.valueOf(System.currentTimeMillis());
         String nonce = UUID.randomUUID().toString().replace("-", "").substring(0, 16);
@@ -94,41 +123,34 @@ public class CallbackPushService {
                     new HttpEntity<>(body, headers), String.class);
             int status = resp.getStatusCodeValue();
             if (status >= 200 && status < 300) {
-                markSuccess(callbackLog, status);
+                markSuccess(claim, status);
             } else {
-                markRetryOrFail(callbackLog, "HTTP " + status, String.valueOf(status));
+                markRetryOrFail(claim, "HTTP " + status, String.valueOf(status));
             }
         } catch (Exception e) {
-            markRetryOrFail(callbackLog, rootMessage(e), null);
+            markRetryOrFail(claim, rootMessage(e), null);
         }
     }
 
-    private void markSuccess(FlwCallbackLog callbackLog, int httpStatus) {
-        callbackLog.setStatus(FlwEnums.CallbackLogStatus.SUCCESS.getCode());
-        callbackLog.setRetryCount(callbackLog.getRetryCount() == null ? 0 : callbackLog.getRetryCount());
-        callbackLog.setLastHttpStatus(String.valueOf(httpStatus));
-        callbackLog.setLastError(null);
-        callbackLog.setPushTime(LocalDateTime.now());
-        saveQuietly(callbackLog);
-        log.debug("[flow-center] 回调推送成功：eventId={} http={}", callbackLog.getEventId(), httpStatus);
+    private void markSuccess(CallbackLogStore.Claim claim, int httpStatus) {
+        if (claim.success(httpStatus)) {
+            log.debug("[flow-center] 回调推送成功：eventId={} http={}", claim.row().getEventId(), httpStatus);
+        }
     }
 
     /**
      * 失败落定（持续重试语义）：越过退避表后按 {@link CallbackProperties#getCapBackoffMinutes()}
      * 封顶继续排期，<b>绝不自动死信</b>；retryCount 首次等于挂起阈值时告警恰一次。
      */
-    private void markRetryOrFail(FlwCallbackLog callbackLog, String error, String httpStatus) {
+    private void markRetryOrFail(CallbackLogStore.Claim claim, String error, String httpStatus) {
+        FlwCallbackLog callbackLog = claim.row();
         int retryCount = (callbackLog.getRetryCount() == null ? 0 : callbackLog.getRetryCount()) + 1;
-        callbackLog.setRetryCount(retryCount);
-        callbackLog.setLastHttpStatus(httpStatus);
-        callbackLog.setLastError(truncate(error));
         java.util.List<Long> backoff = properties.getRetryBackoffMinutes();
         long backoffMinutes = retryCount <= backoff.size()
                 ? backoff.get(retryCount - 1)
                 : Math.max(1, properties.getCapBackoffMinutes());
-        callbackLog.setStatus(FlwEnums.CallbackLogStatus.PENDING.getCode());
-        callbackLog.setNextRetryTime(LocalDateTime.now().plusMinutes(backoffMinutes));
-        saveQuietly(callbackLog);
+        if (!claim.failure(truncate(error), httpStatus, backoffMinutes)) return;
+        callbackLog.setRetryCount(retryCount); // 仅用于日志/告警；数据库已原子 +1，绝不 merge 此快照。
         log.warn("[flow-center] 回调推送失败（第 {} 次重试，{} 分钟后）：eventId={} {}",
                 retryCount, backoffMinutes, callbackLog.getEventId(), error);
         // 挂起告警恰一次：retryCount 每失败 +1 单调递增，等于阈值的时机只有一次；
@@ -163,9 +185,20 @@ public class CallbackPushService {
      * 仍被注册停用场景使用（doPush：注册已停用即无意义重试，置死信便于管理端清理）。
      */
     public void markDead(FlwCallbackLog callbackLog, String error) {
-        callbackLog.setStatus(FlwEnums.CallbackLogStatus.DEAD.getCode());
-        callbackLog.setLastError(truncate(error));
-        saveQuietly(callbackLog);
+        try (CallbackLogStore.Claim claim = logStore.tryClaim(callbackLog.getId())) {
+            if (claim == null) throw new com.yunhai.platform.apistd.BizException(
+                    com.yunhai.platform.utils.exception.ErrorCode.CUSTOM, "回调正在派发，请稍后标记");
+            if (claim.row().getStatus() == FlwEnums.CallbackLogStatus.SUCCESS.getCode()) {
+                throw new com.yunhai.platform.apistd.BizException(
+                        com.yunhai.platform.utils.exception.ErrorCode.PARAM_ERROR, "已成功日志无需标记死信");
+            }
+            markDead(claim, error);
+        }
+    }
+
+    private void markDead(CallbackLogStore.Claim claim, String error) {
+        if (!claim.dead(truncate(error))) return;
+        FlwCallbackLog callbackLog = claim.row();
         log.error("[flow-center] 回调推送进入死信：eventId={} biz={}/{} error={}",
                 callbackLog.getEventId(), callbackLog.getBizType(), callbackLog.getBizId(), error);
         alertDead(callbackLog, error);
@@ -203,14 +236,6 @@ public class CallbackPushService {
         }
     }
 
-    private void saveQuietly(FlwCallbackLog callbackLog) {
-        try {
-            logRepository.save(callbackLog);
-        } catch (Exception e) {
-            log.error("[flow-center] 回调日志落库失败：eventId={} {}", callbackLog.getEventId(), e.toString());
-        }
-    }
-
     /**
      * 回调探测（2026-10-11 管理端增强）：以该注册的 secret 按现行信封签一条未知事件码 flow.ping，
      * 走真实推送管道同步一发——接收方按规范 §6.4 对未知事件码"记日志后 200"，不破坏业务数据。

src/main/java/com/yunhai/flowcenter/callback/CallbackSourcePolicy.java

@@ -0,0 +1,57 @@
+package com.yunhai.flowcenter.callback;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.yunhai.flowcenter.entity.FlwCallbackReg;
+import com.yunhai.flowcenter.entity.FlwInstance;
+import com.yunhai.flowcenter.repository.FlwInstanceRepository;
+import com.yunhai.flowcenter.service.FlowSourcePrincipalResolver;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Component;
+
+/** Same SELF predicate for live fanout, fact recovery, retries and manual re-delivery. */
+@Component @RequiredArgsConstructor
+public class CallbackSourcePolicy {
+    private final FlowSourcePrincipalResolver resolver;
+    private final FlwInstanceRepository instances;
+    private final ObjectMapper mapper;
+    public boolean matches(FlwCallbackReg reg, FlwInstance instance) {
+        if (instance.getSourcePrincipalId() == null) {
+            return true; // Legacy callback migration is explicit; formal facts always use UUID SELF.
+        }
+        if (!instance.getSourcePrincipalId().equals(reg.getSourcePrincipalId())) return false;
+        // Recheck the credential mapping: a reused appKey cannot receive its predecessor's facts.
+        return instance.getSourcePrincipalId().equals(resolver.resolve(reg.getAppKey()).id);
+    }
+    public boolean permits(FlwCallbackReg reg, com.yunhai.flowcenter.entity.FlwCallbackLog delivery) {
+        if (!permits(reg, delivery.getInstanceId(), delivery.getPayloadJson())) return false;
+        FlwInstance instance = instances.findById(delivery.getInstanceId()).orElseThrow(IllegalStateException::new);
+        if (instance.getSourcePrincipalId() == null) return true;
+        try {
+            JsonNode body = mapper.readTree(delivery.getPayloadJson());
+            return java.util.Objects.equals(delivery.getEventId(), body.path("eventId").asText())
+                    && java.util.Objects.equals(delivery.getEvent(), body.path("event").asText())
+                    && java.util.Objects.equals(delivery.getBizType(), body.path("bizType").asText())
+                    && java.util.Objects.equals(delivery.getBizId(), body.path("bizId").asText());
+        } catch (Exception e) { return false; }
+    }
+    public boolean permits(FlwCallbackReg reg, Long instanceId, String payload) {
+        FlwInstance instance = instances.findById(instanceId).orElse(null);
+        if (instance == null || !matches(reg, instance)) return false;
+        if (instance.getSourcePrincipalId() == null) return true;
+        try {
+            JsonNode body = mapper.readTree(payload);
+            JsonNode data = body.path("data");
+            return instance.getSourcePrincipalId().equals(body.path("sourcePrincipalId").asText())
+                    && instance.getApplyRound().equals(body.path("applyRound").asInt())
+                    && instance.getSourcePrincipalId().equals(data.path("sourcePrincipalId").asText())
+                    && instance.getApplyRequestToken().equals(data.path("requestId").asText())
+                    && instance.getApplyPayloadHash().equals(data.path("canonicalHash").asText())
+                    && instance.getDefVersion().equals(data.path("defVersion").asInt())
+                    && instance.getApplyRound().equals(data.path("applyRound").asInt())
+                    && instance.getId().equals(body.path("instanceId").asLong())
+                    && instance.getSchId().equals(body.path("schId").asLong())
+                    && instance.getBizType().equals(body.path("bizType").asText())
+                    && instance.getBizId().equals(body.path("bizId").asText());
+        } catch (Exception e) { return false; }
+    }
+}

src/main/java/com/yunhai/flowcenter/callback/FlwStateEventRecorder.java

@@ -0,0 +1,145 @@
+package com.yunhai.flowcenter.callback;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.yunhai.flowcenter.engine.FlowEvent;
+import com.yunhai.flowcenter.entity.FlwDef;
+import com.yunhai.flowcenter.entity.FlwInstance;
+import com.yunhai.flowcenter.entity.FlwStateEvent;
+import com.yunhai.flowcenter.repository.FlwDefRepository;
+import com.yunhai.flowcenter.repository.FlwInstanceRepository;
+import com.yunhai.flowcenter.repository.FlwStateEventRepository;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.event.EventListener;
+import org.springframework.core.Ordered;
+import org.springframework.core.annotation.Order;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * 状态事件事实记录器（2026-10-25 T05；同日 R2 返修冻结完整信封）：普通同步监听器，与发布方
+ * <b>同一事务</b>执行——状态变更提交成功 ⇔ 不可变事件事实已落库，关闭"提交到 AFTER_COMMIT
+ * 分发之间进程被杀，事件永久丢失"的窗口（审计 E24）。
+ *
+ * <p>职责边界：只落事实行并给 {@link FlowEvent} 赋<b>稳定身份</b>与<b>冻结信封</b>
+ * （eventId 全订阅方/全部重试复用；eventSeq 实例内单调；R2 起信封含 eventId/eventSeq/flowGuid
+ * 全部投递字段——正常分发与崩溃恢复共用同一冻结体，Codex D04"恢复消息体缺 eventId"关闭）；
+ * 回调分发、订阅匹配、推送仍归 {@link CallbackDispatcher}。URGE 站内信不入事件流（既有口径）。
+ * 无事务的发布（fallbackExecution 场景）退化为自动提交直写——事实仍落，只是失去与状态的
+ * 同事务原子性（记录 warn，不静默）。</p>
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class FlwStateEventRecorder {
+
+    private static final DateTimeFormatter TIME = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+
+    private final FlwStateEventRepository eventRepository;
+    private final FlwInstanceRepository instanceRepository;
+    private final FlwDefRepository defRepository;
+    private final ObjectMapper objectMapper;
+
+    @EventListener
+    @Order(Ordered.HIGHEST_PRECEDENCE + 1)
+    public void record(FlowEvent event) {
+        if (event.getType() == FlowEvent.Type.URGE) {
+            return;
+        }
+        if (event.getStableEventId() != null) {
+            return; // 幂等：同一事件对象重复发布不重复落事实
+        }
+        try {
+            FlwInstance instance = instanceRepository.findById(event.getInstanceId()).orElse(null);
+            if (instance == null) {
+                log.warn("[state-event] 事实落库时实例不存在（跳过）：instance={} event={}",
+                        event.getInstanceId(), event.getType().getCode());
+                return;
+            }
+            FlwDef def = instance.getDefId() == null ? null
+                    : defRepository.findById(instance.getDefId()).orElse(null);
+            FlwStateEvent fact = new FlwStateEvent();
+            fact.setEventId(UUID.randomUUID().toString().replace("-", "").toLowerCase());
+            fact.setSchId(instance.getSchId());
+            fact.setInstanceId(instance.getId());
+            fact.setApplyRound(instance.getApplyRound());
+            // T05 发号（终形态）：FOR UPDATE 标量当前读（行 X 锁，任何隔离级别串行正确）
+            // + 锁内回写；不实体 merge（@Version 会 StaleObjectState，实证）。
+            Integer nextSeq = instanceRepository.nextEventSeqForUpdate(instance.getId());
+            if (nextSeq == null) {
+                throw new IllegalStateException("事件序号发号返回空：instance=" + instance.getId());
+            }
+            instanceRepository.setEventSeqCounter(instance.getId(), nextSeq);
+            fact.setEventSeq(nextSeq);
+            fact.setEventType(event.getType().getCode());
+            fact.setTaskId(event.getTaskId());
+            fact.setOperatorId(event.getOperatorId());
+            // R2：冻结完整投递信封（含 eventId/eventSeq/flowGuid）——事实载荷即投递体，
+            // 正常链与恢复链字节级同构，接收端必填字段（eventId/event/bizType/bizId）恒在。
+            fact.setPayloadJson(objectMapper.writeValueAsString(
+                    fullEnvelope(event, instance, def, fact.getEventId(), nextSeq)));
+            eventRepository.save(fact);
+            event.assignStableIdentity(fact.getEventId(), fact.getEventSeq());
+            event.assignStablePayload(fact.getPayloadJson());
+        } catch (Exception e) {
+            // 事实落库失败=状态事务随之一并失败（同事务）或无事务场景仅告警；
+            // 绝不反噬引擎主流程——分发链对账兜底仍在
+            log.error("[state-event] 事件事实落库异常：instance={} event={} {}",
+                    event.getInstanceId(), event.getType().getCode(), e.toString(), e);
+            if (org.springframework.transaction.support.TransactionSynchronizationManager
+                    .isActualTransactionActive()) {
+                // 同事务场景：事实写失败必须让状态一起回滚（事件与状态原子），不吞
+                throw new IllegalStateException("状态事件事实落库失败，事务回滚：" + event.getType().getCode(), e);
+            }
+        }
+    }
+
+    /**
+     * 完整投递信封（R2 冻结体）：{@code eventId/eventSeq/event/schId/flowGuid/instanceId/
+     * instanceVersion/applyRound/bizType/bizId/bizTitle/occurredTime/taskId/operatorUserId/
+     * operatorName/data}。def 缺席（定义已删）时不带 flowGuid——该场合分发器同样不投递。
+     */
+    private Map<String, Object> fullEnvelope(FlowEvent event, FlwInstance instance,
+                                             FlwDef def, String eventId, Integer eventSeq) {
+        Map<String, Object> payload = new HashMap<>();
+        payload.put("eventId", eventId);
+        payload.put("eventSeq", eventSeq);
+        payload.put("event", event.getType().getCode());
+        payload.put("schId", instance.getSchId());
+        if (def != null) {
+            payload.put("flowGuid", def.getFlowGuid());
+        }
+        payload.put("instanceId", instance.getId());
+        payload.put("instanceVersion", instance.getVersion());
+        payload.put("applyRound", instance.getApplyRound());
+        payload.put("bizType", instance.getBizType());
+        payload.put("bizId", instance.getBizId());
+        payload.put("bizTitle", instance.getBizTitle());
+        payload.put("occurredTime", LocalDateTime.now().format(TIME));
+        payload.put("taskId", event.getTaskId());
+        payload.put("operatorUserId", event.getOperatorId());
+        payload.put("operatorName", event.getOperatorName());
+        Map<String, Object> data = new HashMap<>();
+        if (event.getData() != null) {
+            data.putAll(event.getData());
+        }
+        if (event.getRoundIdentity() != null) {
+            data.putAll(event.getRoundIdentity());
+        }
+        if (instance.getSourcePrincipalId() != null) {
+            payload.put("sourcePrincipalId", instance.getSourcePrincipalId());
+            data.put("sourcePrincipalId", instance.getSourcePrincipalId());
+            data.put("requestId", instance.getApplyRequestToken());
+            data.put("defVersion", instance.getDefVersion());
+            data.put("canonicalHash", instance.getApplyPayloadHash());
+            data.put("applyRound", instance.getApplyRound());
+        }
+        payload.put("data", data);
+        return payload;
+    }
+}

src/main/java/com/yunhai/flowcenter/callback/FormalCoordinateReleaseListener.java

@@ -0,0 +1,29 @@
+package com.yunhai.flowcenter.callback;
+import com.yunhai.flowcenter.engine.FlowEvent;
+import com.yunhai.flowcenter.repository.*;
+import lombok.RequiredArgsConstructor;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+
+/** Release in the state transaction; late events can never free a newer round's pointer. */
+@Component @RequiredArgsConstructor
+public class FormalCoordinateReleaseListener {
+    private final FlwInstanceRepository instances;
+    private final FlwBizCoordinateRepository coordinates;
+    @EventListener
+    public void release(FlowEvent event) {
+        if (event.getType() != FlowEvent.Type.COMPLETED && event.getType() != FlowEvent.Type.REJECTED
+                && event.getType() != FlowEvent.Type.WITHDRAWN) return;
+        instances.findById(event.getInstanceId()).ifPresent(instance -> {
+            if (instance.getSourcePrincipalId() != null) {
+                if (!org.springframework.transaction.support.TransactionSynchronizationManager.isActualTransactionActive()) {
+                    throw new IllegalStateException("Formal terminal transition requires the state transaction");
+                }
+                coordinates.release(instance.getSchId(), instance.getSourcePrincipalId(), instance.getBizType(),
+                        instance.getBizId(), instance.getId(), instance.getApplyRound());
+            }
+        });
+    }
+}