移除接收区间补采与完整性旧链路

yeg

2026-09-02 74f55db1acd945f4860b2d0b4bae48e2b918b32b

移除接收区间补采与完整性旧链路

实时接收直接追加成功采样并保留幂等冲突审计，暂停和停止不再创建或关闭接收区间。删除旧调度、完整性接口和 SSE 事件，补充 V28 数据库清理迁移及契约测试。

 ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizer.java

@@ -10,7 +10,6 @@
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore;
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore.CollectorCandidate;
import com.ruoyi.system.mapper.TCollectorMapper;
import com.ruoyi.system.service.ICollectorDisconnectRecoveryService;

/** 将模拟采集器的发现写入 Redis，并同步已登记采集器的实时在线事实。 */
@Component
@@ -19,25 +18,17 @@
    private final BtsDevelopmentSimulator simulator;
    private final DeviceDiscoveryCandidateStore candidateStore;
    private final TCollectorMapper collectorMapper;
    private final ICollectorDisconnectRecoveryService disconnectRecovery;

    @Autowired
    public DevelopmentCollectorStatusSynchronizer(BtsDevelopmentSimulator simulator,
            DeviceDiscoveryCandidateStore candidateStore, TCollectorMapper collectorMapper,
            ICollectorDisconnectRecoveryService disconnectRecovery)
            DeviceDiscoveryCandidateStore candidateStore, TCollectorMapper collectorMapper)
    {
        this.simulator = simulator;
        this.candidateStore = candidateStore;
        this.collectorMapper = collectorMapper;
        this.disconnectRecovery = disconnectRecovery;
    }

    /** 保留单元测试及独立模拟器装配使用的旧构造入口。 */
    public DevelopmentCollectorStatusSynchronizer(BtsDevelopmentSimulator simulator,
            DeviceDiscoveryCandidateStore candidateStore, TCollectorMapper collectorMapper)
    {
        this(simulator, candidateStore, collectorMapper, (codes, at) -> { });
    }

    /** 完整替换 dev-sim 采集器快照，删除配置后候选立即消失。 */
    @Scheduled(fixedDelay = BtsDevelopmentSimulator.POLL_INTERVAL_MS)
@@ -54,7 +45,5 @@
        candidateStore.replaceCollectors(BtsDevelopmentSimulator.ADAPTER_CODE, candidates);
        candidates.forEach(candidate -> collectorMapper.updateRegisteredObservation(candidate.deviceCode(),
                candidate.online() ? "ONLINE" : "OFFLINE"));
        disconnectRecovery.reconcile(simulatorState.collectors().stream()
                .filter(item -> item.online()).map(item -> item.deviceCode()).collect(java.util.stream.Collectors.toSet()), observedAt);
    }
}

 ruoyi-admin/src/main/java/com/ruoyi/config/TestBackfillScheduler.java

@@ -1,24 +1,9 @@
package com.ruoyi.config;

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import com.ruoyi.system.service.ITestBackfillService;

/** 以较低优先级执行历史补采；实时设备观测由既有调度器优先完成。 */
@Component
public class TestBackfillScheduler
{
    private final ITestBackfillService backfill;

    public TestBackfillScheduler(ITestBackfillService backfill)
    {
        this.backfill = backfill;
    }

    @Scheduled(fixedDelay = 3000)
    public void reconcileAndBackfill()
    {
        backfill.reconcileNaturalEnd();
        backfill.resumePendingBackfill();
    }
    /** Ticket28 移除历史补采调度，保留空组件避免旧配置加载失败。 */
    public void reconcileAndBackfill() { }
}

 ruoyi-admin/src/main/java/com/ruoyi/web/controller/system/TTestLiveController.java

@@ -10,7 +10,6 @@
import com.ruoyi.common.core.controller.BaseController;
import com.ruoyi.common.core.domain.AjaxResult;
import com.ruoyi.system.service.ITestLiveService;
import com.ruoyi.system.service.ITestIntegrityService;

/** 测试控制台实时 REST/SSE 入口，仅负责鉴权和协议转发。 */
@RestController
@@ -18,19 +17,14 @@
public class TTestLiveController extends BaseController
{
    private final ITestLiveService liveService;
    private final ITestIntegrityService integrityService;
    public TTestLiveController(ITestLiveService liveService, ITestIntegrityService integrityService)
    { this.liveService = liveService; this.integrityService = integrityService; }
    public TTestLiveController(ITestLiveService liveService)
    { this.liveService = liveService; }

    @PreAuthorize("@ss.hasPermi('system:test:query')")
    @GetMapping("/{testId}/live-state")
    public AjaxResult liveState(@PathVariable Long testId) { return success(liveService.current(testId)); }

    /** 查询测试完整性问题和可追溯的历史异常事实。 */
    @PreAuthorize("@ss.hasPermi('system:test:query')")
    @GetMapping("/{testId}/integrity")
    public AjaxResult integrity(@PathVariable Long testId) { return success(integrityService.selectIssues(testId)); }

    @PreAuthorize("@ss.hasPermi('system:test:query')")
    @GetMapping(value = "/{testId}/live", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter live(@PathVariable Long testId) { return liveService.stream(testId); }

 ruoyi-admin/src/main/resources/db/migration/V28__remove_reception_backfill_integrity_legacy.sql

New file
@@ -0,0 +1,27 @@
-- Ticket28：撤销已废弃的接收区间、历史补采、完整性和采集器断线恢复模型。
-- 原始 t_test_sample 仍保留，成功接收数据继续通过该表追加保存。
SET FOREIGN_KEY_CHECKS = 0;

DROP TABLE t_test_integrity_issue;
DROP TABLE t_test_backfill_cursor;
DROP TABLE t_test_collector_exception_interval;
DROP TABLE t_test_reception_interval;
DROP TABLE t_collector_channel_lease;
DROP TABLE t_channel_lease;

ALTER TABLE t_test_sample
    DROP FOREIGN KEY fk_sample_reception_interval,
    DROP KEY idx_sample_interval,
    DROP COLUMN reception_interval_id;

ALTER TABLE t_test_collector_sample
    DROP FOREIGN KEY fk_collector_sample_reception_interval,
    DROP KEY idx_collector_sample_interval,
    DROP COLUMN reception_interval_id;

ALTER TABLE t_test
    DROP COLUMN backfill_status,
    DROP COLUMN backfill_started_at_utc,
    DROP COLUMN backfill_completed_at_utc;

SET FOREIGN_KEY_CHECKS = 1;

 ruoyi-admin/src/test/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizerTest.java

@@ -16,10 +16,10 @@
    void productionConstructorIsExplicitlyAutowired()
    {
        var constructors = DevelopmentCollectorStatusSynchronizer.class.getDeclaredConstructors();
        assertEquals(2, constructors.length);
        assertEquals(1, constructors.length);
        var production = java.util.Arrays.stream(constructors)
                .filter(item -> item.isAnnotationPresent(Autowired.class)).findFirst().orElseThrow();
        assertEquals(4, production.getParameterCount());
        assertEquals(3, production.getParameterCount());
    }
    @Test
    void synchronizesOnlyConfiguredCollectorsWithoutRegisteringThem() throws Exception

 ruoyi-admin/src/test/java/com/ruoyi/config/TestDeviceBindingMapperContractTest.java

@@ -27,12 +27,14 @@
        assertTrue(mapper.contains("id=\"selectReceptionChannelForUpdate\""));
        assertTrue(mapper.contains("channel_fact.online_status onlineStatus"));
        assertTrue(mapper.contains("channel_fact.last_seen_at_utc lastSeenAtUtc"));
        assertTrue(mapper.contains("then 'RUNNING_ABNORMAL'"));
        assertTrue(mapper.contains("exception_fact.interval_status = 'OPEN'"));
        assertTrue(mapper.contains("set test_status = 'COMPLETED'"));
        assertTrue(mapper.contains("set test_status = 'COMPLETED', reception_status = 'ENDED'"));
        assertTrue(mapper.contains("backfill_status = 'COMPLETED'"));
        assertTrue(mapper.contains("#{testStatus} = 'COMPLETED' and t.test_status in ('COMPLETED', 'COMPLETION_PENDING', 'TERMINATED')"));
        assertFalse(mapper.contains("t_test_reception_interval"));
        assertFalse(mapper.contains("reception_interval_id"));
        assertFalse(mapper.contains("t_test_integrity_issue"));
        assertFalse(mapper.contains("t_test_backfill_cursor"));
        assertFalse(mapper.contains("t_test_collector_exception_interval"));
        assertFalse(mapper.contains("group_concat"));
    }


 ruoyi-admin/src/test/java/com/ruoyi/config/TestPersistedDataQueryContractTest.java

@@ -16,7 +16,7 @@
        assertTrue(mapper.contains("id=\"selectPersistedRecords\""));
        assertTrue(mapper.contains("id=\"selectPersistedSteps\""));
        assertTrue(mapper.contains("id=\"selectPersistedCycles\""));
        assertTrue(mapper.contains("id=\"selectTestSampleContentHash\""));
        assertTrue(mapper.contains("id=\"selectBtsSampleContentHash\""));
        assertTrue(mapper.contains("sample.step_name"));
        assertTrue(mapper.contains("where sample.test_id = #{testId}"));
        assertTrue(mapper.contains("from t_test_sample sample"));

 ruoyi-admin/src/test/java/com/ruoyi/config/Ticket28MigrationContractTest.java

New file
@@ -0,0 +1,39 @@
package com.ruoyi.config;

import static org.junit.jupiter.api.Assertions.assertTrue;

import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import org.junit.jupiter.api.Test;

/** Ticket28 迁移必须移除区间、租约、补采和完整性遗留模型。 */
class Ticket28MigrationContractTest
{
    @Test
    void removesLegacyTablesColumnsAndConstraints()
    {
        String sql = resource("db/migration/V28__remove_reception_backfill_integrity_legacy.sql");
        assertTrue(sql.contains("DROP TABLE t_test_reception_interval"));
        assertTrue(sql.contains("DROP TABLE t_collector_channel_lease"));
        assertTrue(sql.contains("DROP TABLE t_channel_lease"));
        assertTrue(sql.contains("DROP FOREIGN KEY fk_sample_reception_interval"));
        assertTrue(sql.contains("DROP FOREIGN KEY fk_collector_sample_reception_interval"));
        assertTrue(sql.contains("DROP COLUMN reception_interval_id"));
        assertTrue(sql.contains("DROP COLUMN backfill_status"));
        assertTrue(sql.contains("DROP COLUMN backfill_started_at_utc"));
        assertTrue(sql.contains("DROP COLUMN backfill_completed_at_utc"));
    }

