收敛测试暂停继续与停止生命周期

yeg

2026-08-30 1522564d7ddf949131641f220766a25d8d981751

收敛测试暂停继续与停止生命周期

暂停和继续仅控制系统轮询，停止测试进入已终止并释放占用；按审计动作时间过滤结果，移除接收区间业务模型并补齐前后端测试。

 AGENTS.md

@@ -20,9 +20,9 @@

## 业务事务

- 开始接收在一个事务内校验全部检测柜和采集器在线可观测、必填映射完整及区间无冲突，并为各数据来源打开接收区间；任一条件失败则整体回滚。数据库约束或等价事务机制必须阻止并发区间重叠。
- 检测柜通道允许运行中接入，人工结束接收后立即释放业务占用；同一次设备运行可依次分配给多个测试，但实际接收边界不得重叠。
- 全部接收区间关闭后先进入内部“完成确认中”；只有所有来源补采完成且区间内没有未解释缺口、冲突或异常，才聚合为“检测完成”。
- 开始接收在一个事务内校验全部检测柜和采集器在线可观测及必填映射完整；任一条件失败则整体回滚。暂停接收停止该测试的数据请求，继续接收恢复请求，停止测试进入不可恢复终态并释放业务占用。
- 检测柜通道允许运行中接入；同一次设备运行可依次服务多个测试，数据归属按测试级动作时间判断，不建立来源接收区间约束。
- 停止测试或设备自然结束后进入内部“完成确认中”；只有所有来源补采完成且接收时段内没有未解释缺口、冲突或异常，才聚合为“检测完成”。
- “检测完成”和“已终止”是不可恢复终态。异常恢复、数据缺口处置、终止和负责人转交都追加审计。
- 相同“测试 + 通道 + 异常类型”在未恢复期间聚合首次时间、最近时间和次数；恢复后关闭，再次发生创建新事件。

@@ -36,7 +36,7 @@

## API 与实时流

- 普通提交和历史区间查询使用 REST；实时快照、状态变化和异常使用 `SseEmitter`。
- 普通提交和历史数据查询使用 REST；实时快照、状态变化和异常使用 `SseEmitter`。
- SSE 只保留每测试每数据来源最新快照。设备状态、接收状态、测试状态和异常变化不可合并；发送失败关闭连接。重连先发当前事实，历史由 REST 查询。
- 权限由后端统一执行。创建人、负责人和管理员是独立字段和角色；不要依赖前端隐藏按钮实现授权。
- API 时间输出 ISO 8601 带时区，测量值同时返回明确单位。

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

@@ -26,7 +26,7 @@
    public void discover()
    {
        if (!registry.contains(BtsDevelopmentSimulator.ADAPTER_CODE)) return;
        persistence.discover(registry, new DiscoverySource(BtsDevelopmentSimulator.ADAPTER_CODE, null,
        persistence.discoverForReception(registry, new DiscoverySource(BtsDevelopmentSimulator.ADAPTER_CODE, null,
                BtsDevelopmentSimulator.ADAPTER_NAME, BtsDevelopmentSimulator.POLL_INTERVAL_MS));
    }
}

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

@@ -20,7 +20,6 @@
import com.ruoyi.common.enums.BusinessType;
import com.ruoyi.system.domain.TTest;
import com.ruoyi.system.domain.TTestResult;
import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.service.ITTestService;

/** 测试管理 Controller。 */
@@ -96,7 +95,7 @@
        return success(testService.selectTestById(testId));
    }

    /** 独立暂停接收动作；只关闭系统接收区间，不向设备发送控制命令。 */
    /** 独立暂停接收动作；停止该测试数据请求，不向设备发送控制命令。 */
    @PreAuthorize("@ss.hasPermi('system:test:pause')")
    @Log(title = "测试数据接收", businessType = BusinessType.UPDATE)
    @PostMapping("/{testId}/reception/pause")
@@ -106,7 +105,7 @@
        return success(testService.selectTestById(testId));
    }

    /** 独立继续接收动作；从新的成功观测边界打开区间，不向设备发送控制命令。 */
    /** 独立继续接收动作；恢复该测试数据请求，不向设备发送控制命令。 */
    @PreAuthorize("@ss.hasPermi('system:test:resume')")
    @Log(title = "测试数据接收", businessType = BusinessType.UPDATE)
    @PostMapping("/{testId}/reception/resume")
@@ -116,13 +115,14 @@
        return success(testService.selectTestById(testId));
    }

    /** 返回测试各来源的接收区间、动作时间和实际采样边界。 */
    @PreAuthorize("@ss.hasPermi('system:test:query')")
    @GetMapping("/{testId}/reception/intervals")
    public AjaxResult receptionIntervals(@PathVariable Long testId)
    /** 独立停止测试动作；进入不可恢复终态且不向设备发送控制命令。 */
    @PreAuthorize("@ss.hasPermi('system:test:stop')")
    @Log(title = "停止测试", businessType = BusinessType.UPDATE)
    @PostMapping("/{testId}/stop")
    public AjaxResult stopTest(@PathVariable Long testId)
    {
        List<TTestReceptionInterval> intervals = testService.selectReceptionIntervals(testId);
        return success(intervals);
        testService.stopTest(testId, getUserId(), getUsername());
        return success(testService.selectTestById(testId));
    }

    @PreAuthorize("@ss.hasPermi('system:test:remove')")

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

@@ -115,6 +115,16 @@
        assertTrue(v14.contains("CREATE TABLE t_collector_channel_lease"));
        assertTrue(v14.contains("idx_test_reception_status"));
        assertTrue(v14.contains("system:test:start"));

        String v16 = resource("db/migration/V16__simplify_test_reception_lifecycle.sql");
        assertTrue(v16.contains("system:test:stop"));
        assertTrue(v16.contains("旧区间表保留为历史数据"));
        assertFalse(v16.matches("(?s).*\\bDROP\\s+TABLE\\b.*"));

        String testMapper = resource("mapper/system/TTestMapper.xml");
        assertTrue(testMapper.contains("selectReceivingChannelsForAdapter"));
        assertTrue(testMapper.contains("RECEPTION_RESUMED"));
        assertTrue(testMapper.contains("TEST_STOPPED"));
    }

    private static String sha256(String path)

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

