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 30eb842..84b7efc 100644
--- a/src/main/java/com/yunhai/flowcenter/engine/FlowEvent.java
+++ b/src/main/java/com/yunhai/flowcenter/engine/FlowEvent.java
@@ -71,6 +71,46 @@ public class FlowEvent {
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<>();
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/engine/VariableContract.java b/src/main/java/com/yunhai/flowcenter/engine/VariableContract.java
new file mode 100644
index 0000000..d12d817
--- /dev/null
+++ b/src/main/java/com/yunhai/flowcenter/engine/VariableContract.java
@@ -0,0 +1,209 @@
+package com.yunhai.flowcenter.engine;
+
+import com.yunhai.platform.apistd.BizException;
+import com.yunhai.platform.utils.exception.ErrorCode;
+
+import java.math.BigDecimal;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.util.*;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/** D10 定义与发起共用的纯校验器;不转换输入,不依赖 FC/数据库。 */
+public final class VariableContract {
+ private VariableContract() { }
+ private static final Set TYPES = new HashSet<>(Arrays.asList(
+ "string", "number", "boolean", "date", "datetime", "money", "stringArray", "fileArray", "userArray"));
+ private static final String TOKEN = "(?:'(?:[^'\\\\]|\\\\.)*'|\"(?:[^\"\\\\]|\\\\.)*\"|-?\\d+(?:\\.\\d+)?|[A-Za-z_][A-Za-z0-9_]*)";
+ // 先分词以跳过完整字符串;不把函数参数字符串中的 "amount > 3" 当成比较。
+ private static final Pattern TOKENS = Pattern.compile(TOKEN + "|==|!=|>=|<=|>|<|\\S");
+ private static final Set COMPARISONS = new HashSet<>(Arrays.asList("==", "!=", ">=", "<=", ">", "<"));
+
+ public static Set referencedKeys(FlowDefModel model) {
+ Set keys = new LinkedHashSet<>();
+ for (FlowDefModel.Node n : model.nodes) {
+ if (n == null || !n.expressionJudge()) continue;
+ for (FlowDefModel.Edge e : model.outEdges(n.id)) {
+ if (e.condition == null || e.condition.trim().isEmpty()) continue;
+ try {
+ for (String key : ExpressionEvaluator.variableNames(e.condition.trim())) {
+ if (!DefValidator.EXPRESSION_BUILTINS.contains(key)) keys.add(key);
+ }
+ } catch (RuntimeException ex) {
+ // 历史版本保持原有语法异常跳过口径;新契约必须 fail closed。
+ if (model.variables != null) fail("表达式不合法:" + ex.getMessage());
+ }
+ }
+ }
+ return keys;
+ }
+
+ public static void validateDefinition(FlowDefModel model, boolean publishing) {
+ if (model.variables == null) {
+ if (publishing) fail("新发布版本必须声明 variables(无变量请显式声明 [])");
+ return;
+ }
+ Map fields = new LinkedHashMap<>();
+ for (FlowDefModel.Variable v : model.variables) {
+ if (v == null || v.key == null || !v.key.matches("[A-Za-z0-9_]{1,64}")) fail("变量 key 不合法");
+ if (DefValidator.EXPRESSION_BUILTINS.contains(v.key)) fail("变量不能占用内置键:" + v.key);
+ if (fields.put(v.key, v) != null) fail("变量 key 重复:" + v.key);
+ if (!TYPES.contains(v.type)) fail("变量类型不支持:" + v.key + " / " + v.type);
+ if (v.enumValues != null) {
+ if (!"string".equals(v.type) && !"stringArray".equals(v.type)) fail("enum 仅适用于字符串及字符串数组:" + v.key);
+ if (v.enumValues.isEmpty() || v.enumValues.contains(null)
+ || new HashSet<>(v.enumValues).size() != v.enumValues.size()) fail("enum 值域为空或重复:" + v.key);
+ for (String option : v.enumValues) {
+ if (blank(option)) fail("enum 值域不能含空白:" + v.key);
+ validateValue(v, "stringArray".equals(v.type) ? Collections.singletonList(option) : option, true);
+ }
+ }
+ }
+ for (String key : referencedKeys(model)) {
+ if (!fields.containsKey(key)) fail("表达式变量 " + key + " 未声明(不在关联表单字段集内或 variables 中)");
+ }
+ for (FlowDefModel.Node n : model.nodes) {
+ if (n == null || !n.expressionJudge()) continue;
+ for (FlowDefModel.Edge e : model.outEdges(n.id)) {
+ if (e.condition == null) continue;
+ Matcher m = TOKENS.matcher(e.condition);
+ List tokens = new ArrayList<>();
+ while (m.find()) tokens.add(m.group());
+ for (int i = 1; i + 1 < tokens.size(); i++) {
+ if (!COMPARISONS.contains(tokens.get(i))) continue;
+ checkLiteral(fields.get(tokens.get(i-1)), tokens.get(i+1));
+ checkLiteral(fields.get(tokens.get(i+1)), tokens.get(i-1));
+ }
+ }
+ }
+ }
+
+ private static void checkLiteral(FlowDefModel.Variable field, String token) {
+ if (field == null || "null".equals(token)) return;
+ Object value;
+ if (token.startsWith("'") || token.startsWith("\"")) {
+ // Aviator 的转义交由编译器;带转义字面量不做正则域推断。
+ if (token.indexOf('\\') >= 0) return;
+ value = token.substring(1, token.length() - 1);
+ } else if ("true".equals(token) || "false".equals(token)) value = Boolean.valueOf(token);
+ else if (token.matches("-?\\d+(\\.\\d+)?")) value = new BigDecimal(token);
+ else return;
+ try { validateValue(field, value, true); }
+ catch (BizException invalid) { fail("类型桶查②/域查③:" + invalid.getMessage()); }
+ }
+
+ public static void validateInput(FlowDefModel model, Map submitted) {
+ Set refs = referencedKeys(model);
+ if (model.variables == null) {
+ for (String key : refs) {
+ if (!submitted.containsKey(key) || submitted.get(key) == null) fail("流程条件所需变量缺失或为空(key=" + key + ")");
+ }
+ return;
+ }
+ validateDefinition(model, false);
+ Set declared = new HashSet<>();
+ for (FlowDefModel.Variable v : model.variables) {
+ declared.add(v.key);
+ validateValue(v, submitted.get(v.key), Boolean.TRUE.equals(v.required) || refs.contains(v.key));
+ }
+ for (String key : submitted.keySet()) if (!declared.contains(key)) fail("未声明的变量:" + key);
+ }
+
+ /** 必填/表达式引用均需实值;可选 null 可省略,其他非 null 值始终按类型与值域检查。 */
+ private static void validateValue(FlowDefModel.Variable v, Object value, boolean required) {
+ if (value == null) {
+ if (required) fail("变量缺失或为空:" + v.key);
+ return;
+ }
+ if (required && ((value instanceof String && blank((String) value))
+ || (value instanceof Collection && ((Collection>) value).isEmpty()))) fail("变量未填写:" + v.key);
+ // FC 文本/单选和日期控件的清空值为 "";仅非必填且未被表达式引用时视作未填。
+ // 数值控件空值是 null,不能借空字符串绕过 number/money/boolean/数组的类型检查。
+ if (!required && "".equals(value) && Arrays.asList("string", "date", "datetime").contains(v.type)) return;
+ boolean valid = false;
+ switch (v.type) {
+ case "string": valid = value instanceof String; break;
+ case "number": case "money":
+ if (value instanceof Number) {
+ try { new BigDecimal(value.toString()); valid = true; } catch (NumberFormatException ignored) { }
+ }
+ break;
+ case "boolean": valid = value instanceof Boolean; break;
+ case "date":
+ if (value instanceof String && ((String)value).matches("\\d{4}-\\d{2}-\\d{2}")) {
+ try { LocalDate.parse((String)value); valid = true; } catch (RuntimeException ignored) { }
+ }
+ break;
+ case "datetime":
+ if (value instanceof String && ((String)value).matches("\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}(\\.\\d{1,9})?(Z|[+-]\\d{2}:\\d{2})?")) {
+ try { LocalDateTime.parse((String)value); valid = true; }
+ catch (RuntimeException local) {
+ try { OffsetDateTime.parse((String)value); valid = true; } catch (RuntimeException ignored) { }
+ }
+ }
+ break;
+ case "stringArray": case "fileArray": case "userArray":
+ valid = value instanceof List;
+ if (valid) for (Object item : (List>)value) {
+ if ("stringArray".equals(v.type)) {
+ if (!(item instanceof String) || (v.enumValues != null && !v.enumValues.contains(item))) valid = false;
+ } else if (!validRecord(item, "fileArray".equals(v.type))) valid = false;
+ }
+ break;
+ default: break;
+ }
+ if (!valid) fail("变量类型/格式不合法:" + v.key + "(应为 " + v.type + ")");
+ if (value instanceof String && v.enumValues != null && !v.enumValues.contains(value)) fail("变量 " + v.key + " 不在枚举值域 " + v.enumValues + ":" + value);
+ }
+
+ private static boolean validRecord(Object value, boolean file) {
+ if (!(value instanceof Map)) return false;
+ Map, ?> row = (Map, ?>)value;
+ Object id = row.get("id");
+ try {
+ if (!(id instanceof Number) || new BigDecimal(id.toString()).scale() > 0 || ((Number)id).longValue() <= 0) return false;
+ } catch (NumberFormatException invalid) { return false; }
+ return file ? nonblank(row.get("fileName")) && nonblank(row.get("key")) : nonblank(row.get("name"));
+ }
+ private static boolean nonblank(Object value) { return value instanceof String && !blank((String)value); }
+ private static boolean blank(String value) {
+ return value.codePoints().allMatch(c -> Character.isWhitespace(c) || Character.isSpaceChar(c));
+ }
+
+ /** FC 快照是唯一派生来源,保存时传入的 variables 不能覆盖绑定表单的契约。 */
+ public static List fromSnapshot(Map snapshot) {
+ Object fields = snapshot.get("fields");
+ if (!(fields instanceof List)) fail("formContract.fields 必须为数组");
+ List variables = new ArrayList<>();
+ for (Object item : (List>)fields) {
+ if (!(item instanceof Map)) fail("formContract 字段不合法");
+ Map, ?> field = (Map, ?>)item;
+ FlowDefModel.Variable v = new FlowDefModel.Variable();
+ v.key = (String)field.get("key");
+ v.required = Boolean.TRUE.equals(field.get("required"));
+ String type = String.valueOf(field.get("type"));
+ switch (type) {
+ case "text": case "textarea": case "phone": case "select": case "radio": v.type = "string"; break;
+ case "number": case "date": v.type = type; break;
+ case "checkbox": case "multiselect": case "address": v.type = "stringArray"; break;
+ case "image": case "upload": v.type = "fileArray"; break;
+ case "user": case "users": v.type = "userArray"; break;
+ default: fail("FC 字段类型不支持:" + v.key + " / " + type);
+ }
+ if (Arrays.asList("select", "radio", "checkbox", "multiselect").contains(type)) {
+ Object enums = field.get("enum");
+ if (!(enums instanceof List) || ((List>)enums).isEmpty()) fail("FC 枚举值域缺失:" + v.key);
+ v.enumValues = new ArrayList<>();
+ for (Object option : (List>)enums) {
+ if (!(option instanceof String)) fail("FC 枚举必须为字符串:" + v.key);
+ v.enumValues.add((String)option);
+ }
+ }
+ variables.add(v);
+ }
+ return variables;
+ }
+ private static void fail(String message) { throw new BizException(ErrorCode.PARAM_ERROR, message); }
+}
diff --git a/src/main/java/com/yunhai/flowcenter/entity/FlwAdvanceIntent.java b/src/main/java/com/yunhai/flowcenter/entity/FlwAdvanceIntent.java
new file mode 100644
index 0000000..11bd28d
--- /dev/null
+++ b/src/main/java/com/yunhai/flowcenter/entity/FlwAdvanceIntent.java
@@ -0,0 +1,107 @@
+package com.yunhai.flowcenter.entity;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Index;
+import javax.persistence.Table;
+import javax.persistence.Version;
+import javax.persistence.UniqueConstraint;
+
+import java.time.LocalDateTime;
+
+/**
+ * 待推进意图(2026-10-25 T05):任务办理事务内持久化"这个节点该继续推进了"的事实,
+ * 事务外 worker 领取执行——办理事务提交成功 ⇔ 推进意图已持久化,进程崩溃/重启后
+ * 扫描续跑(旧链路 AFTER_COMMIT 三次内存重试,耗尽只 log.error,重启即丢)。
+ *
+ * D09:每个激活批次记录精确任务成员,未办理时以 DONE 等待;办理同事务串行刷新
+ * 本批次的末笔任务及 PENDING 工作。worker 按 id+revision 锁定当前读,推进与结算同事务;
+ * NULL 代次只允许明确接棒或留待有证据的人工回填,不作为本代放行。
+ */
+@Getter
+@Setter
+@Entity
+@Table(name = "flw_advance_intent",
+ uniqueConstraints = @UniqueConstraint(name = "uk_flw_advance_activation", columnNames = "activation_task_id"),
+ indexes = {
+ @Index(name = "idx_flw_advance_intent_due", columnList = "status, next_retry_at"),
+ @Index(name = "idx_flw_advance_intent_instance", columnList = "instance_id")})
+public class FlwAdvanceIntent extends BaseEntity {
+
+ public static final String STATE_PENDING = "PENDING";
+ public static final String STATE_DONE = "DONE";
+ public static final String STATE_EXHAUSTED = "EXHAUSTED";
+
+ @Column(name = "instance_id", nullable = false)
+ private Long instanceId;
+
+ /** 触发推进的节点(TASK_HANDLED 的 nodeId) */
+ @Column(name = "node_id", nullable = false, length = 64)
+ private String nodeId;
+
+ /** First task CREATED in this activation batch; never the last task to finish. NULL = unanchored legacy work. */
+ @Column(name = "activation_task_id")
+ private Long activationTaskId;
+
+ /** Exact membership, including transfers; comma-delimited positive task IDs, e.g. ,12,14,. */
+ @Column(name = "activation_task_ids", columnDefinition = "LONGTEXT")
+ private String activationTaskIds;
+
+ @Column(name = "handled_task_ids", columnDefinition = "LONGTEXT")
+ private String handledTaskIds;
+
+ @Column(name = "trigger_task_id")
+ private Long triggerTaskId;
+
+ @Column(name = "activation_def_id")
+ private Long activationDefId;
+
+ @Column(name = "activation_def_version")
+ private Integer activationDefVersion;
+
+ @Column(name = "activation_apply_round")
+ private Integer activationApplyRound;
+
+ /** A detached worker may act/defer only the revision it actually observed. */
+ @Version
+ @Column(name = "revision", nullable = false)
+ private Long revision = 0L;
+
+ public boolean containsTask(Long id) { return contains(activationTaskIds, id); }
+ public boolean handledTask(Long id) { return contains(handledTaskIds, id); }
+ public void addTask(Long id) { if (!containsTask(id)) activationTaskIds = append(activationTaskIds, id); }
+ public void addHandledTask(Long id) { if (!handledTask(id)) handledTaskIds = append(handledTaskIds, id); }
+ private static boolean contains(String ids, Long id) { return ids != null && id != null && ids.contains("," + id + ","); }
+ private static String append(String ids, Long id) {
+ if (id == null || id <= 0) throw new IllegalArgumentException("Missing activation task identity");
+ return (ids == null ? "," : ids) + id + ",";
+ }
+
+ /** 出边名(可空=引擎默认路由) */
+ @Column(name = "edge_name", length = 128)
+ private String edgeName;
+
+ /** 办理人随行(推进链操作人信封,可空) */
+ @Column(name = "assignee_id")
+ private Integer assigneeId;
+
+ @Column(name = "assignee_name", length = 64)
+ private String assigneeName;
+
+ /** PENDING=待执行;DONE=已执行(含推进判定为无需推进的幂等结论);EXHAUSTED=重试耗尽留人工 */
+ @Column(name = "status", nullable = false, length = 16)
+ private String status = STATE_PENDING;
+
+ @Column(name = "attempts", nullable = false)
+ private Integer attempts = 0;
+
+ /** 退避到期时间(耗尽前按指数退避续扫) */
+ @Column(name = "next_retry_at")
+ private LocalDateTime nextRetryAt;
+
+ @Column(name = "last_error", length = 255)
+ private String lastError;
+}
diff --git a/src/main/java/com/yunhai/flowcenter/entity/FlwApplyCommand.java b/src/main/java/com/yunhai/flowcenter/entity/FlwApplyCommand.java
new file mode 100644
index 0000000..124023f
--- /dev/null
+++ b/src/main/java/com/yunhai/flowcenter/entity/FlwApplyCommand.java
@@ -0,0 +1,24 @@
+package com.yunhai.flowcenter.entity;
+import lombok.Getter;
+import lombok.Setter;
+import javax.persistence.*;
+/** Immutable settled command fact: ACCEPTED or REJECTED. Never repurpose a request ID. */
+@Entity @Getter @Setter
+@Table(name="flw_apply_command", uniqueConstraints=@UniqueConstraint(name="uk_flw_command_request",columnNames={"sch_id","source_principal_id","request_id"}),indexes=@Index(name="idx_flw_command_instance",columnList="instance_id"))
+public class FlwApplyCommand extends BaseEntity {
+ @Column(name="sch_id",nullable=false) private Long schId;
+ @Column(name="source_principal_id",nullable=false,length=36) private String sourcePrincipalId;
+ @Column(name="request_id",nullable=false,length=64) private String requestId;
+ @Column(name="canonical_hash",nullable=false,length=64) private String canonicalHash;
+ @Column(name="canonical_version",nullable=false) private Integer canonicalVersion;
+ @Column(name="biz_type",nullable=false,length=32) private String bizType;
+ @Column(name="biz_id",nullable=false,length=64) private String bizId;
+ @Column(name="flow_guid",nullable=false,length=32) private String flowGuid;
+ @Column(name="state",nullable=false,length=16) private String state;
+ @Column(name="instance_id") private Long instanceId;
+ @Column(name="apply_round") private Integer round;
+ @Column(name="def_version") private Integer defVersion;
+ @Column(name="def_id") private Long defId;
+ @Column(name="reject_code",length=64) private String rejectCode;
+ @Column(name="reject_detail",length=500) private String rejectDetail;
+}
diff --git a/src/main/java/com/yunhai/flowcenter/entity/FlwBizCoordinate.java b/src/main/java/com/yunhai/flowcenter/entity/FlwBizCoordinate.java
new file mode 100644
index 0000000..f1d8027
--- /dev/null
+++ b/src/main/java/com/yunhai/flowcenter/entity/FlwBizCoordinate.java
@@ -0,0 +1,15 @@
+package com.yunhai.flowcenter.entity;
+import lombok.Getter;
+import lombok.Setter;
+import javax.persistence.*;
+@Entity @Getter @Setter
+@Table(name="flw_biz_coordinate", uniqueConstraints=@UniqueConstraint(name="uk_flw_coord", columnNames={"sch_id","source_principal_id","biz_type","biz_id"}))
+public class FlwBizCoordinate extends BaseEntity {
+ @Column(name="sch_id",nullable=false) private Long schId;
+ @Column(name="source_principal_id",nullable=false,length=36) private String sourcePrincipalId;
+ @Column(name="biz_type",nullable=false,length=32) private String bizType;
+ @Column(name="biz_id",nullable=false,length=64) private String bizId;
+ @Column(name="last_round",nullable=false) private Integer lastRound = 0;
+ @Column(name="last_instance_id") private Long lastInstanceId;
+ @Column(name="active_instance_id") private Long activeInstanceId;
+}
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/FlwCommandSource.java b/src/main/java/com/yunhai/flowcenter/entity/FlwCommandSource.java
new file mode 100644
index 0000000..e0b8e35
--- /dev/null
+++ b/src/main/java/com/yunhai/flowcenter/entity/FlwCommandSource.java
@@ -0,0 +1,10 @@
+package com.yunhai.flowcenter.entity;
+import lombok.Getter;
+import lombok.Setter;
+import javax.persistence.*;
+/** Persistent admission mutex. Rows survive application deletion; never recycle. */
+@Entity @Table(name="flw_command_source") @Getter @Setter
+public class FlwCommandSource {
+ @Id @Column(length=36) private String id;
+ @Column(nullable=false) private Integer appId;
+}
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/entity/FlwStateEvent.java b/src/main/java/com/yunhai/flowcenter/entity/FlwStateEvent.java
new file mode 100644
index 0000000..c85c3cc
--- /dev/null
+++ b/src/main/java/com/yunhai/flowcenter/entity/FlwStateEvent.java
@@ -0,0 +1,77 @@
+package com.yunhai.flowcenter.entity;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Index;
+import javax.persistence.Table;
+import javax.persistence.UniqueConstraint;
+
+/**
+ * 流程状态事件事实(2026-10-25 T05):状态/审批动作与不可变事件事实同一数据库事务落库。
+ *
+ * 关闭审计 E24 的丢失窗口:旧链路 AFTER_COMMIT 才按订阅生成 eventId、另事务落回调行——
+ * 提交到分发之间进程被杀,事件永久丢失(订阅方靠对账兜底)。本事实行在发布线程的同一事务内
+ * 写入:状态提交成功 ⇔ 事件事实已持久化;回调分发只是它的消费,崩溃后由 sweep 扫描补分发。
+ *
+ * 身份:{@code eventId} 在事实创建时一次性生成,全部订阅方/全部投递重试复用同一 eventId
+ * (消费端 (event_id, consumer_app_key) 唯一键去重);{@code eventSeq} 为实例内尽力单调序号
+ * (FOR UPDATE 发号器;REQUIRES_NEW 子事务推进拓扑下极端并发窗口可能同值——有序性以
+ * eventSeq+发生时间为准,同实例严格全序需专用发号器,列入后续;不承诺全局副作用严格有序)。
+ */
+@Getter
+@Setter
+@Entity
+@Table(name = "flw_state_event",
+ uniqueConstraints = @UniqueConstraint(name = "uk_flw_state_event_id", columnNames = {"event_id"}),
+ indexes = {
+ @Index(name = "idx_flw_state_event_instance", columnList = "instance_id"),
+ @Index(name = "idx_flw_state_event_dispatch", columnList = "dispatched, create_time"),
+ @Index(name = "idx_flw_state_event_seq", columnList = "instance_id, event_seq")})
+public class FlwStateEvent extends BaseEntity {
+
+ /** 稳定事件身份(32 位 uuid 去横线;投递与重试全链复用) */
+ @Column(name = "event_id", nullable = false, length = 32)
+ private String eventId;
+
+ /** 租户键 */
+ @Column(name = "sch_id", nullable = false)
+ private Long schId;
+
+ /** 实例身份 */
+ @Column(name = "instance_id", nullable = false)
+ private Long instanceId;
+
+ /** 发布时冻结的申报轮次(旧事件不得被 AFTER_COMMIT 重读贴上新轮标签) */
+ @Column(name = "apply_round")
+ private Integer applyRound;
+
+ /** 实例内单调序号(1 起;同实例状态行锁内取 max+1) */
+ @Column(name = "event_seq", nullable = false)
+ private Integer eventSeq;
+
+ /** 事件类型码(FlowEvent.Type#getCode,与回调订阅枚举同值域) */
+ @Column(name = "event_type", nullable = false, length = 32)
+ private String eventType;
+
+ /** 关联任务(办件级事件为 null) */
+ @Column(name = "task_id")
+ private Long taskId;
+
+ /** 操作人(可空) */
+ @Column(name = "operator_id")
+ private Integer operatorId;
+
+ /** 完整载荷快照(与回调信封同构 JSON;发生时状态,不随后续变化) */
+ @Column(name = "payload_json", nullable = false, columnDefinition = "LONGTEXT")
+ private String payloadJson;
+
+ /**
+ * 分发标记:0=待分发(含崩溃窗口恢复);1=已按当前订阅生成了全部回调行(或无可订阅方)。
+ * 分发幂等:重扫前按 (eventId, regId) 查重,不重复建投递行。
+ */
+ @Column(name = "dispatched", nullable = false)
+ private Integer dispatched = 0;
+}
diff --git a/src/main/java/com/yunhai/flowcenter/entity/FlwUsageGrant.java b/src/main/java/com/yunhai/flowcenter/entity/FlwUsageGrant.java
new file mode 100644
index 0000000..d0a913c
--- /dev/null
+++ b/src/main/java/com/yunhai/flowcenter/entity/FlwUsageGrant.java
@@ -0,0 +1,63 @@
+package com.yunhai.flowcenter.entity;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Index;
+import javax.persistence.Table;
+import javax.persistence.UniqueConstraint;
+
+/**
+ * 应用使用授权(2026-10-25 T01):哪个稳定应用主体可以在哪所学校使用哪条流程线。
+ *
+ * 与 {@link FlwApp}(定义保存白名单)和流程定义管理权(FlowAdminGuard 管理角色)三权分立:
+ * 有 flw_app 行不等于能新发起;是应用管理员也不等于能编辑/发布学校流程定义。
+ * 许可键是三元 {@code (schId, flowGuid, appId)}——appId 是 UserBase application_info 主键
+ * (稳定主体,appKey 可重置/注销后不复用历史来源),appKey 快照仅供展示回溯。
+ *
+ * 撤权语义:{@code enabled=0} 只拦截新发起;已受理命令的回执恢复、在途实例推进
+ * 不回读本表(P0-3 收口顺序:认证 → 自有命令回执 → 未命中才查新发起资格)。
+ */
+@Getter
+@Setter
+@Entity
+@Table(name = "flw_usage_grant",
+ uniqueConstraints = @UniqueConstraint(name = "uk_flw_usage_grant",
+ columnNames = {"sch_id", "flow_guid", "app_id"}),
+ indexes = {@Index(name = "idx_flw_usage_grant_sch_flow", columnList = "sch_id, flow_guid"),
+ @Index(name = "idx_flw_usage_grant_app", columnList = "app_id")})
+public class FlwUsageGrant extends BaseEntity {
+ @Column(name="source_principal_id",length=36)
+ private String sourcePrincipalId;
+
+
+ /** 租户键 */
+ @Column(name = "sch_id", nullable = false)
+ private Long schId;
+
+ /** 流程线稳定身份(跨版本不变;版本归属由 flw_def 承载) */
+ @Column(name = "flow_guid", nullable = false, length = 64)
+ private String flowGuid;
+
+ /** 稳定应用主体主键(UserBase application_info.id;非 appKey 凭据索引) */
+ @Column(name = "app_id", nullable = false)
+ private Integer appId;
+
+ /** 授权时的 appKey 快照(展示/审计回溯用;判权不读它) */
+ @Column(name = "app_key_snapshot", length = 64)
+ private String appKeySnapshot;
+
+ /** 应用名快照(授权时 UserBase 返回值;同步不自动刷新) */
+ @Column(name = "app_name_snapshot", length = 128)
+ private String appNameSnapshot;
+
+ /** 1=许可生效;0=撤销(只拦新发起) */
+ @Column(name = "enabled", nullable = false)
+ private Integer enabled = 1;
+
+ /** 备注(谁/为何授予或撤销,人工审计线索) */
+ @Column(name = "remark", length = 255)
+ private String remark;
+}
diff --git a/src/main/java/com/yunhai/flowcenter/model/FlowApplyRetryModels.java b/src/main/java/com/yunhai/flowcenter/model/FlowApplyRetryModels.java
index c8252cd..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} 之上把"中心判出来的是哪一种命中"说清楚 */
@@ -74,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/FormalApplyModels.java b/src/main/java/com/yunhai/flowcenter/model/FormalApplyModels.java
new file mode 100644
index 0000000..74f9a9a
--- /dev/null
+++ b/src/main/java/com/yunhai/flowcenter/model/FormalApplyModels.java
@@ -0,0 +1,34 @@
+package com.yunhai.flowcenter.model;
+
+import lombok.Getter;
+import lombok.Setter;
+
+/** Service-owned v3 contract. No shared SDK DTO changes. */
+public final class FormalApplyModels {
+ private FormalApplyModels() { }
+ @Getter @Setter
+ public static class Request extends OapiModels.ApplyBody {
+ private String requestId;
+ private Long expectedPreviousInstanceId;
+ private Integer expectedPreviousRound;
+ }
+ @Getter @Setter
+ public static class Receipt {
+ private String state;
+ private String requestId;
+ private Long schoolId;
+ private String sourcePrincipalId;
+ private String canonicalHash;
+ private Integer canonicalVersion = 1;
+ private String bizType;
+ private String bizId;
+ private String flowGuid;
+ private Long instanceId;
+ private Integer round;
+ private Integer defVersion;
+ private Integer status;
+ private Integer version;
+ private String rejectCode;
+ private String rejectDetail;
+ }
+}
diff --git a/src/main/java/com/yunhai/flowcenter/model/OapiModels.java b/src/main/java/com/yunhai/flowcenter/model/OapiModels.java
index a48a4b9..611203f 100644
--- a/src/main/java/com/yunhai/flowcenter/model/OapiModels.java
+++ b/src/main/java/com/yunhai/flowcenter/model/OapiModels.java
@@ -109,6 +109,11 @@ public final class OapiModels {
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/FlwAdvanceIntentRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwAdvanceIntentRepository.java
new file mode 100644
index 0000000..6065599
--- /dev/null
+++ b/src/main/java/com/yunhai/flowcenter/repository/FlwAdvanceIntentRepository.java
@@ -0,0 +1,35 @@
+package com.yunhai.flowcenter.repository;
+
+import com.yunhai.flowcenter.entity.FlwAdvanceIntent;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+import org.springframework.data.jpa.repository.Lock;
+import javax.persistence.LockModeType;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Optional;
+
+public interface FlwAdvanceIntentRepository extends JpaRepository {
+
+ /** 到点待执行:PENDING 且(无退避或已到期) */
+ @Query(value="select * from flw_advance_intent where status=:status "
+ + "and (next_retry_at is null or next_retry_at<=:now) order by create_time,id limit 50", nativeQuery=true)
+ List findTop50ByStatusAndNextRetryAtLessThanEqualOrderByCreateTimeAsc(
+ @Param("status") String status, @Param("now") LocalDateTime now);
+
+ List findByInstanceIdAndNodeIdOrderByIdDesc(Long instanceId, String nodeId);
+
+ Optional findFirstByInstanceIdAndNodeIdAndActivationTaskIdIsNotNullOrderByIdDesc(
+ Long instanceId, String nodeId);
+
+ @Lock(LockModeType.PESSIMISTIC_WRITE)
+ @Query("select i from FlwAdvanceIntent i where i.id=:id")
+ Optional findByIdForUpdate(@Param("id") Long id);
+
+ /** 幂等记账:同实例同节点是否已有未完成意图(办理事务内去重,避免并发办理堆积重复行) */
+ @Query("select count(i) > 0 from FlwAdvanceIntent i where i.instanceId = :instanceId "
+ + "and i.nodeId = :nodeId and i.status = 'PENDING'")
+ boolean existsPending(@Param("instanceId") Long instanceId, @Param("nodeId") String nodeId);
+}
diff --git a/src/main/java/com/yunhai/flowcenter/repository/FlwApplyCommandRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwApplyCommandRepository.java
new file mode 100644
index 0000000..0c4a46d
--- /dev/null
+++ b/src/main/java/com/yunhai/flowcenter/repository/FlwApplyCommandRepository.java
@@ -0,0 +1,8 @@
+package com.yunhai.flowcenter.repository;
+import com.yunhai.flowcenter.entity.FlwApplyCommand;
+import org.springframework.data.jpa.repository.JpaRepository;
+import java.util.Optional;
+public interface FlwApplyCommandRepository extends JpaRepository {
+ boolean existsBySchIdAndSourcePrincipalIdAndFlowGuidAndState(Long school, String source, String flow, String state);
+ Optional findBySchIdAndSourcePrincipalIdAndRequestId(Long schId,String sourcePrincipalId,String requestId);
+}
diff --git a/src/main/java/com/yunhai/flowcenter/repository/FlwBizCoordinateRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwBizCoordinateRepository.java
new file mode 100644
index 0000000..c8feb9a
--- /dev/null
+++ b/src/main/java/com/yunhai/flowcenter/repository/FlwBizCoordinateRepository.java
@@ -0,0 +1,14 @@
+package com.yunhai.flowcenter.repository;
+import com.yunhai.flowcenter.entity.FlwBizCoordinate;
+import org.springframework.data.jpa.repository.*;
+import org.springframework.data.repository.query.Param;
+import javax.persistence.LockModeType;
+import java.util.Optional;
+public interface FlwBizCoordinateRepository extends JpaRepository {
+ @Lock(LockModeType.PESSIMISTIC_WRITE)
+ @Query("select c from FlwBizCoordinate c where c.schId=:sch and c.sourcePrincipalId=:source and c.bizType=:type and c.bizId=:biz")
+ Optional lock(@Param("sch") Long sch,@Param("source") String source,@Param("type") String type,@Param("biz") String biz);
+ @Modifying
+ @Query("update FlwBizCoordinate c set c.activeInstanceId=null where c.schId=:sch and c.sourcePrincipalId=:source and c.bizType=:type and c.bizId=:biz and c.activeInstanceId=:instance and c.lastRound=:round")
+ int release(@Param("sch") Long sch,@Param("source") String source,@Param("type") String type,@Param("biz") String biz,@Param("instance") Long instance,@Param("round") Integer round);
+}
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/FlwCommandSourceRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwCommandSourceRepository.java
new file mode 100644
index 0000000..716f64b
--- /dev/null
+++ b/src/main/java/com/yunhai/flowcenter/repository/FlwCommandSourceRepository.java
@@ -0,0 +1,10 @@
+package com.yunhai.flowcenter.repository;
+import com.yunhai.flowcenter.entity.FlwCommandSource;
+import org.springframework.data.jpa.repository.*;
+import org.springframework.data.repository.query.Param;
+import javax.persistence.LockModeType;
+import java.util.Optional;
+public interface FlwCommandSourceRepository extends JpaRepository {
+ @Lock(LockModeType.PESSIMISTIC_WRITE) @Query("select s from FlwCommandSource s where s.id=:id")
+ Optional lock(@Param("id") String id);
+}
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/FlwStateEventRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwStateEventRepository.java
new file mode 100644
index 0000000..3a77769
--- /dev/null
+++ b/src/main/java/com/yunhai/flowcenter/repository/FlwStateEventRepository.java
@@ -0,0 +1,24 @@
+package com.yunhai.flowcenter.repository;
+
+import com.yunhai.flowcenter.entity.FlwStateEvent;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+public interface FlwStateEventRepository extends JpaRepository {
+
+ /** 实例内单调序号取号:调用方持有该实例的状态行锁(同事务状态变更),max+1 串行安全 */
+ @Query("select coalesce(max(e.eventSeq), 0) from FlwStateEvent e where e.instanceId = :instanceId")
+ int maxSeqByInstance(@Param("instanceId") Long instanceId);
+
+ /** 崩溃窗口恢复扫描:未分发且已过缓冲期(给 AFTER_COMMIT 首分发留时间) */
+ List findTop50ByDispatchedAndCreateTimeLessThanEqualOrderByCreateTimeAsc(
+ Integer dispatched, LocalDateTime deadline);
+
+ java.util.Optional findByEventId(String eventId);
+
+ boolean existsByEventId(String eventId);
+}
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/repository/FlwUsageGrantRepository.java b/src/main/java/com/yunhai/flowcenter/repository/FlwUsageGrantRepository.java
new file mode 100644
index 0000000..ee3cd37
--- /dev/null
+++ b/src/main/java/com/yunhai/flowcenter/repository/FlwUsageGrantRepository.java
@@ -0,0 +1,18 @@
+package com.yunhai.flowcenter.repository;
+
+import com.yunhai.flowcenter.entity.FlwUsageGrant;
+import org.springframework.data.jpa.repository.JpaRepository;
+
+import java.util.List;
+import java.util.Optional;
+
+public interface FlwUsageGrantRepository extends JpaRepository {
+ boolean existsBySchIdAndFlowGuidAndSourcePrincipalIdAndEnabled(Long school, String flow, String source, Integer enabled);
+
+
+ Optional findBySchIdAndFlowGuidAndAppId(Long schId, String flowGuid, Integer appId);
+
+ List findBySchIdAndFlowGuidOrderByUpdateTimeDescIdDesc(Long schId, String flowGuid);
+
+ List findBySchIdAndAppIdOrderByUpdateTimeDescIdDesc(Long schId, Integer appId);
+}
diff --git a/src/main/java/com/yunhai/flowcenter/service/ApplyVariableValidator.java b/src/main/java/com/yunhai/flowcenter/service/ApplyVariableValidator.java
new file mode 100644
index 0000000..68ac889
--- /dev/null
+++ b/src/main/java/com/yunhai/flowcenter/service/ApplyVariableValidator.java
@@ -0,0 +1,49 @@
+package com.yunhai.flowcenter.service;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.yunhai.flowcenter.engine.FlowDefModel;
+import com.yunhai.platform.apistd.BizException;
+import com.yunhai.platform.utils.exception.ErrorCode;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.Map;
+import java.util.Set;
+
+/** 发起前按冻结的定义契约校验;历史无声明版本保留原有的引用变量存在性规则。 */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class ApplyVariableValidator {
+
+ private final ObjectMapper objectMapper;
+
+ /** def 的全部 EXPRESSION 出边引用的业务变量键(去重保序) */
+ public Set referencedBusinessKeys(FlowDefModel model) {
+ return com.yunhai.flowcenter.engine.VariableContract.referencedKeys(model);
+ }
+
+ public void requireVariablesPresent(FlowDefModel model, String formDataJson) {
+ com.yunhai.flowcenter.engine.VariableContract.validateInput(model, parse(formDataJson, model.variables != null));
+ }
+
+ /** 与 ConditionRouteService.parseVariables 同源:formData 顶层即变量;非对象 JSON=空(后续逐键报缺失) */
+ private Map parse(String formDataJson, boolean strict) {
+ if (formDataJson == null || formDataJson.trim().isEmpty()) {
+ return new java.util.HashMap<>();
+ }
+ try {
+ Map values = objectMapper.readValue(formDataJson, new TypeReference