    private static String resource(String path)
    {
        try (InputStream stream = Ticket28MigrationContractTest.class.getClassLoader().getResourceAsStream(path))
        {
            if (stream == null) throw new AssertionError("Missing migration resource " + path);
            return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
        }
        catch (Exception exception)
        {
            throw new AssertionError("Unable to read migration resource " + path, exception);
        }
    }
}

 ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/TTestLiveServiceImplTest.java

@@ -14,7 +14,6 @@
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ITTestService;
import com.ruoyi.system.service.ITestLiveService.LiveState;

@@ -66,7 +65,7 @@
        private TestableLiveService(RecordingEmitter emitter)
        {
            super(mock(ITTestService.class), mock(DeviceDiscoveryCandidateStore.class),
                    mock(BtsDevelopmentSimulator.class), mock(TTestMapper.class));
                    mock(BtsDevelopmentSimulator.class));
            this.emitter = emitter;
        }


 ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/TTestServiceImplTest.java

@@ -23,7 +23,6 @@
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.dao.DuplicateKeyException;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.system.domain.TTest;
import com.ruoyi.system.domain.TTestBattery;
@@ -370,7 +369,7 @@
    }

    @Test
    void startReceptionShouldOpenAllSourceIntervalsAtomically()
    void startReceptionShouldPersistReceptionWithoutLegacyIntervals()
    {
        TTest current = validTest();
        current.setTestId(10L);
@@ -388,10 +387,6 @@
        when(testMapper.updateReceptionStarted(eq(10L), any(), anyBoolean(), eq("operator"))).thenReturn(1);

        assertEquals(1, testService.startReception(10L, 7L, "operator"));
        verify(testMapper).insertChannelLease(eq(100L), eq(10L), eq(30L), eq(7L), any());
        verify(testMapper).insertReceptionInterval(eq(10L), eq(30L), eq("CABINET_CHANNEL"), eq(100L), isNull(), any(), eq("RECEPTION_STARTED"));
        verify(testMapper, times(3)).insertCollectorChannelLease(eq(8L), any(), eq(10L), eq(30L), eq(7L), any());
        verify(testMapper, times(3)).insertReceptionInterval(eq(10L), eq(30L), eq("COLLECTOR_CHANNEL"), eq(8L), any(), any(), eq("RECEPTION_STARTED"));
        verify(testMapper).insertTestAudit(eq(10L), eq("RECEPTION_STARTED"), eq(7L), eq("operator"), isNull(), any());
    }

@@ -421,7 +416,7 @@
    }

    @Test
    void startReceptionShouldRollbackWhenDatabaseLeaseIsTaken()
    void startReceptionShouldNotDependOnDatabaseLease()
    {
        TTest current = validTest();
        current.setTestId(10L);
@@ -436,15 +431,10 @@
        when(testMapper.countCollectorEnabled(8L)).thenReturn(1);
        when(testMapper.selectCollectorChannelCount(8L)).thenReturn(4);
        when(testMapper.collectorOnline(8L)).thenReturn(true);
        when(testMapper.insertChannelLease(any(), any(), any(), any(), any()))
                .thenThrow(new DuplicateKeyException("lease already exists"));
        when(testMapper.updateReceptionStarted(eq(10L), any(), anyBoolean(), eq("operator"))).thenReturn(1);

        ServiceException exception = assertThrows(ServiceException.class,
                () -> testService.startReception(10L, 7L, "operator"));
        assertEquals(1, testService.startReception(10L, 7L, "operator"));

        assertEquals("数据来源已被其他测试占用，请刷新后重试", exception.getMessage());
        verify(testMapper, never()).updateReceptionStarted(any(), any(), anyBoolean(), any());
        verify(testMapper, never()).insertTestAudit(anyLong(), any(), anyLong(), any(), any(), any());
    }

    @Test
@@ -463,7 +453,6 @@
                () -> testService.startReception(10L, 7L, "operator"));

        assertEquals("检测柜通道 100 当前离线或不可观测", exception.getMessage());
        verify(testMapper, never()).insertChannelLease(any(), any(), any(), any(), any());
        verify(testMapper, never()).updateReceptionStarted(any(), any(), anyBoolean(), any());
    }

@@ -489,7 +478,6 @@
                () -> testService.startReception(10L, 7L, "operator"));

        assertEquals("每个电池号必须绑定主温度、正电位主通道和负电位主通道", exception.getMessage());
        verify(testMapper, never()).insertChannelLease(any(), any(), any(), any(), any());
        verify(testMapper, never()).updateReceptionStarted(any(), any(), anyBoolean(), any());
    }

@@ -537,7 +525,7 @@
    }

    @Test
    void stopTestShouldCompleteAndReleaseLeases()
    void stopTestShouldCompleteWithoutLegacyLeaseCleanup()
    {
        TTest current = validTest();
        current.setTestId(10L);
@@ -547,28 +535,26 @@

        assertEquals(1, testService.stopTest(10L, 7L, "operator"));
        verify(testMapper).updateReceptionStopped(eq(10L), eq("operator"), any());
        verify(testMapper).closeOpenReceptionIntervals(eq(10L), any(), eq("MANUAL_END"));
        verify(testMapper).deleteChannelLeasesByTestId(10L);
        verify(testMapper).deleteCollectorChannelLeasesByTestId(10L);
        verify(testMapper, never()).deleteChannelLeasesByTestId(anyLong());
        verify(testMapper, never()).deleteCollectorChannelLeasesByTestId(anyLong());
        verify(testMapper).insertTestAudit(eq(10L), eq("TEST_STOPPED"), eq(7L), eq("operator"), isNull(),
                org.mockito.ArgumentMatchers.argThat(value -> value != null && value.contains("COMPLETED")));
    }

    @Test
    void stopTestShouldAuditImmediateReleaseAndClosedIntervals()
    void stopTestShouldAuditImmediateStop()
    {
        TTest current = validTest();
        current.setTestId(10L);
        current.setReceptionStatus("RECEIVING");
        when(testMapper.selectTestByIdForUpdate(10L)).thenReturn(current);
        when(testMapper.updateReceptionStopped(eq(10L), eq("operator"), any())).thenReturn(1);
        when(testMapper.closeOpenReceptionIntervals(eq(10L), any(), eq("MANUAL_END"))).thenReturn(4);

        testService.stopTest(10L, 7L, "operator");

        verify(testMapper).insertTestAudit(eq(10L), eq("TEST_STOPPED"), eq(7L), eq("operator"), isNull(),
                org.mockito.ArgumentMatchers.argThat(value -> value != null
                        && value.contains("closedIntervals") && value.contains("IMMEDIATE")));
                        && value.contains("IMMEDIATE")));
    }

    private void mockEnabledOptions()

 ruoyi-admin/src/test/java/com/ruoyi/web/controller/system/TTestLiveControllerContractTest.java