@@ -3,6 +3,7 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
@@ -22,9 +23,6 @@
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorRealtimeSnapshot;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorRealtimeMeasurement;
import org.springframework.dao.DuplicateKeyException;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.system.domain.TTest;
@@ -35,7 +33,6 @@
import com.ruoyi.system.domain.TTestPlanStep;
import com.ruoyi.system.domain.TTestWarningThreshold;
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ISharedFieldSnapshotService;

@@ -48,9 +45,6 @@

    @Mock
    private ISharedFieldSnapshotService sharedFieldSnapshotService;

    @Mock
    private BtsDevelopmentSimulator collectorSimulator;

    @InjectMocks
    private TTestServiceImpl testService;
@@ -377,12 +371,11 @@
        when(testMapper.countCollectorEnabled(8L)).thenReturn(1);
        when(testMapper.selectCollectorChannelCount(8L)).thenReturn(4);
        when(testMapper.collectorOnline(8L)).thenReturn(true);
        when(testMapper.updateReceptionStarted(eq(10L), any(), eq("operator"))).thenReturn(1);
        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, times(3)).insertCollectorChannelLease(eq(8L), any(), eq(10L), eq(30L), eq(7L), any());
        verify(testMapper, times(4)).insertReceptionInterval(any());
        verify(testMapper).insertTestAudit(eq(10L), eq("RECEPTION_STARTED"), eq(7L), eq("operator"), isNull(), any());
    }

@@ -403,12 +396,12 @@
        when(testMapper.countCollectorEnabled(8L)).thenReturn(1);
        when(testMapper.selectCollectorChannelCount(8L)).thenReturn(4);
        when(testMapper.collectorOnline(8L)).thenReturn(true);
        when(testMapper.updateReceptionStarted(eq(10L), any(), eq("operator"))).thenReturn(1);
        when(testMapper.updateReceptionStarted(eq(10L), any(), anyBoolean(), eq("operator"))).thenReturn(1);

        assertEquals(1, testService.startReception(10L, 7L, "operator"));
        verify(testMapper).insertTestAudit(eq(10L), eq("RECEPTION_STARTED"), eq(7L), eq("operator"), isNull(),
                org.mockito.ArgumentMatchers.argThat(value -> value != null && value.contains("runtimeAttach")
                        && value.contains("运行中接入")));
                        && value.contains("true")));
    }

    @Test
@@ -434,8 +427,7 @@
                () -> testService.startReception(10L, 7L, "operator"));

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

@@ -456,8 +448,7 @@

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

    @Test
@@ -483,7 +474,7 @@

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

    @Test
