Preserve reviewed existing gate prerequisites

Local review candidate only. Explicitly selected inherited prerequisites and prior repair patches; unrelated WIP and independent MQ fixes are excluded. No push or deployment authorized.

src/main/java/com/yunhai/formcenter/entity/AnswerReconcileTask.java

@@ -53,7 +53,8 @@ public class AnswerReconcileTask extends BaseEntity {
 
     /** 终态：不再被派发扫描领取，也不接受进度回退 */
     public static boolean isTerminal(String state) {
-        return STATE_COMPLETED.equals(state) || STATE_FAILED.equals(state) || STATE_CANCELLED.equals(state);
+        return STATE_COMPLETED.equals(state) || STATE_PARTIAL.equals(state)
+                || STATE_FAILED.equals(state) || STATE_CANCELLED.equals(state);
     }
 
     /** 活跃态：同一坐标上只允许一个（单并发），且这些状态要受租约与取消约束 */

src/main/java/com/yunhai/formcenter/entity/FormConsumerGrant.java

@@ -51,10 +51,18 @@ public class FormConsumerGrant extends BaseEntity {
     @Column(name = "read_allowed", nullable = false, columnDefinition = "tinyint")
     private Integer readAllowed = 0;
 
-    /** 订阅授权（答卷变更事件投递）：1=允许 */
+    /** 订阅授权（答卷变更事件的数据订阅权，投影应用/权威对账授权都判它）：1=允许 */
     @Column(name = "subscribe_allowed", nullable = false, columnDefinition = "tinyint")
     private Integer subscribeAllowed = 0;
 
+    /**
+     * MQ broker 推送投递（2026-10-22 决策②：读取/订阅/MQ 推送三权分立）。
+     * 1=该订阅的事件进 outbox 并经 broker 投递；<b>0=HTTP-only（拉取+收据确认），不建 outbox 行</b>，
+     * 避免没接 broker 的消费方在 fc_answer_delivery 里长期 PENDING。翻转让 grant_version 进位。
+     */
+    @Column(name = "push_allowed", nullable = false, columnDefinition = "tinyint")
+    private Integer pushAllowed = 0;
+
     /** 生效开关：0=已撤销（保留行与 grant_version 供审计，撤销后后续读取/发布检查即失败） */
     @Column(name = "enabled", nullable = false, columnDefinition = "tinyint")
     private Integer enabled = 1;

src/main/java/com/yunhai/formcenter/model/FormAccessModels.java

@@ -47,6 +47,8 @@ public final class FormAccessModels {
         private String consumerAppKey;
         private boolean readAllowed;
         private boolean subscribeAllowed;
+        /** MQ broker 推送投递开关：false=HTTP-only（拉取+收据确认），入队侧不建 outbox 行 */
+        private boolean pushAllowed;
         private int grantVersion;
     }
 
@@ -101,11 +103,13 @@ public final class FormAccessModels {
         private String consumerAppKey;
         private boolean readAllowed;
         private boolean subscribeAllowed;
+        /** MQ broker 推送投递开关（运维据此看出哪个消费方在吃推送） */
+        private boolean pushAllowed;
         private boolean enabled;
         private int grantVersion;
     }
 
-    /** grant 授权请求（三元键 + 两把开关；不传开关=只给读，默认拒绝） */
+    /** grant 授权请求（三元键 + 三把开关；不传开关=只给读，默认拒绝；push 依赖 subscribe） */
     @Getter
     @Setter
     public static class GrantRequest {
@@ -114,6 +118,8 @@ public final class FormAccessModels {
         private String consumerAppKey;
         private Boolean readAllowed;
         private Boolean subscribeAllowed;
+        /** MQ推送：缺省/null对已有行保持、新行默认false；显式false才关闭，true依赖subscribe。 */
+        private Boolean pushAllowed;
     }
 
     /** grant 撤销请求 */

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

@@ -193,6 +193,9 @@ public class AnswerDeliveryService {
      * 在<b>调用方事务内</b>为该答卷的当前 revision 建逐订阅投递行。
      *
      * <p>无订阅的简单表单返回空列表——答卷照常提交，不创建任何消费投递记录（B.2 末句）。
+     * <b>HTTP-only 订阅（未开 {@code push_allowed}）同样不建行</b>（2026-10-22 决策②）：
+     * 拉取/收据/对账那条授权还在，但没有 broker 私有队列的消费方进了 outbox 就只能长期 PENDING——
+     * 宁不建行，也不制造积压假象；哪天它真接了 broker，把 push 翻开（grant_version 进位）即可。
      * 撞 {@code uk_delivery_revision} 说明另一个事务已把同一 revision 投出去了（CAS 没挡住的极端形态），
      * 整事务回滚是唯一安全处置，不做"改号重试"。</p>
      */
@@ -207,7 +210,14 @@ public class AnswerDeliveryService {
         }
         String contentHash = snapshots.contentHash(answer);
         LocalDateTime occurredAt = answer.getUpdateTime() == null ? LocalDateTime.now() : answer.getUpdateTime();
+        int queued = 0;
         for (SubscriptionGrant grant : grants) {
+            if (!grant.isPushAllowed()) {
+                log.info("[v2投递] HTTP-only 订阅跳过入队（未开 push_allowed，不建 outbox 行）：consumer={}@{}",
+                        grant.getConsumerAppKey(), grant.getConsumerSchoolId());
+                continue;
+            }
+            queued++;
             AnswerDelivery row = new AnswerDelivery();
             row.setEventId(UUID.randomUUID().toString());
             row.setProtocolVersion(AnswerDeliveryModels.PROTOCOL_VERSION);
@@ -235,9 +245,9 @@ public class AnswerDeliveryService {
                 throw new BizException(ErrorCode.CUSTOM, "并发修改冲突（同一快照修订号已有投递行），本次提交回滚，请重试");
             }
         }
-        log.info("[v2投递] 入队 answer={} rev={} event={} 目标={}", answer.getUuid(),
+        log.info("[v2投递] 入队 answer={} rev={} event={} 目标={}/订阅 {}", answer.getUuid(),
-                answer.getAnswerRevision(), eventType, grants.size());
+                answer.getAnswerRevision(), eventType, queued, grants.size());
-        return grants.size();
+        return queued;
     }
 
     private String writeSnapshot(Answer answer, SubscriptionGrant grant) {
@@ -388,11 +398,15 @@ public class AnswerDeliveryService {
         }
     }
 
-    /** 消费方在该答卷上的当前生效订阅（只走 TASK_1 的判据，本类不拼第二套 ACL） */
+    /**
+     * 消费方在该答卷上的当前生效订阅（只走 TASK_1 的判据，本类不拼第二套 ACL）。
+     * 发布侧同样只认开了 {@code push_allowed} 的 grant：推送被降级回 HTTP-only 的存量行
+     * 在这里复核不到（降级即 grant_version 进位，双保险），按撤权处置而不是照发。
+     */
     private SubscriptionGrant findLiveGrant(AnswerDelivery row) {
         for (SubscriptionGrant grant : accessService.listEnabledSubscriptions(row.getFormUuid(),
                 row.getConsumerSchoolId())) {
-            if (row.getConsumerAppKey().equals(grant.getConsumerAppKey())) {
+            if (row.getConsumerAppKey().equals(grant.getConsumerAppKey()) && grant.isPushAllowed()) {
                 return grant;
             }
         }

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

@@ -283,9 +283,18 @@ public class FormConsumerAccessService {
                 .orElse(null);
         boolean read = Boolean.TRUE.equals(req.getReadAllowed());
         boolean subscribe = Boolean.TRUE.equals(req.getSubscribeAllowed());
+        // 旧客户端没有 pushAllowed；缺省/null 只在新建时默认关闭，已有授权保持原值。
+        boolean push = req.getPushAllowed() == null
+                ? grant != null && isOne(grant.getPushAllowed())
+                : Boolean.TRUE.equals(req.getPushAllowed());
         if (!read && !subscribe) {
             throw new BizException(ErrorCode.PARAM_ERROR, "readAllowed / subscribeAllowed 至少开一把（默认拒绝）");
         }
+        if (push && !subscribe) {
+            // 推送投递的是订阅数据，订阅权没开就谈不上推送（决策②的三权分立）
+            throw new BizException(ErrorCode.PARAM_ERROR,
+                    "pushAllowed 依赖订阅授权；关闭订阅时请显式设置 pushAllowed=false，或保留 subscribeAllowed");
+        }
         if (grant == null) {
             grant = new FormConsumerGrant();
             grant.setFormUuid(form.getUuid());
@@ -294,19 +303,21 @@ public class FormConsumerAccessService {
             grant.setGrantVersion(1);
         } else if (!isOne(grant.getEnabled())
                 || isOne(grant.getReadAllowed()) != read
-                || isOne(grant.getSubscribeAllowed()) != subscribe) {
+                || isOne(grant.getSubscribeAllowed()) != subscribe
-            // 撤销后重申、以及"还开着但能力变了"（如只关订阅）都是新纪元：
+                || isOne(grant.getPushAllowed()) != push) {
+            // 撤销后重申、以及"还开着但能力变了"（如只关订阅、关/开 MQ 推送）都是新纪元：
             // 订阅投递带的是 grantVersion，能力收窄却不进位会让已发出去的事件继续看起来合法
             grant.setGrantVersion(grant.getGrantVersion() + 1);
         }
         grant.setReadAllowed(read ? 1 : 0);
         grant.setSubscribeAllowed(subscribe ? 1 : 0);
+        grant.setPushAllowed(push ? 1 : 0);
         grant.setEnabled(1);
         FormConsumerGrant saved = grantRepository.save(grant);
         log.info("[consumer ACL] 授权 formUuid={} consumerSchoolId={} consumerAppKey={} read={} subscribe={} "
-                        + "grantVersion={}", saved.getFormUuid(), saved.getConsumerSchoolId(),
+                        + "push={} grantVersion={}", saved.getFormUuid(), saved.getConsumerSchoolId(),
                 saved.getConsumerAppKey(), saved.getReadAllowed(), saved.getSubscribeAllowed(),
-                saved.getGrantVersion());
+                saved.getPushAllowed(), saved.getGrantVersion());
         return toGrantView(saved);
     }
 
@@ -363,6 +374,7 @@ public class FormConsumerAccessService {
         view.setConsumerAppKey(grant.getConsumerAppKey());
         view.setReadAllowed(isOne(grant.getReadAllowed()));
         view.setSubscribeAllowed(isOne(grant.getSubscribeAllowed()));
+        view.setPushAllowed(isOne(grant.getPushAllowed()));
         view.setGrantVersion(grant.getGrantVersion() == null ? 0 : grant.getGrantVersion());
         return view;
     }
@@ -374,6 +386,7 @@ public class FormConsumerAccessService {
         view.setConsumerAppKey(grant.getConsumerAppKey());
         view.setReadAllowed(isOne(grant.getReadAllowed()));
         view.setSubscribeAllowed(isOne(grant.getSubscribeAllowed()));
+        view.setPushAllowed(isOne(grant.getPushAllowed()));
         view.setEnabled(isOne(grant.getEnabled()));
         view.setGrantVersion(grant.getGrantVersion() == null ? 0 : grant.getGrantVersion());
         return view;

src/test/java/com/yunhai/formcenter/controller/oapi/OapiFormControllerTest.java

@@ -6,10 +6,12 @@ import com.yunhai.formcenter.model.OapiModels;
 import com.yunhai.formcenter.service.ConsumerPrincipalResolver;
 import com.yunhai.formcenter.service.OapiFormService;
 import com.yunhai.platform.appauth.AppAuthFilter;
+import com.yunhai.platform.openapi.base.OpenApiAppClient;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.util.ReflectionTestUtils;
 import org.springframework.test.web.servlet.setup.MockMvcBuilders;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -48,9 +50,13 @@ class OapiFormControllerTest {
     @BeforeEach
     void setUp() {
         formService = mock(OapiFormService.class);
-        // 密钥注册表用桩替：只认 100018 这一个 appKey，其它一律"未在平台注册"
+        // 验签密钥与平台注册事实是两道门；standalone 不会执行 Spring 字段注入。
         ConsumerPrincipalResolver resolver = new ConsumerPrincipalResolver(
                 appKey -> "100018".equals(appKey) ? "svc-hall-dev-secret-100018" : null);
+        OpenApiAppClient registry = mock(OpenApiAppClient.class);
+        when(registry.getAppByKey("100018"))
+                .thenReturn(FormSchemaValidator.mapper().createObjectNode().put("appKey", "100018"));
+        ReflectionTestUtils.setField(resolver, "appRegistryClient", registry);
         mvc = MockMvcBuilders.standaloneSetup(new OapiFormController(formService, resolver)).build();
     }
 
@@ -139,6 +145,19 @@ class OapiFormControllerTest {
         verifyNoInteractions(formService);
     }
 
+    @Test
+    @DisplayName("自报查询参数不代替验签属性；已验签但未注册也不能进入服务")
+    void selfDeclaredOrUnregisteredCallerNeverReachesService() {
+        assertThat(catchThrowable(() -> mvc.perform(get("/oapi/form/schema")
+                .param("schId", "1001").param("uuid", "u").param("appKey", "100018"))))
+                .hasRootCauseInstanceOf(com.yunhai.platform.apistd.BizException.class);
+        assertThat(catchThrowable(() -> mvc.perform(get("/oapi/form/schema")
+                .param("schId", "1001").param("uuid", "u")
+                .requestAttr(CALLER_ATTR, "100999"))))
+                .hasRootCauseInstanceOf(com.yunhai.platform.apistd.BizException.class);
+        verifyNoInteractions(formService);
+    }
+
     private static OapiModels.SchemaView view(String uuid, int version) {
         OapiModels.SchemaView view = new OapiModels.SchemaView();
         view.setFormUuid(uuid);

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

@@ -193,6 +193,22 @@ class FcAnswerDeliveryTest {
                 .isEqualTo("fc.snapshot.v2." + CONSUMER_APP_KEY + "." + SCH);
     }
 
+    @Test
+    @DisplayName("HTTP-only 订阅（未开 MQ 推送）：数据订阅照常生效但不建 outbox 行，不留长期 PENDING")
+    void httpOnlyGrantCreatesNoOutboxRows() {
+        AnswerDeliveryRepository rows = mock(AnswerDeliveryRepository.class);
+        FormConsumerAccessService access = mock(FormConsumerAccessService.class);
+        when(access.listEnabledSubscriptions(FORM_UUID, SCH)).thenReturn(Collections.singletonList(
+                grant(CONSUMER_APP_KEY, SCH, 7, false)));
+        AnswerDeliveryService service = service(rows, mock(AnswerRepository.class), access,
+                snapshotService(mock(AnswerRepository.class), mock(SubmissionContextRepository.class), access));
+
+        assertThat(service.enqueueInTransaction(answer(1L, "{\"k\":\"v\"}", "draft"),
+                AnswerDeliveryModels.EVENT_CREATED)).isZero();
+
+        verify(rows, never()).saveAndFlush(any());
+    }
+
     @Test
     @DisplayName("无订阅的简单表单：零投递行，答卷照常提交（B.2 末句）")
     void noSubscriptionMeansNoDeliveryRows() {
@@ -260,6 +276,24 @@ class FcAnswerDeliveryTest {
         assertThat(pending.getState()).isEqualTo(AnswerDelivery.STATE_GRANT_REVOKED);
     }
 
+    @Test
+    @DisplayName("MQ 推送被降级回 HTTP-only：发送前复核拦下存量行，不向已关闭推送的消费方发消息")
+    void demotedHttpOnlyGrantStopsDelivery() {
+        AnswerDeliveryRepository rows = mock(AnswerDeliveryRepository.class);
+        AnswerDelivery pending = deliveryRow(AnswerDelivery.STATE_PENDING, 1L, 7);
+        when(rows.findDue(any(LocalDateTime.class), any())).thenReturn(Collections.singletonList(pending));
+        when(rows.findById(pending.getId())).thenReturn(Optional.of(pending));
+        FormConsumerAccessService access = mock(FormConsumerAccessService.class);
+        when(access.listEnabledSubscriptions(FORM_UUID, SCH)).thenReturn(Collections.singletonList(
+                grant(CONSUMER_APP_KEY, SCH, 7, false)));
+        AnswerDeliveryService service = service(rows, mock(AnswerRepository.class), access,
+                snapshotService(mock(AnswerRepository.class), mock(SubmissionContextRepository.class), access));
+        withBroker(service);
+
+        assertThat(service.sendDueBatch()).isZero();
+        assertThat(pending.getState()).isEqualTo(AnswerDelivery.STATE_GRANT_REVOKED);
+    }
+
     @Test
     @DisplayName("confirm 只写 brokerAccepted；consumerApplied 只能由签名回执写（B.4/B.6）")
     void brokerConfirmIsNotBusinessSuccess() {
@@ -613,12 +647,17 @@ class FcAnswerDeliveryTest {
     // ==================== 夹具 ====================
 
     private static SubscriptionGrant grant(String appKey, Long schoolId, int grantVersion) {
+        return grant(appKey, schoolId, grantVersion, true);
+    }
+
+    private static SubscriptionGrant grant(String appKey, Long schoolId, int grantVersion, boolean pushAllowed) {
         SubscriptionGrant grant = new SubscriptionGrant();
         grant.setFormUuid(FORM_UUID);
         grant.setConsumerAppKey(appKey);
         grant.setConsumerSchoolId(schoolId);
         grant.setReadAllowed(true);
         grant.setSubscribeAllowed(true);
+        grant.setPushAllowed(pushAllowed);
         grant.setGrantVersion(grantVersion);
         return grant;
     }

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

@@ -16,10 +16,12 @@ import com.yunhai.formcenter.repository.FormOwnershipRepository;
 import com.yunhai.formcenter.repository.FormRepository;
 import com.yunhai.platform.apistd.BizException;
 import com.yunhai.platform.appauth.AppSecretProvider;
+import com.yunhai.platform.openapi.base.OpenApiAppClient;
 import com.yunhai.platform.utils.exception.ErrorCode;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
 
 import java.time.LocalDateTime;
 import java.util.ArrayList;
@@ -34,6 +36,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyCollection;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 /**
@@ -60,6 +64,7 @@ class FcDiscoveryAccessTest {
 
     private AppSecretProvider secretProvider;
     private ConsumerPrincipalResolver resolver;
+    private OpenApiAppClient registry;
     private FormRepository formRepository;
     private FormOwnershipRepository ownershipRepository;
     private FormConsumerGrantRepository grantRepository;
@@ -76,6 +81,11 @@ class FcDiscoveryAccessTest {
         secretProvider = mock(AppSecretProvider.class);
         when(secretProvider.getSecret(SIGNED_CALLER)).thenReturn("svc-hall-dev-secret-100018");
         resolver = new ConsumerPrincipalResolver(secretProvider);
+        registry = mock(OpenApiAppClient.class);
+        when(registry.getAppByKey(SIGNED_CALLER)).thenReturn(
+                com.yunhai.formcenter.common.FormSchemaValidator.mapper().createObjectNode()
+                        .put("appKey", SIGNED_CALLER));
+        ReflectionTestUtils.setField(resolver, "appRegistryClient", registry);
 
         formRepository = mock(FormRepository.class);
         ownershipRepository = mock(FormOwnershipRepository.class);
@@ -171,6 +181,22 @@ class FcDiscoveryAccessTest {
         assertThat(resolver.isRegisteredAppKey(SIGNED_CALLER)).isTrue();
     }
 
+    @Test
+    @DisplayName("密钥存在不等于已注册：缺注册、错身份、注册服务失败均拒绝")
+    void registeredIdentityMustMatchAndRegistryFailuresStayClosed() {
+        when(registry.getAppByKey(SIGNED_CALLER)).thenReturn(null);
+        assertThatThrownBy(() -> resolver.resolveSignedCaller(SIGNED_CALLER))
+                .isInstanceOf(BizException.class).hasMessageContaining("不可用");
+        when(registry.getAppByKey(SIGNED_CALLER)).thenReturn(
+                com.yunhai.formcenter.common.FormSchemaValidator.mapper().createObjectNode()
+                        .put("appKey", "100999"));
+        assertThatThrownBy(() -> resolver.resolveSignedCaller(SIGNED_CALLER))
+                .isInstanceOf(BizException.class).hasMessageContaining("不可用");
+        when(registry.getAppByKey(SIGNED_CALLER)).thenThrow(new IllegalStateException("offline registry"));
+        assertThatThrownBy(() -> resolver.resolveSignedCaller(SIGNED_CALLER))
+                .isInstanceOf(BizException.class).hasMessageContaining("不可用");
+    }
+
     @Test
     @DisplayName("appCode 同名的应用不自认领：库里 app_code=svc-hall 但调用方无授权 → 拒")
     void appCodeIsFilterNotIdentity() {
@@ -394,8 +420,18 @@ class FcDiscoveryAccessTest {
         otherSchool.setReadAllowed(true);
         assertThatThrownBy(() -> accessService.grant(SCH, otherSchool, false))
                 .isInstanceOf(BizException.class)
-                .hasMessageContaining("只能给本校授予消费授权");
+                .hasMessage("表单不存在：u1");
-        // 超管代某校授权是允许的
+        // 超管也不能把学校资源授给其他消费校：资源归属门独立于管理者身份。
+        assertThatThrownBy(() -> accessService.grant(SCH, otherSchool, true))
+                .isInstanceOf(BizException.class).hasMessage("表单不存在：u1");
+        verify(grantRepository, never()).save(any());
+
+        // 平台资源可授予具体消费校，但校管理员仍只能为自己的消费域授权。
+        givenForm(form("u0", Form.PLATFORM_SCH_ID, "form-center", "public_form", "平台表单", 1));
+        otherSchool.setFormUuid("u0");
+        assertThatThrownBy(() -> accessService.grant(SCH, otherSchool, false))
+                .isInstanceOf(BizException.class).hasMessageContaining("只能给本校授予消费授权");
+        verify(grantRepository, never()).save(any());
         assertThat(accessService.grant(SCH, otherSchool, true).getConsumerSchoolId()).isEqualTo(OTHER_SCH);
 
         GrantRequest platformTenant = new GrantRequest();

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

@@ -581,7 +581,10 @@ class FcReconcileTaskTest {
         assertThat(ack.getAccepted()).isTrue();
         assertThat(ack.getDuplicate()).isTrue();
         verify(resultRepository, never()).save(any(AnswerReconcileResult.class));
+        verify(resultRepository, never()).saveAndFlush(any(AnswerReconcileResult.class));
         verify(taskRepository, never()).advanceTerminal(anyLong(), anyString(), any(), any(), any());
+        verify(taskRepository, never()).finishAttempt(anyLong(), anyInt(), anyString(), any(), any(), any());
+        verify(taskRepository, never()).renewRunning(anyLong(), anyInt(), any());
     }
 
     @Test
@@ -592,7 +595,7 @@ class FcReconcileTaskTest {
         when(taskRepository.findByTaskUuid(TASK_ID)).thenReturn(Optional.of(task));
         when(resultRepository.findByTaskIdAndAttempt(1L, 1)).thenReturn(Optional.empty());
         when(resultRepository.maxAttempt(1L)).thenReturn(2);
-        when(resultRepository.save(any(AnswerReconcileResult.class))).thenAnswer(call -> call.getArgument(0));
+        when(resultRepository.saveAndFlush(any(AnswerReconcileResult.class))).thenAnswer(call -> call.getArgument(0));
         when(grantRepository.findByFormUuidAndConsumerSchoolIdAndConsumerAppKey(FORM_UUID, SCHOOL, CONSUMER))
                 .thenReturn(Optional.of(grant(1L, FORM_UUID, 1)));
 
@@ -601,8 +604,12 @@ class FcReconcileTaskTest {
 
         assertThat(ack.getStaleAttempt()).isTrue();
         assertThat(ack.getAccepted()).isTrue();
-        verify(resultRepository).save(any(AnswerReconcileResult.class));
+        verify(resultRepository).saveAndFlush(any(AnswerReconcileResult.class));
         verify(taskRepository, never()).advanceTerminal(anyLong(), anyString(), any(), any(), any());
+        verify(taskRepository, never()).finishAttempt(anyLong(), anyInt(), anyString(), any(), any(), any());
+        verify(taskRepository, never()).renewRunning(anyLong(), anyInt(), any());
+        assertThat(task.getAttempt()).isEqualTo(3);
+        assertThat(task.getState()).isEqualTo(AnswerReconcileTask.STATE_RUNNING);
     }
 
     @Test
@@ -622,6 +629,7 @@ class FcReconcileTaskTest {
                 report(1, AnswerReconcileTask.STATE_COMPLETED)))
                 .isInstanceOf(BizException.class).hasMessageContaining("不符");
         verify(resultRepository, never()).save(any(AnswerReconcileResult.class));
+        verify(resultRepository, never()).saveAndFlush(any(AnswerReconcileResult.class));
     }
 
     @Test
@@ -632,17 +640,67 @@ class FcReconcileTaskTest {
         when(taskRepository.findByTaskUuid(TASK_ID)).thenReturn(Optional.of(task));
         when(resultRepository.findByTaskIdAndAttempt(eq(1L), anyInt())).thenReturn(Optional.empty());
         when(resultRepository.maxAttempt(1L)).thenReturn(0);
-        when(resultRepository.save(any(AnswerReconcileResult.class))).thenAnswer(call -> call.getArgument(0));
+        when(resultRepository.saveAndFlush(any(AnswerReconcileResult.class))).thenAnswer(call -> call.getArgument(0));
         when(grantRepository.findByFormUuidAndConsumerSchoolIdAndConsumerAppKey(FORM_UUID, SCHOOL, CONSUMER))
                 .thenReturn(Optional.of(grant(1L, FORM_UUID, 1)));
 
-        service.submitResult(caller(CONSUMER), TASK_ID, report(1, AnswerReconcileTask.STATE_PARTIAL));
+        // 先续租、后终态；终态后的 RUNNING 必须幂等吸收，不能恢复执行。
-        verify(taskRepository).advanceTerminal(eq(1L), eq(AnswerReconcileTask.STATE_PARTIAL),
-                any(LocalDateTime.class), isNull(), isNull());
-
         service.submitResult(caller(CONSUMER), TASK_ID, report(1, AnswerReconcileTask.STATE_RUNNING));
-        verify(taskRepository).markRunning(eq(1L), any(LocalDateTime.class));
+        verify(taskRepository).renewRunning(eq(1L), eq(1), any(LocalDateTime.class));
-        verify(taskRepository, times(1)).advanceTerminal(anyLong(), anyString(), any(), any(), any());
+        verify(taskRepository, never()).finishAttempt(anyLong(), anyInt(), anyString(), any(), any(), any());
+        service.submitResult(caller(CONSUMER), TASK_ID, report(1, AnswerReconcileTask.STATE_PARTIAL));
+        verify(taskRepository).finishAttempt(eq(1L), eq(1), eq(AnswerReconcileTask.STATE_PARTIAL),
+                any(LocalDateTime.class), isNull(), isNull());
+        assertThat(task.getState()).isEqualTo(AnswerReconcileTask.STATE_PARTIAL);
+
+        ResultAck duplicate = service.submitResult(caller(CONSUMER), TASK_ID,
+                report(1, AnswerReconcileTask.STATE_RUNNING));
+        assertThat(duplicate.getDuplicate()).isTrue();
+        assertThat(task.getState()).isEqualTo(AnswerReconcileTask.STATE_PARTIAL);
+        verify(taskRepository, times(1)).renewRunning(anyLong(), anyInt(), any());
+        verify(taskRepository, times(1)).finishAttempt(anyLong(), anyInt(), anyString(), any(), any(), any());
+        verify(resultRepository, times(2)).saveAndFlush(any(AnswerReconcileResult.class));
+        verify(taskRepository, never()).markRunning(anyLong(), any());
+        verify(taskRepository, never()).advanceTerminal(anyLong(), anyString(), any(), any(), any());
+    }
+
+    @Test
+    @DisplayName("完整扫描才允许 COMPLETED；与 PARTIAL 保持不同终态")
+    void completedRequiresBothCompleteScans() {
+        AnswerReconcileTask task = task(AnswerReconcileTask.STATE_DISPATCHED);
+        when(taskRepository.findByTaskUuid(TASK_ID)).thenReturn(Optional.of(task));
+        when(resultRepository.saveAndFlush(any(AnswerReconcileResult.class))).thenAnswer(call -> call.getArgument(0));
+        when(grantRepository.findByFormUuidAndConsumerSchoolIdAndConsumerAppKey(FORM_UUID, SCHOOL, CONSUMER))
+                .thenReturn(Optional.of(grant(1L, FORM_UUID, 1)));
+        ResultReport incomplete = report(1, AnswerReconcileTask.STATE_COMPLETED);
+        incomplete.setLocalScanComplete(false);
+        assertThatThrownBy(() -> service.submitResult(caller(CONSUMER), TASK_ID, incomplete))
+                .isInstanceOf(BizException.class).hasMessageContaining("INCOMPLETE_RESULT");
+        verify(resultRepository, never()).saveAndFlush(any());
+        service.submitResult(caller(CONSUMER), TASK_ID, report(1, AnswerReconcileTask.STATE_COMPLETED));
+        verify(taskRepository).finishAttempt(eq(1L), eq(1), eq(AnswerReconcileTask.STATE_COMPLETED),
+                any(LocalDateTime.class), isNull(), isNull());
+        assertThat(task.getState()).isEqualTo(AnswerReconcileTask.STATE_COMPLETED);
+    }
+
+    @Test
+    @DisplayName("PARTIAL 本轮已结束：不再提供清单、快照或缺席证明，也不接受迟到心跳恢复")
+    void partialTaskRejectsFurtherReadsAndProgress() {
+        AnswerReconcileTask task = task(AnswerReconcileTask.STATE_PARTIAL);
+        when(taskRepository.findByTaskUuid(TASK_ID)).thenReturn(Optional.of(task));
+        when(grantRepository.findByFormUuidAndConsumerSchoolIdAndConsumerAppKey(FORM_UUID, SCHOOL, CONSUMER))
+                .thenReturn(Optional.of(grant(1L, FORM_UUID, 1)));
+        assertThatThrownBy(() -> service.manifestPage(caller(CONSUMER), TASK_ID, null, 2))
+                .isInstanceOf(BizException.class).hasMessageContaining("TASK_TERMINAL");
+        assertThatThrownBy(() -> service.taskAnswer(caller(CONSUMER), TASK_ID, ANSWER_A, null))
+                .isInstanceOf(BizException.class).hasMessageContaining("TASK_TERMINAL");
+        ResultAck ack = service.submitResult(caller(CONSUMER), TASK_ID,
+                report(1, AnswerReconcileTask.STATE_RUNNING));
+        assertThat(ack.getDuplicate()).isTrue();
+        assertThat(task.getState()).isEqualTo(AnswerReconcileTask.STATE_PARTIAL);
+        verify(resultRepository, never()).saveAndFlush(any());
+        verify(taskRepository, never()).renewRunning(anyLong(), anyInt(), any());
+        verify(taskRepository, never()).finishAttempt(anyLong(), anyInt(), anyString(), any(), any(), any());
     }
 
     @Test
@@ -662,7 +720,7 @@ class FcReconcileTaskTest {
         stubAccessibleTask();
         when(resultRepository.findByTaskIdAndAttempt(1L, 1)).thenReturn(Optional.empty());
         when(resultRepository.maxAttempt(1L)).thenReturn(0);
-        when(resultRepository.save(any(AnswerReconcileResult.class))).thenAnswer(call -> call.getArgument(0));
+        when(resultRepository.saveAndFlush(any(AnswerReconcileResult.class))).thenAnswer(call -> call.getArgument(0));
         properties.setMaxErrorItems(3);
         List<ReconcileError> errors = new ArrayList<>();
         for (int i = 0; i < 10; i++) {
@@ -678,7 +736,7 @@ class FcReconcileTaskTest {
 
         ArgumentCaptor<AnswerReconcileResult> saved = ArgumentCaptor.forClass(AnswerReconcileResult.class);
         service.submitResult(caller(CONSUMER), TASK_ID, report);
-        verify(resultRepository).save(saved.capture());
+        verify(resultRepository).saveAndFlush(saved.capture());
 
         AnswerReconcileResult row = saved.getValue();
         assertThat(row.getErrorsJson()).doesNotContain("冒充身份");

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

@@ -0,0 +1,220 @@
+package com.yunhai.formcenter.service;
+
+import com.yunhai.formcenter.entity.Form;
+import com.yunhai.formcenter.entity.FormConsumerGrant;
+import com.yunhai.formcenter.model.FormAccessModels.GrantRequest;
+import com.yunhai.formcenter.model.FormAccessModels.GrantView;
+import com.yunhai.formcenter.model.FormAccessModels.SubscriptionGrant;
+import com.yunhai.formcenter.repository.FormConsumerGrantRepository;
+import com.yunhai.formcenter.repository.FormOwnershipRepository;
+import com.yunhai.formcenter.repository.FormRepository;
+import com.yunhai.platform.apistd.BizException;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * 读取授权与 MQ 订阅的分权（2026-10-22 决策②）：
+ * {@code read_allowed}（拉取）/ {@code subscribe_allowed}（数据订阅：投影与对账授权的判据）/
+ * {@code push_allowed}（MQ broker 推送投递）三把开关独立——
+ * <b>HTTP-only 消费方保留 read+subscribe，但 push 不开就绝不建 outbox 行</b>，
+ * 避免无 broker 消费方在 fc_answer_delivery 里长期 PENDING。
+ */
+@DisplayName("消费授权三开关 - push_allowed 闸门（读取/订阅/MQ 推送分权）")
+class FormConsumerGrantPushGateTest {
+
+    private static final long DOMAIN = 5L;
+    private static final String FORM_UUID = "f0e1d2c3b4a5968778695a4b3c2d1e0f";
+    private static final String APP_KEY = "100018";
+
+    private FormOwnershipRepository ownership;
+    private FormConsumerGrantRepository grants;
+    private FormRepository forms;
+    private FormConsumerAccessService service;
+
+    @BeforeEach
+    void setUp() {
+        ownership = mock(FormOwnershipRepository.class);
+        grants = mock(FormConsumerGrantRepository.class);
+        forms = mock(FormRepository.class);
+        service = new FormConsumerAccessService(ownership, grants, forms);
+
+        Form form = new Form();
+        form.setUuid(FORM_UUID);
+        form.setSchId(DOMAIN);
+        when(forms.findByUuid(FORM_UUID)).thenReturn(Optional.of(form));
+        when(grants.save(any())).thenAnswer(call -> call.getArgument(0));
+    }
+
+    private GrantRequest request(Boolean read, Boolean subscribe, Boolean push) {
+        GrantRequest req = new GrantRequest();
+        req.setFormUuid(FORM_UUID);
+        req.setConsumerSchoolId(DOMAIN);
+        req.setConsumerAppKey(APP_KEY);
+        req.setReadAllowed(read);
+        req.setSubscribeAllowed(subscribe);
+        req.setPushAllowed(push);
+        return req;
+    }
+
+    private FormConsumerGrant existing(int grantVersion, int read, int subscribe, int push) {
+        FormConsumerGrant grant = new FormConsumerGrant();
+        grant.setId(1L);
+        grant.setFormUuid(FORM_UUID);
+        grant.setConsumerSchoolId(DOMAIN);
+        grant.setConsumerAppKey(APP_KEY);
+        grant.setReadAllowed(read);
+        grant.setSubscribeAllowed(subscribe);
+        grant.setPushAllowed(push);
+        grant.setEnabled(1);
+        grant.setGrantVersion(grantVersion);
+        when(grants.findByFormUuidAndConsumerSchoolIdAndConsumerAppKey(FORM_UUID, DOMAIN, APP_KEY))
+                .thenReturn(Optional.of(grant));
+        return grant;
+    }
+
+    @Test
+    @DisplayName("新 grant 的 push_allowed 默认 0：存量行不加配置就是 HTTP-only，不推送")
+    void newGrantDefaultsToHttpOnly() {
+        assertThat(new FormConsumerGrant().getPushAllowed()).isZero();
+    }
+
+    @Test
+    @DisplayName("push 依赖订阅数据权：subscribe 没开就给 push 直接拒")
+    void pushRequiresSubscribe() {
+        assertThatThrownBy(() -> service.grant(DOMAIN, request(true, null, true), false))
+                .isInstanceOf(BizException.class).hasMessageContaining("订阅");
+    }
+
+    @Test
+    @DisplayName("read+subscribe+push 三开：落库 push=1，grantVersion 从 1 起")
+    void tripleGrantWritesPushFlag() {
+        GrantView view = service.grant(DOMAIN, request(true, true, true), false);
+        assertThat(view.isPushAllowed()).isTrue();
+        assertThat(view.getGrantVersion()).isEqualTo(1);
+    }
+
+    @Test
+    @DisplayName("push 降级（关推送、保留 read+subscribe）也是能力变化：grantVersion 必须进位")
+    void demotingPushBumpsEpoch() {
+        FormConsumerGrant grant = existing(3, 1, 1, 1);
+
+        GrantView view = service.grant(DOMAIN, request(true, true, false), false);
+
+        assertThat(grant.getPushAllowed()).isZero();
+        assertThat(grant.getGrantVersion()).isEqualTo(4);
+        assertThat(view.isPushAllowed()).isFalse();
+    }
+
+    @Test
+    @DisplayName("push 升级同样新纪元：存量 PENDING 行的旧 grantVersion 从此发不出去")
+    void enablingPushBumpsEpoch() {
+        FormConsumerGrant grant = existing(3, 1, 1, 0);
+
+        service.grant(DOMAIN, request(true, true, true), false);
+
+        assertThat(grant.getPushAllowed()).isEqualTo(1);
+        assertThat(grant.getGrantVersion()).isEqualTo(4);
+    }
+
+    @Test
+    @DisplayName("三开关都没变就不进位：重复授权不制造新纪元")
+    void unchangedFlagsKeepEpoch() {
+        FormConsumerGrant grant = existing(3, 1, 1, 1);
+
+        service.grant(DOMAIN, request(true, true, true), false);
+
+        assertThat(grant.getGrantVersion()).isEqualTo(3);
+    }
+
+    @Test
+    @DisplayName("投递目标清单带 push 标志：入队侧据此跳过 HTTP-only 消费方")
+    void subscriptionListCarriesPushFlag() {
+        FormConsumerGrant httpOnly = existing(7, 1, 1, 0);
+        when(grants.findEnabledSubscriptions(FORM_UUID, DOMAIN))
+                .thenReturn(Collections.singletonList(httpOnly));
+
+        SubscriptionGrant view = service.listEnabledSubscriptions(FORM_UUID, DOMAIN).get(0);
+
+        assertThat(view.isSubscribeAllowed()).isTrue();
+        assertThat(view.isPushAllowed()).isFalse();
+        assertThat(view.getGrantVersion()).isEqualTo(7);
+    }
+
+    @Test
+    @DisplayName("管理面 grant 视图回显 push 标志：运维看得到哪个消费方在推送")
+    void grantViewCarriesPushFlag() {
+        // existing() 内部也有打桩，先完成它再 stub findLiveByForm（嵌套 when 会 UnfinishedStubbing）
+        FormConsumerGrant row = existing(2, 1, 1, 1);
+        when(grants.findLiveByForm(FORM_UUID)).thenReturn(Collections.singletonList(row));
+
+        GrantView view = service.listGrants(DOMAIN, FORM_UUID).get(0);
+
+        assertThat(view.isPushAllowed()).isTrue();
+    }
+
+    @Test
+    @DisplayName("旧客户端缺省/null push：保留现值且不无故推进授权纪元")
+    void omittedAndNullPushPreserveExistingPermission() throws Exception {
+        FormConsumerGrant grant = existing(3, 1, 1, 1);
+        for (String suffix : new String[]{"", ",\"pushAllowed\":null"}) {
+            GrantRequest req = com.yunhai.formcenter.common.FormSchemaValidator.mapper().readValue(
+                    "{\"formUuid\":\"" + FORM_UUID + "\",\"consumerSchoolId\":5,"
+                            + "\"consumerAppKey\":\"" + APP_KEY + "\",\"readAllowed\":true,"
+                            + "\"subscribeAllowed\":true" + suffix + "}", GrantRequest.class);
+            GrantView saved = service.grant(DOMAIN, req, false);
+            assertThat(saved.isPushAllowed()).isTrue();
+            assertThat(grant.getPushAllowed()).isEqualTo(1);
+            assertThat(saved.getGrantVersion()).isEqualTo(3);
+        }
+    }
+
+    @Test
+    void omittedPushKeepsHttpOnlyAndDefaultsNewGrantToFalse() {
+        GrantView created = service.grant(DOMAIN, request(true, true, null), false);
+        assertThat(created.isPushAllowed()).isFalse();
+        assertThat(created.getGrantVersion()).isEqualTo(1);
+        FormConsumerGrant grant = existing(4, 1, 1, 0);
+        assertThat(service.grant(DOMAIN, request(true, true, null), false).isPushAllowed()).isFalse();
+        assertThat(grant.getGrantVersion()).isEqualTo(4);
+    }
+
+    @Test
+    void regrantPreservesPushButStartsNewEpoch() {
+        FormConsumerGrant grant = existing(4, 1, 1, 1);
+        grant.setEnabled(0);
+        GrantView saved = service.grant(DOMAIN, request(true, true, null), false);
+        assertThat(saved.isPushAllowed()).isTrue();
+        assertThat(saved.isEnabled()).isTrue();
+        assertThat(saved.getGrantVersion()).isEqualTo(5);
+    }
+
+    @Test
+    @DisplayName("合并后push=true而关订阅：拒绝且不改任何旧授权值")
+    void inheritedPushRequiresExplicitDisableBeforeRemovingSubscription() {
+        FormConsumerGrant grant = existing(7, 1, 1, 1);
+        assertThatThrownBy(() -> service.grant(DOMAIN, request(true, false, null), false))
+                .isInstanceOf(BizException.class).hasMessageContaining("显式设置 pushAllowed=false");
+        assertThat(grant.getReadAllowed()).isEqualTo(1);
+        assertThat(grant.getSubscribeAllowed()).isEqualTo(1);
+        assertThat(grant.getPushAllowed()).isEqualTo(1);
+        assertThat(grant.getEnabled()).isEqualTo(1);
+        assertThat(grant.getGrantVersion()).isEqualTo(7);
+        verify(grants, never()).save(any());
+        GrantView saved = service.grant(DOMAIN, request(true, false, false), false);
+        assertThat(saved.isSubscribeAllowed()).isFalse();
+        assertThat(saved.isPushAllowed()).isFalse();
+        assertThat(saved.getGrantVersion()).isEqualTo(8);
+    }
+}