@@ -23,15 +23,6 @@
                .contains("system:test:query"));
    }

    @Test
    void integrityMappingIsAuthenticatedSurface()
    {
        Method integrity = method("integrity", Long.class);
        assertArrayEquals(new String[] { "/{testId}/integrity" }, integrity.getAnnotation(GetMapping.class).value());
        assertTrue(integrity.getAnnotation(org.springframework.security.access.prepost.PreAuthorize.class).value()
                .contains("system:test:query"));
    }

    private Method method(String name, Class<?>... types)
    {
        try { return TTestLiveController.class.getDeclaredMethod(name, types); }

 ruoyi-system/src/main/java/com/ruoyi/system/domain/TTest.java

@@ -39,9 +39,6 @@
    private Date endedAtUtc;
    /** 是否从已运行设备的后续观测边界接入。 */
    private Boolean runtimeAttach;
    private String backfillStatus;
    private Date backfillStartedAtUtc;
    private Date backfillCompletedAtUtc;
    private String delFlag;
    private String batteryNo;
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
@@ -58,7 +55,6 @@
    @NotEmpty(message = "阈值配置不能为空")
    private List<TTestWarningThreshold> thresholds;
    private List<TTestFieldSnapshot> fieldSnapshots;
    private List<TTestReceptionInterval> receptionIntervals;
    private List<TTestResult> resultSummary;
    private String logicStepMappingsJson;
    private List<TTestLogicStepMapping> logicStepMappings;
@@ -95,12 +91,6 @@
    public void setEndedAtUtc(Date value) { endedAtUtc = value; }
    public Boolean getRuntimeAttach() { return runtimeAttach; }
    public void setRuntimeAttach(Boolean value) { runtimeAttach = value; }
    public String getBackfillStatus() { return backfillStatus; }
    public void setBackfillStatus(String value) { backfillStatus = value; }
    public Date getBackfillStartedAtUtc() { return backfillStartedAtUtc; }
    public void setBackfillStartedAtUtc(Date value) { backfillStartedAtUtc = value; }
    public Date getBackfillCompletedAtUtc() { return backfillCompletedAtUtc; }
    public void setBackfillCompletedAtUtc(Date value) { backfillCompletedAtUtc = value; }
    public String getDelFlag() { return delFlag; }
    public void setDelFlag(String value) { delFlag = value; }
    public String getBatteryNo() { return batteryNo; }
@@ -119,8 +109,6 @@
    public void setThresholds(List<TTestWarningThreshold> value) { thresholds = value; }
    public List<TTestFieldSnapshot> getFieldSnapshots() { return fieldSnapshots; }
    public void setFieldSnapshots(List<TTestFieldSnapshot> value) { fieldSnapshots = value; }
    public List<TTestReceptionInterval> getReceptionIntervals() { return receptionIntervals; }
    public void setReceptionIntervals(List<TTestReceptionInterval> value) { receptionIntervals = value; }
    public List<TTestResult> getResultSummary() { return resultSummary; }
    public void setResultSummary(List<TTestResult> value) { resultSummary = value; }
    public String getLogicStepMappingsJson() { return logicStepMappingsJson; }

 ruoyi-system/src/main/java/com/ruoyi/system/mapper/TTestMapper.java

@@ -11,12 +11,12 @@
import com.ruoyi.system.domain.TTestPlanStep;
import com.ruoyi.system.domain.TTestResult;
import com.ruoyi.system.domain.TTestWarningThreshold;
import com.ruoyi.system.domain.TTestCollectorSource;
import com.ruoyi.system.domain.TTestCollectorExceptionInterval;
import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.domain.TTestLogicStepMapping;
import com.ruoyi.system.domain.TTestCollectorSource;
import com.ruoyi.system.domain.TTestBackfillCursor;
import com.ruoyi.system.domain.TTestIntegrityIssue;
import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.domain.TTestLogicStepMapping;
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.TTestDataRevisionAction;
import com.ruoyi.system.domain.TTestConsoleImportRecord;
@@ -57,7 +57,6 @@
    List<TTestCollectorChannel> selectTestCollectorChannels(Long testId);
    List<TTestWarningThreshold> selectTestThresholds(Long testId);
    /** 按来源和起始边界顺序返回测试接收区间。 */
    List<TTestReceptionInterval> selectTestReceptionIntervals(Long testId);
    /** 查询当前正在接收的测试，供接收调度绑定测试事实。 */
    List<Long> selectReceivingTestIds();
    List<java.util.Map<String, Object>> selectPersistedRecords(@org.apache.ibatis.annotations.Param("testId") Long testId);
@@ -86,23 +85,8 @@
    /** 锁定检测柜通道的最新可观测事实，供开始接收校验使用。 */
    TDetectionChannel selectReceptionChannelForUpdate(Long channelId);
    /** 锁定当前检测柜通道租约；返回空表示可接入。 */
    TTestChannelBinding selectChannelLeaseForUpdate(Long channelId);
    int insertChannelLease(@Param("channelId") Long channelId, @Param("testId") Long testId,
            @Param("bindingId") Long bindingId, @Param("acquiredBy") Long acquiredBy,
            @Param("acquiredAtUtc") Date acquiredAtUtc);
    int insertCollectorChannelLease(@Param("collectorId") Long collectorId, @Param("channelCode") String channelCode,
            @Param("testId") Long testId, @Param("bindingId") Long bindingId,
            @Param("acquiredBy") Long acquiredBy, @Param("acquiredAtUtc") Date acquiredAtUtc);
    /** 为每个绑定来源追加一个开放接收区间；区间关闭后仅保留历史事实。 */
    int insertReceptionInterval(@Param("testId") Long testId, @Param("bindingId") Long bindingId,
            @Param("sourceType") String sourceType, @Param("sourceId") Long sourceId,
            @Param("sourceChannelCode") String sourceChannelCode, @Param("actionAtUtc") Date actionAtUtc,
            @Param("boundaryNote") String boundaryNote);
    /** 在停止动作事务内关闭该测试全部开放区间，返回实际关闭数量。 */
    int closeOpenReceptionIntervals(@Param("testId") Long testId, @Param("endBoundaryAtUtc") Date endBoundaryAtUtc,
            @Param("boundaryNote") String boundaryNote);
    int deleteChannelLeasesByTestId(Long testId);
    int deleteCollectorChannelLeasesByTestId(Long testId);
    int hasReceivingTestForAdapter(@Param("adapterCode") String adapterCode);
    int hasPausedOrEndedTestForAdapter(@Param("adapterCode") String adapterCode);
    List<TDetectionChannel> selectReceivingChannelsForAdapter(@Param("adapterCode") String adapterCode);
@@ -117,33 +101,12 @@
            @Param("endedAtUtc") Date endedAtUtc);

    /** 查询仍在接收中的采集器来源。 */
    List<TTestCollectorSource> selectReceivingCollectorSources();
    /** 查询来源当前开放的断线区间。 */
    TTestCollectorExceptionInterval selectOpenCollectorException(@Param("testId") Long testId,
            @Param("batteryRelationId") Long batteryRelationId, @Param("collectorId") Long collectorId,
            @Param("channelCode") String channelCode);
    int insertCollectorExceptionInterval(TTestCollectorExceptionInterval interval);
    int closeCollectorExceptionInterval(@Param("intervalId") Long intervalId,
            @Param("recoveredAtUtc") Date recoveredAtUtc, @Param("missingToUtc") Date missingToUtc);
    List<TTestCollectorExceptionInterval> selectCollectorExceptionIntervals(Long testId);
    String selectBtsSampleContentHash(@Param("channelId") Long channelId, @Param("btsTestId") String btsTestId,
            @Param("sequenceId") Long sequenceId);

    /** 查询需要自然结束或继续补采的测试。 */
    List<TTest> selectTestsForNaturalEnd();
    List<TTest> selectTestsForBackfill();
    /** 按稳定来源身份锁定补采游标，避免应用重启重复推进。 */
    TTestBackfillCursor selectBackfillCursorForUpdate(@Param("testId") Long testId,
            @Param("bindingId") Long bindingId, @Param("sourceType") String sourceType,
            @Param("sourceId") Long sourceId, @Param("sourceChannelCode") String sourceChannelCode);
    int insertBackfillCursor(TTestBackfillCursor cursor);
    int updateBackfillCursor(TTestBackfillCursor cursor);
    int updateTestBackfillStarted(@Param("testId") Long testId, @Param("updatedAtUtc") Date updatedAtUtc);
    int updateTestBackfillWaiting(@Param("testId") Long testId, @Param("updatedAtUtc") Date updatedAtUtc,
            @Param("error") String error);
    int updateTestBackfillCompleted(@Param("testId") Long testId, @Param("updatedAtUtc") Date updatedAtUtc);
    int updateTestCompleted(@Param("testId") Long testId, @Param("updatedAtUtc") Date updatedAtUtc);
    int updateTestNaturallyEnded(@Param("testId") Long testId, @Param("endedAtUtc") Date endedAtUtc);
    String selectSampleContentHash(@Param("channelId") Long channelId, @Param("btsTestId") String btsTestId,
            @Param("sequenceId") Long sequenceId);
    /** 按测试事实查询通道序号，避免补采重复写入已即时接收的采样。 */
    String selectTestSampleContentHash(@Param("testId") Long testId, @Param("channelId") Long channelId,
            @Param("sequenceId") Long sequenceId);