@@ -497,88 +488,50 @@
    }

    @Test
    void pauseReceptionShouldCloseAllOpenIntervalsAndAppendAudit()
    void pauseReceptionShouldStopPollingAndAppendAudit()
    {
        TTest current = validTest();
        current.setTestId(10L);
        current.setReceptionStatus("RECEIVING");
        when(testMapper.selectTestByIdForUpdate(10L)).thenReturn(current);
        TTestReceptionInterval cabinet = receptionInterval(1L, "CABINET_CHANNEL", 100L, null);
        TTestReceptionInterval collector = receptionInterval(2L, "COLLECTOR_CHANNEL", 8L, "S-01");
        when(testMapper.selectOpenReceptionIntervalsForUpdate(10L)).thenReturn(List.of(cabinet, collector));
        TDetectionChannel channel = new TDetectionChannel();
        channel.setChannelId(100L); channel.setOnlineStatus("ONLINE"); channel.setLastSeenAtUtc(Instant.parse("2026-08-29T12:00:00Z"));
        when(testMapper.selectReceptionChannelForUpdate(100L)).thenReturn(channel);
        when(testMapper.collectorOnline(8L)).thenReturn(true);
        mockCollectorBoundary();
        when(testMapper.closeReceptionInterval(anyLong(), any(), any())).thenReturn(1);
        when(testMapper.updateReceptionPaused(eq(10L), eq("operator"), any())).thenReturn(1);

        assertEquals(1, testService.pauseReception(10L, 7L, "operator"));
        verify(testMapper, times(2)).closeReceptionInterval(anyLong(), any(), any());
        verify(testMapper).updateReceptionPaused(eq(10L), eq("operator"), any());
        verify(testMapper).insertTestAudit(eq(10L), eq("RECEPTION_PAUSED"), eq(7L), eq("operator"), isNull(), any());
        verify(testMapper).insertTestAudit(eq(10L), eq("RECEPTION_PAUSED"), eq(7L), eq("operator"), isNull(),
                org.mockito.ArgumentMatchers.argThat(value -> value != null && value.contains("STOPPED")));
    }

    @Test
    void pauseReceptionShouldFailBeforeWritesWhenBoundaryUnavailable()
    void resumeReceptionShouldRestartPollingAndAppendAudit()
    {
        TTest current = validTest();
        current.setTestId(10L); current.setReceptionStatus("RECEIVING");
        current.setTestId(10L);
        current.setReceptionStatus("PAUSED");
        when(testMapper.selectTestByIdForUpdate(10L)).thenReturn(current);
        TTestReceptionInterval interval = receptionInterval(1L, "CABINET_CHANNEL", 100L, null);
        when(testMapper.selectOpenReceptionIntervalsForUpdate(10L)).thenReturn(List.of(interval));
        when(testMapper.selectReceptionChannelForUpdate(100L)).thenReturn(null);

        ServiceException exception = assertThrows(ServiceException.class,
                () -> testService.pauseReception(10L, 7L, "operator"));
        assertEquals("检测柜通道 100 当前无法确认暂停接收边界", exception.getMessage());
        verify(testMapper, never()).closeReceptionInterval(anyLong(), any(), any());
        verify(testMapper, never()).updateReceptionPaused(anyLong(), any(), any());
    }

    @Test
    void resumeReceptionShouldOpenNewIntervalsAndAppendAudit()
    {
        TTest current = validTest();
        current.setTestId(10L); current.setReceptionStatus("PAUSED");
        current.getBatteries().get(0).getBinding().setBindingId(30L);
        when(testMapper.selectTestByIdForUpdate(10L)).thenReturn(current);
        TDetectionChannel channel = new TDetectionChannel();
        channel.setChannelId(100L); channel.setOnlineStatus("ONLINE"); channel.setLastSeenAtUtc(Instant.parse("2026-08-29T12:00:00Z"));
        when(testMapper.selectReceptionChannelForUpdate(100L)).thenReturn(channel);
        when(testMapper.countCollectorEnabled(8L)).thenReturn(1);
        when(testMapper.collectorOnline(8L)).thenReturn(true);
        when(testMapper.selectCollectorChannelCount(8L)).thenReturn(4);
        mockCollectorBoundary();
        when(testMapper.updateReceptionResumed(eq(10L), eq("operator"), any())).thenReturn(1);

        assertEquals(1, testService.resumeReception(10L, 7L, "operator"));
        verify(testMapper, times(4)).insertReceptionInterval(any());
        verify(testMapper).updateReceptionResumed(eq(10L), eq("operator"), any());
        verify(testMapper).insertTestAudit(eq(10L), eq("RECEPTION_RESUMED"), eq(7L), eq("operator"), isNull(), any());
        verify(testMapper).insertTestAudit(eq(10L), eq("RECEPTION_RESUMED"), eq(7L), eq("operator"), isNull(),
                org.mockito.ArgumentMatchers.argThat(value -> value != null && value.contains("RESUMED")));
    }

    private TTestReceptionInterval receptionInterval(Long id, String sourceType, Long sourceId, String channelCode)
    @Test
    void stopTestShouldTerminateAndReleaseLeases()
    {
        TTestReceptionInterval interval = new TTestReceptionInterval();
        interval.setIntervalId(id); interval.setTestId(10L); interval.setSourceType(sourceType);
        interval.setSourceId(sourceId); interval.setSourceChannelCode(channelCode); interval.setIntervalStatus("OPEN");
        return interval;
    }
        TTest current = validTest();
        current.setTestId(10L);
        current.setReceptionStatus("PAUSED");
        when(testMapper.selectTestByIdForUpdate(10L)).thenReturn(current);
        when(testMapper.updateReceptionStopped(eq(10L), eq("operator"), any())).thenReturn(1);

    private void mockCollectorBoundary()
    {
        Instant deviceTime = Instant.parse("2026-08-29T12:00:00Z");
        CollectorRealtimeMeasurement measurement = new CollectorRealtimeMeasurement("dev-sim", "collector-8", "S-01",
                1L, deviceTime, "2026-08-29 20:00:00", deviceTime, "PRIMARY_TEMPERATURE", new BigDecimal("25"), "℃");
        CollectorRealtimeMeasurement positive = new CollectorRealtimeMeasurement("dev-sim", "collector-8", "S-02",
                1L, deviceTime, "2026-08-29 20:00:00", deviceTime, "PRIMARY_POSITIVE_POTENTIAL", new BigDecimal("1"), "V");
        CollectorRealtimeMeasurement negative = new CollectorRealtimeMeasurement("dev-sim", "collector-8", "S-03",
                1L, deviceTime, "2026-08-29 20:00:00", deviceTime, "PRIMARY_NEGATIVE_POTENTIAL", new BigDecimal("1"), "V");
        when(collectorSimulator.collectorRealtime(8)).thenReturn(
                new CollectorRealtimeSnapshot("collector-8", true, 1000,
                        List.of(measurement, positive, negative)));
        assertEquals(1, testService.stopTest(10L, 7L, "operator"));
        verify(testMapper).updateReceptionStopped(eq(10L), eq("operator"), any());
        verify(testMapper).deleteChannelLeasesByTestId(10L);
        verify(testMapper).deleteCollectorChannelLeasesByTestId(10L);
        verify(testMapper).insertTestAudit(eq(10L), eq("TEST_STOPPED"), eq(7L), eq("operator"), isNull(),
                org.mockito.ArgumentMatchers.argThat(value -> value != null && value.contains("TERMINATED")));
    }

    private void mockEnabledOptions()

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

@@ -4,7 +4,6 @@
import java.lang.reflect.Method;
import org.junit.jupiter.api.Test;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.security.access.prepost.PreAuthorize;

class TTestControllerContractTest
@@ -25,8 +24,9 @@
        assertPermission("resumeReception", "@ss.hasPermi('system:test:resume')", Long.class);
        assertEquals("/{testId}/reception/resume", TTestController.class.getMethod("resumeReception", Long.class)
                .getAnnotation(PostMapping.class).value()[0]);
        assertEquals("/{testId}/reception/intervals", TTestController.class.getMethod("receptionIntervals", Long.class)
                .getAnnotation(GetMapping.class).value()[0]);
        assertPermission("stopTest", "@ss.hasPermi('system:test:stop')", Long.class);
        assertEquals("/{testId}/stop", TTestController.class.getMethod("stopTest", Long.class)
                .getAnnotation(PostMapping.class).value()[0]);
    }

    private void assertPermission(String methodName, String permission, Class<?>... parameterTypes) throws Exception

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

File was deleted

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

@@ -12,7 +12,6 @@
import com.ruoyi.system.domain.TTestResult;
import com.ruoyi.system.domain.TTestWarningThreshold;
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.domain.vo.TestBatteryOptionVo;
import com.ruoyi.system.domain.vo.TestChannelOptionVo;
import com.ruoyi.system.domain.vo.TestCollectorOptionVo;
@@ -64,7 +63,7 @@
            @Param("actorId") Long actorId, @Param("actorName") String actorName,
            @Param("previousValue") String previousValue, @Param("newValue") String newValue);

    /** 锁定检测柜通道的最新可观测事实，供接收边界确认使用。 */
    /** 锁定检测柜通道的最新可观测事实，供开始接收校验使用。 */
    TDetectionChannel selectReceptionChannelForUpdate(Long channelId);
    /** 锁定当前检测柜通道租约；返回空表示可接入。 */
    TTestChannelBinding selectChannelLeaseForUpdate(Long channelId);
