problems = new ArrayList<>();
++ VariableContract.validateDefinition(model, false);
+
+ if (model.nodes == null || model.nodes.isEmpty()) {
+ throw new BizException(ErrorCode.PARAM_ERROR, "流程定义缺少节点");
+@@ -248,7 +249,7 @@ public final class DefValidator {
+ * (命名/唯一/同分支上下文由度约束段复用 {@link #validateOutlets} 口径)。
+ *
+ * EXPRESSION 模式(2026-10-10 四议题)分层校验:语法错/出口缺条件 → 发布期硬拦
+- * (表单变量是子应用运行时传的,发布期无从核对 schema,一律放行、缺失按 null 注入求值)。
++ * 新契约的变量引用/类型/值域由 VariableContract 统一检查;历史无声明版本保持兼容。
+ */
+ private static void validateCondition(FlowDefModel model, FlowDefModel.Node n, Set enabledRouterKeys,
+ List problems) {
+diff --git a/src/main/java/com/yunhai/flowcenter/engine/FlowAdvanceService.java b/src/main/java/com/yunhai/flowcenter/engine/FlowAdvanceService.java
+index 08f14ca..ef0e950 100644
+--- a/src/main/java/com/yunhai/flowcenter/engine/FlowAdvanceService.java
++++ b/src/main/java/com/yunhai/flowcenter/engine/FlowAdvanceService.java
+@@ -1,81 +1,201 @@
+ package com.yunhai.flowcenter.engine;
+
+-import com.yunhai.flowcenter.entity.FlwDef;
+-import com.yunhai.flowcenter.entity.FlwInstance;
+-import com.yunhai.flowcenter.entity.FlwTask;
++import com.yunhai.flowcenter.entity.*;
+ import com.yunhai.flowcenter.enums.FlwEnums;
+-import com.yunhai.flowcenter.repository.FlwDefRepository;
+-import com.yunhai.flowcenter.repository.FlwInstanceRepository;
+-import com.yunhai.flowcenter.repository.FlwTaskRepository;
++import com.yunhai.flowcenter.repository.*;
+ import lombok.RequiredArgsConstructor;
+-import lombok.extern.slf4j.Slf4j;
+ import org.springframework.stereotype.Service;
+-import org.springframework.transaction.annotation.Propagation;
+-import org.springframework.transaction.annotation.Transactional;
++import org.springframework.transaction.annotation.*;
+
+-/**
+- * 节点放行推进(AFTER_COMMIT 监听器经代理调用本 bean,REQUIRES_NEW 生效——
+- * 事务已提交后的同步监听阶段没有活动事务,@Modifying 批量语句必须有新事务承载):
+- * 该节点待办清零且办件在途才推进;并发冲突由 flw_instance.version 乐观锁暴露,监听器层负责重试。
+- */
+-@Slf4j
++import javax.persistence.EntityManager;
++import javax.persistence.PersistenceContext;
++import java.time.LocalDateTime;
++import java.util.*;
++
++/** D09: durable activation membership, serialized last-handler payload, and revision-scoped settlement. */
+ @Service
+ @RequiredArgsConstructor
+ public class FlowAdvanceService {
+-
+ private final FlwInstanceRepository instanceRepository;
+ private final FlwDefRepository defRepository;
+ private final FlwTaskRepository taskRepository;
+ private final FlowEngine flowEngine;
++ private final FlwAdvanceIntentRepository intentRepository;
++ @PersistenceContext private EntityManager entityManager;
++
++ /** The listener groups the existing totalTasks TASK_CREATED events within their creation transaction. */
++ @Transactional(propagation = Propagation.MANDATORY)
++ public Long registerTask(FlowEvent event, Long activationIntentId) {
++ String nodeId = (String) event.getData().get("nodeId");
++ FlwAdvanceIntent intent;
++ if (activationIntentId == null) {
++ FlwInstance instance = instanceRepository.findById(event.getInstanceId()).orElseThrow(IllegalStateException::new);
++ intent = new FlwAdvanceIntent();
++ intent.setInstanceId(instance.getId());intent.setNodeId(nodeId);
++ intent.setActivationTaskId(event.getTaskId());
++ intent.setActivationDefId(instance.getDefId());intent.setActivationDefVersion(instance.getDefVersion());
++ intent.setActivationApplyRound(instance.getApplyRound());
++ intent.setStatus(FlwAdvanceIntent.STATE_DONE); // Registration only; no advance work before handling.
++ intent.setNextRetryAt(LocalDateTime.now());
++ } else {
++ intent = intentRepository.findById(activationIntentId).orElseThrow(IllegalStateException::new);
++ if (!Objects.equals(intent.getInstanceId(), event.getInstanceId()) || !Objects.equals(intent.getNodeId(), nodeId)) {
++ throw new IllegalStateException("Activation registration identity mismatch");
++ }
++ }
++ intent.addTask(event.getTaskId());
++ return intentRepository.save(intent).getId();
++ }
++
++ /** TRANSFERRED lacks the parent task ID. Accept only a uniquely proven parent activation, never guess by max ID. */
++ @Transactional(propagation = Propagation.MANDATORY)
++ public void registerTransfer(FlowEvent event) {
++ String nodeId = (String) event.getData().get("fromNodeId");
++ List generations = intentRepository.findByInstanceIdAndNodeIdOrderByIdDesc(event.getInstanceId(), nodeId);
++ Set parents = new HashSet<>();
++ List candidates = taskRepository.findTransferParentCandidates(event.getTaskId(), event.getOperatorId());
++ for (Long taskId : candidates) {
++ FlwAdvanceIntent parent = memberOf(generations, taskId);
++ if (parent == null) throw new IllegalStateException("D09 transfer parent requires audited activation backfill");
++ parents.add(parent.getId());
++ }
++ if (parents.size() != 1) throw new IllegalStateException("D09 ambiguous transfer activation; parent identity is required");
++ FlwAdvanceIntent parent = lockIntent(parents.iterator().next());
++ parent.addTask(event.getTaskId());
++ intentRepository.save(parent);
++ }
++
++ /** Called synchronously in the task transaction, after claimAndFinish. The head lock orders commits, not clocks/IDs. */
++ @Transactional(propagation = Propagation.MANDATORY)
++ public void recordHandled(FlowEvent event) {
++ String nodeId = (String) event.getData().get("nodeId");
++ FlwAdvanceIntent candidate = memberOf(intentRepository.findByInstanceIdAndNodeIdOrderByIdDesc(event.getInstanceId(), nodeId), event.getTaskId());
++ if (candidate == null) throw new IllegalStateException("D09 task has no proven activation; audited backfill required");
++ FlwAdvanceIntent intent = lockIntent(candidate.getId());
++ if (intent.handledTask(event.getTaskId())) return; // Duplicate/late publication cannot rewind the last handler.
++ intent.addHandledTask(event.getTaskId());intent.setTriggerTaskId(event.getTaskId());
++ intent.setEdgeName((String) event.getData().get("edgeName"));
++ intent.setAssigneeId((Integer) event.getData().get("assigneeId"));
++ intent.setAssigneeName((String) event.getData().get("assigneeName"));
++ intent.setStatus(FlwAdvanceIntent.STATE_PENDING);intent.setAttempts(0);
++ intent.setNextRetryAt(LocalDateTime.now());intent.setLastError(null);
++ intentRepository.save(intent);
++ }
++
++ public static final class IntentRef {
++ public final Long id;
++ public final Long revision;
++ public IntentRef(Long id, Long revision) { this.id = id; this.revision = revision; }
++ }
++
++ @Transactional(propagation = Propagation.REQUIRES_NEW, readOnly = true)
++ public IntentRef pendingForTask(Long instanceId, String nodeId, Long taskId) {
++ FlwAdvanceIntent intent = memberOf(intentRepository.findByInstanceIdAndNodeIdOrderByIdDesc(instanceId, nodeId), taskId);
++ return intent == null || !FlwAdvanceIntent.STATE_PENDING.equals(intent.getStatus()) ? null
++ : new IntentRef(intent.getId(), intent.getRevision());
++ }
+
+ /**
+- * @param nodeId 待放行节点
+- * @param edgeName 多出口节点所选出口(办理事件携带);可空——兜底从该节点
+- * 最近一条已同意任务的 edge_name 列重放(推进在 AFTER_COMMIT 新事务重查,
+- * 事件传参只是快路径,落库的值才是权威)
+- * @param operator 本节点办理人(TASK_HANDLED 事件 data 穿透;推进产生的
+- * TASK_CREATED/COMPLETED 操作人信封字段来源),可空
+- * @return 是否实际推进了
++ * One transaction owns validation, engine effects and settlement. RC sees the last committed task status;
++ * the head lock serializes against last-handler refresh, and instance @Version fences reactivation writes.
+ */
+- @Transactional(propagation = Propagation.REQUIRES_NEW)
+- public boolean tryAdvance(Long instanceId, String nodeId, String edgeName, FlowOperator operator) {
+- FlwInstance instance = instanceRepository.findById(instanceId).orElse(null);
+- if (instance == null
+- || instance.getStatus() != FlwEnums.InstanceStatus.APPROVING.getCode()) {
++ @Transactional(propagation = Propagation.REQUIRES_NEW, isolation = Isolation.READ_COMMITTED)
++ public boolean tryAdvance(Long intentId, Long observedRevision) {
++ FlwAdvanceIntent intent = lockIntent(intentId);
++ if (!ownsRevision(intent, observedRevision)) return false;
++ FlwAdvanceIntent latest = intentRepository.findFirstByInstanceIdAndNodeIdAndActivationTaskIdIsNotNullOrderByIdDesc(
++ intent.getInstanceId(), intent.getNodeId()).orElse(null);
++ if (latest != null && !Objects.equals(latest.getId(), intent.getId())) {
++ done(intent, "D09_STALE_ACTIVATION: successor=" + latest.getId());
++ handoff(latest.getId());
+ return false;
+ }
+- // 已推进过的节点不再放行(currentNodes 由引擎在放行时移除,乐观锁保证互斥)
+- String current = instance.getCurrentNodes();
+- boolean active = current != null && java.util.Arrays.stream(current.split(","))
+- .anyMatch(nodeId::equals);
+- if (!active) {
++ if (intent.getActivationTaskId() == null || !intent.containsTask(intent.getActivationTaskId())
++ || intent.getTriggerTaskId() == null || !intent.handledTask(intent.getTriggerTaskId())) {
++ hold(intent, "D09_UNANCHORED: audited cohort/last-handler backfill required; not executed");
+ return false;
+ }
+- long pending = taskRepository.findByInstanceIdAndStatus(instanceId,
+- FlwEnums.TaskStatus.PENDING.getCode()).stream()
+- .filter(t -> nodeId.equals(t.getNodeId()))
+- .count();
+- if (pending > 0) {
+- return false;
++ FlwInstance instance = instanceRepository.findById(intent.getInstanceId()).orElse(null);
++ if (instance == null || instance.getStatus() != FlwEnums.InstanceStatus.APPROVING.getCode()
++ || !active(instance, intent.getNodeId())) {
++ done(intent, null);return false;
+ }
+- FlwDef def = defRepository.findById(instance.getDefId()).orElse(null);
+- if (def == null) {
+- log.error("[flow-center] 办件定义缺失,推进中止:instance={} defId={}",
+- instanceId, instance.getDefId());
+- return false;
++ if (!Objects.equals(instance.getDefId(), intent.getActivationDefId())
++ || !Objects.equals(instance.getDefVersion(), intent.getActivationDefVersion())
++ || !Objects.equals(instance.getApplyRound(), intent.getActivationApplyRound())) {
++ hold(intent, "D09_IDENTITY_MISMATCH: no proven successor; audited backfill required");return false;
+ }
+- String chosen = edgeName;
+- if (chosen == null) {
+- chosen = taskRepository.findByInstanceIdAndStatus(instanceId,
+- FlwEnums.TaskStatus.APPROVED.getCode()).stream()
+- .filter(t -> nodeId.equals(t.getNodeId()) && t.getEdgeName() != null)
+- .max(java.util.Comparator.comparing(
+- FlwTask::getHandleTime, java.util.Comparator.nullsFirst(java.util.Comparator.naturalOrder())))
+- .map(FlwTask::getEdgeName)
+- .orElse(null);
++ boolean pending = taskRepository.findByInstanceIdAndStatus(instance.getId(), FlwEnums.TaskStatus.PENDING.getCode())
++ .stream().anyMatch(t -> intent.getNodeId().equals(t.getNodeId()));
++ if (pending) {
++ // A future handler holds/obtains this same head lock and must re-arm it in its task transaction.
++ done(intent, null);return false;
+ }
+- flowEngine.continueAdvance(instance, def, nodeId, chosen, operator);
+- instanceRepository.save(instance);
++ FlwTask trigger = taskRepository.findById(intent.getTriggerTaskId()).orElse(null);
++ if (trigger == null || !intent.containsTask(trigger.getId())
++ || !Objects.equals(trigger.getInstanceId(), instance.getId())
++ || !Objects.equals(trigger.getNodeId(), intent.getNodeId())
++ || !Integer.valueOf(FlwEnums.TaskStatus.APPROVED.getCode()).equals(trigger.getStatus())) {
++ hold(intent, "D09_INVALID_LAST_HANDLER: not executed");return false;
++ }
++ FlwDef def = defRepository.findById(instance.getDefId()).orElseThrow(() -> new IllegalStateException("Missing activation definition"));
++ flowEngine.continueAdvance(instance, def, intent.getNodeId(), trigger.getEdgeName(),
++ FlowOperator.of(trigger.getAssigneeId(), trigger.getAssigneeName()));
++ if (active(instance, intent.getNodeId())) throw new IllegalStateException("Activation did not advance; retaining recoverable intent");
++ instanceRepository.saveAndFlush(instance); // An optimistic conflict rolls back engine writes AND settlement.
++ done(intent, null);
+ return true;
+ }
++
++ /** Never merge a worker's detached row. A refreshed revision owns its own retry budget. */
++ @Transactional(propagation = Propagation.REQUIRES_NEW)
++ public void defer(Long intentId, Long observedRevision, String error) {
++ FlwAdvanceIntent intent = lockIntent(intentId);
++ if (!ownsRevision(intent, observedRevision)) return;
++ intent.setAttempts(intent.getAttempts() + 1);intent.setLastError(truncate(error));
++ if (intent.getAttempts() >= 6) intent.setStatus(FlwAdvanceIntent.STATE_EXHAUSTED);
++ else intent.setNextRetryAt(LocalDateTime.now().plusSeconds(Math.min(1800, 30L * (1L << (intent.getAttempts() - 1)))));
++ intentRepository.save(intent);
++ }
++
++ private FlwAdvanceIntent lockIntent(Long id) {
++ // Queries in a handling transaction may already have loaded a stale RR snapshot into the identity map.
++ // A lock upgrade on that managed row checks its obsolete @Version before refresh can run.
++ // Flush our own changes, detach ONLY this row, then load a new managed row with a locking current read.
++ entityManager.flush();
++ FlwAdvanceIntent cached = entityManager.find(FlwAdvanceIntent.class, id);
++ if (cached != null) entityManager.detach(cached);
++ return intentRepository.findByIdForUpdate(id).orElse(null);
++ }
++
++ private void handoff(Long successorId) {
++ FlwAdvanceIntent successor = lockIntent(successorId);
++ // A stale worker must not consume its successor. Re-arm a previously waiting head, retaining its own payload.
++ if (successor != null && successor.getTriggerTaskId() != null && FlwAdvanceIntent.STATE_DONE.equals(successor.getStatus())) {
++ successor.setStatus(FlwAdvanceIntent.STATE_PENDING);successor.setAttempts(0);
++ successor.setNextRetryAt(LocalDateTime.now());successor.setLastError(null);intentRepository.save(successor);
++ }
++ }
++
++ private static FlwAdvanceIntent memberOf(List intents, Long taskId) {
++ FlwAdvanceIntent found = null;
++ for (FlwAdvanceIntent intent : intents) if (intent.containsTask(taskId)) {
++ if (found != null && !Objects.equals(found.getId(), intent.getId())) throw new IllegalStateException("Task belongs to multiple activations");
++ found = intent;
++ }
++ return found;
++ }
++ private static boolean ownsRevision(FlwAdvanceIntent intent, Long revision) {
++ return intent != null && FlwAdvanceIntent.STATE_PENDING.equals(intent.getStatus()) && Objects.equals(intent.getRevision(), revision);
++ }
++ private static boolean active(FlwInstance instance, String nodeId) {
++ return instance.getCurrentNodes() != null && Arrays.asList(instance.getCurrentNodes().split(",")).contains(nodeId);
++ }
++ private void done(FlwAdvanceIntent intent, String reason) {
++ intent.setStatus(FlwAdvanceIntent.STATE_DONE);intent.setLastError(reason);intentRepository.save(intent);
++ }
++ private void hold(FlwAdvanceIntent intent, String reason) {
++ intent.setStatus(FlwAdvanceIntent.STATE_EXHAUSTED);intent.setLastError(reason);intentRepository.save(intent);
++ }
++ private static String truncate(String s) { return s == null || s.length() <= 255 ? s : s.substring(0,255); }
+ }
+diff --git a/src/main/java/com/yunhai/flowcenter/engine/FlowDefModel.java b/src/main/java/com/yunhai/flowcenter/engine/FlowDefModel.java
+index db0076c..64c2f67 100644
+--- a/src/main/java/com/yunhai/flowcenter/engine/FlowDefModel.java
++++ b/src/main/java/com/yunhai/flowcenter/engine/FlowDefModel.java
+@@ -56,6 +56,18 @@ public class FlowDefModel {
+ public List nodes;
+ public List edges;
+ public Settings settings;
++ /** null = 历史无契约版本;新发布必须显式声明(无业务变量时为 [])。 */
++ public List variables;
++
++ @JsonIgnoreProperties(ignoreUnknown = true)
++ public static class Variable {
++ public String key;
++ /** string/number/boolean/date/datetime/money/stringArray/fileArray/userArray */
++ public String type;
++ public Boolean required;
++ @com.fasterxml.jackson.annotation.JsonProperty("enum")
++ public List enumValues;
++ }
+ /**
+ * 画布坐标(nodeId → 左上角),仅前端设计器/查看器复用;引擎不消费。
+ * 缺失时前端自行做自动布局,故可选、可空。
+diff --git a/src/main/java/com/yunhai/flowcenter/engine/FlowEngine.java b/src/main/java/com/yunhai/flowcenter/engine/FlowEngine.java
+index 6c9bb7f..d1b7543 100644
+--- a/src/main/java/com/yunhai/flowcenter/engine/FlowEngine.java
++++ b/src/main/java/com/yunhai/flowcenter/engine/FlowEngine.java
+@@ -45,6 +45,9 @@ public class FlowEngine {
+ /** 发起后首推:从 START 激活(instance 已建好、status=10)。operator=null:TASK_CREATED/COMPLETED 的操作人回退发起人 */
+ public void startAdvance(FlwInstance instance, FlwDef def) {
+ FlowDefModel model = parseModel(def);
++ // D11:整份定义的固定 USERS 先整体预检,非法人不得等到后续节点才暴露(会留下卡单)。
++ // 校验失败抛业务异常,由发起事务整体回滚(零实例零任务)。
++ assigneeResolver.requireEligibleFixedAssignees(instance.getSchId(), model);
+ setCurrentNodes(instance, null);
+ advance(instance, model, model.start, null, null);
+ }
+diff --git a/src/main/java/com/yunhai/flowcenter/engine/FlowEvent.java b/src/main/java/com/yunhai/flowcenter/engine/FlowEvent.java
+index f500a59..84b7efc 100644
+--- a/src/main/java/com/yunhai/flowcenter/engine/FlowEvent.java
++++ b/src/main/java/com/yunhai/flowcenter/engine/FlowEvent.java
+@@ -69,6 +69,55 @@ public class FlowEvent {
+ private final String operatorName;
+ /** 事件明细(result/opinion/rejectNode/task.created 的 n/m 等),可为空 */
+ private final Map data;
++ private Map roundIdentity;
++
++ // ===== T05 稳定事件身份(2026-10-25) =====
++ /** 事实行 event_id:发布线程同步监听器在状态事务内落事实行时一次性赋值,全订阅方/重试复用 */
++ private String stableEventId;
++ /** 事实行 event_seq(实例内单调序号) */
++ private Integer stableEventSeq;
++ /**
++ * R2(Codex D04)冻结信封:事实落库时一次性构建的完整投递载荷 JSON(含 eventId/eventSeq/
++ * flowGuid 等全部信封字段)。正常分发与崩溃恢复分发共用同一份冻结体——恢复消息体不再缺
++ * eventId/eventSeq,两条路径字节级同构。事实行缺席(存量路径)为 null,由分发器现场构建。
++ */
++ private String stablePayloadJson;
++
++ public void assignStableIdentity(String eventId, Integer eventSeq) {
++ if (this.stableEventId != null) {
++ return; // 只赋一次:幂等重入(同事件多监听器)不换身份
++ }
++ this.stableEventId = eventId;
++ this.stableEventSeq = eventSeq;
++ }
++
++ public String getStableEventId() {
++ return stableEventId;
++ }
++
++ public Integer getStableEventSeq() {
++ return stableEventSeq;
++ }
++
++ /** 随稳定身份一并冻结完整投递信封(只赋一次;身份与信封同源同事务) */
++ public void assignStablePayload(String payloadJson) {
++ if (this.stablePayloadJson != null) {
++ return;
++ }
++ this.stablePayloadJson = payloadJson;
++ }
++
++ public String getStablePayloadJson() {
++ return stablePayloadJson;
++ }
++
++ public void captureRoundIdentity(Integer round, String requestToken) {
++ if (roundIdentity != null) return;
++ Map identity = new HashMap<>();
++ identity.put("applyRound", round);
++ identity.put("requestToken", requestToken);
++ roundIdentity = java.util.Collections.unmodifiableMap(identity);
++ }
+
+ public static FlowEvent of(Type type, Long instanceId) {
+ return new FlowEvent(type, instanceId, null, null, null, new HashMap<>());
+diff --git a/src/main/java/com/yunhai/flowcenter/engine/FlowNodeAdvanceListener.java b/src/main/java/com/yunhai/flowcenter/engine/FlowNodeAdvanceListener.java
+index 1ac00c3..d0e8a93 100644
+--- a/src/main/java/com/yunhai/flowcenter/engine/FlowNodeAdvanceListener.java
++++ b/src/main/java/com/yunhai/flowcenter/engine/FlowNodeAdvanceListener.java
+@@ -1,50 +1,102 @@
+ package com.yunhai.flowcenter.engine;
+
++import com.yunhai.flowcenter.entity.FlwAdvanceIntent;
++import com.yunhai.flowcenter.repository.FlwAdvanceIntentRepository;
+ 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.orm.ObjectOptimisticLockingFailureException;
++import org.springframework.scheduling.annotation.Scheduled;
+ import org.springframework.stereotype.Component;
+ import org.springframework.transaction.event.TransactionPhase;
+ import org.springframework.transaction.event.TransactionalEventListener;
++import org.springframework.transaction.support.TransactionSynchronization;
++import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+-/**
+- * 节点放行推进监听器:TASK_HANDLED 事件 AFTER_COMMIT 后在新事务里重查——
+- * 该节点待办是否清零、办件是否仍在途,全部成立才继续推进(下一节点/JOIN 合并/END)。
+- * 会签并发办理时各方提交的快照互相看不到对方未提交的办理行,事务内判断
+- * "节点是否全部办结"必假,提交后重查才是唯一可靠时机;推进串行化由办件乐观锁兜底
+- * (重试循环在监听器侧,跨 bean 调用 {@link FlowAdvanceService#tryAdvance} 保证 REQUIRES_NEW 生效)。
+- */
++import java.time.LocalDateTime;
++import java.util.HashMap;
++import java.util.Map;
++
++/** TASK_CREATED establishes membership; TASK_HANDLED refreshes that same activation in the handling transaction. */
+ @Slf4j
+ @Component
+ @RequiredArgsConstructor
+ public class FlowNodeAdvanceListener {
+-
+ private final FlowAdvanceService advanceService;
++ private final FlwAdvanceIntentRepository intentRepository;
++
++ @EventListener
++ @Order(Ordered.HIGHEST_PRECEDENCE + 2)
++ public void persistIntent(FlowEvent event) {
++ if (event.getData() == null) return;
++ if (event.getType() == FlowEvent.Type.TASK_CREATED) {
++ ActivationBatch batch = batch();
++ if (!batch.tasks.add(event.getTaskId())) return;
++ String key = event.getInstanceId() + ":" + event.getData().get("nodeId");
++ int total = ((Number) event.getData().get("totalTasks")).intValue();
++ if (total <= 0) throw new IllegalStateException("Missing activation batch size");
++ Registration registration = batch.nodes.get(key);
++ if (registration == null) { registration = new Registration(total);batch.nodes.put(key, registration); }
++ if (registration.total != total) throw new IllegalStateException("Activation batch size changed");
++ registration.intentId = advanceService.registerTask(event, registration.intentId);
++ if (--registration.remaining == 0) batch.nodes.remove(key);
++ } else if (event.getType() == FlowEvent.Type.TRANSFERRED) {
++ advanceService.registerTransfer(event);
++ } else if (event.getType() == FlowEvent.Type.TASK_HANDLED && event.getData().get("nodeId") != null) {
++ advanceService.recordHandled(event);
++ }
++ }
+
+ @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true)
+ public void onTaskHandled(FlowEvent event) {
+- if (event.getType() != FlowEvent.Type.TASK_HANDLED || event.getData() == null) {
+- return;
+- }
++ if (event.getType() != FlowEvent.Type.TASK_HANDLED || event.getData() == null) return;
+ String nodeId = (String) event.getData().get("nodeId");
+- if (nodeId == null) {
+- return;
+- }
+- String edgeName = (String) event.getData().get("edgeName");
+- // 办理人随事件 data 穿透进推进链:新 TASK_CREATED/COMPLETED 的操作人信封字段(2026-09-20)
+- FlowOperator operator = FlowOperator.of(
+- (Integer) event.getData().get("assigneeId"),
+- (String) event.getData().get("assigneeName"));
++ if (nodeId == null) return;
++ FlowAdvanceService.IntentRef ref = advanceService.pendingForTask(event.getInstanceId(), nodeId, event.getTaskId());
++ if (ref == null) return;
+ for (int attempt = 1; attempt <= 3; attempt++) {
+- try {
+- advanceService.tryAdvance(event.getInstanceId(), nodeId, edgeName, operator);
+- return;
+- } catch (ObjectOptimisticLockingFailureException e) {
+- log.warn("[flow-center] 推进冲突重试 {}/3:instance={} node={}",
+- attempt, event.getInstanceId(), nodeId);
++ try { advanceService.tryAdvance(ref.id, ref.revision);return; }
++ catch (ObjectOptimisticLockingFailureException conflict) {
++ log.warn("[flow-center] D09 advance conflict {}/3: intent={} revision={}", attempt, ref.id, ref.revision);
++ } catch (Exception failure) {
++ advanceService.defer(ref.id, ref.revision, failure.toString());return;
+ }
+ }
+- log.error("[flow-center] 推进重试耗尽(请人工检查):instance={} node={}",
+- event.getInstanceId(), nodeId);
++ advanceService.defer(ref.id, ref.revision, "AFTER_COMMIT retries exhausted; sweep will resume");
++ }
++
++ @Scheduled(fixedDelay = 30_000)
++ public void sweepIntents() {
++ for (FlwAdvanceIntent detached : intentRepository.findTop50ByStatusAndNextRetryAtLessThanEqualOrderByCreateTimeAsc(
++ FlwAdvanceIntent.STATE_PENDING, LocalDateTime.now())) {
++ Long id = detached.getId(), revision = detached.getRevision();
++ try { advanceService.tryAdvance(id, revision); }
++ catch (Exception failure) { advanceService.defer(id, revision, failure.toString()); }
++ }
++ }
++
++ /** Synchronizations are suspended with REQUIRES_NEW; an outer activation batch cannot leak into its child. */
++ private static ActivationBatch batch() {
++ if (!TransactionSynchronizationManager.isSynchronizationActive()) throw new IllegalStateException("Activation requires its creation transaction");
++ for (TransactionSynchronization sync : TransactionSynchronizationManager.getSynchronizations()) {
++ if (sync instanceof ActivationBatch) return (ActivationBatch) sync;
++ }
++ ActivationBatch created = new ActivationBatch();
++ TransactionSynchronizationManager.registerSynchronization(created);return created;
++ }
++ private static final class ActivationBatch implements TransactionSynchronization {
++ final Map nodes = new HashMap<>();
++ final java.util.Set tasks = new java.util.HashSet<>();
++ @Override public void beforeCommit(boolean readOnly) {
++ if (!nodes.isEmpty()) throw new IllegalStateException("Incomplete activation membership; creation transaction must roll back");
++ }
++ }
++ private static final class Registration {
++ final int total;
++ int remaining;
++ Long intentId;
++ Registration(int total) { this.total = total;this.remaining = total; }
+ }
+ }
+diff --git a/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackLog.java b/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackLog.java
+index a2cf36c..85d38c0 100644
+--- a/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackLog.java
++++ b/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackLog.java
+@@ -12,14 +12,16 @@ import javax.persistence.UniqueConstraint;
+ import java.time.LocalDateTime;
+
+ /**
+- * 回调推送日志 = 重试队列(概览 §4.9):eventId 唯一(重试不变,业务方按它幂等去重);
++ * 回调推送日志 = 重试队列(概览 §4.9;2026-10-25 R2 幂等键改复合):eventId 全订阅方复用
++ * (重试不变,业务方按它幂等去重),(event_id, reg_id) 复合唯一——同一事件对同一订阅方只建
++ * 一行,多订阅方各一行;并发 exists→insert 竞态由复合 uk 兜底(DuplicateKeyException=已建行)。
+ * payload_json 完整快照可重放;索引 (status, next_retry_time) 是扫描主路径。
+ */
+ @Getter
+ @Setter
+ @Entity
+ @Table(name = "flw_callback_log",
+- uniqueConstraints = @UniqueConstraint(name = "uk_flw_log_event", columnNames = "event_id"),
++ uniqueConstraints = @UniqueConstraint(name = "uk_flw_log_event_reg", columnNames = {"event_id", "reg_id"}),
+ indexes = {
+ @Index(name = "idx_flw_log_scan", columnList = "status, next_retry_time"),
+ @Index(name = "idx_flw_log_inst", columnList = "instance_id")})
+diff --git a/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackReg.java b/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackReg.java
+index 5e04140..7f83a56 100644
+--- a/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackReg.java
++++ b/src/main/java/com/yunhai/flowcenter/entity/FlwCallbackReg.java
+@@ -21,6 +21,9 @@ import javax.persistence.UniqueConstraint;
+ columnNames = {"sch_id", "app_key", "flow_guid"}),
+ indexes = @Index(name = "idx_flw_reg_flow", columnList = "sch_id, flow_guid, status"))
+ public class FlwCallbackReg extends BaseEntity {
++ @Column(name="source_principal_id",length=36)
++ private String sourcePrincipalId;
++
+
+ /** 租户键 */
+ @Column(name = "sch_id", nullable = false)
+diff --git a/src/main/java/com/yunhai/flowcenter/entity/FlwDef.java b/src/main/java/com/yunhai/flowcenter/entity/FlwDef.java
+index edc7724..9204b9a 100644
+--- a/src/main/java/com/yunhai/flowcenter/entity/FlwDef.java
++++ b/src/main/java/com/yunhai/flowcenter/entity/FlwDef.java
+@@ -74,6 +74,15 @@ public class FlwDef extends BaseEntity {
+ @Column(name = "form_version")
+ private Integer formVersion;
+
++ /**
++ * 线级停发标志(2026-10-25 T03):1=该 flowGuid 整线停发——阻止新发起/预览,不影响在途实例
++ * (在途按实例固定 defId 原版推进)。与版本级 status(DISABLED) 正交:线级停发后 enable()
++ * 恢复的只是版本状态,线仍不可新发起;enableLine 才恢复线可用性,且不改写任何发布内容字节。
++ * 批量 UPDATE 全版本行同值,读任一行即读线状态。
++ */
++ @Column(name = "line_offline", nullable = false, columnDefinition = "INT NOT NULL DEFAULT 0")
++ private Integer lineOffline = 0;
++
+ /** §5.1 流程定义 JSON */
+ @Column(name = "content_json", nullable = false, columnDefinition = "LONGTEXT")
+ private String contentJson;
+diff --git a/src/main/java/com/yunhai/flowcenter/entity/FlwInstance.java b/src/main/java/com/yunhai/flowcenter/entity/FlwInstance.java
+index 42be69b..b8e8143 100644
+--- a/src/main/java/com/yunhai/flowcenter/entity/FlwInstance.java
++++ b/src/main/java/com/yunhai/flowcenter/entity/FlwInstance.java
+@@ -14,8 +14,9 @@ import java.time.LocalDateTime;
+
+ /**
+ * 办件实例(概览 §4.3 + V2.0 Part D):托管模式下业务对象留在业务方,本表只存
+- * biz_title/biz_url/biz_snapshot 快照与关联键;uk(sch_id, biz_type, biz_id) 幂等——
+- * 同一业务对象一条办件,重复代发起返回原办件;驳回/撤回后重提复用同一行(R4 重启语义)。
++ * biz_title/biz_url/biz_snapshot 快照与关联键。正式命令每轮独立实例,唯一域为
++ * (school, sourcePrincipalId, bizType, bizId, applyRound),单在途由坐标表事务指针保护。
++ * sourcePrincipalId=NULL / legacySlot=1 仅保留迁移前旧入口的一业务一行语义;正式模式关闭旧写入口。
+ *
+ * version 是乐观锁(推进串行化)兼回调 payload 的 instanceVersion(业务方按它丢弃乱序旧事件),
+ * 由 JPA @Version 在 flush 时自增。
+@@ -24,12 +25,19 @@ import java.time.LocalDateTime;
+ @Setter
+ @Entity
+ @Table(name = "flw_instance",
+- uniqueConstraints = @UniqueConstraint(name = "uk_flw_instance_biz",
+- columnNames = {"sch_id", "biz_type", "biz_id"}),
++ uniqueConstraints = {
++ @UniqueConstraint(name = "uk_flw_instance_legacy", columnNames = {"sch_id", "biz_type", "biz_id", "legacy_slot"}),
++ @UniqueConstraint(name = "uk_flw_instance_round", columnNames = {"sch_id", "source_principal_id", "biz_type", "biz_id", "apply_round"})},
+ indexes = {
+ @Index(name = "idx_flw_inst_submitter", columnList = "sch_id, submitter_id, status"),
+ @Index(name = "idx_flw_inst_status_time", columnList = "sch_id, status, submit_time")})
+ public class FlwInstance extends BaseEntity {
++ @Column(name="source_principal_id",length=36,updatable=false)
++ private String sourcePrincipalId;
++ /** 1 for legacy writers, NULL for formal commands. Migration must drop the old three-column UK. */
++ @Column(name="legacy_slot")
++ private Integer legacySlot = 1;
++
+
+ /** 租户键 */
+ @Column(name = "sch_id", nullable = false)
+@@ -141,7 +149,7 @@ public class FlwInstance extends BaseEntity {
+ private String applyRequestToken;
+
+ /**
+- * 当前发起轮次(1 起;随新一轮请求标识进位)。
++ * 当前发起轮次(1 起)。正式命令由服务端坐标锁内分配,不接受客户端 attempt 分轮。
+ *
+ *
轮次守卫是单调的:请求带来的 attempt 低于行上的 applyRound 就是陈旧轮次的迟到重投,
+ * 只回读现状。两侧(消费方命令行 + 本行)都持久化,崩溃恢复后仍能判出"哪一轮才是当前轮"。
+@@ -149,6 +157,22 @@ public class FlwInstance extends BaseEntity {
+ @Column(name = "apply_round")
+ private Integer applyRound;
+
++ /**
++ * 受理时规范载荷指纹(2026-10-25 T04,SHA-256 十六进制):同 requestToken 必须同指纹——
++ * 同 ID 不同载荷是两份命令,撞在同一个 token 上按冲突拒绝,不伪装成幂等成功。
++ * 存量行(指纹列为 NULL)跳过该核对(守卫缺省是关,与 token/round 存量纪律一致)。
++ */
++ @Column(name = "apply_payload_hash", length = 64)
++ private String applyPayloadHash;
++
++ /**
++ * 事件序号计数器(2026-10-25 T05):实例内事件事实 eventSeq 的发号器。发号时在事务内
++ * 悲观锁住本行(findByIdForUpdate 当前读)再 +1——REPEATABLE READ 下对聚合做快照读会
++ * 拿到陈旧 max(FlowEngineIT 会签并发实证),行上计数器+行锁才是串行安全的。
++ */
++ @Column(name = "event_seq_counter")
++ private Integer eventSeqCounter;
++
+ /**
+ * 台账展示位(F11-4):该办件是否「路由挂起」(审批中且最新 attempt 的路由判断行无出口)。
+ * 非持久列——由管理端办件分页(FlowAdminController.instancePage)按页查询后填充,
+diff --git a/src/main/java/com/yunhai/flowcenter/model/FlowApplyRetryModels.java b/src/main/java/com/yunhai/flowcenter/model/FlowApplyRetryModels.java
+index 946caf3..d31fc8e 100644
+--- a/src/main/java/com/yunhai/flowcenter/model/FlowApplyRetryModels.java
++++ b/src/main/java/com/yunhai/flowcenter/model/FlowApplyRetryModels.java
+@@ -39,6 +39,11 @@ public final class FlowApplyRetryModels {
+ * 这样"网络重投"与"用户授权重提"在中心侧也是两个不同的事实。
+ */
+ private Boolean retryOnly;
++ /**
++ * 期望的前一轮次(T04 服务端轮次,可选):显式重提必须声明基于第 N 轮;
++ * 服务端在坐标锁内核对并只分配 N+1,不信客户端 attempt。null=既有语义。
++ */
++ private Integer expectedPrevRound;
+ }
+
+ /** v2 响应:在旧 {instanceId,duplicated} 之上把"中心判出来的是哪一种命中"说清楚 */
+@@ -60,6 +65,9 @@ public final class FlowApplyRetryModels {
+ private String branch;
+ /** 行上的当前轮次(消费方据此校正自己的命令表认知) */
+ private Integer applyRound;
++ /** 同一响应的真实状态/实例版本;分支判据保持不变。 */
++ private Integer status;
++ private Integer version;
+ }
+
+ /** 分支码常量(与 {@link ApplyResultV2#branch} 同值域) */
+@@ -71,6 +79,11 @@ public final class FlowApplyRetryModels {
+ public static final String REPLAY_ONLY = "REPLAY_ONLY";
+ public static final String RESTARTED = "RESTARTED";
+ public static final String CONCURRENT_CREATED = "CONCURRENT_CREATED";
++ /**
++ * 冲突(2026-10-25 T04):同 token 异载荷、或带 token 撞上他人/其他轮次的在途与通过件——
++ * 调用方(OapiFlowService)把该分支翻译成业务异常,不伪装幂等成功。
++ */
++ public static final String CONFLICT = "CONFLICT";
+
+ private Branch() {
+ }
+diff --git a/src/main/java/com/yunhai/flowcenter/model/OapiModels.java b/src/main/java/com/yunhai/flowcenter/model/OapiModels.java
+index 4760a6c..611203f 100644
+--- a/src/main/java/com/yunhai/flowcenter/model/OapiModels.java
++++ b/src/main/java/com/yunhai/flowcenter/model/OapiModels.java
+@@ -107,6 +107,13 @@ public final class OapiModels {
+ @Data
+ public static class InstanceVo {
+ private Long instanceId;
++ private Integer applyRound;
++ private String requestToken;
++ /**
++ * 来源应用快照(2026-10-25 T04):UNKNOWN 恢复链按"稳定来源+requestToken"核验命令归属的
++ * 依据——查得实例但 applyAppKey 与本应用不符=他人办件,不得认领(P0-4 关闭点)。
++ */
++ private String applyAppKey;
+ private Integer status;
+ private Integer version;
+ private String nodeSummary;
+diff --git a/src/main/java/com/yunhai/flowcenter/repository/FlwCallbackLogRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwCallbackLogRepository.java
+index 03839d6..ee0a176 100644
+--- a/src/main/java/com/yunhai/flowcenter/repository/FlwCallbackLogRepository.java
++++ b/src/main/java/com/yunhai/flowcenter/repository/FlwCallbackLogRepository.java
+@@ -13,6 +13,9 @@ import java.util.List;
+
+ public interface FlwCallbackLogRepository extends JpaRepository {
+
++ /** T05 投递幂等:同一事件事实对同一订阅方只建一行(补分发与正常分发竞争时去重) */
++ boolean existsByEventIdAndRegId(String eventId, Long regId);
++
+ /** 重推扫描主路径:待推/重试中且到点 */
+ List findTop50ByStatusAndNextRetryTimeLessThanEqualOrderByNextRetryTimeAsc(
+ Integer status, LocalDateTime now);
+diff --git a/src/main/java/com/yunhai/flowcenter/repository/FlwCcRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwCcRepository.java
+index 72c2488..cf8552a 100644
+--- a/src/main/java/com/yunhai/flowcenter/repository/FlwCcRepository.java
++++ b/src/main/java/com/yunhai/flowcenter/repository/FlwCcRepository.java
+@@ -14,7 +14,7 @@ public interface FlwCcRepository extends JpaRepository {
+ /** 抄送我的(用户面页签,走 user+sent_time 索引;appKey 可空=不按应用过滤,口径=定义归属 flw_def.app_key) */
+ @Query("select c from FlwCc c join FlwInstance i on i.id = c.instanceId join FlwDef d on d.id = i.defId "
+ + "where c.schId = :schId and c.userId = :userId "
+- + "and (:appKey is null or d.appKey = :appKey) order by c.sentTime desc")
++ + "and (:appKey is null or i.sourcePrincipalId = :appKey or (i.sourcePrincipalId is null and i.applyAppKey = :appKey)) order by c.sentTime desc")
+ Page pageCc(@Param("schId") Long schId, @Param("userId") Integer userId,
+ @Param("appKey") String appKey, Pageable pageable);
+
+diff --git a/src/main/java/com/yunhai/flowcenter/repository/FlwDefRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwDefRepository.java
+index 83f11b7..c582182 100644
+--- a/src/main/java/com/yunhai/flowcenter/repository/FlwDefRepository.java
++++ b/src/main/java/com/yunhai/flowcenter/repository/FlwDefRepository.java
+@@ -32,12 +32,22 @@ public interface FlwDefRepository extends JpaRepository {
+
+ List findBySchIdOrderByUpdateTimeDesc(Long schId);
+
+- /** 已发布流程列表(oapi list 数据源):每条流程线取最高已发布版本 */
+- @Query("select d from FlwDef d where d.schId = :schId and d.status = 1 "
++ /**
++ * 已发布流程列表(oapi list 数据源):每条流程线取最高已发布版本。
++ * 2026-10-25 T03:外层追加 {@code lineOffline=0}——线级停发的整线不出现在新发起预览源,
++ * 且不存在"停发后回退旧版"的形态(内层仍是该线已发布最高版本,停发不改变版本事实)。
++ */
++ @Query("select d from FlwDef d where d.schId = :schId and d.status = 1 and d.lineOffline = 0 "
+ + "and d.version = (select max(d2.version) from FlwDef d2 where d2.schId = d.schId "
+ + "and d2.flowGuid = d.flowGuid and d2.status = 1) order by d.updateTime desc")
+ List findPublishedLatestPerGuid(@Param("schId") Long schId);
+
++ /** 线级停发/恢复(T03):批量 UPDATE 全版本行同值;返回受影响行数 */
++ @org.springframework.data.jpa.repository.Modifying
++ @Query("update FlwDef d set d.lineOffline = :offline where d.schId = :schId and d.flowGuid = :flowGuid")
++ int updateLineOffline(@Param("schId") Long schId, @Param("flowGuid") String flowGuid,
++ @Param("offline") Integer offline);
++
+ /** 在途统计(议题二④):一次取回多条流程线的全部版本行 */
+ List findBySchIdAndFlowGuidIn(Long schId, Collection flowGuids);
+ }
+diff --git a/src/main/java/com/yunhai/flowcenter/repository/FlwInstanceRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwInstanceRepository.java
+index 1cc57d6..05e6cd0 100644
+--- a/src/main/java/com/yunhai/flowcenter/repository/FlwInstanceRepository.java
++++ b/src/main/java/com/yunhai/flowcenter/repository/FlwInstanceRepository.java
+@@ -12,8 +12,35 @@ import java.util.List;
+ import java.util.Optional;
+
+ public interface FlwInstanceRepository extends JpaRepository {
++ @org.springframework.data.jpa.repository.Lock(javax.persistence.LockModeType.PESSIMISTIC_WRITE)
++ @Query("select i from FlwInstance i where i.id=:id")
++ Optional findFormalPreviousForUpdate(@Param("id") Long id);
++
++
++ /**
++ * T05 事件序号发号(2026-10-25 FlowEngineIT 会签并发实证三轮修正,终形态):
++ * ①FOR UPDATE 标量当前读发号(取行 X 锁,任何隔离级别都看到已提交最新值)→
++ * ②按读到的值回写计数器(值已在我锁内计算,无竞态)→③事实行按该值落库。
++ * 纯快照读 max+1 会撞 uk;实体 merge 会触发 @Version 冲突;自增 UPDATE 后普通 SELECT
++ * 的读回在并发窗口同样不可靠——FOR UPDATE 标量读是唯一两全形态。
++ */
++ @org.springframework.data.jpa.repository.Query(
++ value = "select coalesce(event_seq_counter, 0) + 1 from flw_instance where id = :id for update",
++ nativeQuery = true)
++ Integer nextEventSeqForUpdate(@Param("id") Long id);
++
++ @org.springframework.data.jpa.repository.Modifying
++ @org.springframework.data.jpa.repository.Query(
++ value = "update flw_instance set event_seq_counter = :v where id = :id", nativeQuery = true)
++ int setEventSeqCounter(@Param("id") Long id, @Param("v") Integer v);
++
++ Optional findFirstBySchIdAndSourcePrincipalIdAndBizTypeAndBizIdOrderByApplyRoundDesc(Long school, String source, String type, String biz);
++
++ @Query("select i from FlwInstance i where i.schId=:school and i.sourcePrincipalId=:source and (:bizType is null or i.bizType=:bizType) and (:status is null or i.status=:status) order by i.id desc")
++ Page pageSelf(@Param("school") Long school,@Param("source") String source,@Param("bizType") String bizType,@Param("status") Integer status, Pageable pageable);
+
+ /** biz uk 幂等主键 */
++ @Query("select i from FlwInstance i where i.schId=?1 and i.bizType=?2 and i.bizId=?3 and i.sourcePrincipalId is null")
+ Optional findBySchIdAndBizTypeAndBizId(Long schId, String bizType, String bizId);
+
+ List findBySchIdAndSubmitterIdAndStatusOrderByIdDesc(Long schId, Integer submitterId, Integer status);
+diff --git a/src/main/java/com/yunhai/flowcenter/repository/FlwTaskRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwTaskRepository.java
+index dc21dfc..f196414 100644
+--- a/src/main/java/com/yunhai/flowcenter/repository/FlwTaskRepository.java
++++ b/src/main/java/com/yunhai/flowcenter/repository/FlwTaskRepository.java
+@@ -13,6 +13,12 @@ import java.util.List;
+
+ public interface FlwTaskRepository extends JpaRepository {
+
++ /** Additive D09 query: the real transfer parent is in this set. Ambiguous generations must be rejected. */
++ @Query(value="select t.id from flw_task t join flw_task n on n.id=:newTaskId "
++ + "where t.instance_id=n.instance_id and t.node_id=n.node_id and t.status=5 "
++ + "and t.assignee_id=:operatorId and t.handle_time=n.arrive_time", nativeQuery=true)
++ List findTransferParentCandidates(@Param("newTaskId") Long newTaskId, @Param("operatorId") Integer operatorId);
++
+ List findByInstanceIdOrderByIdAsc(Long instanceId);
+
+ List findByInstanceIdAndStatus(Long instanceId, Integer status);
+@@ -52,10 +58,13 @@ public interface FlwTaskRepository extends JpaRepository {
+ @Query("update FlwTask t set t.status = 3 where t.instanceId = :instanceId and t.status = 0 and t.id <> :exceptTaskId")
+ int invalidatePendingExcept(@Param("instanceId") Long instanceId, @Param("exceptTaskId") Long exceptTaskId);
+
++ @Query("select t from FlwTask t join FlwInstance i on i.id=t.instanceId where t.schId=:school and i.sourcePrincipalId=:source and t.assigneeId=:user and t.status=0 and (:type is null or i.bizType=:type) order by t.id asc")
++ Page pageSelfTodo(@Param("school") Long school,@Param("source") String source,@Param("user") Integer user,@Param("type") String type, Pageable pageable);
++
+ /** 我的待办(主路径,走 assignee 索引;appKey 可空=不按应用过滤,口径=办件定义归属应用 flw_def.app_key) */
+ @Query("select t from FlwTask t join FlwInstance i on i.id = t.instanceId join FlwDef d on d.id = i.defId "
+ + "where t.schId = :schId and t.assigneeId = :assigneeId and t.status = 0 "
+- + "and (:appKey is null or d.appKey = :appKey) "
++ + "and (:appKey is null or i.sourcePrincipalId = :appKey or (i.sourcePrincipalId is null and i.applyAppKey = :appKey)) "
+ + "order by t.arriveTime asc, t.id asc")
+ Page pageTodo(@Param("schId") Long schId, @Param("assigneeId") Integer assigneeId,
+ @Param("appKey") String appKey, Pageable pageable);
+@@ -84,7 +93,7 @@ public interface FlwTaskRepository extends JpaRepository {
+ @Query("select t from FlwTask t join FlwInstance i on i.id = t.instanceId join FlwDef d on d.id = i.defId "
+ + "where t.schId = :schId and t.assigneeId = :assigneeId "
+ + "and t.status in :statuses "
+- + "and (:appKey is null or d.appKey = :appKey) "
++ + "and (:appKey is null or i.sourcePrincipalId = :appKey or (i.sourcePrincipalId is null and i.applyAppKey = :appKey)) "
+ + "and t.id = (select max(t2.id) from FlwTask t2 where t2.instanceId = t.instanceId "
+ + " and t2.assigneeId = :assigneeId and t2.status in :statuses) "
+ + "order by t.handleTime desc, t.id desc")
+diff --git a/src/main/java/com/yunhai/flowcenter/service/FlowCallbackAdminService.java b/src/main/java/com/yunhai/flowcenter/service/FlowCallbackAdminService.java
+index e8d717d..1725612 100644
+--- a/src/main/java/com/yunhai/flowcenter/service/FlowCallbackAdminService.java
++++ b/src/main/java/com/yunhai/flowcenter/service/FlowCallbackAdminService.java
+@@ -33,6 +33,7 @@ public class FlowCallbackAdminService {
+ private final FlwCallbackLogRepository logRepository;
+ private final CallbackPushService pushService;
+ private final CallbackProperties callbackProperties;
++ private final com.yunhai.flowcenter.callback.CallbackLogStore logStore;
+
+ public List listRegs(Long schId) {
+ return regRepository.findBySchIdOrderByUpdateTimeDesc(schId);
+@@ -104,7 +105,6 @@ public class FlowCallbackAdminService {
+ * 管理端手工标记死信(持续重试语义下死信的唯一人工入口:人工判定事件无意义时止血)。
+ * 标记后可经既有 rePush 复活。
+ */
+- @Transactional
+ public void markDead(Long schId, Long logId) {
+ FlwCallbackLog callbackLog = logRepository.findById(logId)
+ .orElseThrow(() -> new BizException(ErrorCode.PARAM_ERROR, "推送日志不存在:" + logId));
+@@ -119,7 +119,6 @@ public class FlowCallbackAdminService {
+ }
+
+ /** 手工重推(死信/失败日志):置回待推并立即触发一次推送 */
+- @Transactional
+ public void rePush(Long schId, List logIds) {
+ if (logIds == null || logIds.isEmpty()) {
+ throw new BizException(ErrorCode.PARAM_ERROR, "请选择要重推的日志");
+@@ -128,20 +127,9 @@ public class FlowCallbackAdminService {
+ if (logs.size() != logIds.size()) {
+ throw new BizException(ErrorCode.PARAM_ERROR, "存在不属于该校或不存在 dead/pending 日志");
+ }
+- for (FlwCallbackLog callbackLog : logs) {
+- if (callbackLog.getStatus() == FlwEnums.CallbackLogStatus.SUCCESS.getCode()) {
+- continue;
+- }
+- callbackLog.setStatus(FlwEnums.CallbackLogStatus.PENDING.getCode());
+- callbackLog.setRetryCount(0);
+- callbackLog.setNextRetryTime(LocalDateTime.now());
+- logRepository.save(callbackLog);
+- }
+- logRepository.flush();
+- for (FlwCallbackLog callbackLog : logs) {
+- if (callbackLog.getStatus() != null) {
+- pushService.pushAsync(callbackLog.getId());
+- }
++ // 短事务提交并释放会话锁后才执行 HTTP;不能把管理端旧实体全量回写覆盖刚成功的状态。
++ for (Long id : logStore.resetForReplay(schId, logIds)) {
++ pushService.pushAsync(id);
+ }
+ }
+
+diff --git a/src/main/java/com/yunhai/flowcenter/service/FlowDefAdminService.java b/src/main/java/com/yunhai/flowcenter/service/FlowDefAdminService.java
+index 001e9d1..da109e3 100644
+--- a/src/main/java/com/yunhai/flowcenter/service/FlowDefAdminService.java
++++ b/src/main/java/com/yunhai/flowcenter/service/FlowDefAdminService.java
+@@ -3,10 +3,12 @@ package com.yunhai.flowcenter.service;
+ import com.fasterxml.jackson.databind.ObjectMapper;
+ import com.yunhai.flowcenter.engine.DefValidator;
+ import com.yunhai.flowcenter.engine.FlowDefModel;
++import com.yunhai.flowcenter.engine.VariableContract;
+ import com.yunhai.flowcenter.engine.FlowEngine;
+ import com.yunhai.flowcenter.entity.FlwAssigneeMap;
+ import com.yunhai.flowcenter.entity.FlwDef;
+ import com.yunhai.flowcenter.enums.FlwEnums;
++import com.yunhai.flowcenter.enums.FormBindingErrorCode;
+ import com.yunhai.flowcenter.repository.FlwAssigneeMapRepository;
+ import com.yunhai.flowcenter.repository.FlwDefRepository;
+ import com.yunhai.platform.apistd.BizException;
+@@ -192,12 +194,13 @@ public class FlowDefAdminService {
+ boolean pairing = formUuid != null && !clearing && formVersion == null;
+ boolean versionOnly = formUuid == null && formVersion != null;
+ if (pairing || versionOnly) {
+- throw new BizException(ErrorCode.PARAM_ERROR,
++ throw new BizException(formVersion == null ? FormBindingErrorCode.VERSION_REQUIRED
++ : FormBindingErrorCode.INVALID_COORDINATE,
+ "关联表单必须成对提交 formUuid 与 formVersion(只传一个拒):当前 formUuid="
+ + (formUuid == null ? "未传" : formUuid.trim()) + ",formVersion=" + formVersion);
+ }
+ if (formUuid != null && !clearing) {
+- checkedSchema(schId, formUuid.trim(), formVersion); // Validate before the first save.
++ checkedSchema(schId, formUuid.trim(), formVersion, "保存"); // Validate before the first save.
+ }
+ FlwDef saved = save(schId, id, code, name, contentJson, baseGuid, appKey, changeLog);
+ if (formUuid != null) {
+@@ -216,7 +219,7 @@ public class FlowDefAdminService {
+ * 2026-10-20 C1 第一刀:def 关联了表单(form_uuid 非空)时,发布期拉 form-center schema 快照
+ * 嵌入版本行 content_json.formContract(与 defId 快照纪律同构:在途办件按发起时快照校验),
+ * 并做表达式键查①红牌(变量 ⊆ 快照字段 ∪ 内置变量)。拉取失败=发布阻断(可重试);
+- * 运行期对 form-center 零依赖。form_uuid 为空(存量线/Mode C)行为逐字不变。 */
++ * 运行期对 form-center 零依赖;新发布的无绑定版本必须显式声明 variables。 */
+ @Transactional
+ public FlwDef publish(Long schId, Long id) {
+ FlwDef def = requireDef(schId, id);
+@@ -227,86 +230,33 @@ public class FlowDefAdminService {
+ throw new BizException(ErrorCode.PARAM_ERROR, "停用版本请走启用;修改表单关联须保存为新版本草稿");
+ }
+ FlowDefModel model = FlowDefModel.parse(objectMapper, def.getContentJson());
+- DefValidator.validate(model, enabledDeptKeys(schId), enabledRouterKeys());
+ if (def.getFormUuid() != null && !def.getFormUuid().trim().isEmpty()) {
+- Map snapshot;
+- try {
+- // 管理链始终使用确切版本;缺版本须重新选择,不回落当前发布版。
+- snapshot = checkedSchema(schId, def.getFormUuid().trim(), def.getFormVersion());
+- } catch (FormCenterSchemaClient.SchemaUnavailableException e) {
+- throw new BizException(ErrorCode.PARAM_ERROR,
+- "关联表单快照拉取失败,发布已阻断(可重试):" + e.getMessage());
+- } catch (FormCenterSchemaClient.SchemaContractViolationException e) {
+- throw new BizException(ErrorCode.PARAM_ERROR,
+- "关联表单版本核验失败,发布已阻断(重试无意义,请重新选择表单与版本):" + e.getMessage());
+- }
+- List problems = new ArrayList<>();
+- validateExpressionKeysAgainstSnapshot(model, snapshot, problems);
+- validateExpressionTypesAndDomains(model, snapshot, problems);
+- if (!problems.isEmpty()) {
+- throw new BizException(ErrorCode.PARAM_ERROR,
+- "发布校验失败(表达式三查):\n" + String.join("\n", problems));
+- }
+- def.setContentJson(embedFormContract(def.getContentJson(), snapshot));
++ Map snapshot = checkedSchema(schId, def.getFormUuid().trim(),
++ def.getFormVersion(), "发布");
++ model.variables = VariableContract.fromSnapshot(snapshot);
++ VariableContract.validateDefinition(model, true);
++ DefValidator.validate(model, enabledDeptKeys(schId), enabledRouterKeys());
++ def.setContentJson(embedFormContract(def.getContentJson(), snapshot, model.variables));
++ } else if (def.getFormVersion() != null) {
++ throw new BizException(FormBindingErrorCode.INVALID_COORDINATE, "关联表单版本必须同时绑定 formUuid");
++ }
++ if (def.getFormUuid() == null || def.getFormUuid().trim().isEmpty()) {
++ VariableContract.validateDefinition(model, true);
++ DefValidator.validate(model, enabledDeptKeys(schId), enabledRouterKeys());
+ }
+ def.setStatus(FlwEnums.DefStatus.PUBLISHED.getCode());
+ def.setPublishTime(LocalDateTime.now());
+ return defRepository.save(def);
+ }
+
+- /**
+- * 表达式键查①(C1 三查之一,红牌):对每条 EXPRESSION 判断的出边表达式,
+- * 变量集 ⊆ 快照字段 keys ∪ 内置变量({@link DefValidator#EXPRESSION_BUILTINS})。
+- * 不执行表达式、不改沙箱(口径与 v2 方案 §4.7 一致)。
+- */
+- private void validateExpressionKeysAgainstSnapshot(FlowDefModel model, Map snapshot,
+- List problems) {
+- @SuppressWarnings("unchecked")
+- List