@@ -162,8 +125,8 @@
    Long selectBtsSampleId(@Param("channelId") Long channelId, @Param("btsTestId") String btsTestId,
            @Param("sequenceId") Long sequenceId);
    int insertCollectorSample(@Param("testId") Long testId, @Param("bindingId") Long bindingId,
            @Param("intervalId") Long intervalId,
            @Param("collectorId") Long collectorId, @Param("channelCode") String channelCode,
            @Param("intervalId") Long intervalId, @Param("collectorId") Long collectorId,
            @Param("channelCode") String channelCode,
            @Param("sequenceId") Long sequenceId, @Param("deviceTimeUtc") Date deviceTimeUtc,
            @Param("deviceTimeRaw") String deviceTimeRaw, @Param("receivedAtUtc") Date receivedAtUtc,
            @Param("metricCode") String metricCode, @Param("metricValue") java.math.BigDecimal metricValue,
@@ -173,6 +136,61 @@
    String selectCollectorSampleContentHash(@Param("collectorId") Long collectorId,
            @Param("channelCode") String channelCode, @Param("sequenceId") Long sequenceId,
            @Param("metricCode") String metricCode);
    int insertDataRevisionAction(TTestDataRevisionAction action);
    List<TTestDataRevisionAction> selectDataRevisionActions(TTestDataRevisionAction query);
    TTestDataRevisionAction selectDataRevisionActionById(Long actionId);
    int upsertDataOverride(TTestDataOverride override);
    List<TTestDataOverride> selectDataOverrides(@Param("testId") Long testId,
            @Param("dataType") String dataType);
    int deleteBtsSamplesByIds(@Param("testId") Long testId, @Param("sampleIds") List<Long> sampleIds);
    int deleteBtsMeasurementsBySampleIds(@Param("sampleIds") List<Long> sampleIds);
    int deleteCollectorSamplesByIds(@Param("testId") Long testId, @Param("sampleIds") List<Long> sampleIds);
    Long selectBtsSampleIdByDeviceChannelSequence(@Param("testId") Long testId,
            @Param("deviceId") Long deviceId, @Param("channelNo") Long channelNo,
            @Param("sequenceId") Long sequenceId);
    List<Long> selectBtsSampleIdsByDeviceChannelStep(@Param("testId") Long testId,
            @Param("deviceId") Long deviceId, @Param("channelNo") Long channelNo,
            @Param("cycleNo") Integer cycleNo, @Param("stepNo") Integer stepNo);
    List<Long> selectBtsSampleIdsByDeviceChannelCycle(@Param("testId") Long testId,
            @Param("deviceId") Long deviceId, @Param("channelNo") Long channelNo,
            @Param("cycleNo") Integer cycleNo);
    int insertConsoleImportRecord(TTestConsoleImportRecord record);
    int insertSampleConflict(@Param("testId") Long testId, @Param("channelId") Long channelId,
            @Param("detail") String detail);
    int countTestBatteryNo(@Param("testId") Long testId, @Param("batteryNo") String batteryNo);
    /** 兼容已移除的历史实现源码；当前运行链路不再调用这些方法。 */
    TTestChannelBinding selectChannelLeaseForUpdate(Long channelId);
    int insertChannelLease(@Param("channelId") Long channelId, @Param("testId") Long testId,
            @Param("bindingId") Long bindingId, @Param("acquiredBy") Long acquiredBy,
            @Param("acquiredAtUtc") Date acquiredAtUtc);
    int insertCollectorChannelLease(@Param("collectorId") Long collectorId, @Param("channelCode") String channelCode,
            @Param("testId") Long testId, @Param("bindingId") Long bindingId,
            @Param("acquiredBy") Long acquiredBy, @Param("acquiredAtUtc") Date acquiredAtUtc);
    int deleteChannelLeasesByTestId(Long testId);
    int deleteCollectorChannelLeasesByTestId(Long testId);
    List<TTestCollectorSource> selectReceivingCollectorSources();
    TTestCollectorExceptionInterval selectOpenCollectorException(@Param("testId") Long testId,
            @Param("batteryRelationId") Long batteryRelationId, @Param("collectorId") Long collectorId,
            @Param("channelCode") String channelCode);
    int insertCollectorExceptionInterval(TTestCollectorExceptionInterval interval);
    int closeCollectorExceptionInterval(@Param("intervalId") Long intervalId,
            @Param("recoveredAtUtc") Date recoveredAtUtc, @Param("missingToUtc") Date missingToUtc);
    List<TTestCollectorExceptionInterval> selectCollectorExceptionIntervals(Long testId);
    List<TTest> selectTestsForNaturalEnd();
    List<TTest> selectTestsForBackfill();
    TTestBackfillCursor selectBackfillCursorForUpdate(@Param("testId") Long testId,
            @Param("bindingId") Long bindingId, @Param("sourceType") String sourceType,
            @Param("sourceId") Long sourceId, @Param("sourceChannelCode") String sourceChannelCode);
    int insertBackfillCursor(TTestBackfillCursor cursor);
    int updateBackfillCursor(TTestBackfillCursor cursor);
    int updateTestBackfillStarted(@Param("testId") Long testId, @Param("updatedAtUtc") Date updatedAtUtc);
    int updateTestBackfillWaiting(@Param("testId") Long testId, @Param("updatedAtUtc") Date updatedAtUtc,
            @Param("error") String error);
    int updateTestBackfillCompleted(@Param("testId") Long testId, @Param("updatedAtUtc") Date updatedAtUtc);
    int updateTestCompleted(@Param("testId") Long testId, @Param("updatedAtUtc") Date updatedAtUtc);
    int updateTestNaturallyEnded(@Param("testId") Long testId, @Param("endedAtUtc") Date endedAtUtc);
    String selectSampleContentHash(@Param("channelId") Long channelId, @Param("btsTestId") String btsTestId,
            @Param("sequenceId") Long sequenceId);
    int insertBackfillConflict(@Param("testId") Long testId, @Param("channelId") Long channelId,
            @Param("detail") String detail);
    List<TTestIntegrityIssue> selectIntegrityIssues(Long testId);
@@ -198,25 +216,9 @@
    int resolveIntegrityIssue(@Param("issueId") Long issueId, @Param("resolvedAtUtc") Date resolvedAtUtc);
    int countOpenIntegrityIssues(Long testId);
    int countOpenExceptions(Long testId);
    int closeOpenReceptionIntervals(@Param("testId") Long testId, @Param("endBoundaryAtUtc") Date endBoundaryAtUtc,
            @Param("boundaryNote") String boundaryNote);
    List<TTestReceptionInterval> selectTestReceptionIntervals(Long testId);

    int insertDataRevisionAction(TTestDataRevisionAction action);
    List<TTestDataRevisionAction> selectDataRevisionActions(TTestDataRevisionAction query);
    TTestDataRevisionAction selectDataRevisionActionById(Long actionId);
    int upsertDataOverride(TTestDataOverride override);
    List<TTestDataOverride> selectDataOverrides(@Param("testId") Long testId,
            @Param("dataType") String dataType);
    int deleteBtsSamplesByIds(@Param("testId") Long testId, @Param("sampleIds") List<Long> sampleIds);
    int deleteBtsMeasurementsBySampleIds(@Param("sampleIds") List<Long> sampleIds);
    int deleteCollectorSamplesByIds(@Param("testId") Long testId, @Param("sampleIds") List<Long> sampleIds);
    Long selectBtsSampleIdByDeviceChannelSequence(@Param("testId") Long testId,
            @Param("deviceId") Long deviceId, @Param("channelNo") Long channelNo,
            @Param("sequenceId") Long sequenceId);
    List<Long> selectBtsSampleIdsByDeviceChannelStep(@Param("testId") Long testId,
            @Param("deviceId") Long deviceId, @Param("channelNo") Long channelNo,
            @Param("cycleNo") Integer cycleNo, @Param("stepNo") Integer stepNo);
    List<Long> selectBtsSampleIdsByDeviceChannelCycle(@Param("testId") Long testId,
            @Param("deviceId") Long deviceId, @Param("channelNo") Long channelNo,
            @Param("cycleNo") Integer cycleNo);
    int insertConsoleImportRecord(TTestConsoleImportRecord record);
    int countTestBatteryNo(@Param("testId") Long testId, @Param("batteryNo") String batteryNo);
    /** ����兼容旧源码的声明；Ticket28 运行链路不再调用。 */
}

 ruoyi-system/src/main/java/com/ruoyi/system/service/ICollectorDisconnectRecoveryService.java

@@ -3,9 +3,10 @@
import java.time.Instant;
import java.util.Set;


/** 维护接收中采集器来源的断线异常与恢复边界。 */
public interface ICollectorDisconnectRecoveryService
{
    /** 根据当前模拟器在线设备编码对账所有接收中来源。 */
    void reconcile(Set<String> onlineCollectorCodes, Instant observedAt);
    /** 根据当前模拟器在线设备编码对账所有接收中来源。 */
}

 ruoyi-system/src/main/java/com/ruoyi/system/service/ITestBackfillService.java

@@ -3,12 +3,12 @@
/** 设备自然结束后的区间历史补采与恢复协调接口。 */
public interface ITestBackfillService
{
    /** 对接收中的测试检查检测柜自然结束并关闭其接收边界。 */
    void reconcileNaturalEnd();
    void resumePendingBackfill();
    void endNaturally(Long testId, String reason);
    /** 对接收中的测试检查检测柜自然结束并关闭其接收边界。 */

    /** 从持久化游标继续所有处于完成确认中的测试补采。 */
    void resumePendingBackfill();

    /** 立即结束指定测试的接收并进入完成确认中，供自然结束协调器使用。 */
    void endNaturally(Long testId, String reason);
}

 ruoyi-system/src/main/java/com/ruoyi/system/service/ITestIntegrityService.java

@@ -3,6 +3,7 @@
import java.util.List;
import com.ruoyi.system.domain.TTestIntegrityIssue;


/** 查询测试区间完整性事实。 */
public interface ITestIntegrityService
{

 ruoyi-system/src/main/java/com/ruoyi/system/service/ITestLiveService.java

@@ -3,7 +3,6 @@
import java.time.Instant;
import java.util.List;
import java.util.Map;
import com.ruoyi.system.domain.TTestIntegrityIssue;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

/** 提供测试控制台的双状态实时事实与鉴权 SSE。 */
@@ -15,17 +14,11 @@
    /** 打开实时 SSE；连接初次发送当前事实，后续只推送最新快照。 */
    SseEmitter stream(Long testId);
    record LiveState(Long testId, String testStatus, String receptionStatus, Instant updatedAt,
            List<LiveSource> sources, List<Map<String, Object>> exceptions,
            List<TTestIntegrityIssue> integrityIssues, TestPlanSnapshot testPlan)
            List<LiveSource> sources, List<Map<String, Object>> exceptions, TestPlanSnapshot testPlan)
    {
        public LiveState(Long testId, String testStatus, String receptionStatus, Instant updatedAt,
                List<LiveSource> sources, List<Map<String, Object>> exceptions)
        { this(testId, testStatus, receptionStatus, updatedAt, sources, exceptions, List.of(), null); }

        public LiveState(Long testId, String testStatus, String receptionStatus, Instant updatedAt,
                List<LiveSource> sources, List<Map<String, Object>> exceptions,
                List<TTestIntegrityIssue> integrityIssues)
        { this(testId, testStatus, receptionStatus, updatedAt, sources, exceptions, integrityIssues, null); }
        { this(testId, testStatus, receptionStatus, updatedAt, sources, exceptions, null); }
    }
    record TestPlanSnapshot(int currentCycleNo, int currentStepNo, int plannedCycleCount,
            List<TestPlanRow> rows) { }

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/CollectorDisconnectRecoveryServiceImpl.java

@@ -13,7 +13,6 @@
import com.ruoyi.system.service.ICollectorDisconnectRecoveryService;

/** 将采集器在线事实对账为可恢复的来源级异常区间。 */
@Service
public class CollectorDisconnectRecoveryServiceImpl implements ICollectorDisconnectRecoveryService
{
    private final TTestMapper mapper;

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java

@@ -24,8 +24,6 @@
import com.ruoyi.device.model.Measurement;
import com.ruoyi.system.domain.TTest;
import com.ruoyi.system.domain.TTestBattery;
import com.ruoyi.system.domain.TTestCollectorExceptionInterval;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ITestLiveService;
import com.ruoyi.system.service.ITestLiveService.TestPlanRow;
import com.ruoyi.system.service.ITestLiveService.TestPlanSnapshot;
@@ -38,13 +36,12 @@
    private final ITTestService tests;
    private final DeviceDiscoveryCandidateStore candidates;
    private final BtsDevelopmentSimulator simulator;
    private final TTestMapper testMapper;
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    private final ConcurrentMap<String, LiveSource> lastSources = new ConcurrentHashMap<>();