@@ -74,19 +73,18 @@
    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);
    int hasReceivingTestForAdapter(@Param("adapterCode") String adapterCode);
    int hasPausedOrEndedTestForAdapter(@Param("adapterCode") String adapterCode);
    List<TDetectionChannel> selectReceivingChannelsForAdapter(@Param("adapterCode") String adapterCode);
    boolean collectorOnline(Long collectorId);
    int updateReceptionStarted(@Param("testId") Long testId, @Param("startedAtUtc") Date startedAtUtc,
            @Param("updateBy") String updateBy);
    int insertReceptionInterval(TTestReceptionInterval interval);
    List<TTestReceptionInterval> selectOpenReceptionIntervalsForUpdate(Long testId);
    List<TTestReceptionInterval> selectReceptionIntervals(Long testId);
    TTestReceptionInterval selectLatestReceptionIntervalForUpdate(@Param("testId") Long testId,
            @Param("sourceType") String sourceType, @Param("sourceId") Long sourceId,
            @Param("sourceChannelCode") String sourceChannelCode);
    int closeReceptionInterval(@Param("intervalId") Long intervalId, @Param("endBoundaryAtUtc") Date endBoundaryAtUtc,
            @Param("boundaryNote") String boundaryNote);
            @Param("runtimeAttach") boolean runtimeAttach, @Param("updateBy") String updateBy);
    int updateReceptionPaused(@Param("testId") Long testId, @Param("updateBy") String updateBy,
            @Param("updatedAtUtc") Date updatedAtUtc);
    int updateReceptionResumed(@Param("testId") Long testId, @Param("updateBy") String updateBy,
            @Param("updatedAtUtc") Date updatedAtUtc);
    int updateReceptionStopped(@Param("testId") Long testId, @Param("updateBy") String updateBy,
            @Param("endedAtUtc") Date endedAtUtc);
}

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

@@ -26,17 +26,17 @@
    /** 旧服务入口已停用；调用方须使用带操作人的原子开始接收方法。 */
    int startTest(Long testId, String updateBy);

    /** 原子开始接收：校验全部来源并从实际观测边界打开区间。 */
    /** 原子开始接收：校验全部来源并开启测试数据请求。 */
    int startReception(Long testId, Long operatorId, String updateBy);

    /** 原子暂停接收：关闭全部开放区间并保留设备实时观察。 */
    /** 原子暂停接收：停止测试数据请求并保留最后实时快照。 */
    int pauseReception(Long testId, Long operatorId, String updateBy);

    /** 原子继续接收：从新的成功观测边界打开全部来源区间。 */
    /** 原子继续接收：恢复测试数据请求。 */
    int resumeReception(Long testId, Long operatorId, String updateBy);

    /** 查询测试各数据来源的接收区间和实际边界。 */
    List<com.ruoyi.system.domain.TTestReceptionInterval> selectReceptionIntervals(Long testId);
    /** 原子停止测试：停止数据请求并进入不可恢复终态。 */
    int stopTest(Long testId, Long operatorId, String updateBy);

    /** 仅删除未开始接收的测试并追加审计记录。 */
    int deleteTestByIds(Long[] testIds, Long operatorId, String updateBy);

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