    public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates,
            BtsDevelopmentSimulator simulator, TTestMapper testMapper)
    { this.tests = tests; this.candidates = candidates; this.simulator = simulator; this.testMapper = testMapper; }
            BtsDevelopmentSimulator simulator)
    { this.tests = tests; this.candidates = candidates; this.simulator = simulator; }

    @Override public LiveState current(Long testId)
    {
@@ -60,41 +57,16 @@
            if (battery.getCollectorId() != null)
                sources.add(snapshotFor(testId, collectorSource(battery), freezeSnapshots));
        }
        List<TTestCollectorExceptionInterval> persisted = testMapper.selectCollectorExceptionIntervals(testId);
        if (persisted == null) persisted = List.of();
        List<Map<String, Object>> exceptions = new ArrayList<>(persisted.stream().map(interval -> {
            Map<String, Object> item = new LinkedHashMap<>();
            item.put("sourceType", "COLLECTOR");
            item.put("sourceId", interval.getCollectorId());
            item.put("collectorCode", interval.getCollectorCode());
            item.put("collectorName", interval.getCollectorName());
            item.put("channelCode", interval.getCollectorChannelCode());
            item.put("measurementName", interval.getMeasurementName());
            item.put("exceptionType", interval.getExceptionType());
            item.put("status", interval.getIntervalStatus());
            item.put("openedAt", interval.getOpenedAtUtc());
            item.put("missingFrom", interval.getMissingFromUtc());
            item.put("recoveredAt", interval.getRecoveredAtUtc());
            item.put("missingTo", interval.getMissingToUtc());
            item.put("detail", interval.getDetail());
            return item;
        }).toList());
        List<Map<String, Object>> exceptions = new ArrayList<>();
        exceptions.addAll(sources.stream().filter(source -> !"COLLECTOR".equals(source.sourceType())
                && !"ONLINE".equals(source.deviceStatus())
                && !"WORKING".equals(source.deviceStatus()) && !"PAUSED".equals(source.deviceStatus()))
                .map(source -> Map.<String, Object>of("sourceType", source.sourceType(), "sourceId", source.sourceId(),
                        "batteryNo", source.batteryNo(), "status", source.deviceStatus(), "message", "数据来源当前不可观测"))
                .toList());
        boolean abnormal = persisted.stream().anyMatch(item -> "OPEN".equals(item.getIntervalStatus()));
        String testStatus = abnormal && "RUNNING".equals(test.getTestStatus()) ? "RUNNING_ABNORMAL" : test.getTestStatus();
        List<com.ruoyi.system.domain.TTestIntegrityIssue> integrity = new ArrayList<>();
        List<com.ruoyi.system.domain.TTestIntegrityIssue> persistedIntegrity = testMapper.selectIntegrityIssues(testId);
        if (persistedIntegrity != null) integrity.addAll(persistedIntegrity);
        List<com.ruoyi.system.domain.TTestIntegrityIssue> persistedExceptions = testMapper.selectTestExceptionFacts(testId);
        if (persistedExceptions != null) integrity.addAll(persistedExceptions);
        String testStatus = test.getTestStatus();
        TestPlanSnapshot testPlan = detectorTestPlan(test);
        return new LiveState(testId, testStatus, test.getReceptionStatus(), Instant.now(), sources, exceptions,
                integrity == null ? List.of() : integrity, testPlan);
        return new LiveState(testId, testStatus, test.getReceptionStatus(), Instant.now(), sources, exceptions, testPlan);
    }

    private TestPlanSnapshot detectorTestPlan(TTest test)
@@ -169,7 +141,6 @@
        emitter.send(SseEmitter.event().name("reception-status").data(Map.of("testId", testId, "receptionStatus", state.receptionStatus())));
        emitter.send(SseEmitter.event().name("test-status").data(Map.of("testId", testId, "testStatus", state.testStatus())));
        emitter.send(SseEmitter.event().name("exception").data(state.exceptions()));
        emitter.send(SseEmitter.event().name("integrity").data(state.integrityIssues()));
        emitter.send(SseEmitter.event().name("heartbeat").data(Map.of("at", state.updatedAt())));
    }


 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestServiceImpl.java

@@ -183,8 +183,6 @@
                throw new ServiceException("检测柜通道 " + binding.getChannelId() + " 当前离线或不可观测");
            // 只有设备事实为运行中时才标记“运行中接入”；设备暂停属于独立设备状态。
            runtimeAttach |= "WORKING".equalsIgnoreCase(channel.getWorkStatus());
            if (testMapper.selectChannelLeaseForUpdate(binding.getChannelId()) != null)
                throw new ServiceException("检测柜通道 " + binding.getChannelId() + " 已被其他测试占用");
            if (battery.getCollectorId() == null || testMapper.countCollectorEnabled(battery.getCollectorId()) == 0)
                throw new ServiceException("电池号 " + battery.getBatteryNo() + " 的采集器不存在或未启用");
            if (!testMapper.collectorOnline(battery.getCollectorId()))
@@ -197,20 +195,6 @@

        try
        {
            for (TTestBattery battery : current.getBatteries())
            {
                TTestChannelBinding binding = battery.getBinding();
                testMapper.insertChannelLease(binding.getChannelId(), testId, binding.getBindingId(), operatorId, actionAt);
                testMapper.insertReceptionInterval(testId, binding.getBindingId(), "CABINET_CHANNEL",
                        binding.getChannelId(), null, actionAt, "RECEPTION_STARTED");
                for (TTestCollectorChannel mapping : mutable(battery.getCollectorChannels()))
                {
                    testMapper.insertCollectorChannelLease(battery.getCollectorId(), mapping.getChannelCode(), testId,
                            binding.getBindingId(), operatorId, actionAt);
                    testMapper.insertReceptionInterval(testId, binding.getBindingId(), "COLLECTOR_CHANNEL",
                            battery.getCollectorId(), mapping.getChannelCode(), actionAt, "RECEPTION_STARTED");
                }
            }
            if (testMapper.updateReceptionStarted(testId, actionAt, runtimeAttach, updateBy) != 1)
                throw new ServiceException("测试状态已变更，请刷新后重试");
            testMapper.insertTestAudit(testId, "RECEPTION_STARTED", operatorId, updateBy, null,
@@ -238,7 +222,6 @@
        if (testMapper.updateReceptionPaused(testId, updateBy, actionAt) != 1)
            throw new ServiceException("测试状态已变更，请刷新后重试");
        // 暂停边界不属于本次测试的确认接收区间，继续时由新的区间重新开始计时。
        testMapper.closeOpenReceptionIntervals(testId, actionAt, "RECEPTION_PAUSED");
        testMapper.insertTestAudit(testId, "RECEPTION_PAUSED", operatorId, updateBy, null,
                JSON.toJSONString(Map.of("receptionStatus", "PAUSED", "polling", "STOPPED")));
        return 1;
@@ -259,15 +242,6 @@
        Date actionAt = new Date();
        if (testMapper.updateReceptionResumed(testId, updateBy, actionAt) != 1)
            throw new ServiceException("测试状态已变更，请刷新后重试");
        for (TTestBattery battery : current.getBatteries())
        {
            TTestChannelBinding binding = battery.getBinding();
            testMapper.insertReceptionInterval(testId, binding.getBindingId(), "CABINET_CHANNEL",
                    binding.getChannelId(), null, actionAt, "RECEPTION_RESUMED");
            for (TTestCollectorChannel mapping : mutable(battery.getCollectorChannels()))
                testMapper.insertReceptionInterval(testId, binding.getBindingId(), "COLLECTOR_CHANNEL",
                        battery.getCollectorId(), mapping.getChannelCode(), actionAt, "RECEPTION_RESUMED");
        }
        testMapper.insertTestAudit(testId, "RECEPTION_RESUMED", operatorId, updateBy, null,
                JSON.toJSONString(Map.of("receptionStatus", "RECEIVING", "polling", "RESUMED")));
        return 1;
@@ -286,12 +260,9 @@
        Date actionAt = new Date();
        if (testMapper.updateReceptionStopped(testId, updateBy, actionAt) != 1)
            throw new ServiceException("测试状态已变更，请刷新后重试");
        int closedIntervals = testMapper.closeOpenReceptionIntervals(testId, actionAt, "MANUAL_END");
        testMapper.deleteChannelLeasesByTestId(testId);
        testMapper.deleteCollectorChannelLeasesByTestId(testId);
        testMapper.insertTestAudit(testId, "TEST_STOPPED", operatorId, updateBy, null,
                JSON.toJSONString(Map.of("testStatus", "COMPLETED", "receptionStatus", "ENDED",
                        "polling", "STOPPED", "closedIntervals", closedIntervals, "release", "IMMEDIATE")));
                        "polling", "STOPPED", "release", "IMMEDIATE")));
        return 1;
    }

@@ -411,8 +382,6 @@
        test.setBatteryNo(joinBatteryNos(batteries));
        List<TTestWarningThreshold> loadedThresholds = testMapper.selectTestThresholds(testId);
        test.setThresholds(loadedThresholds == null ? mutable(test.getThresholds()) : mutable(loadedThresholds));
        List<com.ruoyi.system.domain.TTestReceptionInterval> intervals = testMapper.selectTestReceptionIntervals(testId);
        test.setReceptionIntervals(intervals == null ? List.of() : intervals);
        List<TTestFieldSnapshot> loadedSnapshots = sharedFieldSnapshotService.selectByTestId(testId);
        test.setFieldSnapshots(loadedSnapshots == null ? mutable(test.getFieldSnapshots()) : loadedSnapshots);
        String mappingJson = test.getLogicStepMappingsJson();

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestBackfillServiceImpl.java

@@ -33,7 +33,6 @@
import com.ruoyi.system.service.ITTestService;