@@ -135,7 +135,7 @@
        try
        {
            registry.synchronize(toRuntimeConfig(adapter));
            AdapterDiscoveryResult result = persistence.discover(registry, new DiscoverySource(
            AdapterDiscoveryResult result = persistence.discoverForReception(registry, new DiscoverySource(
                    adapter.getAdapterCode(), adapter.getAdapterId(), adapter.getAdapterName(),
                    adapter.getPollIntervalMs()));
            mapper.updateDiscoverySuccess(adapterId, result.observedAtUtc());
@@ -191,7 +191,10 @@
        {
            Instant lastAttempt = adapter.getLastAttemptAtUtc();
            if (lastAttempt != null && Duration.between(lastAttempt, now).toMillis() < adapter.getPollIntervalMs()) continue;
            try { discover(adapter.getAdapterId()); }
            try
            {
                if (persistence.shouldPollReception(adapter.getAdapterCode())) discover(adapter.getAdapterId());
            }
            catch (ServiceException ignored) { /* 失败状态已持久化，下一轮按周期重试。 */ }
        }
    }

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

@@ -16,8 +16,10 @@
import com.ruoyi.device.model.DeviceWorkStatus;
import com.ruoyi.device.model.RealtimeSnapshot;
import com.ruoyi.device.registry.DeviceAdapterRegistry;
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.vo.AdapterDiscoveryResult;
import com.ruoyi.system.mapper.TDetectionCabinetMapper;
import com.ruoyi.system.mapper.TTestMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@@ -28,17 +30,45 @@
    private static final Duration MINIMUM_CANDIDATE_LIFETIME = Duration.ofSeconds(30);
    private final DeviceDiscoveryCandidateStore candidateStore;
    private final TDetectionCabinetMapper detectionCabinetMapper;
    private final TTestMapper testMapper;

    @Autowired
    public DeviceDiscoveryPersistenceService(DeviceDiscoveryCandidateStore candidateStore,
            TDetectionCabinetMapper detectionCabinetMapper)
    {
        this(candidateStore, detectionCabinetMapper, null);
    }

    @Autowired
    public DeviceDiscoveryPersistenceService(DeviceDiscoveryCandidateStore candidateStore,
            TDetectionCabinetMapper detectionCabinetMapper, TTestMapper testMapper)
    {
        this.candidateStore = candidateStore;
        this.detectionCabinetMapper = detectionCabinetMapper;
        this.testMapper = testMapper;
    }

    /** 执行设备发现和可用通道查询，并将本轮观测写入 Redis 候选。 */
    public AdapterDiscoveryResult discover(DeviceAdapterRegistry registry, DiscoverySource source)
    {
        return discoverNow(registry, source);
    }

    /** 调度器观测；暂停或停止测试且没有其他接收测试时，不再向该适配器请求实时数据。 */
    public AdapterDiscoveryResult discoverForReception(DeviceAdapterRegistry registry, DiscoverySource source)
    {
        if (!shouldPollReception(source.adapterCode()))
            return new AdapterDiscoveryResult(0, 0, Instant.now());
        return discoverNow(registry, source);
    }

    /** 判断适配器是否仍需为活动测试发起实时请求。 */
    public boolean shouldPollReception(String adapterCode)
    {
        return testMapper == null || testMapper.hasReceivingTestForAdapter(adapterCode) > 0
                || testMapper.hasPausedOrEndedTestForAdapter(adapterCode) == 0;
    }

    private AdapterDiscoveryResult discoverNow(DeviceAdapterRegistry registry, DiscoverySource source)
    {
        DiscoveryObservation observation = registry.execute(source.adapterCode(), runtime -> {
            List<DetectedChannel> channels = runtime.discoverChannels();
@@ -46,8 +76,9 @@
                    .filter(DetectedChannel::available)
                    .map(DetectedChannel::identity)
                    .collect(java.util.stream.Collectors.toCollection(LinkedHashSet::new));
            Map<ChannelIdentity, RealtimeSnapshot> snapshots = available.isEmpty()
                    ? Map.of() : runtime.inquire(available);
            Set<ChannelIdentity> requested = receivingIdentities(source.adapterCode(), available);
            Map<ChannelIdentity, RealtimeSnapshot> snapshots = requested.isEmpty()
                    ? Map.of() : runtime.inquire(requested);
            return new DiscoveryObservation(channels, snapshots);
        });
        Instant observedAt = Instant.now();
@@ -73,9 +104,22 @@
        return new AdapterDiscoveryResult(grouped.size(), detected.size(), observedAt);
    }

    private Set<ChannelIdentity> receivingIdentities(String adapterCode, Set<ChannelIdentity> available)
    {
        if (testMapper == null) return available;
        List<TDetectionChannel> receiving = testMapper.selectReceivingChannelsForAdapter(adapterCode);
        if (receiving == null || receiving.isEmpty()) return Set.of();
        Set<ChannelIdentity> identities = receiving.stream()
                .map(channel -> new ChannelIdentity(adapterCode, channel.getDevtype(), channel.getDevid(),
                        channel.getSubdevid(), channel.getChlid()))
                .collect(java.util.stream.Collectors.toCollection(LinkedHashSet::new));
        identities.retainAll(available);
        return identities;
    }

    /**
     * 发现候选与正式设备是两份事实；正式登记后仍需把成功观测同步到 MySQL，
     * 否则开始接收无法取得可靠边界。离线轮询只更新当前状态，不覆盖最后成功观测时间。
     * 发现候选与正式设备是两份事实；正式登记后仍需把成功观测同步到 MySQL。
     * 离线轮询只更新当前状态，不覆盖最后成功观测时间；暂停期间不会进入此路径。
     */
    private void synchronizeRegisteredFacts(DiscoverySource source, List<DetectedChannel> detected,
            Map<ChannelIdentity, RealtimeSnapshot> snapshots, Instant observedAt)

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

@@ -11,6 +11,8 @@
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import jakarta.annotation.PreDestroy;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
@@ -33,6 +35,7 @@
    private final DeviceDiscoveryCandidateStore candidates;
    private final BtsDevelopmentSimulator simulator;
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    private final ConcurrentMap<String, LiveSource> lastSources = new ConcurrentHashMap<>();

    public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates,
            BtsDevelopmentSimulator simulator)
@@ -43,10 +46,14 @@
        TTest test = tests.selectTestById(testId);
        List<CabinetCandidate> cabinets = candidates.snapshot(Instant.now()).cabinets();
        List<LiveSource> sources = new ArrayList<>();
        boolean freezeSnapshots = "PAUSED".equals(test.getReceptionStatus())
                || "ENDED".equals(test.getReceptionStatus());
        for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
        {
            if (battery.getBinding() != null) sources.add(cabinetSource(battery, cabinets));
            if (battery.getCollectorId() != null) sources.add(collectorSource(battery));
            if (battery.getBinding() != null)
                sources.add(snapshotFor(testId, cabinetSource(battery, cabinets), freezeSnapshots));
            if (battery.getCollectorId() != null)
                sources.add(snapshotFor(testId, collectorSource(battery), freezeSnapshots));
        }
        List<Map<String, Object>> exceptions = sources.stream().filter(source -> !"ONLINE".equals(source.deviceStatus())
                && !"WORKING".equals(source.deviceStatus()) && !"PAUSED".equals(source.deviceStatus()))
@@ -56,6 +63,14 @@
        return new LiveState(testId, test.getTestStatus(), test.getReceptionStatus(), Instant.now(), sources, exceptions);
    }

    private LiveSource snapshotFor(Long testId, LiveSource fresh, boolean freeze)
    {
        String key = testId + ":" + fresh.sourceType() + ":" + fresh.sourceId() + ":" + fresh.batteryNo();
        if (freeze) return lastSources.getOrDefault(key, fresh);
        lastSources.put(key, fresh);
        return fresh;
    }

    @Override public SseEmitter stream(Long testId)
    {
        SseEmitter emitter = new SseEmitter(0L);

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

@@ -25,10 +25,7 @@
import com.ruoyi.system.domain.TTestFieldSnapshot;
import com.ruoyi.system.domain.TTestWarningThreshold;
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.domain.vo.TestConfigurationOptionsVo;
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorRealtimeMeasurement;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ITTestService;
import com.ruoyi.system.service.ISharedFieldSnapshotService;
@@ -51,9 +48,6 @@

    @Autowired
    private ISharedFieldSnapshotService sharedFieldSnapshotService;

    @Autowired(required = false)
    private BtsDevelopmentSimulator collectorSimulator;

    @Override
    public TTest selectTestById(Long testId)
@@ -172,7 +166,6 @@
        if (current.getBatteries().isEmpty()) throw new ServiceException("测试未配置电池号");

        Date actionAt = new Date();
        List<TTestReceptionInterval> intervals = new ArrayList<>();
        Set<Long> channelIds = new HashSet<>();
        Set<String> collectorChannels = new HashSet<>();
        boolean runtimeAttach = false;
@@ -186,8 +179,6 @@
            TDetectionChannel channel = testMapper.selectReceptionChannelForUpdate(binding.getChannelId());
            if (channel == null || !isEnabled(channel.getStatus()) || !isOnline(channel.getOnlineStatus()))
                throw new ServiceException("检测柜通道 " + binding.getChannelId() + " 当前离线或不可观测");
            if (channel.getLastSeenAtUtc() == null)
                throw new ServiceException("检测柜通道 " + binding.getChannelId() + " 尚未取得成功观测边界");
            runtimeAttach |= "WORKING".equalsIgnoreCase(channel.getWorkStatus())
                    || "PAUSED".equalsIgnoreCase(channel.getWorkStatus());
            if (testMapper.selectChannelLeaseForUpdate(binding.getChannelId()) != null)
@@ -200,14 +191,6 @@
            validateCollectorChannels(battery.getCollectorId(), collectorChannelCount == null ? 0 : collectorChannelCount,
                    battery.getCollectorChannels(), collectorChannels);

            intervals.add(interval(current, binding.getBindingId(), "CABINET_CHANNEL", binding.getChannelId(), null,
                    actionAt, Date.from(channel.getLastSeenAtUtc()), runtimeAttach
                            ? "运行中接入：采用检测柜最近成功观测边界" : "采用检测柜最近成功观测边界"));
            for (TTestCollectorChannel mapping : mutable(battery.getCollectorChannels()))
            {
                intervals.add(interval(current, binding.getBindingId(), "COLLECTOR_CHANNEL", battery.getCollectorId(),
                        mapping.getChannelCode(), actionAt, actionAt, "采集器开始接收确认边界"));
            }
        }

        try
@@ -220,12 +203,10 @@
                    testMapper.insertCollectorChannelLease(battery.getCollectorId(), mapping.getChannelCode(), testId,
                            binding.getBindingId(), operatorId, actionAt);
            }
            for (TTestReceptionInterval interval : intervals) testMapper.insertReceptionInterval(interval);
            if (testMapper.updateReceptionStarted(testId, actionAt, updateBy) != 1)
            if (testMapper.updateReceptionStarted(testId, actionAt, runtimeAttach, updateBy) != 1)
                throw new ServiceException("测试状态已变更，请刷新后重试");
            testMapper.insertTestAudit(testId, "RECEPTION_STARTED", operatorId, updateBy, null,
                    JSON.toJSONString(Map.of("receptionStatus", "RECEIVING", "intervalCount", intervals.size(),
                            "runtimeAttach", runtimeAttach, "intervals", intervals)));
                    JSON.toJSONString(Map.of("receptionStatus", "RECEIVING", "runtimeAttach", runtimeAttach)));
            return 1;
        }
        catch (DuplicateKeyException exception)
@@ -245,29 +226,11 @@
        if (!"RECEIVING".equals(current.getReceptionStatus()))
            throw new ServiceException("当前测试不是接收中状态");

        List<TTestReceptionInterval> openIntervals = mutable(testMapper.selectOpenReceptionIntervalsForUpdate(testId));
        if (openIntervals.isEmpty()) throw new ServiceException("当前测试没有开放的接收区间");
        Date actionAt = new Date();
        Map<Long, Date> boundaries = new HashMap<>();
        // 先确认所有来源都能取得边界，再写入任何一条关闭记录，保证整项动作原子失败。
        for (TTestReceptionInterval open : openIntervals)
        {
            Date boundary = confirmBoundary(open, actionAt);
            boundaries.put(open.getIntervalId(), boundary);
            open.setEndBoundaryAtUtc(boundary);
            open.setIntervalStatus("CLOSED");
        }
        for (TTestReceptionInterval open : openIntervals)
        {
            if (testMapper.closeReceptionInterval(open.getIntervalId(), boundaries.get(open.getIntervalId()),
                    "暂停接收：采用来源最近成功观测边界") != 1)
                throw new ServiceException("接收区间已变更，请刷新后重试");
        }
        if (testMapper.updateReceptionPaused(testId, updateBy, actionAt) != 1)
            throw new ServiceException("测试状态已变更，请刷新后重试");
        testMapper.insertTestAudit(testId, "RECEPTION_PAUSED", operatorId, updateBy, null,
                JSON.toJSONString(Map.of("receptionStatus", "PAUSED", "intervalCount", openIntervals.size(),
                        "intervals", openIntervals)));
                JSON.toJSONString(Map.of("receptionStatus", "PAUSED", "polling", "STOPPED")));
        return 1;
    }