/** 以测试动作边界筛选来源历史，并以数据库游标保证补采可恢复。 */
@Service
public class TestBackfillServiceImpl implements ITestBackfillService
{
    private static final int PAGE_SIZE = 1000;
@@ -195,14 +194,14 @@
        if (testScoped != null)
        {
            if (!testScoped.equalsIgnoreCase(HexFormat.of().formatHex(hash)))
                mapper.insertBackfillConflict(test.getTestId(), binding.getChannelId(),
                mapper.insertSampleConflict(test.getTestId(), binding.getChannelId(),
                        "测试内采样内容冲突 seqid=" + sample.sequenceId());
            return;
        }
        String old = mapper.selectSampleContentHash(binding.getChannelId(), btsTestId, sample.sequenceId());
        if (old != null && !old.equalsIgnoreCase(HexFormat.of().formatHex(hash)))
        {
            mapper.insertBackfillConflict(test.getTestId(), binding.getChannelId(),
            mapper.insertSampleConflict(test.getTestId(), binding.getChannelId(),
                    "BTS testid=" + btsTestId + ", seqid=" + sample.sequenceId());
            return;
        }
@@ -255,7 +254,7 @@
                record.sequenceId(), record.metricCode());
        if (old != null && !old.equalsIgnoreCase(HexFormat.of().formatHex(hash)))
        {
            mapper.insertBackfillConflict(test.getTestId(), null,
            mapper.insertSampleConflict(test.getTestId(), null,
                    "采集器=" + record.collectorCode() + ", seqid=" + record.sequenceId());
            return;
        }

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java

@@ -6,8 +6,6 @@
import java.util.Date;
import java.util.HexFormat;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
@@ -15,7 +13,6 @@
import com.ruoyi.system.domain.TTest;
import com.ruoyi.system.domain.TTestBattery;
import com.ruoyi.system.domain.TTestChannelBinding;
import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ITTestService;
import com.ruoyi.system.service.ITestDataReceptionService;
@@ -43,6 +40,9 @@
        if (testIds == null) return;
        for (Long testId : testIds)
        {
            // 锁住测试行直到本轮采样写入完成，暂停/停止动作只能在本轮接收提交后生效。
            TTest locked = mapper.selectTestByIdForUpdate(testId);
            if (locked == null || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
            TTest test = tests.selectTestById(testId);
            if (test == null || !"RECEIVING".equals(test.getReceptionStatus())) continue;
            captureTest(test);
@@ -55,45 +55,43 @@
        {
            TTestChannelBinding binding = battery.getBinding();
            if (binding == null || binding.getDevid() == null || binding.getChlid() == null) continue;
            Optional<TTestReceptionInterval> open = mapper.selectTestReceptionIntervals(test.getTestId()).stream()
                    .filter(item -> Objects.equals(item.getBindingId(), binding.getBindingId())
                            && "CABINET_CHANNEL".equals(item.getSourceType())
                            && Objects.equals(item.getSourceId(), binding.getChannelId())
                            && "OPEN".equals(item.getIntervalStatus()))
                    .findFirst();
            if (open.isEmpty()) continue;
            Date start = open.get().getStartBoundaryAtUtc();
            List<BtsHistoryRecord> records;
            try { records = simulator.btsRecords(binding.getDevid(), binding.getChlid()); }
            catch (IllegalArgumentException ignored) { continue; }
            for (BtsHistoryRecord record : records)
            {
                if (start != null && record.deviceTime().isBefore(start.toInstant())) continue;
                save(test, battery, binding, open.get(), record);
                save(test, binding, record);
            }
        }
    }

    private void save(TTest test, TTestBattery battery, TTestChannelBinding binding,
            TTestReceptionInterval interval, BtsHistoryRecord record)
    private void save(TTest test, TTestChannelBinding binding,
            BtsHistoryRecord record)
    {
        // 模拟器历史 testid 按通道固定；接收事实必须再带测试作用域，避免顺序测试共用全局幂等键。
        String btsTestId = "SIM-" + binding.getDevid() + "-" + binding.getChlid() + "-T" + test.getTestId();
        byte[] hash = hash(record);
        String old = mapper.selectSampleContentHash(binding.getChannelId(), btsTestId, record.sequenceId());
        String old = mapper.selectBtsSampleContentHash(binding.getChannelId(), btsTestId, record.sequenceId());
        if (old != null)
        {
            if (!old.equalsIgnoreCase(HexFormat.of().formatHex(hash)))
                mapper.insertBackfillConflict(test.getTestId(), binding.getChannelId(),
                mapper.insertSampleConflict(test.getTestId(), binding.getChannelId(),
                        "实时接收内容冲突 testid=" + btsTestId + ", seqid=" + record.sequenceId());
            return;
        }
        mapper.insertBtsSample(test.getTestId(), binding.getBindingId(), binding.getChannelId(), interval.getIntervalId(),
        int inserted = mapper.insertBtsSample(test.getTestId(), binding.getBindingId(), binding.getChannelId(), null,
                btsTestId, record.sequenceId(), Date.from(record.deviceTime()),
                record.deviceTime().toString(), new Date(), record.cycleNo(), record.stepNo(), record.stepType(),
                record.stepName(), null,
                BigDecimal.valueOf(record.voltageV()), BigDecimal.valueOf(record.currentA()),
                BigDecimal.valueOf(record.capacityAh()), BigDecimal.valueOf(record.energyWh()), hash);
        if (inserted == 0)
        {
            String actual = mapper.selectBtsSampleContentHash(binding.getChannelId(), btsTestId, record.sequenceId());
            if (actual != null && !actual.equalsIgnoreCase(HexFormat.of().formatHex(hash)))
                mapper.insertSampleConflict(test.getTestId(), binding.getChannelId(),
                        "实时接收并发内容冲突 testid=" + btsTestId + ", seqid=" + record.sequenceId());
        }
    }

    private byte[] hash(BtsHistoryRecord record)

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestIntegrityServiceImpl.java

@@ -9,7 +9,6 @@
import com.ruoyi.system.service.ITestIntegrityService;

/** 完整性查询只返回持久化问题事实，不把实时快照当作历史结果。 */
@Service
public class TestIntegrityServiceImpl implements ITestIntegrityService
{
    private final TTestMapper mapper;

 ruoyi-system/src/main/resources/mapper/system/TTestMapper.xml

@@ -19,9 +19,6 @@
        <result property="startedAtUtc" column="started_at_utc" />
        <result property="endedAtUtc" column="ended_at_utc" />
        <result property="runtimeAttach" column="runtime_attach" />
        <result property="backfillStatus" column="backfill_status" />
        <result property="backfillStartedAtUtc" column="backfill_started_at_utc" />
        <result property="backfillCompletedAtUtc" column="backfill_completed_at_utc" />
        <result property="delFlag" column="del_flag" />
        <result property="createBy" column="create_by" />
        <result property="createTime" column="create_time" />
@@ -114,54 +111,18 @@
        <result property="exceptionInfo" column="exception_info" />
        <result property="delFlag" column="del_flag" />
    </resultMap>
    <resultMap id="TTestCollectorExceptionIntervalResultMap" type="TTestCollectorExceptionInterval">
        <id property="intervalId" column="interval_id" />
        <result property="testId" column="test_id" />
        <result property="batteryRelationId" column="battery_relation_id" />
        <result property="collectorId" column="collector_id" />
        <result property="collectorCode" column="collector_code" />
        <result property="collectorName" column="collector_name" />
        <result property="collectorChannelCode" column="collector_channel_code" />
        <result property="measurementName" column="measurement_name" />
        <result property="exceptionType" column="exception_type" />
        <result property="openedAtUtc" column="opened_at_utc" />
        <result property="missingFromUtc" column="missing_from_utc" />
        <result property="recoveredAtUtc" column="recovered_at_utc" />
        <result property="missingToUtc" column="missing_to_utc" />
        <result property="intervalStatus" column="interval_status" />
        <result property="detail" column="detail" />
    </resultMap>
    <resultMap id="TTestReceptionIntervalResultMap" type="TTestReceptionInterval">
        <id property="intervalId" column="interval_id" />
        <result property="testId" column="test_id" />
        <result property="bindingId" column="binding_id" />
        <result property="sourceType" column="source_type" />
        <result property="sourceId" column="source_id" />
        <result property="sourceChannelCode" column="source_channel_code" />
        <result property="actionAtUtc" column="action_at_utc" />
        <result property="startBoundaryAtUtc" column="start_boundary_at_utc" />
        <result property="endBoundaryAtUtc" column="end_boundary_at_utc" />
        <result property="boundaryNote" column="boundary_note" />
        <result property="intervalStatus" column="interval_status" />
    </resultMap>

    <sql id="selectTestBase">
        select t.test_id, t.test_name, t.creator_id, creator.nick_name creator_name,
               t.principal_id, principal.nick_name principal_name,
               t.test_type_id, ty.test_type_name, t.test_purpose,
               case when t.test_status = 'RUNNING'
                         and exists (select 1 from t_test_collector_exception_interval exception_fact
                                     where exception_fact.test_id = t.test_id
                                       and exception_fact.interval_status = 'OPEN')
                    then 'RUNNING_ABNORMAL'
                    else t.test_status end test_status,
               t.test_status,
               t.reception_status, t.monitor_timeout_minutes,
               t.started_at_utc, t.ended_at_utc, t.del_flag,
               t.create_by, t.create_time, t.update_by, t.update_time, t.remark,
               cast(t.logic_step_mappings_json as char) logic_step_mappings_json,
               (select count(1) from t_test_battery b where b.test_id = t.test_id and b.del_flag = 0) battery_count,
               t.runtime_attach runtime_attach, t.backfill_status, t.backfill_started_at_utc,
               t.backfill_completed_at_utc
               t.runtime_attach runtime_attach
        from t_test t
        left join sys_user creator on creator.user_id = t.creator_id and creator.del_flag = '0'
        left join sys_user principal on principal.user_id = t.principal_id and principal.del_flag = '0'
@@ -178,12 +139,7 @@
            <if test="batteryNo != null and batteryNo != ''">and exists (select 1 from t_test_battery fb where fb.test_id = t.test_id and fb.battery_no = #{batteryNo} and fb.del_flag = 0)</if>
            <if test="testStatus != null and testStatus != ''">
                and (
                    (#{testStatus} = 'RUNNING_ABNORMAL' and t.test_status = 'RUNNING'
                        and exists (select 1 from t_test_collector_exception_interval exception_fact
                                    where exception_fact.test_id = t.test_id and exception_fact.interval_status = 'OPEN'))
                    or (#{testStatus} = 'RUNNING' and t.test_status = 'RUNNING'
                        and not exists (select 1 from t_test_collector_exception_interval exception_fact
                                        where exception_fact.test_id = t.test_id and exception_fact.interval_status = 'OPEN'))
                    (#{testStatus} = 'RUNNING' and t.test_status = 'RUNNING')
                    or (#{testStatus} = 'COMPLETED' and t.test_status in ('COMPLETED', 'COMPLETION_PENDING', 'TERMINATED'))
                    or (#{testStatus} not in ('RUNNING', 'RUNNING_ABNORMAL', 'COMPLETED') and t.test_status = #{testStatus})
                )
@@ -359,13 +315,6 @@
    <select id="selectTestThresholds" parameterType="Long" resultMap="TTestThresholdResultMap">
        select threshold_id, test_id, metric_code, min_value, max_value, unit, severity, del_flag
        from t_test_warning_threshold where test_id = #{testId} and del_flag = 0 order by threshold_id
    </select>
    <select id="selectTestReceptionIntervals" parameterType="Long" resultMap="TTestReceptionIntervalResultMap">
        select interval_id, test_id, binding_id, source_type, source_id, source_channel_code,
               action_at_utc, start_boundary_at_utc, end_boundary_at_utc, boundary_note, interval_status
        from t_test_reception_interval
        where test_id = #{testId}
        order by source_type, source_id, source_channel_code, start_boundary_at_utc, interval_id
    </select>
    <select id="selectReceivingTestIds" resultType="long">
        select test_id from t_test where del_flag = 0 and reception_status = 'RECEIVING' order by test_id
@@ -547,90 +496,21 @@
        for update
    </select>

    <select id="selectTestsForNaturalEnd" resultMap="TTestResultMap">
        <include refid="selectTestBase" />
        where t.del_flag = 0 and t.reception_status in ('RECEIVING', 'PAUSED')
        order by t.test_id
    </select>
    <select id="selectTestsForBackfill" resultMap="TTestResultMap">
        <include refid="selectTestBase" />
        where t.del_flag = 0 and t.reception_status = 'ENDED'
          and t.test_status in ('COMPLETION_PENDING', 'COMPLETED')
          and coalesce(t.backfill_status, 'NOT_STARTED') &lt;&gt; 'COMPLETED'
        order by t.ended_at_utc, t.test_id
    </select>
    <resultMap id="TTestBackfillCursorResultMap" type="com.ruoyi.system.domain.TTestBackfillCursor">
        <id property="cursorId" column="cursor_id" />
        <result property="testId" column="test_id" />
        <result property="bindingId" column="binding_id" />
        <result property="sourceType" column="source_type" />
        <result property="sourceId" column="source_id" />
        <result property="sourceChannelCode" column="source_channel_code" />
        <result property="protocolTestId" column="protocol_test_id" />
        <result property="nextPosition" column="next_position" />
        <result property="sourceStatus" column="source_status" />
        <result property="protocolMetadata" column="protocol_metadata" />
        <result property="lastError" column="last_error" />
        <result property="updatedAtUtc" column="updated_at_utc" />
    </resultMap>
    <select id="selectBackfillCursorForUpdate" resultMap="TTestBackfillCursorResultMap">
        select * from t_test_backfill_cursor
        where test_id = #{testId} and binding_id = #{bindingId}
          and source_type = #{sourceType} and source_id = #{sourceId}
          and (source_channel_code = #{sourceChannelCode} or (source_channel_code is null and #{sourceChannelCode} is null))
        for update
    </select>
    <insert id="insertBackfillCursor" parameterType="com.ruoyi.system.domain.TTestBackfillCursor" useGeneratedKeys="true" keyProperty="cursorId">
        insert into t_test_backfill_cursor
        (test_id, binding_id, source_type, source_id, source_channel_code, protocol_test_id,
         next_position, source_status, protocol_metadata, last_error, updated_at_utc)
        values (#{testId}, #{bindingId}, #{sourceType}, #{sourceId}, #{sourceChannelCode}, #{protocolTestId},
                #{nextPosition}, #{sourceStatus}, cast(#{protocolMetadata} as json), #{lastError}, #{updatedAtUtc})
    </insert>
    <update id="updateBackfillCursor" parameterType="com.ruoyi.system.domain.TTestBackfillCursor">
        update t_test_backfill_cursor set protocol_test_id = #{protocolTestId}, next_position = #{nextPosition},
            source_status = #{sourceStatus}, protocol_metadata = cast(#{protocolMetadata} as json),
            last_error = #{lastError}, updated_at_utc = #{updatedAtUtc}
        where cursor_id = #{cursorId}
    </update>
    <update id="updateTestBackfillStarted">
        update t_test set test_status = 'COMPLETION_PENDING', backfill_status = 'RUNNING',
            backfill_started_at_utc = coalesce(backfill_started_at_utc, #{updatedAtUtc}),
            update_time = #{updatedAtUtc} where test_id = #{testId} and reception_status = 'ENDED'
    </update>
    <update id="updateTestBackfillWaiting">
        update t_test set backfill_status = 'WAITING', update_time = #{updatedAtUtc}, remark = #{error}
        where test_id = #{testId} and reception_status = 'ENDED'
    </update>
    <update id="updateTestBackfillCompleted">
        update t_test set backfill_status = 'COMPLETED', backfill_completed_at_utc = #{updatedAtUtc},
            update_time = #{updatedAtUtc} where test_id = #{testId} and reception_status = 'ENDED'
    </update>
    <update id="updateTestCompleted">
        update t_test set test_status = 'COMPLETED', backfill_status = 'COMPLETED',
            backfill_completed_at_utc = #{updatedAtUtc}, update_time = #{updatedAtUtc}
        update t_test set test_status = 'COMPLETED', update_time = #{updatedAtUtc}
        where test_id = #{testId} and reception_status = 'ENDED' and test_status = 'COMPLETION_PENDING'
    </update>
    <update id="updateTestNaturallyEnded">
        update t_test set test_status = 'COMPLETION_PENDING', reception_status = 'ENDED',
            ended_at_utc = #{endedAtUtc}, backfill_status = 'NOT_STARTED', update_time = #{endedAtUtc}
        where test_id = #{testId} and reception_status in ('RECEIVING', 'PAUSED')
    </update>
    <select id="selectSampleContentHash" resultType="string">
    <select id="selectBtsSampleContentHash" resultType="string">
        select hex(content_sha256) from t_test_sample
        where channel_id = #{channelId} and bts_test_id = #{btsTestId} and seq_id = #{sequenceId}
    </select>
    <select id="selectTestSampleContentHash" resultType="string">
        select hex(content_sha256) from t_test_sample
        where test_id = #{testId} and channel_id = #{channelId} and seq_id = #{sequenceId}
        order by sample_id limit 1
    </select>
    <insert id="insertBtsSample">
        insert ignore into t_test_sample
        (test_id, binding_id, channel_id, reception_interval_id, bts_test_id, seq_id, device_time_utc,
        (test_id, binding_id, channel_id, bts_test_id, seq_id, device_time_utc,
         device_time_raw, received_at_utc, cycle_no, step_no, step_type, step_name, test_time_ms, voltage_v,
         current_a, capacity_ah, energy_wh, content_sha256)
        values (#{testId}, #{bindingId}, #{channelId}, #{intervalId}, #{btsTestId}, #{sequenceId}, #{deviceTimeUtc},
        values (#{testId}, #{bindingId}, #{channelId}, #{btsTestId}, #{sequenceId}, #{deviceTimeUtc},
                #{deviceTimeRaw}, #{receivedAtUtc}, #{cycleNo}, #{stepNo}, #{stepType}, #{stepName}, #{testTimeMs}, #{voltageV},
                #{currentA}, #{capacityAh}, #{energyWh}, #{contentSha256})
    </insert>
@@ -645,88 +525,11 @@
    </select>
    <insert id="insertCollectorSample">
        insert ignore into t_test_collector_sample
        (test_id, binding_id, reception_interval_id, collector_id, collector_channel_code, sequence_id, device_time_utc,
        (test_id, binding_id, collector_id, collector_channel_code, sequence_id, device_time_utc,
         device_time_raw, server_received_at_utc, metric_code, metric_value, unit, content_sha256)
        values (#{testId}, #{bindingId}, #{intervalId}, #{collectorId}, #{channelCode}, #{sequenceId}, #{deviceTimeUtc},
        values (#{testId}, #{bindingId}, #{collectorId}, #{channelCode}, #{sequenceId}, #{deviceTimeUtc},
                #{deviceTimeRaw}, #{receivedAtUtc}, #{metricCode}, #{metricValue}, #{unit}, #{contentSha256})
    </insert>
    <resultMap id="TTestIntegrityIssueResultMap" type="com.ruoyi.system.domain.TTestIntegrityIssue">
        <id property="issueId" column="issue_id"/><result property="testId" column="test_id"/>
        <result property="intervalId" column="interval_id"/><result property="bindingId" column="binding_id"/>
        <result property="sourceType" column="source_type"/><result property="sourceId" column="source_id"/>
        <result property="sourceChannelCode" column="source_channel_code"/><result property="issueType" column="issue_type"/>
        <result property="issueStatus" column="issue_status"/><result property="missingFromSequence" column="missing_from_sequence"/>
        <result property="missingToSequence" column="missing_to_sequence"/><result property="retryCount" column="retry_count"/>
        <result property="detail" column="detail"/><result property="firstDetectedAtUtc" column="first_detected_at_utc"/>
        <result property="lastDetectedAtUtc" column="last_detected_at_utc"/><result property="resolvedAtUtc" column="resolved_at_utc"/>
    </resultMap>
    <select id="selectIntegrityIssues" resultMap="TTestIntegrityIssueResultMap">
        select * from t_test_integrity_issue where test_id = #{testId}
        order by issue_status, last_detected_at_utc desc, issue_id desc
    </select>
    <select id="selectTestExceptionFacts" resultMap="TTestIntegrityIssueResultMap">
        select exception_id issue_id, test_id, null interval_id, null binding_id,
               case when channel_id is null then 'UNKNOWN_SOURCE' else 'CABINET_CHANNEL' end source_type,
               channel_id source_id, null source_channel_code, exception_type issue_type,
               exception_status issue_status, null missing_from_sequence, null missing_to_sequence,
               occurrence_count retry_count, detail, first_occurred_at_utc first_detected_at_utc,
               last_occurred_at_utc last_detected_at_utc, handled_at_utc resolved_at_utc
        from t_test_exception where test_id = #{testId}
        order by exception_status, last_occurred_at_utc desc, exception_id desc
    </select>
    <select id="selectBtsSequencesByInterval" resultType="long">
        select seq_id from t_test_sample where reception_interval_id = #{intervalId} order by seq_id
    </select>
    <select id="selectCollectorSequencesByInterval" resultType="long">
        select distinct sequence_id from t_test_collector_sample where reception_interval_id = #{intervalId} order by sequence_id
    </select>
    <select id="selectOpenIntegrityIssue" resultMap="TTestIntegrityIssueResultMap">
        select * from t_test_integrity_issue where interval_id = #{intervalId} and issue_status = 'OPEN'
          and source_type = #{sourceType} and source_id = #{sourceId}
          and (source_channel_code = #{sourceChannelCode} or (source_channel_code is null and #{sourceChannelCode} is null))
          and (missing_from_sequence = #{missingFrom} or (missing_from_sequence is null and #{missingFrom} is null))
          and (missing_to_sequence = #{missingTo} or (missing_to_sequence is null and #{missingTo} is null)) limit 1
    </select>
    <select id="selectOpenIntegrityIssues" resultMap="TTestIntegrityIssueResultMap">
        select * from t_test_integrity_issue where interval_id = #{intervalId} and issue_status = 'OPEN'
          and source_type = #{sourceType} and source_id = #{sourceId}
          and (source_channel_code = #{sourceChannelCode} or (source_channel_code is null and #{sourceChannelCode} is null))
        order by issue_id
    </select>
    <insert id="insertIntegrityIssue" parameterType="com.ruoyi.system.domain.TTestIntegrityIssue" useGeneratedKeys="true" keyProperty="issueId">
        insert into t_test_integrity_issue
        (test_id, interval_id, binding_id, source_type, source_id, source_channel_code, issue_type, issue_status,
         missing_from_sequence, missing_to_sequence, retry_count, detail, first_detected_at_utc, last_detected_at_utc)
        values (#{testId}, #{intervalId}, #{bindingId}, #{sourceType}, #{sourceId}, #{sourceChannelCode}, #{issueType}, 'OPEN',
                #{missingFromSequence}, #{missingToSequence}, #{retryCount}, #{detail}, #{firstDetectedAtUtc}, #{lastDetectedAtUtc})
    </insert>
    <update id="touchIntegrityIssue">
        update t_test_integrity_issue set retry_count = #{retryCount}, detail = #{detail}, last_detected_at_utc = #{detectedAtUtc}
        where issue_id = #{issueId} and issue_status = 'OPEN'
    </update>
    <update id="rewindBackfillCursor">
        update t_test_backfill_cursor set next_position = least(next_position, #{nextPosition}),
            source_status = 'PENDING', last_error = '区间序号缺口待重试', updated_at_utc = utc_timestamp(3)
        where test_id = #{testId} and binding_id = #{bindingId} and source_type = #{sourceType}
          and source_id = #{sourceId}
          and (source_channel_code = #{sourceChannelCode} or (source_channel_code is null and #{sourceChannelCode} is null))
    </update>
    <update id="resolveIntegrityIssues">
        update t_test_integrity_issue set issue_status = 'RESOLVED', resolved_at_utc = #{resolvedAtUtc}
        where test_id = #{testId} and interval_id = #{intervalId} and issue_status = 'OPEN'
          and source_type = #{sourceType} and source_id = #{sourceId}
          and (source_channel_code = #{sourceChannelCode} or (source_channel_code is null and #{sourceChannelCode} is null))
    </update>
    <update id="resolveIntegrityIssue">
        update t_test_integrity_issue set issue_status = 'RESOLVED', resolved_at_utc = #{resolvedAtUtc}
        where issue_id = #{issueId} and issue_status = 'OPEN'
    </update>
    <select id="countOpenIntegrityIssues" resultType="int">
        select count(1) from t_test_integrity_issue where test_id = #{testId} and issue_status = 'OPEN'
    </select>
    <select id="countOpenExceptions" resultType="int">
        select count(1) from t_test_exception where test_id = #{testId} and exception_status = 'OPEN'
    </select>
    <select id="selectCollectorSampleId" resultType="long">
        select sample_id from t_test_collector_sample
        where collector_id = #{collectorId} and collector_channel_code = #{channelCode}
@@ -737,50 +540,15 @@
        where collector_id = #{collectorId} and collector_channel_code = #{channelCode}
          and sequence_id = #{sequenceId} and metric_code = #{metricCode}
    </select>
    <insert id="insertBackfillConflict">
    <insert id="insertSampleConflict">
        insert into t_test_exception
        (test_id, channel_id, exception_type, exception_status, severity,
         first_occurred_at_utc, last_occurred_at_utc, occurrence_count, detail)
        values (#{testId}, #{channelId}, 'HISTORY_CONTENT_CONFLICT', 'OPEN', 'WARNING', utc_timestamp(3),
        values (#{testId}, #{channelId}, 'SAMPLE_CONTENT_CONFLICT', 'OPEN', 'WARNING', utc_timestamp(3),
                utc_timestamp(3), 1, #{detail})
        on duplicate key update last_occurred_at_utc = utc_timestamp(3), occurrence_count = occurrence_count + 1,
            detail = values(detail)
    </insert>
    <select id="selectChannelLeaseForUpdate" parameterType="Long" resultMap="TTestBindingResultMap">
        select binding.binding_id, binding.test_id, binding.battery_relation_id, binding.channel_id,
               binding.claim_status
        from t_channel_lease lease
        join t_test_channel_binding binding on binding.binding_id = lease.binding_id
        where lease.channel_id = #{channelId}
        for update
    </select>
    <insert id="insertChannelLease">
        insert into t_channel_lease(channel_id, test_id, binding_id, acquired_by, acquired_at_utc)
        values(#{channelId}, #{testId}, #{bindingId}, #{acquiredBy}, #{acquiredAtUtc})
    </insert>
    <insert id="insertCollectorChannelLease">
        insert into t_collector_channel_lease(collector_id, channel_code, test_id, binding_id, acquired_by, acquired_at_utc)
        values(#{collectorId}, #{channelCode}, #{testId}, #{bindingId}, #{acquiredBy}, #{acquiredAtUtc})
    </insert>
    <insert id="insertReceptionInterval">
        insert into t_test_reception_interval
        (test_id, binding_id, source_type, source_id, source_channel_code,
         action_at_utc, start_boundary_at_utc, boundary_note, interval_status)
        values(#{testId}, #{bindingId}, #{sourceType}, #{sourceId}, #{sourceChannelCode},
               #{actionAtUtc}, #{actionAtUtc}, #{boundaryNote}, 'OPEN')
    </insert>
    <update id="closeOpenReceptionIntervals">
        update t_test_reception_interval
        set end_boundary_at_utc = #{endBoundaryAtUtc},
            boundary_note = #{boundaryNote}, interval_status = 'CLOSED'
        where test_id = #{testId} and interval_status = 'OPEN'
    </update>
    <delete id="deleteChannelLeasesByTestId">
        delete from t_channel_lease where test_id = #{testId}
    </delete>
    <delete id="deleteCollectorChannelLeasesByTestId">
        delete from t_collector_channel_lease where test_id = #{testId}
    </delete>
    <select id="hasReceivingTestForAdapter" resultType="int">
        select count(1)
        from t_test_channel_binding binding
@@ -788,15 +556,6 @@
        join t_detection_channel channel_fact on channel_fact.channel_id = binding.channel_id
        where channel_fact.adapter_code = #{adapterCode}
          and test_fact.reception_status = 'RECEIVING'
          and test_fact.del_flag = 0 and channel_fact.del_flag = 0
    </select>
    <select id="hasPausedOrEndedTestForAdapter" resultType="int">
        select count(1)
        from t_test_channel_binding binding
        join t_test test_fact on test_fact.test_id = binding.test_id
        join t_detection_channel channel_fact on channel_fact.channel_id = binding.channel_id
        where channel_fact.adapter_code = #{adapterCode}
          and test_fact.reception_status in ('PAUSED', 'ENDED')
          and test_fact.del_flag = 0 and channel_fact.del_flag = 0
    </select>
    <select id="selectReceivingChannelsForAdapter" resultType="com.ruoyi.system.domain.TDetectionChannel">
@@ -838,54 +597,10 @@
    <update id="updateReceptionStopped">
        update t_test
        set test_status = 'COMPLETED', reception_status = 'ENDED', ended_at_utc = #{endedAtUtc},
            backfill_status = 'COMPLETED',
            update_by = #{updateBy}, update_time = #{endedAtUtc}
        where test_id = #{testId} and reception_status in ('RECEIVING', 'PAUSED') and del_flag = 0
    </update>

    <select id="selectReceivingCollectorSources" resultType="com.ruoyi.system.domain.TTestCollectorSource">
        select test_fact.test_id testId, battery.relation_id batteryRelationId, battery.battery_no batteryNo,
               collector.collector_id collectorId, collector.collector_code collectorCode,
               mapping.channel_code channelCode
        from t_test test_fact
        join t_test_battery battery on battery.test_id = test_fact.test_id and battery.del_flag = 0
        join t_collector collector on collector.collector_id = battery.collector_id
            and collector.del_flag = 0
        join t_test_collector_channel mapping on mapping.battery_relation_id = battery.relation_id
            and mapping.test_id = test_fact.test_id
        where test_fact.reception_status = 'RECEIVING' and test_fact.del_flag = 0
        order by test_fact.test_id, battery.relation_id, mapping.mapping_id
    </select>
    <select id="selectOpenCollectorException" resultMap="TTestCollectorExceptionIntervalResultMap">
        select * from t_test_collector_exception_interval
        where test_id = #{testId} and battery_relation_id = #{batteryRelationId}
          and collector_id = #{collectorId} and collector_channel_code = #{channelCode}
          and exception_type = 'COLLECTOR_DISCONNECTED' and interval_status = 'OPEN'
        order by interval_id desc limit 1
    </select>
    <insert id="insertCollectorExceptionInterval" parameterType="TTestCollectorExceptionInterval" useGeneratedKeys="true" keyProperty="intervalId">
        insert into t_test_collector_exception_interval
        (test_id, battery_relation_id, collector_id, collector_channel_code, exception_type,
         opened_at_utc, missing_from_utc, detail)
        values (#{testId}, #{batteryRelationId}, #{collectorId}, #{collectorChannelCode},
                #{exceptionType}, #{openedAtUtc}, #{missingFromUtc}, #{detail})
    </insert>
    <update id="closeCollectorExceptionInterval">
        update t_test_collector_exception_interval
        set recovered_at_utc = #{recoveredAtUtc}, missing_to_utc = #{missingToUtc}, interval_status = 'CLOSED'
        where interval_id = #{intervalId} and interval_status = 'OPEN'
    </update>
    <select id="selectCollectorExceptionIntervals" parameterType="Long" resultMap="TTestCollectorExceptionIntervalResultMap">
        select exception_fact.*, collector.collector_code, collector.collector_name,
               mapping.measurement_name
        from t_test_collector_exception_interval exception_fact
        left join t_collector collector on collector.collector_id = exception_fact.collector_id
        left join t_test_collector_channel mapping on mapping.battery_relation_id = exception_fact.battery_relation_id
            and mapping.collector_id = exception_fact.collector_id
            and mapping.channel_code = exception_fact.collector_channel_code
        where exception_fact.test_id = #{testId} and exception_fact.interval_status = 'OPEN'
        order by exception_fact.opened_at_utc, exception_fact.interval_id
    </select>
    <resultMap id="TTestDataOverrideResultMap" type="com.ruoyi.system.domain.TTestDataOverride">
        <id property="overrideId" column="override_id"/><result property="testId" column="test_id"/>
        <result property="dataType" column="data_type"/><result property="stableIdentity" column="stable_identity"/>