@@ -281,131 +244,32 @@
        if (current == null) throw new ServiceException("测试不存在");
        if (!"PAUSED".equals(current.getReceptionStatus()))
            throw new ServiceException("当前测试不是暂停接收状态");
        loadConfiguration(current);
        if (current.getBatteries().isEmpty()) throw new ServiceException("测试未配置电池号");

        Date actionAt = new Date();
        List<TTestReceptionInterval> intervals = new ArrayList<>();
        Set<Long> channelIds = new HashSet<>();
        Set<String> collectorChannels = new HashSet<>();
        for (TTestBattery battery : current.getBatteries())
        {
            TTestChannelBinding binding = battery.getBinding();
            if (binding == null || binding.getChannelId() == null)
                throw new ServiceException("电池号 " + battery.getBatteryNo() + " 未绑定检测柜通道");
            if (!channelIds.add(binding.getChannelId())) throw new ServiceException("检测柜通道重复绑定，无法继续接收");
            TDetectionChannel channel = testMapper.selectReceptionChannelForUpdate(binding.getChannelId());
            if (channel == null || !isEnabled(channel.getStatus()) || !isOnline(channel.getOnlineStatus())
                    || channel.getLastSeenAtUtc() == null)
                throw new ServiceException("检测柜通道 " + binding.getChannelId() + " 当前无法确认继续接收边界");
            if (isEndedWorkStatus(channel.getWorkStatus()))
                throw new ServiceException("检测柜通道 " + binding.getChannelId() + " 已结束，不能继续接收");
            if (battery.getCollectorId() == null || testMapper.countCollectorEnabled(battery.getCollectorId()) == 0
                    || !testMapper.collectorOnline(battery.getCollectorId()))
                throw new ServiceException("采集器 " + battery.getCollectorId() + " 当前无法确认继续接收边界");
            Integer count = testMapper.selectCollectorChannelCount(battery.getCollectorId());
            validateCollectorChannels(battery.getCollectorId(), count == null ? 0 : count,
                    battery.getCollectorChannels(), collectorChannels);
            Date cabinetBoundary = Date.from(channel.getLastSeenAtUtc());
            ensureAfterPreviousBoundary(current.getTestId(), "CABINET_CHANNEL", binding.getChannelId(), null,
                    cabinetBoundary);
            intervals.add(interval(current, binding.getBindingId(), "CABINET_CHANNEL", binding.getChannelId(), null,
                    actionAt, cabinetBoundary, "继续接收：采用检测柜最近成功观测边界"));
            for (TTestCollectorChannel mapping : mutable(battery.getCollectorChannels()))
            {
                Date collectorBoundary = collectorBoundary(battery.getCollectorId(), mapping.getChannelCode());
                ensureAfterPreviousBoundary(current.getTestId(), "COLLECTOR_CHANNEL", battery.getCollectorId(),
                        mapping.getChannelCode(), collectorBoundary);
                intervals.add(interval(current, binding.getBindingId(), "COLLECTOR_CHANNEL", battery.getCollectorId(),
                        mapping.getChannelCode(), actionAt, collectorBoundary, "继续接收：采用采集器最近成功观测边界"));
            }
        }
        for (TTestReceptionInterval interval : intervals) testMapper.insertReceptionInterval(interval);
        if (testMapper.updateReceptionResumed(testId, updateBy, actionAt) != 1)
            throw new ServiceException("测试状态已变更，请刷新后重试");
        testMapper.insertTestAudit(testId, "RECEPTION_RESUMED", operatorId, updateBy, null,
                JSON.toJSONString(Map.of("receptionStatus", "RECEIVING", "intervalCount", intervals.size(),
                        "intervals", intervals)));
                JSON.toJSONString(Map.of("receptionStatus", "RECEIVING", "polling", "RESUMED")));
        return 1;
    }

    @Override
    public List<TTestReceptionInterval> selectReceptionIntervals(Long testId)
    @Transactional(rollbackFor = Exception.class)
    public int stopTest(Long testId, Long operatorId, String updateBy)
    {
        if (testId == null) throw new ServiceException("测试ID不能为空");
        selectTestById(testId);
        return mutable(testMapper.selectReceptionIntervals(testId));
    }

    private Date confirmBoundary(TTestReceptionInterval interval, Date actionAt)
    {
        if ("CABINET_CHANNEL".equals(interval.getSourceType()))
        {
            TDetectionChannel channel = testMapper.selectReceptionChannelForUpdate(interval.getSourceId());
            if (channel == null || !isEnabled(channel.getStatus()) || !isOnline(channel.getOnlineStatus())
                    || channel.getLastSeenAtUtc() == null)
                throw new ServiceException("检测柜通道 " + interval.getSourceId() + " 当前无法确认暂停接收边界");
            Date boundary = Date.from(channel.getLastSeenAtUtc());
            if (interval.getStartBoundaryAtUtc() != null && boundary.before(interval.getStartBoundaryAtUtc()))
                throw new ServiceException("检测柜通道 " + interval.getSourceId() + " 的暂停边界早于区间开始");
            return boundary;
        }
        if (!testMapper.collectorOnline(interval.getSourceId()))
            throw new ServiceException("采集器 " + interval.getSourceId() + " 当前无法确认暂停接收边界");
        Date boundary = collectorBoundary(interval.getSourceId(), interval.getSourceChannelCode());
        if (interval.getStartBoundaryAtUtc() != null && boundary.before(interval.getStartBoundaryAtUtc()))
            throw new ServiceException("采集器 " + interval.getSourceId() + " 的暂停边界早于区间开始");
        return boundary;
    }

    private Date collectorBoundary(Long collectorId, String channelCode)
    {
        if (collectorSimulator == null) throw new ServiceException("采集器 " + collectorId + " 当前无法取得成功观测边界");
        try
        {
            return collectorSimulator.collectorRealtime(collectorId.intValue()).measurements().stream()
                    .filter(item -> channelCode == null || channelCode.equals(item.channelCode()))
                    .map(CollectorRealtimeMeasurement::deviceTime).filter(java.util.Objects::nonNull)
                    .max(java.time.Instant::compareTo).map(Date::from)
                    .orElseThrow(() -> new ServiceException("采集器 " + collectorId + " 当前无法取得成功观测边界"));
        }
        catch (RuntimeException exception)
        {
            if (exception instanceof ServiceException serviceException) throw serviceException;
            throw new ServiceException("采集器 " + collectorId + " 当前无法取得成功观测边界");
        }
    }

    private void ensureAfterPreviousBoundary(Long testId, String sourceType, Long sourceId,
            String sourceChannelCode, Date boundary)
    {
        TTestReceptionInterval previous = testMapper.selectLatestReceptionIntervalForUpdate(testId, sourceType,
                sourceId, sourceChannelCode);
        if (previous != null && previous.getEndBoundaryAtUtc() != null
                && !boundary.after(previous.getEndBoundaryAtUtc()))
            throw new ServiceException("数据来源尚未取得新的接收边界，请稍后重试");
    }

    private boolean isEndedWorkStatus(String workStatus)
    {
        return "STOPPED".equalsIgnoreCase(workStatus) || "FINISHED".equalsIgnoreCase(workStatus)
                || "ENDED".equalsIgnoreCase(workStatus);
    }

    private TTestReceptionInterval interval(TTest test, Long bindingId, String sourceType, Long sourceId,
            String sourceChannelCode, Date actionAt, Date boundaryAt, String note)
    {
        TTestReceptionInterval interval = new TTestReceptionInterval();
        interval.setTestId(test.getTestId());
        interval.setBindingId(bindingId);
        interval.setSourceType(sourceType);
        interval.setSourceId(sourceId);
        interval.setSourceChannelCode(sourceChannelCode);
        interval.setActionAtUtc(actionAt);
        interval.setStartBoundaryAtUtc(boundaryAt);
        interval.setBoundaryNote(note);
        interval.setIntervalStatus("OPEN");
        return interval;
        if (operatorId == null) throw new ServiceException("操作人不能为空");
        TTest current = testMapper.selectTestByIdForUpdate(testId);
        if (current == null) throw new ServiceException("测试不存在");
        if (NOT_STARTED.equals(current.getReceptionStatus()) || "ENDED".equals(current.getReceptionStatus()))
            throw new ServiceException("当前测试不能停止");
        Date actionAt = new Date();
        if (testMapper.updateReceptionStopped(testId, updateBy, actionAt) != 1)
            throw new ServiceException("测试状态已变更，请刷新后重试");
        testMapper.deleteChannelLeasesByTestId(testId);
        testMapper.deleteCollectorChannelLeasesByTestId(testId);
        testMapper.insertTestAudit(testId, "TEST_STOPPED", operatorId, updateBy, null,
                JSON.toJSONString(Map.of("testStatus", "TERMINATED", "receptionStatus", "ENDED", "polling", "STOPPED")));
        return 1;
    }

    private boolean isOnline(String status)

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

@@ -117,8 +117,7 @@
               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,
               (select count(1) from t_test_battery b where b.test_id = t.test_id and b.del_flag = 0) battery_count,
               exists (select 1 from t_test_reception_interval ri where ri.test_id = t.test_id
                       and ri.boundary_note like '运行中接入%') runtime_attach
               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'
@@ -274,6 +273,19 @@
        select result.*, battery.test_id from t_test_result result
        join t_test_battery battery on battery.relation_id = result.relation_id
        where battery.test_id = #{testId} and result.del_flag = 0
          and exists (
              select 1 from t_test_audit start_action
              where start_action.test_id = battery.test_id
                and start_action.event_type in ('RECEPTION_STARTED', 'RECEPTION_RESUMED')
                and start_action.occurred_at_utc &lt;= result.result_time
                and not exists (
                    select 1 from t_test_audit pause_action
                    where pause_action.test_id = battery.test_id
                      and pause_action.event_type in ('RECEPTION_PAUSED', 'TEST_STOPPED')
                      and pause_action.occurred_at_utc &gt; start_action.occurred_at_utc
                      and pause_action.occurred_at_utc &lt;= result.result_time
                )
          )
        <if test="relationId != null">and result.relation_id = #{relationId}</if>
        <if test="channelNo != null">and result.channel_no = #{channelNo}</if>
        <if test="beginResultTime != null">and result.result_time &gt;= #{beginResultTime}</if>
@@ -380,6 +392,43 @@
        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>
    <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
        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 = '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">
        select distinct channel_fact.channel_id, channel_fact.adapter_id, channel_fact.adapter_code,
               channel_fact.cabinet_id, channel_fact.devtype, channel_fact.devid,
               channel_fact.subdevid, channel_fact.chlid, channel_fact.display_no,
               channel_fact.online_status, channel_fact.work_status, channel_fact.last_seen_at_utc,
               channel_fact.status
        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 = 'RECEIVING'
          and test_fact.del_flag = 0 and channel_fact.del_flag = 0 and channel_fact.status = '0'
    </select>
    <select id="collectorOnline" resultType="boolean">
        select count(1) &gt; 0 from t_collector
        where collector_id = #{collectorId} and status = '0' and del_flag = 0
@@ -388,54 +437,10 @@
    <update id="updateReceptionStarted">
        update t_test
        set test_status = 'RUNNING', reception_status = 'RECEIVING',
            runtime_attach = #{runtimeAttach},
            started_at_utc = coalesce(started_at_utc, #{startedAtUtc}),
            update_by = #{updateBy}, update_time = #{startedAtUtc}
        where test_id = #{testId} and reception_status = 'NOT_STARTED' and del_flag = 0
    </update>
    <insert id="insertReceptionInterval" parameterType="com.ruoyi.system.domain.TTestReceptionInterval" useGeneratedKeys="true" keyProperty="intervalId">
        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},
               #{startBoundaryAtUtc}, #{boundaryNote}, #{intervalStatus})
    </insert>
    <select id="selectOpenReceptionIntervalsForUpdate" parameterType="Long"
            resultType="com.ruoyi.system.domain.TTestReceptionInterval">
        select interval_id intervalId, test_id testId, binding_id bindingId, source_type sourceType,
               source_id sourceId, source_channel_code sourceChannelCode, action_at_utc actionAtUtc,
               start_boundary_at_utc startBoundaryAtUtc, end_boundary_at_utc endBoundaryAtUtc,
               boundary_note boundaryNote, interval_status intervalStatus
        from t_test_reception_interval
        where test_id = #{testId} and interval_status = 'OPEN'
        order by interval_id
        for update
    </select>
    <select id="selectReceptionIntervals" parameterType="Long"
            resultType="com.ruoyi.system.domain.TTestReceptionInterval">
        select interval_id intervalId, test_id testId, binding_id bindingId, source_type sourceType,
               source_id sourceId, source_channel_code sourceChannelCode, action_at_utc actionAtUtc,
               start_boundary_at_utc startBoundaryAtUtc, end_boundary_at_utc endBoundaryAtUtc,
               boundary_note boundaryNote, interval_status intervalStatus
        from t_test_reception_interval
        where test_id = #{testId}
        order by interval_id
    </select>
    <select id="selectLatestReceptionIntervalForUpdate" resultType="com.ruoyi.system.domain.TTestReceptionInterval">
        select interval_id intervalId, test_id testId, binding_id bindingId, source_type sourceType,
               source_id sourceId, source_channel_code sourceChannelCode, action_at_utc actionAtUtc,
               start_boundary_at_utc startBoundaryAtUtc, end_boundary_at_utc endBoundaryAtUtc,
               boundary_note boundaryNote, interval_status intervalStatus
        from t_test_reception_interval
        where test_id = #{testId} 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 interval_id desc
        limit 1
        for update
    </select>
    <update id="closeReceptionInterval">
        update t_test_reception_interval
        set end_boundary_at_utc = #{endBoundaryAtUtc}, boundary_note = #{boundaryNote}, interval_status = 'CLOSED'
        where interval_id = #{intervalId} and interval_status = 'OPEN'
    </update>
    <update id="updateReceptionPaused">
        update t_test
@@ -447,4 +452,10 @@
        set test_status = 'RUNNING', reception_status = 'RECEIVING', update_by = #{updateBy}, update_time = #{updatedAtUtc}
        where test_id = #{testId} and reception_status = 'PAUSED' and del_flag = 0
    </update>
    <update id="updateReceptionStopped">
        update t_test
        set test_status = 'TERMINATED', reception_status = 'ENDED', ended_at_utc = #{endedAtUtc},
            update_by = #{updateBy}, update_time = #{endedAtUtc}
        where test_id = #{testId} and reception_status in ('RECEIVING', 'PAUSED') and del_flag = 0
    </update>
</mapper>