实现新威 7.6 直连取数与逐行接收

yeg

2026-09-28 147428971be43eed90a79f3849f15000d3855390

实现新威 7.6 直连取数与逐行接收

- 新增 battery-device-neware 模块：只读直连厂商 MySQL 的 NEWARE_MYSQL 适配器，
  含独立数据源、语句白名单，以及"从未发出过非查询语句"的自动化断言；
  连接口令只从环境变量或启动参数读取，不入库。
- 适配器实例类型按存储形态放宽，来源形态的非敏感配置落到 t_device_adapter.source_config（迁移 V22）。
- 取数按通道身份里的适配器编码从注册表取实例：新增聚合适配器并声明为首选，
  实例缺失时明确失败、不回退到其他实例，发现与取数重新走同一条路。
- 接收水位进入观测契约：新增 currentWatermark / samplesSince，只取严格晚于水位的数据行；
  同秒多行按来源行身份各自成立（槽位留空），重复同内容幂等、同身份不同内容不覆盖。
- 接收调度器脱离 dev-simulator profile；厂商来源的发现候选不按时间过期并按最近数据时间倒序展示。
- 检测柜列表新增数据来源，控制台实时区与数据列表实时页展示最近数据时间与通道工作状态。

 .gitignore

@@ -4,6 +4,7 @@
/battery-business/target/
/battery-device-api/target/
/battery-device-bts/target/
/battery-device-neware/target/
/battery-device-simulator/target/
/ruoyi-framework/target/
/ruoyi-generator/target/

 battery-business/src/main/java/com/battery/business/domain/TDetectionCabinet.java

@@ -30,6 +30,9 @@
    private Long adapterId;
    private String adapterCode;

    /** 数据来源名称，仅供展示，用于区分不同机器上的同名柜号；不落库。 */
    private String adapterName;

    private Integer vendorDeviceType;
    private Integer vendorDeviceId;
    private Instant lastSeenAtUtc;
@@ -107,7 +110,9 @@
    public Long getAdapterId() { return adapterId; }
    public void setAdapterId(Long adapterId) { this.adapterId = adapterId; }
    public String getAdapterCode() { return adapterCode; }
    public void setAdapterCode(String adapterCode) { this.adapterCode = adapterCode; }
    public void setAdapterCode(String value) { this.adapterCode = value; }
    public String getAdapterName() { return adapterName; }
    public void setAdapterName(String value) { adapterName = value; }
    public Integer getVendorDeviceType() { return vendorDeviceType; }
    public void setVendorDeviceType(Integer value) { vendorDeviceType = value; }
    public Integer getVendorDeviceId() { return vendorDeviceId; }

 battery-business/src/main/java/com/battery/business/domain/TDeviceAdapter.java

@@ -49,6 +49,10 @@
    private Boolean readOnly;

    private String capabilities;

    /** 来源形态配置（JSON），承载各存储形态互不相交的非敏感字段；连接口令不入库。 */
    private String sourceConfig;

    private String status;
    /** 适配器连接状态，仅允许 ONLINE/OFFLINE。 */
    private String connectionStatus;
@@ -78,6 +82,8 @@
    public void setReadOnly(Boolean readOnly) { this.readOnly = readOnly; }
    public String getCapabilities() { return capabilities; }
    public void setCapabilities(String capabilities) { this.capabilities = capabilities; }
    public String getSourceConfig() { return sourceConfig; }
    public void setSourceConfig(String sourceConfig) { this.sourceConfig = sourceConfig; }
    public String getStatus() { return status; }
    public void setStatus(String status) { this.status = status; }
    public String getConnectionStatus() { return connectionStatus; }

 battery-business/src/main/java/com/battery/business/domain/vo/DiscoveredCabinetVo.java

@@ -1,6 +1,12 @@
package com.battery.business.domain.vo;

/** 原检测柜选择弹窗使用的设备摘要。 */
import java.time.Instant;

/**
 * 检测柜选择弹窗使用的设备摘要。
 * {@code lastDataAt} 是该设备最近一次采样行的设备时间（来源里没有采样行时为空），
 * 由操作员据此判断哪几台柜子现在真的在产出数据；登记与否仍由操作员决定。
 */
public record DiscoveredCabinetVo(
        String deviceCode,
        String ip,
@@ -9,7 +15,8 @@
        String manufacturer,
        String model,
        Integer channelCount,
        String onlineStatus)
        String onlineStatus,
        Instant lastDataAt)
{
    public String getDeviceCode() { return deviceCode; }
    public String getIp() { return ip; }
@@ -19,4 +26,5 @@
    public String getModel() { return model; }
    public Integer getChannelCount() { return channelCount; }
    public String getOnlineStatus() { return onlineStatus; }
    public Instant getLastDataAt() { return lastDataAt; }
}

 battery-business/src/main/java/com/battery/business/mapper/TTestMapper.java

@@ -89,6 +89,9 @@
    int insertRealtimeRecord(TTestRealtimeRecord record);
    String selectRealtimeRecordContentHash(@Param("testId") Long testId, @Param("testBatteryId") Long testBatteryId,
            @Param("captureSlotUtc") Date captureSlotUtc);
    /** 按来源行身份查询已保存的实时记录摘要；逐行接收据此做幂等与冲突判定。 */
    String selectRealtimeRecordHashBySourceRecord(@Param("testBatteryId") Long testBatteryId,
            @Param("sourceIdentity") String sourceIdentity, @Param("sourceRecordId") String sourceRecordId);
    int insertStepRecord(TTestStepRecord record);
    int insertCycleRecord(TTestCycleRecord record);
    String selectStepRecordContentHash(@Param("testBatteryId") Long testBatteryId,

 battery-business/src/main/java/com/battery/business/service/impl/DeviceAdapterRuntimeCoordinator.java

@@ -2,14 +2,15 @@

import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
import org.springframework.context.annotation.Profile;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import com.battery.business.service.IDeviceAdapterService;

/** 在单采集进程内恢复启用实例，并按各自周期执行只读发现。 */
/**
 * 在单采集进程内恢复启用实例，并按各自周期执行只读发现。
 * 不限定 profile：真实来源的发现与接收要在非开发 profile 下运行。
 */
@Component
@Profile({ "dev-simulator", "bts" })
public class DeviceAdapterRuntimeCoordinator
{
    private final IDeviceAdapterService service;

 battery-business/src/main/java/com/battery/business/service/impl/DeviceAdapterServiceImpl.java

@@ -32,6 +32,10 @@
    private static final String DISABLED = "1";
    private static final String READ_ONLY_CAPABILITIES =
            "{\"discovery\":true,\"realtimeObservation\":true,\"historyDownload\":true,\"control\":false}";
    /** 适配器类型按存储形态划分；BTS 保留给开发模拟器与将来的协议路线（ADR-0065）。 */
    private static final List<String> SUPPORTED_TYPES = List.of("BTS", "NEWARE_MYSQL");
    /** 直连厂商存储的来源，其通道登记清单是长期事实，候选不按时间过期。 */
    private static final List<String> PERSISTENT_CANDIDATE_TYPES = List.of("NEWARE_MYSQL");

    private final TDeviceAdapterMapper mapper;
    private final DeviceAdapterRegistry registry;
@@ -136,7 +140,7 @@
            registry.synchronize(toRuntimeConfig(adapter));
            AdapterDiscoveryResult result = persistence.discoverForReception(registry, new DiscoverySource(
                    adapter.getAdapterCode(), adapter.getAdapterId(), adapter.getAdapterName(),
                    adapter.getPollIntervalMs()));
                    adapter.getPollIntervalMs(), PERSISTENT_CANDIDATE_TYPES.contains(adapter.getAdapterType())));
            mapper.updateDiscoverySuccess(adapterId, result.observedAtUtc());
            return result;
        }
@@ -214,8 +218,9 @@
        adapter.setAdapterType(StringUtils.trim(adapter.getAdapterType()).toUpperCase(Locale.ROOT));
        adapter.setHost(StringUtils.trim(adapter.getHost()));
        adapter.setTimezoneId(StringUtils.trim(adapter.getTimezoneId()));
        adapter.setSourceConfig(StringUtils.trim(adapter.getSourceConfig()));
        adapter.setRemark(StringUtils.trim(adapter.getRemark()));
        if (!"BTS".equals(adapter.getAdapterType()))
        if (!SUPPORTED_TYPES.contains(adapter.getAdapterType()))
            throw error("不支持的适配器类型", DeviceAdapterErrorCodes.UNSUPPORTED_TYPE);
        if (!Boolean.TRUE.equals(adapter.getReadOnly()))
            throw error("当前里程碑只允许只读设备适配器", DeviceAdapterErrorCodes.INVALID_CONFIG);
@@ -243,7 +248,21 @@
    {
        return new AdapterInstanceConfig(adapter.getAdapterCode(), adapter.getAdapterType(), adapter.getHost(),
                adapter.getPort(), adapter.getPollIntervalMs(), adapter.getTimezoneId(),
                ENABLED.equals(adapter.getStatus()), parseCapabilities(adapter.getCapabilities()));
                ENABLED.equals(adapter.getStatus()), parseCapabilities(adapter.getCapabilities()),
                parseSourceConfig(adapter.getSourceConfig()));
    }

    /** 来源形态配置只承载非敏感项；连接口令由环境变量或启动参数提供，不入库。 */
    private static java.util.Map<String, String> parseSourceConfig(String sourceConfig)
    {
        if (StringUtils.isBlank(sourceConfig)) return java.util.Map.of();
        JSONObject value = JSON.parseObject(sourceConfig);
        if (value == null) return java.util.Map.of();
        java.util.Map<String, String> result = new java.util.LinkedHashMap<>();
        value.forEach((key, item) -> {
            if (item != null) result.put(String.valueOf(key), String.valueOf(item));
        });
        return result;
    }

    private AdapterCapabilities parseCapabilities(String capabilities)

 battery-business/src/main/java/com/battery/business/service/impl/DeviceDiscoveryPersistenceService.java

@@ -28,6 +28,11 @@
public class DeviceDiscoveryPersistenceService
{
    private static final Duration MINIMUM_CANDIDATE_LIFETIME = Duration.ofSeconds(30);
    /**
     * 直连厂商存储的来源，其通道登记清单是长期存在的事实，没有"过期"的对应语义：
     * 候选是否出现在界面上只由"已登记且未删除"决定，这个时间只是让候选一直有效。
     */
    static final Instant CANDIDATE_NEVER_EXPIRES = Instant.parse("9999-12-31T23:59:59Z");
    private final DeviceDiscoveryCandidateStore candidateStore;
    private final TDetectionCabinetMapper detectionCabinetMapper;
    private final TTestMapper testMapper;
@@ -75,7 +80,8 @@
                .map(item -> new CabinetCandidate(item.adapterCode(), item.adapterId(), item.deviceCode(),
                        item.devtype(), item.devid(), item.ip(), item.mac(), item.name(), item.manufacturer(), item.model(),
                        item.channels().stream().map(channel -> new ChannelCandidate(channel.identity(), false,
                                DeviceWorkStatus.UNOBSERVABLE, failedAt)).toList(), failedAt, item.expiresAt()))
                                DeviceWorkStatus.UNOBSERVABLE, failedAt, null, channel.lastDataAt())).toList(),
                        failedAt, item.expiresAt(), item.lastDataAt()))
                .toList());
        detectionCabinetMapper.markRegisteredFactsUnobservable(adapterCode);
        detectionCabinetMapper.markRegisteredCabinetsOffline(adapterCode);
@@ -108,9 +114,10 @@
                        LinkedHashMap::new, java.util.stream.Collectors.toList()));
        Duration lifetime = Duration.ofMillis(Math.max(MINIMUM_CANDIDATE_LIFETIME.toMillis(),
                (long) source.pollIntervalMs() * 3));
        Instant expiresAt = source.persistentCandidates() ? CANDIDATE_NEVER_EXPIRES : observedAt.plus(lifetime);
        List<CabinetCandidate> candidates = grouped.entrySet().stream()
                .map(entry -> toCandidate(source, entry.getKey(), entry.getValue(), snapshots,
                        observedAt, observedAt.plus(lifetime)))
                        observedAt, expiresAt))
                .toList();
        candidateStore.recordCabinets(candidates);
        if (detected.isEmpty()) markUnavailable(source.adapterCode());
@@ -185,16 +192,27 @@
            RealtimeSnapshot snapshot = snapshots.get(item.identity());
            DeviceWorkStatus status = snapshot == null || snapshot.workStatus() == null
                    ? DeviceWorkStatus.UNKNOWN : snapshot.workStatus();
            return new ChannelCandidate(item.identity(), item.available(), status, observedAt, snapshots.get(item.identity()));
            return new ChannelCandidate(item.identity(), item.available(), status, observedAt,
                    snapshots.get(item.identity()), item.lastDataAt());
        }).toList();
        Instant lastDataAt = channels.stream().map(DetectedChannel::lastDataAt)
                .filter(java.util.Objects::nonNull).max(Instant::compareTo).orElse(null);
        return new CabinetCandidate(source.adapterCode(), source.adapterId(),
                source.adapterCode() + "-" + key.devtype() + "-" + key.devid(),
                key.devtype(), key.devid(), ip, metadata.macAddress(), name,
                metadata.manufacturer(), metadata.model(), channelCandidates, observedAt, expiresAt);
                metadata.manufacturer(), metadata.model(), channelCandidates, observedAt, expiresAt, lastDataAt);
    }

    /** 发现来源只携带候选标识和过期计算所需配置，不复用数据库实体。 */
    public record DiscoverySource(String adapterCode, Long adapterId, String adapterName, int pollIntervalMs) { }
    public record DiscoverySource(String adapterCode, Long adapterId, String adapterName, int pollIntervalMs,
            boolean persistentCandidates)
    {
        /** 兼容按轮询过期计算候选的既有来源。 */
        public DiscoverySource(String adapterCode, Long adapterId, String adapterName, int pollIntervalMs)
        {
            this(adapterCode, adapterId, adapterName, pollIntervalMs, false);
        }
    }

    private record DiscoveryObservation(List<DetectedChannel> channels,
            Map<ChannelIdentity, RealtimeSnapshot> snapshots) { }

 battery-business/src/main/java/com/battery/business/service/impl/RoutingTestDataObservationAdapter.java

New file
@@ -0,0 +1,130 @@
package com.battery.business.service.impl;

import java.util.List;
import java.util.Map;
import java.util.Set;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Service;
import com.battery.device.api.TestDataObservationAdapter;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.model.DetectedChannel;
import com.battery.device.api.model.DeviceObservation;
import com.battery.device.api.registry.DeviceAdapterRegistry;

/**
 * 把取数请求按通道身份里的适配器编码送到它所属的来源实例。
 *
 * <p>这是持久化与实时展示路径上唯一的取数入口：同一个进程里可以注册多个来源实例，
 * 每个通道的请求只送到它自己的实例；实例缺失或未提供测试数据观测契约时明确失败，
 * 不回退到其他实例。发现不走这里，它按实例在 {@link DeviceDiscoveryPersistenceService} 中执行。
 *
 * <p>声明为 {@code @Primary}：来源实例可能自己也实现了观测契约（开发模拟器就是这样，
 * 它同时是 Spring Bean），业务侧注入的必须是这个按编码解析的入口，而不是某台设备的实例。
 */
@Service
@Primary
public class RoutingTestDataObservationAdapter implements TestDataObservationAdapter
{
    /** 聚合路由不是来源实例，这个编码只用于说明"按通道身份路由"。 */
    public static final String ADAPTER_CODE = "by-source";

    private final DeviceAdapterRegistry registry;

    public RoutingTestDataObservationAdapter(DeviceAdapterRegistry registry)
    {
        this.registry = registry;
    }

    @Override
    public String adapterCode()
    {
        return ADAPTER_CODE;
    }

    @Override
    public AdapterCapabilities capabilities()
    {
        return AdapterCapabilities.observeOnly();
    }

    /** 聚合适配器不承载发现；发现按来源实例执行，避免在这条路上猜一个设备。 */
    @Override
    public List<DetectedChannel> discoverChannels()
    {
        throw new UnsupportedOperationException("设备发现按来源实例执行，不在聚合适配器上执行");
    }

    /** 一次观测里的通道可能来自多个来源实例，逐个按各自的编码请求，不能按第一个通道决定全部。 */
    @Override
    public Map<ChannelIdentity, DeviceObservation> observe(Set<ChannelIdentity> channels)
    {
        if (channels == null || channels.isEmpty()) return Map.of();
        Map<ChannelIdentity, DeviceObservation> result = new java.util.LinkedHashMap<>();
        for (Map.Entry<String, List<ChannelIdentity>> group : groupBySource(channels).entrySet())
        {
            Set<ChannelIdentity> groupChannels = new java.util.LinkedHashSet<>(group.getValue());
            result.putAll(registry.executeTestData(group.getKey(), adapter -> adapter.observe(groupChannels)));
        }
        return Map.copyOf(result);
    }

    private static Map<String, List<ChannelIdentity>> groupBySource(Set<ChannelIdentity> channels)
    {
        Map<String, List<ChannelIdentity>> grouped = new java.util.LinkedHashMap<>();
        for (ChannelIdentity channel : channels)
            grouped.computeIfAbsent(channel.adapterCode(), ignored -> new java.util.ArrayList<>()).add(channel);
        return grouped;
    }

    @Override
    public CabinetRealtime latestCabinet(ChannelIdentity channel)
    {
        return registry.executeTestData(channel.adapterCode(), adapter -> adapter.latestCabinet(channel));
    }

    @Override
    public CollectorRealtime latestCollector(String adapterCode, String collectorCode)
    {
        // 采集器测量属于某个来源实例的辅助测量，按该实例请求；实例缺失时明确失败。
        // 来源编码为空说明测试电池连检测柜绑定都没有，此时来源未知：不猜也不回退，报无法请求。
        if (adapterCode == null || adapterCode.isBlank()) return null;
        return registry.executeTestData(adapterCode, adapter -> adapter.latestCollector(adapterCode, collectorCode));
    }

    @Override
    public List<StepRecord> stepRecords(ChannelIdentity channel)
    {
        return registry.executeTestData(channel.adapterCode(), adapter -> adapter.stepRecords(channel));
    }

    @Override
    public List<CycleRecord> cycleRecords(ChannelIdentity channel)
    {
        return registry.executeTestData(channel.adapterCode(), adapter -> adapter.cycleRecords(channel));
    }

    @Override
    public long persistenceIntervalMillis(ChannelIdentity channel)
    {
        return registry.executeTestData(channel.adapterCode(), adapter -> adapter.persistenceIntervalMillis(channel));
    }

    @Override
    public TestPlan testPlan(ChannelIdentity channel)
    {
        return registry.executeTestData(channel.adapterCode(), adapter -> adapter.testPlan(channel));
    }

    @Override
    public SampleWatermark currentWatermark(ChannelIdentity channel)
    {
        return registry.executeTestData(channel.adapterCode(), adapter -> adapter.currentWatermark(channel));
    }

    @Override
    public List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark watermark)
    {
        return registry.executeTestData(channel.adapterCode(), adapter -> adapter.samplesSince(channel, watermark));
    }
}

 battery-business/src/main/java/com/battery/business/service/impl/TDetectionCabinetServiceImpl.java

@@ -18,6 +18,7 @@
import com.battery.business.domain.TDetectionChannel;
import com.battery.business.domain.vo.DiscoveredCabinetVo;
import com.battery.business.mapper.TDetectionCabinetMapper;
import com.battery.business.service.IDeviceAdapterService;
import com.battery.business.service.ITDetectionCabinetService;
import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore;
import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore.CabinetCandidate;
@@ -39,6 +40,7 @@

    private final TDetectionCabinetMapper detectionCabinetMapper;
    private final DeviceDiscoveryCandidateStore candidateStore;
    private final IDeviceAdapterService adapterService;

    /**
     * 构造检测柜业务服务。
@@ -48,8 +50,42 @@
    public TDetectionCabinetServiceImpl(TDetectionCabinetMapper detectionCabinetMapper,
            DeviceDiscoveryCandidateStore candidateStore)
    {
        this(detectionCabinetMapper, candidateStore, null);
    }

    /**
     * 构造检测柜业务服务。
     *
     * @param detectionCabinetMapper 检测柜数据访问对象
     * @param candidateStore 发现候选存储
     * @param adapterService 适配器实例服务；为空时列表不展示数据来源名称
     */
    @org.springframework.beans.factory.annotation.Autowired
    public TDetectionCabinetServiceImpl(TDetectionCabinetMapper detectionCabinetMapper,
            DeviceDiscoveryCandidateStore candidateStore,
            @org.springframework.lang.Nullable IDeviceAdapterService adapterService)
    {
        this.detectionCabinetMapper = detectionCabinetMapper;
        this.candidateStore = candidateStore;
        this.adapterService = adapterService;
    }

    /** 数据来源名称按适配器编码解析；代码装配实例没有数据库记录时退回展示编码。 */
    private Map<String, String> adapterNames()
    {
        if (adapterService == null) return Map.of();
        return adapterService.selectList(new com.battery.business.domain.TDeviceAdapter()).stream()
                .filter(item -> item.getAdapterCode() != null && item.getAdapterName() != null)
                .collect(java.util.stream.Collectors.toMap(
                        com.battery.business.domain.TDeviceAdapter::getAdapterCode,
                        com.battery.business.domain.TDeviceAdapter::getAdapterName, (left, right) -> right));
    }

    /** 适配器编码可能缺失（历史登记），缺失时不查询来源名称。 */
    private void applyAdapterName(TDetectionCabinet cabinet, Map<String, String> adapterNames)
    {
        if (cabinet == null || cabinet.getAdapterCode() == null) return;
        cabinet.setAdapterName(adapterNames.get(cabinet.getAdapterCode()));
    }

    /**
@@ -66,6 +102,7 @@
        {
            throw new ServiceException("检测柜不存在或已删除");
        }
        applyAdapterName(detectionCabinet, adapterNames());
        return applyRealtimeState(detectionCabinet, candidatesByCode());
    }

@@ -87,8 +124,10 @@
        detectionCabinet.setOnlineStatus(null);
        detectionCabinet.setTestStatus(null);
        Map<String, CabinetCandidate> candidates = candidatesByCode();
        Map<String, String> adapterNames = adapterNames();
        return detectionCabinetMapper.selectDetectionCabinetList(detectionCabinet).stream()
                .map(item -> applyRealtimeState(item, candidates))
                .peek(item -> applyAdapterName(item, adapterNames))
                .filter(item -> StringUtils.isBlank(requestedOnlineStatus)
                        || requestedOnlineStatus.equals(item.getOnlineStatus()))
                .filter(item -> StringUtils.isBlank(requestedTestStatus)
@@ -97,8 +136,10 @@
    }

    /**
     * 读取未过期且未登记的候选，离线设备仍保留身份并可选择。
     * 发现只返回设备事实，不从通道工作状态推断系统测试关系。
     * 读取未登记且未删除的候选，离线设备仍保留身份并可选择。
     * 候选不按时间过期（厂商存储里的通道登记是长期事实），只按"已登记且未删除"排除；
     * 每条附带最近数据时间并按它倒序，便于操作员判断哪几台真的在产出数据。
     * 发现只返回设备事实，不从通道工作状态推断系统测试关系，也不自动判定"在用"。
     *
     * @return 可供选择的检测柜设备摘要
     */
@@ -110,7 +151,11 @@
                .filter(candidate -> !registeredCodes.contains(candidate.deviceCode()))
                .map(candidate -> new DiscoveredCabinetVo(
                            candidate.deviceCode(), candidate.ip(), candidate.mac(), candidate.name(),
                            candidate.manufacturer(), candidate.model(), candidate.channels().size(), candidate.online() ? "在线" : "离线"))
                            candidate.manufacturer(), candidate.model(), candidate.channels().size(),
                            candidate.online() ? "在线" : "离线", candidate.lastDataAt()))
                .sorted(java.util.Comparator.comparing(DiscoveredCabinetVo::lastDataAt,
                        java.util.Comparator.nullsLast(java.util.Comparator.reverseOrder()))
                        .thenComparing(DiscoveredCabinetVo::deviceCode))
                .toList();
    }


 battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java

@@ -191,6 +191,10 @@
    private Map<String, Object> snapshot(ChannelCandidate channel)
    {
        Map<String, Object> value = new LinkedHashMap<>();
        // 通道工作状态是独立设备事实：读库来源下不可观测，界面把它当正常取值展示而不是异常。
        value.put("workStatus", channel.workStatus() == null ? null : channel.workStatus().name());
        // 最近数据时间取来源里最近一次采样行的设备时间，与展示轮询节奏无关。
        value.put("lastDataAt", channel.lastDataAt() == null ? null : channel.lastDataAt().toEpochMilli());
        if (channel.latestSnapshot() == null) return value;
        value.put("source", "检测柜");
        value.put("observedAt", channel.latestSnapshot().receivedAt().toEpochMilli());
@@ -232,7 +236,8 @@
            return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(),
                    online ? "ONLINE" : "OFFLINE", Map.of());
        }
        var collector = adapter.latestCollector(collectorCode);
        var collector = adapter.latestCollector(
                battery.getBinding() == null ? null : battery.getBinding().getAdapterCode(), collectorCode);
        if (collector != null)
        {
            Map<String, Object> value = new LinkedHashMap<>();

 battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java

@@ -23,6 +23,7 @@
import com.battery.device.api.TestDataObservationAdapter.CycleRecord;
import com.battery.device.api.TestDataObservationAdapter.CollectorMeasurement;
import com.battery.device.api.TestDataObservationAdapter.CollectorRealtime;
import com.battery.device.api.TestDataObservationAdapter.SampleWatermark;
import com.battery.device.api.TestDataObservationAdapter.StepRecord;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.business.domain.TTest;
@@ -47,6 +48,9 @@
    private final Clock clock;
    private final Map<String, Instant> lastSnapshotAt = new java.util.concurrent.ConcurrentHashMap<>();
    private final Map<Long, Instant> lastRealtimeCheckAt = new java.util.concurrent.ConcurrentHashMap<>();
    /** 每个测试电池各自的接收水位；只存在内存中，重启后按来源里的最新一行重新取水位。 */
    private final Map<String, SampleCursor> sampleCursors = new java.util.concurrent.ConcurrentHashMap<>();


    @Autowired
    public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, @Nullable TestDataObservationAdapter adapter,
@@ -95,11 +99,127 @@

    private void saveRealtime(TTest test, TTestBattery battery)
    {
        // 柜端和采集器端分别判断成功；同一槽位只写一行，失败来源的身份、时间和指标保持为空。
        if (adapter == null) return;
        TTestChannelBinding binding = battery.getBinding(); if (binding == null || binding.getDevid() == null || binding.getChlid() == null) return;
        // 来源以设备数据行为接收单位时，柜端事实逐行落库（同秒多行都保留），槽位只留给采集器端。
        boolean rowBased = persistCabinetSamples(test, battery, binding);
        saveRealtimeSlot(test, battery, binding, !rowBased);
    }

    /**
     * 按来源水位逐行接收柜端事实。水位是"开始接收"或"继续接收"这一刻来源里的最新一行，
     * 因此水位之前的数据行即使仍在来源里也不进本测试，暂停期间的空档也不会被补回来。
     *
     * @return 该来源是否按设备数据行接收；来源不支持或被拒绝时返回 false，调用方回落到轮询槽位
     */
    private boolean persistCabinetSamples(TTest test, TTestBattery battery, TTestChannelBinding binding)
    {
        String key = test.getTestId() + ":" + battery.getTestBatteryId();
        Instant boundary = test.getReceptionBoundaryAtUtc() == null ? null : test.getReceptionBoundaryAtUtc().toInstant();
        SampleCursor cursor = sampleCursors.computeIfAbsent(key, ignored -> new SampleCursor());
        synchronized (cursor)
        {
            // 开始接收与继续接收都会推进接收边界：边界变化即重新取水位，暂停期间的数据不补采。
            if (!java.util.Objects.equals(cursor.boundary, boundary))
            {
                cursor.boundary = boundary; cursor.watermark = null; cursor.rowBased = null; cursor.sampledAt = null;
            }
            if (Boolean.FALSE.equals(cursor.rowBased)) return false;
            ChannelIdentity channel = channelIdentity(binding);
            Instant now = clock.instant();
            long interval = persistenceInterval(binding);
            // 厂商写库节奏是 60 秒，取数不必更快；展示节奏由控制台轮询单独承担。
            if (cursor.sampledAt != null && now.isBefore(cursor.sampledAt.plusMillis(interval)))
                return Boolean.TRUE.equals(cursor.rowBased);
            try
            {
                if (cursor.watermark == null)
                {
                    cursor.sampledAt = now;
                    SampleWatermark watermark = adapter.currentWatermark(channel);
                    if (watermark == null)
                    {
                        // 来源不按设备数据行接收（例如开发模拟器）：柜端事实继续走轮询槽位。
                        cursor.rowBased = false;
                        return false;
                    }
                    cursor.watermark = watermark;
                    cursor.rowBased = true;
                    return true;
                }
                List<CabinetRealtime> rows = adapter.samplesSince(channel, cursor.watermark);
                cursor.sampledAt = now;
                for (CabinetRealtime row : rows)
                {
                    insertCabinetSample(test, battery, binding, row);
                    cursor.watermark = new SampleWatermark(row.deviceTime(), row.sequenceId());
                }
                return true;
            }
            catch (RuntimeException exception)
            {
                // 实例缺失、来源不可达或读取失败：明确失败并留痕，下一轮重试；不回退到其他实例。
                log.warn("柜端逐行接收失败 testId={}, bindingId={}, adapterCode={}", test.getTestId(),
                        binding.getBindingId(), binding.getAdapterCode(), exception);
                cursor.sampledAt = now;
                return Boolean.TRUE.equals(cursor.rowBased);
            }
        }
    }

    /**
     * 把一行设备事实写成一条实时记录；来源行身份保证重复接收幂等、同身份不同内容报冲突。
     * 归属边界由水位在设备时间里判定（见 {@link #persistCabinetSamples}），
     * 这里不用服务器取数时间反推设备时间。
     */
    private void insertCabinetSample(TTest test, TTestBattery battery, TTestChannelBinding binding, CabinetRealtime row)
    {
        if (row.deviceTime() == null || row.serverReceivedAt() == null) return;
        Date observedAt = Date.from(row.serverReceivedAt());
        evaluateCabinet(test, binding, "VOLTAGE", BigDecimal.valueOf(row.voltageV()), "V", observedAt);
        evaluateCabinet(test, binding, "CURRENT", BigDecimal.valueOf(row.currentA()), "A", observedAt);
        evaluateCabinet(test, binding, "CAPACITY", BigDecimal.valueOf(row.capacityAh()), "Ah", observedAt);
        evaluateCabinet(test, binding, "ENERGY", BigDecimal.valueOf(row.energyWh()), "Wh", observedAt);
        String identity = sourceIdentity(binding);
        String recordId = row.sourceRecordId() == null || row.sourceRecordId().isBlank()
                ? "cabinet:" + row.sequenceId() : row.sourceRecordId();
        TTestRealtimeRecord value = new TTestRealtimeRecord();
        value.setTestId(test.getTestId()); value.setTestBatteryId(battery.getTestBatteryId());
        value.setBindingId(binding.getBindingId()); value.setAdapterCode(binding.getAdapterCode());
        // 设备数据行不是系统轮询槽位，槽位留空：同秒多行因此可以各自成立（ADR-0051 的槽位可空机制）。
        value.setCaptureSlotUtc(null); value.setServerReceivedAtUtc(observedAt);
        value.setCabinetDevtype(binding.getDevtype()); value.setCabinetDevid(binding.getDevid());
        value.setCabinetSubdevid(binding.getSubdevid()); value.setCabinetChlid(binding.getChlid());
        value.setCabinetAdapterTimezoneId(row.timezoneId());
        value.setCabinetSequenceId(row.sequenceId());
        value.setCabinetDeviceTimeUtc(Date.from(row.deviceTime()));
        value.setCabinetDeviceTimeRaw(row.deviceTimeRaw());
        value.setCycleNo(row.cycleNo()); value.setStepNo(row.stepNo()); value.setStepType(row.stepType());
        value.setStepName(row.stepName());
        value.setVoltageV(BigDecimal.valueOf(row.voltageV())); value.setCurrentA(BigDecimal.valueOf(row.currentA()));
        value.setCapacityAh(BigDecimal.valueOf(row.capacityAh())); value.setEnergyWh(BigDecimal.valueOf(row.energyWh()));
        Map<String, Object> extension = new LinkedHashMap<>();
        if (row.relativeTimeSeconds() != null) extension.put("relativeTimeSeconds", row.relativeTimeSeconds());
        value.setExtensionMeasurementsJson(JSON.toJSONString(extension));
        value.setSourceIdentity(identity); value.setSourceRecordId(recordId);
        value.setContentSha256(hash(value));
        // 重复接收的同内容记录幂等忽略；同身份但内容不同时保留原事实，不覆盖。
        String existing = mapper.selectRealtimeRecordHashBySourceRecord(battery.getTestBatteryId(), identity, recordId);
        if (existing != null)
        {
            if (!existing.equalsIgnoreCase(HexFormat.of().formatHex(value.getContentSha256())))
                log.warn("实时记录内容冲突，拒绝覆盖 testId={}, testBatteryId={}, sourceRecordId={}",
                        test.getTestId(), battery.getTestBatteryId(), recordId);
            return;
        }
        mapper.insertRealtimeRecord(value);
    }

    private void saveRealtimeSlot(TTest test, TTestBattery battery, TTestChannelBinding binding, boolean includeCabinet)
    {
        // 柜端和采集器端分别判断成功；同一槽位只写一行，失败来源的身份、时间和指标保持为空。
        Instant now = clock.instant(); long interval = persistenceInterval(binding); String key = test.getTestId() + ":" + battery.getTestBatteryId();
        CabinetRealtime cabinet = latestCabinet(binding);
        CabinetRealtime cabinet = includeCabinet ? latestCabinet(binding) : null;
        if (cabinet != null && !accepted(test, cabinet.deviceTime(), cabinet.serverReceivedAt())) cabinet = null;
        CollectorRealtime collector = latestCollector(battery);
        Map<String, CollectorMeasurement> mapped = mappedCollectorMetrics(test, battery, collector);
@@ -192,8 +312,9 @@
        }
    }

    private CabinetRealtime latestCabinet(TTestChannelBinding b) { try { return adapter.latestCabinet(new ChannelIdentity(b.getAdapterCode(), b.getDevtype(), b.getDevid(), b.getSubdevid(), b.getChlid())); } catch (RuntimeException ex) { return null; } }
    private CollectorRealtime latestCollector(TTestBattery b) { if (b.getCollectorId() == null) return null; try { return adapter.latestCollector(mapper.selectCollectorCode(b.getCollectorId())); } catch (RuntimeException ex) { return null; } }
    private CabinetRealtime latestCabinet(TTestChannelBinding b) { try { return adapter.latestCabinet(channelIdentity(b)); } catch (RuntimeException ex) { return null; } }
    private static ChannelIdentity channelIdentity(TTestChannelBinding b) { return new ChannelIdentity(b.getAdapterCode(), b.getDevtype(), b.getDevid(), b.getSubdevid(), b.getChlid()); }
    private CollectorRealtime latestCollector(TTestBattery b) { if (b.getCollectorId() == null || b.getBinding() == null) return null; try { return adapter.latestCollector(b.getBinding().getAdapterCode(), mapper.selectCollectorCode(b.getCollectorId())); } catch (RuntimeException ex) { return null; } }
    private Map<String, CollectorMeasurement> mappedCollectorMetrics(TTest test, TTestBattery battery, CollectorRealtime source)
    {
        Map<String, CollectorMeasurement> result = new LinkedHashMap<>();
@@ -222,7 +343,7 @@
                .ifPresent(sequence -> parts.add("collector:" + sequence));
        return String.join(":", parts);
    }
    private long persistenceInterval(TTestChannelBinding b) { try { return Math.max(100L, adapter.persistenceIntervalMillis(new ChannelIdentity(b.getAdapterCode(), b.getDevtype(), b.getDevid(), b.getSubdevid(), b.getChlid()))); } catch (RuntimeException ex) { return 60000L; } }
    private long persistenceInterval(TTestChannelBinding b) { try { return Math.max(100L, adapter.persistenceIntervalMillis(channelIdentity(b))); } catch (RuntimeException ex) { return 60000L; } }
    private static String sourceIdentity(TTestChannelBinding b) { return String.join(":", String.valueOf(b.getAdapterCode()), String.valueOf(b.getDevtype()), String.valueOf(b.getDevid()), String.valueOf(b.getSubdevid()), String.valueOf(b.getChlid())); }
    private static String timezone(TTestChannelBinding b) { return b.getAdapterTimezoneId() == null || b.getAdapterTimezoneId().isBlank() ? "Asia/Hong_Kong" : b.getAdapterTimezoneId(); }
    private static String blankToNull(String value) { return value == null || value.isBlank() ? null : value; }
@@ -241,4 +362,13 @@
        try { return MessageDigest.getInstance("SHA-256").digest(JSON.toJSONBytes(content)); }
        catch (Exception ex) { throw new IllegalStateException("无法计算测试数据内容摘要", ex); }
    }

    /** 单个测试电池的逐行接收状态：接收边界、来源水位、是否按行接收与上次取数时间。 */
    private static final class SampleCursor
    {
        private Instant boundary;
        private SampleWatermark watermark;
        private Boolean rowBased;
        private Instant sampledAt;
    }
}

 battery-business/src/main/resources/mapper/battery/TDeviceAdapterMapper.xml

@@ -11,6 +11,7 @@
        <result property="pollIntervalMs" column="poll_interval_ms"/>
        <result property="timezoneId" column="timezone_id"/>
        <result property="capabilities" column="capabilities"/>
        <result property="sourceConfig" column="source_config"/>
        <result property="readOnly" column="read_only"/>
        <result property="status" column="status"/>
        <result property="connectionStatus" column="connection_status"/>
@@ -28,7 +29,7 @@

    <sql id="adapterColumns">
        adapter_id, adapter_code, adapter_name, adapter_type, host, port, poll_interval_ms, timezone_id,
        capabilities, (json_extract(capabilities, '$.control') = false) read_only, status,
        capabilities, source_config, (json_extract(capabilities, '$.control') = false) read_only, status,
        connection_status, last_error_code, last_error_message, last_attempt_at_utc, last_observed_at_utc,
        create_by, create_time, update_by, update_time, remark, del_flag
    </sql>
@@ -61,27 +62,29 @@
    <!-- 应用装配层提供托管配置；持久化层仅负责按稳定编码幂等恢复。 -->
    <insert id="upsertManagedDeviceAdapter" parameterType="TDeviceAdapter">
        insert into t_device_adapter(adapter_code, adapter_name, adapter_type, host, port, poll_interval_ms,
            timezone_id, capabilities, status, connection_status, create_by, create_time, update_by, update_time,
            timezone_id, capabilities, source_config, status, connection_status, create_by, create_time, update_by, update_time,
            remark, del_flag)
        values(#{adapterCode}, #{adapterName}, #{adapterType}, #{host}, #{port}, #{pollIntervalMs}, #{timezoneId},
            #{capabilities}, '0', 'OFFLINE', 'system', sysdate(), 'system', sysdate(), #{remark}, 0)
            #{capabilities}, cast(#{sourceConfig} as json), '0', 'OFFLINE', 'system', sysdate(), 'system', sysdate(), #{remark}, 0)
        on duplicate key update adapter_name=values(adapter_name), adapter_type=values(adapter_type),
            host=values(host), port=values(port), poll_interval_ms=values(poll_interval_ms),
            timezone_id=values(timezone_id), capabilities=values(capabilities), status='0',
            timezone_id=values(timezone_id), capabilities=values(capabilities), source_config=values(source_config),
            status='0',
            connection_status='OFFLINE', last_error_code=null, last_error_message=null,
            update_by='system', update_time=sysdate(), remark=values(remark), del_flag=0
    </insert>

    <insert id="insertDeviceAdapter" parameterType="TDeviceAdapter" useGeneratedKeys="true" keyProperty="adapterId">
        insert into t_device_adapter(adapter_code, adapter_name, adapter_type, host, port, poll_interval_ms,
            timezone_id, capabilities, status, connection_status, create_by, create_time, remark, del_flag)
            timezone_id, capabilities, source_config, status, connection_status, create_by, create_time, remark, del_flag)
        values(#{adapterCode}, #{adapterName}, #{adapterType}, #{host}, #{port}, #{pollIntervalMs},
            #{timezoneId}, #{capabilities}, #{status}, 'OFFLINE', #{createBy}, sysdate(), #{remark}, 0)
            #{timezoneId}, #{capabilities}, cast(#{sourceConfig} as json), #{status}, 'OFFLINE', #{createBy}, sysdate(), #{remark}, 0)
    </insert>

    <update id="updateDeviceAdapter" parameterType="TDeviceAdapter">
        update t_device_adapter set adapter_name=#{adapterName}, adapter_type=#{adapterType}, host=#{host}, port=#{port},
            poll_interval_ms=#{pollIntervalMs}, timezone_id=#{timezoneId}, capabilities=#{capabilities},
            source_config=cast(#{sourceConfig} as json),
            connection_status='OFFLINE', last_error_code=null, last_error_message=null,
            update_by=#{updateBy}, update_time=sysdate()
        where adapter_id=#{adapterId} and del_flag=0

 battery-business/src/main/resources/mapper/battery/TTestMapper.xml

@@ -382,6 +382,12 @@
        where test_id = #{testId} and test_battery_id = #{testBatteryId} and capture_slot_utc = #{captureSlotUtc}
        limit 1
    </select>
    <select id="selectRealtimeRecordHashBySourceRecord" resultType="string">
        select lower(hex(content_sha256)) from t_test_realtime_record
        where test_battery_id = #{testBatteryId} and source_identity = #{sourceIdentity}
          and source_record_id = #{sourceRecordId}
        limit 1
    </select>
    <insert id="insertStepRecord" parameterType="com.battery.business.domain.TTestStepRecord" useGeneratedKeys="true" keyProperty="stepRecordId">
        insert ignore into t_test_step_record
        (test_id, test_battery_id, binding_id, adapter_code, source_identity, source_record_id, dedupe_key, adapter_timezone_id,

 battery-device-api/src/main/java/com/battery/device/api/TestDataObservationAdapter.java

@@ -12,8 +12,12 @@
    /** 返回指定检测柜通道最近一次实时观测；没有成功观测时返回空值。 */
    CabinetRealtime latestCabinet(ChannelIdentity channel);

    /** 返回采集器独立实时快照；不得通过历史下载补采。 */
    CollectorRealtime latestCollector(String collectorCode);
    /**
     * 返回采集器独立实时快照；不得通过历史下载补采。
     * 采集器测量按"同一来源的辅助测量"表达（ADR-0064），所以必须指名是哪个来源实例，
     * 否则多实例下无法确定该向哪台机器请求。
     */
    CollectorRealtime latestCollector(String adapterCode, String collectorCode);

    /** 返回设备直接提供的工步统计，不从实时采样推导。 */
    List<StepRecord> stepRecords(ChannelIdentity channel);
@@ -27,14 +31,49 @@
    /** 返回设备计划方案；设备未提供时保持为空，不推导计划。 */
    default TestPlan testPlan(ChannelIdentity channel) { return null; }

    /**
     * 读取通道当前的接收水位。水位是来源里已存在的最新一行（设备绝对时间 + 协议序号），
     * 用于把"开始接收"和"继续接收"切在这一时刻之后：水位之前的数据行即使仍在来源里也不进本测试。
     *
     * <p>返回空值表示该来源不按设备数据行接收（例如由系统轮询产生槽位的来源），此时不启用逐行接收。
     * 设备时间为空的水位表示这一刻来源里没有任何可用行，之后出现的行全部属于本次接收。
     * 来源不可达时必须抛出异常，不能返回空值——空值会被误当作"来源不按行接收"。
     */
    default SampleWatermark currentWatermark(ChannelIdentity channel) { return null; }

    /**
     * 返回严格晚于水位的设备数据行，按设备时间与协议序号升序。
     * 同一绝对秒内的多行各自成立，调用方不得按秒合并。
     * 水位为空表示调用方尚未取得水位，此时返回空列表：不把来源里的设备历史并入本次接收。
     */
    default List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark watermark) { return List.of(); }

    /** 接收水位：设备绝对时间与协议序号的二元组；序号在同一厂商测试内有效。 */
    record SampleWatermark(Instant deviceTime, long sequenceId) { }

    record TestPlan(int currentCycleNo, int currentStepNo, int plannedCycleCount, List<PlanRow> rows) { }
    record PlanRow(String rowType, int stepNo, Integer cyclePosition, String stepName, double durationSeconds,
            Double voltageV, String executionStatus, Integer cycleCount, Long recordCount) { }

    /** 柜端成功观测；相对时长使用设备提供的秒数，缺失不按采样时间推算。 */
    /**
     * 柜端成功观测；相对时长使用设备提供的秒数，缺失不按采样时间推算。
     * {@code sourceRecordId} 是来源行的稳定身份（如厂商测试编号加协议序号），
     * 供幂等与冲突判定使用；来源不提供时保持为空，调用方退化为按协议序号识别。
     */
    record CabinetRealtime(long sequenceId, Instant deviceTime, double currentA, double voltageV,
            double capacityAh, double energyWh, int cycleNo, int stepNo, String stepName, String stepType,
            Instant serverReceivedAt, String deviceTimeRaw, String timezoneId, Long relativeTimeSeconds) { }
            Instant serverReceivedAt, String deviceTimeRaw, String timezoneId, Long relativeTimeSeconds,
            String sourceRecordId)
    {
        /** 兼容不携带来源行身份的既有适配器。 */
        public CabinetRealtime(long sequenceId, Instant deviceTime, double currentA, double voltageV,
                double capacityAh, double energyWh, int cycleNo, int stepNo, String stepName, String stepType,
                Instant serverReceivedAt, String deviceTimeRaw, String timezoneId, Long relativeTimeSeconds)
        {
            this(sequenceId, deviceTime, currentA, voltageV, capacityAh, energyWh, cycleNo, stepNo, stepName,
                    stepType, serverReceivedAt, deviceTimeRaw, timezoneId, relativeTimeSeconds, null);
        }
    }

    record CollectorRealtime(String collectorCode, boolean online, int sampleIntervalMs,
            List<CollectorMeasurement> measurements, String timezoneId)

 battery-device-api/src/main/java/com/battery/device/api/discovery/DeviceDiscoveryCandidateStore.java

@@ -43,22 +43,43 @@
        }
    }

    /**
     * 未登记的候选设备。{@code lastDataAt} 取该设备各通道最近的采样行时间，
     * 供操作员判断哪几台设备现在真的在产出数据；它不改变候选的取舍，登记范围由操作员决定。
     */
    record CabinetCandidate(String adapterCode, Long adapterId, String deviceCode, int devtype, int devid,
            String ip, String mac, String name, String manufacturer, String model,
            List<ChannelCandidate> channels, Instant observedAt, Instant expiresAt)
            List<ChannelCandidate> channels, Instant observedAt, Instant expiresAt, Instant lastDataAt)
    {
        public CabinetCandidate { channels = List.copyOf(channels); }
        public boolean online() { return channels.stream().anyMatch(ChannelCandidate::online); }

        /** 兼容不含最近数据时间的既有候选。 */
        public CabinetCandidate(String adapterCode, Long adapterId, String deviceCode, int devtype, int devid,
                String ip, String mac, String name, String manufacturer, String model,
                List<ChannelCandidate> channels, Instant observedAt, Instant expiresAt)
        {
            this(adapterCode, adapterId, deviceCode, devtype, devid, ip, mac, name, manufacturer, model,
                    channels, observedAt, expiresAt, null);
        }
    }

    /** 通道候选；{@code lastDataAt} 是该通道来源里最近一次采样行的设备时间。 */
    record ChannelCandidate(ChannelIdentity identity, boolean online, DeviceWorkStatus workStatus,
            Instant observedAt, RealtimeSnapshot latestSnapshot)
            Instant observedAt, RealtimeSnapshot latestSnapshot, Instant lastDataAt)
    {
        /** 兼容仅有状态的旧发现候选。 */
        public ChannelCandidate(ChannelIdentity identity, boolean online, DeviceWorkStatus workStatus,
                Instant observedAt)
        {
            this(identity, online, workStatus, observedAt, null);
            this(identity, online, workStatus, observedAt, null, null);
        }

        /** 兼容不含最近数据时间的既有候选。 */
        public ChannelCandidate(ChannelIdentity identity, boolean online, DeviceWorkStatus workStatus,
                Instant observedAt, RealtimeSnapshot latestSnapshot)
        {
            this(identity, online, workStatus, observedAt, latestSnapshot, null);
        }
    }


 battery-device-api/src/main/java/com/battery/device/api/model/DetectedChannel.java

@@ -1,10 +1,23 @@
package com.battery.device.api.model;

import java.time.Instant;

/**
 * 发现结果中的一个通道。
 * {@code lastDataAt} 是来源里最近一次采样行的设备时间；它是"最近是否有新采样行写入"这一条依据的取值，
 * 与设备结束时间或通道水位线列无关。来源不提供该事实时保持为空。
 */
public record DetectedChannel(ChannelIdentity identity, String serverIp, boolean available,
        DetectedDeviceMetadata deviceMetadata)
        DetectedDeviceMetadata deviceMetadata, Instant lastDataAt)
{
    public DetectedChannel(ChannelIdentity identity, String serverIp, boolean available)
    {
        this(identity, serverIp, available, DetectedDeviceMetadata.empty());
        this(identity, serverIp, available, DetectedDeviceMetadata.empty(), null);
    }

    public DetectedChannel(ChannelIdentity identity, String serverIp, boolean available,
            DetectedDeviceMetadata deviceMetadata)
    {
        this(identity, serverIp, available, deviceMetadata, null);
    }
}

 battery-device-api/src/main/java/com/battery/device/api/registry/AdapterInstanceConfig.java

@@ -2,6 +2,7 @@

import java.time.ZoneId;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import com.battery.device.api.model.AdapterCapabilities;

@@ -14,8 +15,12 @@
        int pollIntervalMs,
        String timezoneId,
        boolean enabled,
        AdapterCapabilities capabilities)
        AdapterCapabilities capabilities,
        Map<String, String> sourceConfig)
{
    /**
     * 来源形态配置按存储形态互不相交，只承载非敏感项；连接口令由环境变量或启动参数提供，不下沉到数据库。
     */
    public AdapterInstanceConfig
    {
        adapterCode = requireText(adapterCode, "adapterCode");
@@ -23,6 +28,7 @@
        host = requireText(host, "host");
        timezoneId = requireText(timezoneId, "timezoneId");
        capabilities = Objects.requireNonNull(capabilities, "capabilities");
        sourceConfig = sourceConfig == null ? Map.of() : Map.copyOf(sourceConfig);
        if (port < 1 || port > 65535) throw new IllegalArgumentException("port must be between 1 and 65535");
        if (pollIntervalMs < 1) throw new IllegalArgumentException("pollIntervalMs must be positive");
        ZoneId.of(timezoneId);
@@ -30,6 +36,19 @@
            throw new IllegalArgumentException("MVP adapter must support discovery and remain read-only");
    }

    /** 兼容不含来源形态配置的既有装配。 */
    public AdapterInstanceConfig(String adapterCode, String adapterType, String host, int port, int pollIntervalMs,
            String timezoneId, boolean enabled, AdapterCapabilities capabilities)
    {
        this(adapterCode, adapterType, host, port, pollIntervalMs, timezoneId, enabled, capabilities, Map.of());
    }

    /** 读取来源形态的非敏感配置项；缺少该项时返回空值，是否必需由适配器决定。 */
    public String sourceOption(String key)
    {
        return sourceConfig.get(key);
    }

    private static String requireText(String value, String field)
    {
        String normalized = Objects.requireNonNull(value, field).trim();

 battery-device-api/src/main/java/com/battery/device/api/registry/DeviceAdapterRegistry.java

@@ -8,6 +8,7 @@
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import com.battery.device.api.DeviceAdapter;
import com.battery.device.api.TestDataObservationAdapter;

/** 管理启用设备适配器的运行时实例及其会话生命周期。 */
public final class DeviceAdapterRegistry implements AutoCloseable
@@ -34,18 +35,36 @@
            return;
        }

        RegisteredAdapter previous;
        synchronized (this)
        {
            RegisteredAdapter current = adapters.get(config.adapterCode());
            if (current != null && current.config.equals(config)) return;

            DeviceAdapterFactory factory = factories.get(normalizeType(config.adapterType()));
            if (factory == null)
            if (!factories.containsKey(normalizeType(config.adapterType())))
                throw new AdapterRegistryException("Unsupported device adapter type: " + config.adapterType());
        }

            DeviceAdapter replacement = factory.create(config);
            previous = adapters.put(config.adapterCode(), new RegisteredAdapter(config, replacement));
        DeviceAdapterFactory factory = factories.get(normalizeType(config.adapterType()));
        register(config, factory.create(config));
    }

    /**
     * 注册已构造的运行时实例。代码装配实例与数据库配置实例共用同一份实例表，
     * 使发现与取数按同一适配器编码解析，避免同编码出现两个实例。
     */
    public void register(AdapterInstanceConfig config, DeviceAdapter adapter)
    {
        Objects.requireNonNull(config, "config");
        Objects.requireNonNull(adapter, "adapter");
        if (!config.enabled())
        {
            remove(config.adapterCode());
            return;
        }

        RegisteredAdapter previous;
        synchronized (this)
        {
            previous = adapters.put(config.adapterCode(), new RegisteredAdapter(config, adapter));
        }
        closeWhenIdle(previous);
    }
@@ -54,14 +73,7 @@
    public <T> T execute(String adapterCode, Function<DeviceAdapter, T> operation)
    {
        Objects.requireNonNull(operation, "operation");
        RegisteredAdapter registered;
        synchronized (this)
        {
            registered = adapters.get(adapterCode);
            if (registered == null)
                throw new AdapterRegistryException("Device adapter is not enabled: " + adapterCode);
            registered.operationLock.lock();
        }
        RegisteredAdapter registered = acquire(adapterCode);
        try
        {
            return operation.apply(registered.adapter);
@@ -72,6 +84,41 @@
        }
    }

    /**
     * 在实例级串行边界内按测试数据观测契约执行。
     * 实例缺失或未提供该契约时明确失败，不回退到其他实例。
     */
    public <T> T executeTestData(String adapterCode, Function<TestDataObservationAdapter, T> operation)
    {
        Objects.requireNonNull(operation, "operation");
        RegisteredAdapter registered = acquire(adapterCode);
        try
        {
            if (!(registered.adapter instanceof TestDataObservationAdapter observation))
                throw new AdapterRegistryException(
                        "Device adapter does not expose test data observation: " + adapterCode);
            return operation.apply(observation);
        }
        finally
        {
            registered.operationLock.unlock();
        }
    }

    /** 取用实例并持有其串行锁；调用方必须在 finally 中释放。 */
    private RegisteredAdapter acquire(String adapterCode)
    {
        RegisteredAdapter registered;
        synchronized (this)
        {
            registered = adapters.get(adapterCode);
            if (registered == null)
                throw new AdapterRegistryException("Device adapter is not enabled: " + adapterCode);
            registered.operationLock.lock();
        }
        return registered;
    }

    public synchronized boolean contains(String adapterCode)
    {
        return adapters.containsKey(adapterCode);

 battery-device-api/src/test/java/com/battery/device/api/registry/DeviceAdapterRegistryTest.java

@@ -1,5 +1,6 @@
package com.battery.device.api.registry;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -15,6 +16,7 @@
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.jupiter.api.Test;
import com.battery.device.api.DeviceAdapter;
import com.battery.device.api.TestDataObservationAdapter;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.model.DeviceControlCommand;
@@ -153,6 +155,139 @@
        }
    }

    /**
     * 多来源实例：取数按适配器编码解析到各自实例；实例缺失时明确失败，不回退到其他实例。
     */
    @Test
    void routesTestDataObservationToTheInstanceNamedByTheChannelAndFailsWithoutFallback()
    {
        try (DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of(new ObservableFactory())))
        {
            registry.register(observableConfig("source-a", 1000), new ObservableAdapter("source-a", 1000L));
            registry.register(observableConfig("source-b", 2000), new ObservableAdapter("source-b", 2000L));

            assertEquals(1000L, registry.executeTestData("source-a",
                    adapter -> adapter.latestCabinet(channel("source-a")).sequenceId()).longValue());
            assertEquals(2000L, registry.executeTestData("source-b",
                    adapter -> adapter.latestCabinet(channel("source-b")).sequenceId()).longValue());

            // 实例缺失或未提供扩展观测契约时明确失败，不能落到别的实例上。
            assertThrows(AdapterRegistryException.class,
                    () -> registry.executeTestData("source-c", adapter -> adapter.latestCabinet(channel("source-a"))));
            assertThrows(AdapterRegistryException.class,
                    () -> registry.executeTestData("lab-bts", adapter -> adapter.adapterCode()));
        }
    }

    /** 代码装配的实例与工厂实例共用同一份实例表，发现与取数因此按同一编码解析。 */
    @Test
    void codeAssembledInstanceIsVisibleToBothAccessors()
    {
        try (DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of()))
        {
            ObservableAdapter assembled = new ObservableAdapter("assembled", 42L);
            registry.register(observableConfig("assembled", 3000), assembled);

            assertSame(assembled, registry.execute("assembled", adapter -> adapter));
            assertSame(assembled, registry.executeTestData("assembled", adapter -> adapter));
        }
    }

    private static ChannelIdentity channel(String adapterCode)
    {
        return new ChannelIdentity(adapterCode, 22, 1, 1, 1);
    }

    private static AdapterInstanceConfig observableConfig(String adapterCode, int pollIntervalMs)
    {
        return new AdapterInstanceConfig(adapterCode, "NEWARE_MYSQL", "127.0.0.1", 3306, pollIntervalMs,
                "Asia/Hong_Kong", true, AdapterCapabilities.observeOnly());
    }

    private static final class ObservableFactory implements DeviceAdapterFactory
    {
        @Override
        public String adapterType()
        {
            return "NEWARE_MYSQL";
        }

        @Override
        public DeviceAdapter create(AdapterInstanceConfig config)
        {
            return new ObservableAdapter(config.adapterCode(), 0L);
        }
    }

    /** 同时实现发现与测试数据观测契约的最小实例，用于验证按编码路由。 */
    private static final class ObservableAdapter implements DeviceAdapter, TestDataObservationAdapter
    {
        private final String code;
        private final long sequenceId;

        private ObservableAdapter(String code, long sequenceId)
        {
            this.code = code;
            this.sequenceId = sequenceId;
        }

        @Override
        public String adapterCode()
        {
            return code;
        }

        @Override
        public AdapterCapabilities capabilities()
        {
            return AdapterCapabilities.observeOnly();
        }

        @Override
        public List<com.battery.device.api.model.DetectedChannel> discoverChannels()
        {
            return List.of();
        }

        @Override
        public java.util.Map<ChannelIdentity, com.battery.device.api.model.RealtimeSnapshot> inquire(
                java.util.Set<ChannelIdentity> channels)
        {
            return java.util.Map.of();
        }

        @Override
        public TestDataObservationAdapter.CabinetRealtime latestCabinet(ChannelIdentity channel)
        {
            return new TestDataObservationAdapter.CabinetRealtime(sequenceId, java.time.Instant.EPOCH, 0, 0, 0, 0,
                    0, 0, null, null, java.time.Instant.EPOCH, null, "Asia/Hong_Kong", null);
        }

        @Override
        public TestDataObservationAdapter.CollectorRealtime latestCollector(String adapterCode, String collectorCode)
        {
            return null;
        }

        @Override
        public List<TestDataObservationAdapter.StepRecord> stepRecords(ChannelIdentity channel)
        {
            return List.of();
        }

        @Override
        public List<TestDataObservationAdapter.CycleRecord> cycleRecords(ChannelIdentity channel)
        {
            return List.of();
        }

        @Override
        public long persistenceIntervalMillis(ChannelIdentity channel)
        {
            return 60000L;
        }
    }

    private static AdapterInstanceConfig config(String host, boolean enabled)
    {
        return new AdapterInstanceConfig("lab-bts", "BTS", host, 502, 5000,

 battery-device-neware/pom.xml

New file
@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent><groupId>com.ruoyi</groupId><artifactId>ruoyi</artifactId><version>3.9.2</version></parent>
    <groupId>com.battery</groupId><artifactId>battery-device-neware</artifactId>
    <description>新威上位机数据库的只读直连适配器</description>
    <dependencies>
        <dependency><groupId>com.battery</groupId><artifactId>battery-device-api</artifactId></dependency>
        <dependency><groupId>com.mysql</groupId><artifactId>mysql-connector-j</artifactId></dependency>
        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency>
    </dependencies>
</project>

 battery-device-neware/src/main/java/com/battery/device/neware/NewareMysqlAdapter.java

New file
@@ -0,0 +1,466 @@
package com.battery.device.neware;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.Clock;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder;
import java.time.temporal.ChronoField;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Pattern;
import javax.sql.DataSource;
import com.battery.device.api.DeviceAdapter;
import com.battery.device.api.DeviceAdapterException;
import com.battery.device.api.TestDataObservationAdapter;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.model.DetectedChannel;
import com.battery.device.api.model.DetectedDeviceMetadata;
import com.battery.device.api.model.DeviceConnectionStatus;
import com.battery.device.api.model.DeviceWorkStatus;
import com.battery.device.api.model.RealtimeSnapshot;
import com.battery.device.api.registry.AdapterInstanceConfig;

/**
 * 直连新威上位机 MySQL 库的只读适配器（存储形态 {@code NEWARE_MYSQL}）。
 *
 * <p>现场确认的取数事实（见 {@code docs/commissioning/现场对接清单.md}）：
 * <ul>
 *   <li>发现候选来自 {@code channel} 通道登记表，主键 {@code (dev_uid, unit_id, chl_id)}；
 *       设备号 {@code dev_uid = devtype × 10000 + devid}。登记表不等于"当前在用"，
 *       所以只把"最近数据时间"作为事实交出，登记哪些候选由操作员决定。</li>
 *   <li>逐点数据在 {@code data_<设备六位>_<年>_<月>_<日>} 分表里，表名由 {@code test} 表的
 *       {@code main_first_table} / {@code main_second_table} 直接给出，不按日期推算。</li>
 *   <li>最新测试按 {@code start_time} 取，不能按 {@code test_id} 排序（实测不单调）。</li>
 *   <li>活跃与否只看有没有新的采样行；{@code end_time} 是否为空与 8.0 的通道水位线列都不作为依据。</li>
 *   <li>厂商固定每 60 秒写一行，实时展示节奏与数据产生节奏天生分离。</li>
 *   <li>辅助表族实测没有数据，采集器辅助测量只能走导入，因此 {@link #latestCollector} 保持为空。</li>
 * </ul>
 *
 * <p>单位换算按实测公式：电压与电流除以 10000（原单位 0.1 mV / 0.1 mA），容量与能量除以 36000000
 * （原单位 1e-4 C / 1e-4 J）。工步类型保留设备给出的数字，不按电流符号猜测方向；
 * 内阻与温度的单位语义仍未确认，因此不对外暴露。
 */
public final class NewareMysqlAdapter implements DeviceAdapter, TestDataObservationAdapter
{
    public static final String ADAPTER_TYPE = "NEWARE_MYSQL";
    /** 连接口令的默认环境变量名；也可由实例配置的 passwordEnv 指定。 */
    public static final String DEFAULT_PASSWORD_ENV = "BATTERY_VENDOR_DB_PASSWORD";
    /** 厂商实测写库周期；它同时是持久化周期，展示节奏不随之加快。 */
    public static final long SAMPLE_INTERVAL_MS = 60000L;
    private static final int SAMPLE_BATCH_LIMIT = 500;
    private static final Pattern SAFE_TABLE_NAME = Pattern.compile("[A-Za-z0-9_]{1,64}");
    /** 设备时间按适配器时区解释；来源列是秒精度 datetime，小数部分按需接受。 */
    private static final DateTimeFormatter DEVICE_TIME = new DateTimeFormatterBuilder()
            .appendPattern("yyyy-MM-dd HH:mm:ss")
            .optionalStart().appendFraction(ChronoField.NANO_OF_SECOND, 1, 9, true).optionalEnd()
            .toFormatter(Locale.ROOT);
    /** 水位回写来源时使用秒精度文本，与来源列一致；同秒内的区分由协议序号承担。 */
    private static final DateTimeFormatter DEVICE_TIME_TEXT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss",
            Locale.ROOT);
    private static final BigDecimal VOLTAGE_CURRENT_DIVISOR = BigDecimal.valueOf(10000);
    private static final BigDecimal CAPACITY_ENERGY_DIVISOR = BigDecimal.valueOf(36000000);

    private final AdapterInstanceConfig config;
    private final DataSource dataSource;
    private final Runnable connectionReset;
    private final Clock clock;
    private final ZoneId deviceZone;
    /** 最近数据时间最快也要一分钟才可能变化，缓存使发现清单不必每轮重读全部通道。 */
    private final long lastDataCacheMillis;
    private final Map<String, CachedLastData> lastDataCache = new ConcurrentHashMap<>();

    public NewareMysqlAdapter(AdapterInstanceConfig config, DataSource dataSource)
    {
        this(config, dataSource, () -> { }, Clock.systemUTC(), SAMPLE_INTERVAL_MS);
    }

    public NewareMysqlAdapter(AdapterInstanceConfig config, DataSource dataSource, Runnable connectionReset,
            Clock clock, long lastDataCacheMillis)
    {
        this.config = config;
        this.dataSource = dataSource;
        this.connectionReset = connectionReset;
        this.clock = clock;
        this.deviceZone = ZoneId.of(config.timezoneId());
        this.lastDataCacheMillis = Math.max(0, lastDataCacheMillis);
    }

    /** 由工厂使用：把驱动连接包上只读白名单，并在查询失败后作废连接。 */
    public static NewareMysqlAdapter create(AdapterInstanceConfig config, SingleConnectionVendorDataSource source)
    {
        return new NewareMysqlAdapter(config, new ReadOnlyVendorDataSource(source, config.adapterCode()),
                source::invalidate, Clock.systemUTC(), SAMPLE_INTERVAL_MS);
    }

    @Override
    public String adapterCode()
    {
        return config.adapterCode();
    }

    @Override
    public AdapterCapabilities capabilities()
    {
        return config.capabilities();
    }

    /** 发现候选来自厂商库的通道登记表；每条候选附带该通道最近的采样行时间。 */
    @Override
    public List<DetectedChannel> discoverChannels()
    {
        List<ChannelKey> registered = query("select dev_uid, unit_id, chl_id from channel "
                + "order by dev_uid, unit_id, chl_id", this::readRegisteredChannels);
        Map<Long, List<ChannelKey>> byDevice = new LinkedHashMap<>();
        for (ChannelKey key : registered)
            byDevice.computeIfAbsent(key.devUid(), ignored -> new ArrayList<>()).add(key);
        List<DetectedChannel> channels = new ArrayList<>();
        Instant observedAt = clock.instant();
        for (Map.Entry<Long, List<ChannelKey>> entry : byDevice.entrySet())
        {
            Map<ChannelKey, VendorTest> latestTests = latestTests(entry.getKey());
            for (ChannelKey key : entry.getValue())
            {
                VendorTest latest = latestTests.get(key);
                Instant lastDataAt = latest == null ? null : cachedLastSampleAt(key, latest, observedAt);
                channels.add(new DetectedChannel(identity(key), config.host(), true,
                        DetectedDeviceMetadata.empty(), lastDataAt));
            }
        }
        return List.copyOf(channels);
    }

    /** 实时快照取自该通道最新测试的最新一行；没有可用行时不返回该通道，避免沿用过期观测。 */
    @Override
    public Map<ChannelIdentity, RealtimeSnapshot> inquire(Set<ChannelIdentity> channels)
    {
        if (channels == null || channels.isEmpty()) return Map.of();
        Map<ChannelIdentity, RealtimeSnapshot> result = new LinkedHashMap<>();
        for (ChannelIdentity channel : channels)
        {
            CabinetRealtime latest = latestCabinet(channel);
            if (latest == null) continue;
            result.put(channel, snapshot(channel, latest));
        }
        return Map.copyOf(result);
    }

    @Override
    public CabinetRealtime latestCabinet(ChannelIdentity channel)
    {
        ChannelKey key = channelKey(channel);
        VendorTest latest = latestTest(key);
        if (latest == null) return null;
        SampleRow row = lastSample(key, latest);
        return row == null ? null : cabinetRealtime(row);
    }

    /** 新威辅助表族实测没有数据，采集器辅助测量只能靠导入（ADR-0064）。 */
    @Override
    public CollectorRealtime latestCollector(String adapterCode, String collectorCode)
    {
        return null;
    }

    /** 工步层记录由本单之后的一张单实现，这里不返回从采样推导出来的工步。 */
    @Override
    public List<StepRecord> stepRecords(ChannelIdentity channel)
    {
        return List.of();
    }

    /** 循环汇总表不存在，聚合配方须先用真实数据核对，这里不返回推导值。 */
    @Override
    public List<CycleRecord> cycleRecords(ChannelIdentity channel)
    {
        return List.of();
    }

    /** 厂商固定 60 秒写库；持久化周期与之对齐，不做倍速。 */
    @Override
    public long persistenceIntervalMillis(ChannelIdentity channel)
    {
        return SAMPLE_INTERVAL_MS;
    }

    /** 水位是来源里已存在的最新一行：设备绝对时间与协议序号。 */
    @Override
    public SampleWatermark currentWatermark(ChannelIdentity channel)
    {
        ChannelKey key = channelKey(channel);
        VendorTest latest = latestTest(key);
        if (latest == null) return new SampleWatermark(null, 0L);
        SampleRow row = lastSample(key, latest);
        return row == null ? new SampleWatermark(null, 0L) : new SampleWatermark(row.deviceTime(), row.sequenceId());
    }

    /**
     * 严格晚于水位的设备数据行。同一绝对秒内的多行各自成立（实测工步切换点会在同一秒写入多行），
     * 所以按 {@code (test_atime, seq_id)} 过滤与排序，不按秒合并。
     *
     * <p>设备时间为空的水位表示边界时该通道在来源里没有任何行，之后出现的行全部属于本次接收。
     * 厂商侧测试编号不参与边界判定，只写进来源行身份；接收边界是本系统的接收边界。
     * 若测试期间分表滚动，两张表都会被读取，避免滚动点整段缺失。
     */
    @Override
    public List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark watermark)
    {
        if (watermark == null) return List.of();
        ChannelKey key = channelKey(channel);
        VendorTest latest = latestTest(key);
        if (latest == null) return List.of();
        boolean bounded = watermark.deviceTime() != null;
        String rawWatermark = bounded ? deviceTimeText(watermark.deviceTime()) : null;
        List<SampleRow> rows = new ArrayList<>();
        for (String table : latest.tables())
        {
            String predicate = bounded
                    ? "and (test_atime > ? or (test_atime = ? and seq_id > ?)) "
                    : "";
            Object[] parameters = bounded
                    ? new Object[] { key.unitId(), key.chlId(), rawWatermark, rawWatermark, watermark.sequenceId() }
                    : new Object[] { key.unitId(), key.chlId() };
            rows.addAll(query("select test_id, seq_id, cycle, step_id, step_type, test_time, test_vol, test_cur, "
                    + "test_capchg, test_engchg, test_capdchg, test_engdchg, test_atime from " + quote(table)
                    + " where unit_id = ? and chl_id = ? " + predicate
                    + "order by test_atime, seq_id limit " + SAMPLE_BATCH_LIMIT,
                    this::readSampleRows, parameters));
        }
        rows.sort(Comparator.comparing(SampleRow::deviceTime).thenComparingLong(SampleRow::sequenceId));
        return rows.stream().map(this::cabinetRealtime).toList();
    }

    // ------------------------------------------------------------------ 来源读取

    private List<ChannelKey> readRegisteredChannels(ResultSet rows) throws SQLException
    {
        List<ChannelKey> channels = new ArrayList<>();
        while (rows.next())
            channels.add(new ChannelKey(rows.getLong("dev_uid"), rows.getInt("unit_id"), rows.getInt("chl_id")));
        return channels;
    }

    /** 按起始时间取每台设备每个通道的最新测试；test_id 实测不单调，不能用来排序。 */
    private Map<ChannelKey, VendorTest> latestTests(long devUid)
    {
        return query("select unit_id, chl_id, test_id, main_first_table, main_second_table from test "
                + "where dev_uid = ? and start_time = (select max(start_time) from test newer "
                + "where newer.dev_uid = test.dev_uid and newer.unit_id = test.unit_id and newer.chl_id = test.chl_id) "
                + "order by unit_id, chl_id, test_id", rows -> {
                    Map<ChannelKey, VendorTest> result = new LinkedHashMap<>();
                    while (rows.next())
                    {
                        ChannelKey key = new ChannelKey(devUid, rows.getInt("unit_id"), rows.getInt("chl_id"));
                        result.putIfAbsent(key, vendorTest(rows.getLong("test_id"),
                                rows.getString("main_first_table"), rows.getString("main_second_table")));
                    }
                    return result;
                }, devUid);
    }

    private VendorTest latestTest(ChannelKey key)
    {
        List<VendorTest> tests = query("select test_id, main_first_table, main_second_table from test "
                + "where dev_uid = ? and unit_id = ? and chl_id = ? order by start_time desc limit 1",
                rows -> rows.next() ? List.of(vendorTest(rows.getLong("test_id"), rows.getString("main_first_table"),
                        rows.getString("main_second_table"))) : List.<VendorTest>of(),
                key.devUid(), key.unitId(), key.chlId());
        return tests.isEmpty() ? null : tests.get(0);
    }

    /** 测试数据可能跨分表滚动，两张表都要看；优先取第二张表，滚动后它就是最新的那张。 */
    private SampleRow lastSample(ChannelKey key, VendorTest test)
    {
        List<String> tables = new ArrayList<>(test.tables());
        Collections.reverse(tables);
        for (String table : tables)
        {
            List<SampleRow> rows = query("select test_id, seq_id, cycle, step_id, step_type, test_time, test_vol, "
                    + "test_cur, test_capchg, test_engchg, test_capdchg, test_engdchg, test_atime from " + quote(table)
                    + " where unit_id = ? and chl_id = ? and test_id = ? order by seq_id desc limit 1",
                    this::readSampleRows, key.unitId(), key.chlId(), test.testId());
            if (!rows.isEmpty()) return rows.get(0);
        }
        return null;
    }

    /** 最近数据时间来自采样行本身：没有新采样行就是没有，不看结束时间也不看通道水位线列。 */
    private Instant cachedLastSampleAt(ChannelKey key, VendorTest test, Instant now)
    {
        String cacheKey = key.devUid() + ":" + key.unitId() + ":" + key.chlId() + ":" + test.testId();
        CachedLastData cached = lastDataCache.get(cacheKey);
        if (cached != null && now.isBefore(cached.expiresAt())) return cached.lastDataAt();
        SampleRow row = lastSample(key, test);
        Instant lastDataAt = row == null ? null : row.deviceTime();
        if (lastDataCacheMillis > 0)
            lastDataCache.put(cacheKey, new CachedLastData(lastDataAt, now.plusMillis(lastDataCacheMillis)));
        return lastDataAt;
    }

    private List<SampleRow> readSampleRows(ResultSet rows) throws SQLException
    {
        List<SampleRow> result = new ArrayList<>();
        while (rows.next())
        {
            String raw = rows.getString("test_atime");
            LocalDateTime deviceTime = parseDeviceTime(raw);
            result.add(new SampleRow(rows.getLong("test_id"), rows.getLong("seq_id"), rows.getInt("cycle"),
                    rows.getInt("step_id"), rows.getString("step_type"), rows.getLong("test_time"),
                    rows.getBigDecimal("test_vol"), rows.getBigDecimal("test_cur"), rows.getBigDecimal("test_capchg"),
                    rows.getBigDecimal("test_engchg"), rows.getBigDecimal("test_capdchg"),
                    rows.getBigDecimal("test_engdchg"), raw, deviceTime.atZone(deviceZone).toInstant()));
        }
        return result;
    }

    private CabinetRealtime cabinetRealtime(SampleRow row)
    {
        // 充放电容量按电流方向取对应的一组：设备界面对正在放电的通道显示放电容量。
        boolean charging = row.current().signum() >= 0;
        BigDecimal capacity = charging ? row.capacityCharge() : row.capacityDischarge();
        BigDecimal energy = charging ? row.energyCharge() : row.energyDischarge();
        return new CabinetRealtime(row.sequenceId(), row.deviceTime(), scaled(row.current(), VOLTAGE_CURRENT_DIVISOR),
                scaled(row.voltage(), VOLTAGE_CURRENT_DIVISOR), scaled(capacity, CAPACITY_ENERGY_DIVISOR),
                scaled(energy, CAPACITY_ENERGY_DIVISOR), row.cycle(), row.stepId(), null, row.stepType(),
                clock.instant(), row.deviceTimeRaw(), deviceZone.getId(),
                row.testTimeMillis() == null ? null : row.testTimeMillis() / 1000,
                "neware:" + row.testId() + ":" + row.sequenceId());
    }

    private RealtimeSnapshot snapshot(ChannelIdentity channel, CabinetRealtime latest)
    {
        // 读库路线下通道工作状态不可观测（ADR-0063）：不按数据行推断，也不沿用上一次状态。
        return new RealtimeSnapshot(channel, DeviceWorkStatus.UNOBSERVABLE, latest.cycleNo(), latest.stepNo(),
                latest.stepType(), null, BigDecimal.valueOf(latest.currentA()), BigDecimal.valueOf(latest.voltageV()),
                BigDecimal.valueOf(latest.capacityAh()), BigDecimal.valueOf(latest.energyWh()), null,
                latest.relativeTimeSeconds(), null, latest.serverReceivedAt(), Map.of(),
                DeviceConnectionStatus.ONLINE, latest.deviceTime(), List.of(), null);
    }

    private static double scaled(BigDecimal value, BigDecimal divisor)
    {
        if (value == null) return 0d;
        // 保留 12 位小数与业务列精度一致，避免在换算环节提前丢失设备原始分辨率。
        return value.divide(divisor, 12, RoundingMode.HALF_UP).doubleValue();
    }

    private ChannelIdentity identity(ChannelKey key)
    {
        // 设备号编码 devtype × 10000 + devid；unit_id 是该柜内的单元，作为 subdevid 保留。
        return new ChannelIdentity(config.adapterCode(), (int) (key.devUid() / 10000),
                (int) (key.devUid() % 10000), key.unitId(), key.chlId());
    }

    private ChannelKey channelKey(ChannelIdentity channel)
    {
        if (!config.adapterCode().equals(channel.adapterCode()))
            throw new DeviceAdapterException("通道身份不属于本实例：" + channel.adapterCode());
        return new ChannelKey(channel.devtype() * 10000L + channel.devid(), channel.subdevid(), channel.chlid());
    }

    private static VendorTest vendorTest(long testId, String firstTable, String secondTable)
    {
        List<String> tables = new ArrayList<>(2);
        if (isSafeTable(firstTable)) tables.add(firstTable.trim());
        if (isSafeTable(secondTable) && !tables.contains(secondTable.trim())) tables.add(secondTable.trim());
        return new VendorTest(testId, tables);
    }

    private static boolean isSafeTable(String table)
    {
        return table != null && SAFE_TABLE_NAME.matcher(table.trim()).matches();
    }

    private static String quote(String table)
    {
        // 表名来自厂商库自身的 test 记录，必须先按白名单校验再拼接，避免把库内容当 SQL。
        if (!isSafeTable(table)) throw new DeviceAdapterException("来源表名不合法：" + table);
        return "`" + table.trim() + "`";
    }

    private LocalDateTime parseDeviceTime(String raw)
    {
        if (raw == null || raw.isBlank())
            throw new DeviceAdapterException("PARSE_FAILED", "设备时间为空，无法解释为设备事实", null);
        String normalized = raw.trim().replace('T', ' ');
        if (normalized.endsWith("Z")) normalized = normalized.substring(0, normalized.length() - 1).trim();
        try
        {
            return LocalDateTime.parse(normalized, DEVICE_TIME);
        }
        catch (RuntimeException exception)
        {
            throw new DeviceAdapterException("PARSE_FAILED", "设备时间无法解释：" + raw, exception);
        }
    }

    private String deviceTimeText(Instant instant)
    {
        return instant.atZone(deviceZone).toLocalDateTime().format(DEVICE_TIME_TEXT);
    }

    // ------------------------------------------------------------------ JDBC

    /**
     * 厂商数据源交出的是长期复用的一条连接，这里只关闭语句与结果集，不关闭连接本身；
     * 查询失败时作废连接，下一轮取数重新建立。
     */
    private <T> T query(String sql, ResultSetReader<T> reader, Object... parameters)
    {
        try
        {
            Connection connection = dataSource.getConnection();
            try (PreparedStatement statement = connection.prepareStatement(sql))
            {
                for (int index = 0; index < parameters.length; index++)
                    statement.setObject(index + 1, parameters[index]);
                try (ResultSet rows = statement.executeQuery())
                {
                    return reader.read(rows);
                }
            }
        }
        catch (NonQueryStatementException exception)
        {
            throw exception;
        }
        catch (SQLException exception)
        {
            connectionReset.run();
            throw new DeviceAdapterException("CONNECTION_FAILED",
                    "厂商库读取失败：" + config.host() + ":" + config.port() + "，原因：" + exception.getMessage(),
                    exception);
        }
    }

    private interface ResultSetReader<T>
    {
        T read(ResultSet rows) throws SQLException;
    }

    private record ChannelKey(long devUid, int unitId, int chlId) { }

    private record VendorTest(long testId, List<String> tables) { }

    private record CachedLastData(Instant lastDataAt, Instant expiresAt) { }

    private record SampleRow(long testId, long sequenceId, int cycle, int stepId, String stepType,
            Long testTimeMillis, BigDecimal voltage, BigDecimal current, BigDecimal capacityCharge,
            BigDecimal energyCharge, BigDecimal capacityDischarge, BigDecimal energyDischarge,
            String deviceTimeRaw, Instant deviceTime) { }
}

 battery-device-neware/src/main/java/com/battery/device/neware/NewareMysqlAdapterFactory.java

New file
@@ -0,0 +1,78 @@
package com.battery.device.neware;

import java.util.Map;
import com.battery.device.api.DeviceAdapter;
import com.battery.device.api.DeviceAdapterException;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.registry.AdapterInstanceConfig;
import com.battery.device.api.registry.DeviceAdapterFactory;

/**
 * 按存储形态 {@code NEWARE_MYSQL} 构造直连厂商库的只读适配器实例。
 *
 * <p>连接口令只从环境变量或启动参数读取（变量名由实例配置的 {@code passwordEnv} 指定，
 * 缺省为 {@link NewareMysqlAdapter#DEFAULT_PASSWORD_ENV}），不进数据库也不进仓库。
 * 账号固定为现场取得的服务端只读账号，全权限账号不进入配置。
 */
public final class NewareMysqlAdapterFactory implements DeviceAdapterFactory
{
    public static final String DATABASE_OPTION = "database";
    public static final String USER_OPTION = "user";
    public static final String PASSWORD_ENV_OPTION = "passwordEnv";
    private static final String DEFAULT_DATABASE = "bts63";
    private static final String READ_ONLY_USER = "visitor";

    @Override
    public String adapterType()
    {
        return NewareMysqlAdapter.ADAPTER_TYPE;
    }

    @Override
    public DeviceAdapter create(AdapterInstanceConfig config)
    {
        String database = option(config.sourceConfig(), DATABASE_OPTION, DEFAULT_DATABASE);
        String user = option(config.sourceConfig(), USER_OPTION, READ_ONLY_USER);
        String password = credential(option(config.sourceConfig(), PASSWORD_ENV_OPTION,
                NewareMysqlAdapter.DEFAULT_PASSWORD_ENV));
        SingleConnectionVendorDataSource source = new SingleConnectionVendorDataSource(
                jdbcUrl(config.host(), config.port(), database), user, password);
        return NewareMysqlAdapter.create(config, source);
    }

    /** 只读账号口令；启动参数优先于环境变量，便于本机排查而不必改环境。 */
    static String credential(String variableName)
    {
        if (variableName == null || variableName.isBlank())
            throw new DeviceAdapterException("未配置厂商库口令的取值变量名");
        String value = System.getenv(variableName);
        if (value == null || value.isBlank()) value = System.getProperty(variableName);
        if (value == null || value.isBlank())
            throw new DeviceAdapterException("缺少厂商库连接口令：请设置环境变量 " + variableName);
        return value;
    }

    /**
     * MySQL 5.5 那台不支持任何会话级只读机制，因此 URL 中不得出现 transaction_read_only
     * （ADR-0062）；只读由服务端账号与语句白名单两层保证。
     */
    static String jdbcUrl(String host, int port, String database)
    {
        return "jdbc:mysql://" + host + ":" + port + "/" + database
                + "?connectTimeout=5000&socketTimeout=10000&useSSL=false&allowPublicKeyRetrieval=true"
                + "&zeroDateTimeBehavior=convertToNull&connectionTimeZone=Asia/Hong_Kong"
                + "&forceConnectionTimeZoneToSession=false";
    }

    private static String option(Map<String, String> sourceConfig, String key, String fallback)
    {
        String value = sourceConfig.get(key);
        return value == null || value.isBlank() ? fallback : value.trim();
    }

    /** 供装配层声明该形态的运行时能力：可发现、可观测实时值、只读。 */
    public static AdapterCapabilities capabilities()
    {
        return AdapterCapabilities.observeOnly();
    }
}

 battery-device-neware/src/main/java/com/battery/device/neware/NonQueryStatementException.java

New file
@@ -0,0 +1,12 @@
package com.battery.device.neware;

/** 非查询语句在到达驱动之前被拒绝；抛出时不会产生任何服务端效果。 */
public class NonQueryStatementException extends RuntimeException
{
    private static final long serialVersionUID = 1L;

    public NonQueryStatementException(String message)
    {
        super(message);
    }
}

 battery-device-neware/src/main/java/com/battery/device/neware/ReadOnlyStatementAudit.java

New file
@@ -0,0 +1,45 @@
package com.battery.device.neware;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * 厂商数据源上的语句审计：记录已转发到驱动的语句和被白名单挡下的语句。
 *
 * <p>它让"该数据源从未发出过非查询语句"成为可断言的事实，而不是纪律性约定；
 * 断言对应的测试见 {@code ReadOnlyVendorDataSourceTest}。
 */
public final class ReadOnlyStatementAudit
{
    private final List<String> forwardedStatements = new CopyOnWriteArrayList<>();
    private final List<String> rejectedStatements = new CopyOnWriteArrayList<>();

    void recordForwarded(String statement)
    {
        forwardedStatements.add(abbreviate(statement));
    }

    void recordRejected(String statement)
    {
        rejectedStatements.add(abbreviate(statement));
    }

    /** 已被转发到驱动的语句。 */
    public List<String> forwardedStatements()
    {
        return List.copyOf(forwardedStatements);
    }

    /** 被白名单挡下、从未到达驱动的语句。 */
    public List<String> rejectedStatements()
    {
        return List.copyOf(rejectedStatements);
    }

    private static String abbreviate(String statement)
    {
        if (statement == null) return "";
        String trimmed = statement.trim();
        return trimmed.length() <= 200 ? trimmed : trimmed.substring(0, 200);
    }
}

 battery-device-neware/src/main/java/com/battery/device/neware/ReadOnlyVendorDataSource.java

New file
@@ -0,0 +1,237 @@
package com.battery.device.neware;

import java.io.PrintWriter;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.sql.CallableStatement;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.SQLFeatureNotSupportedException;
import java.sql.Statement;
import java.util.List;
import java.util.Locale;
import java.util.logging.Logger;
import javax.sql.DataSource;

/**
 * 厂商数据源的只读防线：在 JDBC 调用到达驱动之前按语句白名单放行。
 *
 * <p>只允许 {@code SELECT} / {@code SHOW} / {@code DESCRIBE}（含 {@code DESC}）；
 * 其余语句（含存储过程调用）直接抛 {@link NonQueryStatementException}，不转发给驱动，
 * 因此在服务端不产生任何效果。被拒绝的语句记入 {@link ReadOnlyStatementAudit}，
 * 使"该数据源从未发出过非查询语句"可以被测试断言。
 *
 * <p>防线是结构性的，不是数学上的：任何绕过本数据源另开连接的代码仍能写。
 * 服务端只读账号（见现场对接清单第七节）是第一道防线，本层是第二道。
 */
public final class ReadOnlyVendorDataSource implements DataSource
{
    private static final List<String> ALLOWED_PREFIXES = List.of("select", "show", "describe", "desc");

    private final DataSource delegate;
    private final String sourceLabel;
    private final ReadOnlyStatementAudit audit = new ReadOnlyStatementAudit();

    public ReadOnlyVendorDataSource(DataSource delegate, String sourceLabel)
    {
        if (delegate == null) throw new IllegalArgumentException("delegate data source must not be null");
        this.delegate = delegate;
        this.sourceLabel = sourceLabel == null || sourceLabel.isBlank() ? "vendor" : sourceLabel.trim();
    }

    /** 返回本数据源上的语句审计，用于断言从未发出过非查询语句。 */
    public ReadOnlyStatementAudit audit()
    {
        return audit;
    }

    @Override
    public Connection getConnection() throws SQLException
    {
        return guard(delegate.getConnection());
    }

    @Override
    public Connection getConnection(String username, String password) throws SQLException
    {
        return guard(delegate.getConnection(username, password));
    }

    private Connection guard(Connection connection)
    {
        return (Connection) Proxy.newProxyInstance(connection.getClass().getClassLoader(),
                new Class<?>[] { Connection.class }, (proxy, method, args) -> {
                    switch (method.getName())
                    {
                        case "prepareStatement" ->
                        {
                            String sql = (String) args[0];
                            requireQuery(sql);
                            return guardStatement((Statement) invoke(connection, method, args), sql, true);
                        }
                        case "prepareCall" -> throw new NonQueryStatementException(
                                "厂商数据源禁止存储过程调用：" + abbreviate((String) args[0]));
                        case "createStatement" ->
                        {
                            return guardStatement((Statement) invoke(connection, method, args), null, false);
                        }
                        default -> { }
                    }
                    return invoke(connection, method, args);
                });
    }

    private Statement guardStatement(Statement statement, String preparedSql, boolean prepared)
    {
        Class<?> statementType = prepared ? PreparedStatement.class : Statement.class;
        return (Statement) Proxy.newProxyInstance(statement.getClass().getClassLoader(),
                new Class<?>[] { statementType }, (proxy, method, args) -> {
                    String sql = sqlArgument(method, args);
                    if (sql != null) requireQuery(sql);
                    return invoke(statement, method, args);
                });
    }

    /**
     * 取出本次调用携带的 SQL 文本：语句级方法带 SQL 时按该 SQL 判定，
     * 预编译语句的无参执行按 prepare 时的 SQL 判定（已在 prepareStatement 中放行）。
     */
    private static String sqlArgument(Method method, Object[] args)
    {
        if (args == null || args.length == 0 || !(args[0] instanceof String sql)) return null;
        return switch (method.getName())
        {
            case "execute", "executeUpdate", "executeLargeUpdate", "executeQuery", "addBatch", "executeBatch",
                    "executeLargeBatch" -> sql;
            default -> null;
        };
    }

    private void requireQuery(String sql)
    {
        if (isQuery(sql))
        {
            audit.recordForwarded(sql);
            return;
        }
        audit.recordRejected(sql);
        throw new NonQueryStatementException("厂商数据源只允许查询语句：" + abbreviate(sql));
    }

    /** 只按语句首关键字判定，避免用正则近似 SQL 语法。 */
    static boolean isQuery(String sql)
    {
        if (sql == null) return false;
        String normalized = stripLeadingComments(sql).toLowerCase(Locale.ROOT);
        if (normalized.isEmpty()) return false;
        for (String prefix : ALLOWED_PREFIXES)
        {
            if (!normalized.startsWith(prefix)) continue;
            int boundary = prefix.length();
            if (normalized.length() == boundary || Character.isWhitespace(normalized.charAt(boundary))
                    || normalized.charAt(boundary) == '(' || normalized.charAt(boundary) == '*'
                    || normalized.charAt(boundary) == '`')
                return true;
        }
        return false;
    }

    private static String stripLeadingComments(String sql)
    {
        String remaining = sql.trim();
        while (true)
        {
            if (remaining.startsWith("/*"))
            {
                int end = remaining.indexOf("*/");
                if (end < 0) return remaining;
                remaining = remaining.substring(end + 2).trim();
                continue;
            }
            if (remaining.startsWith("--") || remaining.startsWith("#"))
            {
                int end = remaining.indexOf('\n');
                if (end < 0) return "";
                remaining = remaining.substring(end + 1).trim();
                continue;
            }
            return remaining;
        }
    }

    private static String abbreviate(String sql)
    {
        if (sql == null) return "";
        String trimmed = sql.trim();
        return trimmed.length() <= 200 ? trimmed : trimmed.substring(0, 200);
    }

    private static Object invoke(Object target, Method method, Object[] args) throws Throwable
    {
        try
        {
            return method.invoke(target, args);
        }
        catch (InvocationTargetException exception)
        {
            throw exception.getTargetException();
        }
    }

    @Override
    public PrintWriter getLogWriter() throws SQLException
    {
        return delegate.getLogWriter();
    }

    @Override
    public void setLogWriter(PrintWriter out) throws SQLException
    {
        delegate.setLogWriter(out);
    }

    @Override
    public void setLoginTimeout(int seconds) throws SQLException
    {
        delegate.setLoginTimeout(seconds);
    }

    @Override
    public int getLoginTimeout() throws SQLException
    {
        return delegate.getLoginTimeout();
    }

    @Override
    public Logger getParentLogger() throws SQLFeatureNotSupportedException
    {
        return delegate.getParentLogger();
    }

    @Override
    public <T> T unwrap(Class<T> iface) throws SQLException
    {
        if (iface.isInstance(this)) return iface.cast(this);
        throw new SQLFeatureNotSupportedException("厂商数据源不暴露未受约束的驱动对象");
    }

    @Override
    public boolean isWrapperFor(Class<?> iface)
    {
        return iface.isInstance(this);
    }

    @Override
    public String toString()
    {
        return "ReadOnlyVendorDataSource[" + sourceLabel + "]";
    }

    /** 便于测试与排查的来源说明。 */
    public String sourceLabel()
    {
        return sourceLabel;
    }
}

 battery-device-neware/src/main/java/com/battery/device/neware/SingleConnectionVendorDataSource.java

New file
@@ -0,0 +1,110 @@
package com.battery.device.neware;

import java.io.PrintWriter;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.SQLFeatureNotSupportedException;
import java.util.Properties;
import java.util.logging.Logger;
import javax.sql.DataSource;

/**
 * 由 DriverManager 直连厂商库的极简数据源：只被设备适配器使用，不注册进 Spring，
 * 因此不参与任何业务事务，Druid 主数据源的回滚不会波及厂商库。
 *
 * <p>连接复用一条并在查询失败时作废，避免厂商服务端重启后长期持有失效连接；
 * 厂商库实测按固定 60 秒写库，取数频率远低于连接复用带来的收益门槛。
 */
public final class SingleConnectionVendorDataSource implements DataSource
{
    private final String url;
    private final Properties credentials;
    private Connection connection;

    public SingleConnectionVendorDataSource(String url, String user, String password)
    {
        if (url == null || url.isBlank()) throw new IllegalArgumentException("vendor jdbc url must not be blank");
        this.url = url;
        this.credentials = new Properties();
        this.credentials.setProperty("user", user == null ? "" : user);
        this.credentials.setProperty("password", password == null ? "" : password);
    }

    @Override
    public synchronized Connection getConnection() throws SQLException
    {
        if (connection == null || connection.isClosed())
        {
            connection = DriverManager.getConnection(url, credentials);
        }
        return connection;
    }

    @Override
    public Connection getConnection(String username, String password) throws SQLException
    {
        return DriverManager.getConnection(url, username, password);
    }

    /** 作废当前连接，使下一次取数重新建立；用于查询失败后的自愈。 */
    public synchronized void invalidate()
    {
        if (connection == null) return;
        try
        {
            connection.close();
        }
        catch (SQLException ignored)
        {
            // 连接已失效，无需处理。
        }
        finally
        {
            connection = null;
        }
    }

    @Override
    public PrintWriter getLogWriter() throws SQLException
    {
        return DriverManager.getLogWriter();
    }

    @Override
    public void setLogWriter(PrintWriter out) throws SQLException
    {
        DriverManager.setLogWriter(out);
    }

    @Override
    public void setLoginTimeout(int seconds) throws SQLException
    {
        DriverManager.setLoginTimeout(seconds);
    }

    @Override
    public int getLoginTimeout() throws SQLException
    {
        return DriverManager.getLoginTimeout();
    }

    @Override
    public Logger getParentLogger() throws SQLFeatureNotSupportedException
    {
        return Logger.getLogger("com.battery.device.neware");
    }

    @Override
    public <T> T unwrap(Class<T> iface) throws SQLException
    {
        if (iface.isInstance(this)) return iface.cast(this);
        throw new SQLFeatureNotSupportedException("不支持解包为 " + iface);
    }

    @Override
    public boolean isWrapperFor(Class<?> iface)
    {
        return iface.isInstance(this);
    }
}

 battery-device-neware/src/test/java/com/battery/device/neware/FakeVendorDatabase.java

New file
@@ -0,0 +1,333 @@
package com.battery.device.neware;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import javax.sql.DataSource;

/**
 * 可控的假厂商库：按新威 7.6 实测表结构（{@code channel} / {@code test} / {@code data_*}）回答适配器的查询，
 * 并把每一条到达"驱动"的语句原样记录下来。
 *
 * <p>它只回答适配器实际使用的查询形态，因此测试断言的是外部行为（发现清单、最新测试、分表定位、
 * 单位换算、水位过滤），不是 SQL 文本本身；记录下来的语句用于断言只读防线没有漏过非查询语句。
 */
final class FakeVendorDatabase
{
    record ChannelRow(long devUid, int unitId, int chlId) { }

    record TestRow(long devUid, int unitId, int chlId, long testId, String startTime,
            String firstTable, String secondTable) { }

    record SampleRow(String table, long testId, int unitId, int chlId, long seqId, int cycle, int stepId,
            String stepType, long testTimeMillis, int voltage, int current, int capacityCharge, int energyCharge,
            int capacityDischarge, int energyDischarge, String testAtime) { }

    private final List<String> statements = new java.util.concurrent.CopyOnWriteArrayList<>();
    private final List<ChannelRow> channels = new ArrayList<>();
    private final List<TestRow> tests = new ArrayList<>();
    private final List<SampleRow> samples = new ArrayList<>();

    FakeVendorDatabase registeredChannel(long devUid, int unitId, int chlId)
    {
        channels.add(new ChannelRow(devUid, unitId, chlId));
        return this;
    }

    FakeVendorDatabase test(TestRow row)
    {
        tests.add(row);
        return this;
    }

    FakeVendorDatabase sample(SampleRow row)
    {
        samples.add(row);
        return this;
    }

    /** 到达"驱动"的语句原文，按调用顺序。 */
    List<String> statements()
    {
        return List.copyOf(statements);
    }

    DataSource dataSource()
    {
        return new DataSource()
        {
            @Override
            public Connection getConnection()
            {
                return connection();
            }

            @Override
            public Connection getConnection(String username, String password)
            {
                return connection();
            }

            @Override
            public java.io.PrintWriter getLogWriter()
            {
                return null;
            }

            @Override
            public void setLogWriter(java.io.PrintWriter out)
            {
            }

            @Override
            public void setLoginTimeout(int seconds)
            {
            }

            @Override
            public int getLoginTimeout()
            {
                return 0;
            }

            @Override
            public java.util.logging.Logger getParentLogger()
            {
                return java.util.logging.Logger.getLogger("fake-vendor");
            }

            @Override
            public <T> T unwrap(Class<T> iface) throws SQLException
            {
                throw new SQLException("不支持解包");
            }

            @Override
            public boolean isWrapperFor(Class<?> iface)
            {
                return false;
            }
        };
    }

    private Connection connection()
    {
        return (Connection) Proxy.newProxyInstance(getClass().getClassLoader(), new Class<?>[] { Connection.class },
                (proxy, method, args) -> {
                    if ("prepareStatement".equals(method.getName())) return statement((String) args[0]);
                    if ("createStatement".equals(method.getName())) return plainStatement();
                    return defaultValue(method);
                });
    }

    /** 语句级执行：SQL 在执行时才出现，因此在这里才记录；探测类语句不回数据。 */
    private java.sql.Statement plainStatement()
    {
        return (java.sql.Statement) Proxy.newProxyInstance(getClass().getClassLoader(),
                new Class<?>[] { java.sql.Statement.class }, (proxy, method, args) -> {
                    if ("executeQuery".equals(method.getName()) || "execute".equals(method.getName())
                            || "executeUpdate".equals(method.getName()))
                    {
                        statements.add((String) args[0]);
                        return resultSet(List.of());
                    }
                    return defaultValue(method);
                });
    }

    private PreparedStatement statement(String sql)
    {
        statements.add(sql);
        List<Object> parameters = new ArrayList<>();
        return (PreparedStatement) Proxy.newProxyInstance(getClass().getClassLoader(),
                new Class<?>[] { PreparedStatement.class }, (proxy, method, args) -> {
                    switch (method.getName())
                    {
                        case "setObject", "setString", "setLong", "setInt" ->
                        {
                            int index = (Integer) args[0];
                            while (parameters.size() < index) parameters.add(null);
                            parameters.set(index - 1, args[1]);
                            return null;
                        }
                        case "executeQuery" -> { return resultSet(resolve(sql, parameters)); }
                        case "close" -> { return null; }
                        default -> { return defaultValue(method); }
                    }
                });
    }

    /** 按适配器使用的查询形态分派；参数顺序与适配器绑定顺序一致。 */
    private List<Map<String, Object>> resolve(String sql, List<Object> parameters)
    {
        String normalized = sql.toLowerCase().replaceAll("\\s+", " ");
        if (normalized.contains("from channel"))
            return channels.stream().sorted(Comparator.comparingLong(ChannelRow::devUid)
                    .thenComparingInt(ChannelRow::unitId).thenComparingInt(ChannelRow::chlId))
                    .map(row -> {
                        Map<String, Object> value = new LinkedHashMap<>();
                        value.put("dev_uid", row.devUid());
                        value.put("unit_id", row.unitId());
                        value.put("chl_id", row.chlId());
                        return value;
                    })
                    .toList();
        if (normalized.contains("from test") && normalized.contains("max(start_time)"))
            return latestTests(longValue(parameters, 0));
        if (normalized.contains("from test"))
            return latestTestForChannel(longValue(parameters, 0), intValue(parameters, 1), intValue(parameters, 2));
        if (normalized.contains("order by test_atime, seq_id limit"))
            return samplesAfter(tableOf(sql), parameters);
        return lastSampleOfTest(tableOf(sql), intValue(parameters, 0), intValue(parameters, 1),
                longValue(parameters, 2));
    }

    private List<Map<String, Object>> latestTests(long devUid)
    {
        Map<String, TestRow> latest = new LinkedHashMap<>();
        tests.stream().filter(row -> row.devUid() == devUid)
                .sorted(Comparator.comparing(TestRow::startTime).reversed().thenComparingLong(TestRow::testId))
                .forEach(row -> latest.putIfAbsent(row.unitId() + ":" + row.chlId(), row));
        return latest.values().stream().sorted(Comparator.comparingInt(TestRow::unitId)
                .thenComparingInt(TestRow::chlId)).map(this::testColumns).toList();
    }

    private List<Map<String, Object>> latestTestForChannel(long devUid, int unitId, int chlId)
    {
        return tests.stream().filter(row -> row.devUid() == devUid && row.unitId() == unitId && row.chlId() == chlId)
                .sorted(Comparator.comparing(TestRow::startTime).reversed().thenComparingLong(TestRow::testId))
                .limit(1).map(this::testColumns).toList();
    }

    private List<Map<String, Object>> lastSampleOfTest(String table, int unitId, int chlId, long testId)
    {
        return samples.stream().filter(row -> row.table().equals(table) && row.unitId() == unitId
                        && row.chlId() == chlId && row.testId() == testId)
                .sorted(Comparator.comparingLong(SampleRow::seqId).reversed()).limit(1)
                .map(this::sampleColumns).toList();
    }

    private List<Map<String, Object>> samplesAfter(String table, List<Object> parameters)
    {
        int unitId = intValue(parameters, 0), chlId = intValue(parameters, 1);
        // 没有带时间的下界时（水位表示界时来源里没有行）返回该通道的全部行。
        boolean bounded = parameters.size() >= 5;
        String watermark = bounded ? String.valueOf(parameters.get(2)) : null;
        long watermarkSeq = bounded ? longValue(parameters, 4) : 0L;
        return samples.stream().filter(row -> row.table().equals(table) && row.unitId() == unitId
                        && row.chlId() == chlId
                        && (!bounded || after(row.testAtime(), row.seqId(), watermark, watermarkSeq)))
                .sorted(Comparator.comparing(SampleRow::testAtime).thenComparingLong(SampleRow::seqId))
                .map(this::sampleColumns).toList();
    }

    private static boolean after(String atime, long seqId, String watermark, long watermarkSeq)
    {
        int time = atime.compareTo(watermark);
        return time > 0 || (time == 0 && seqId > watermarkSeq);
    }

    private static String tableOf(String sql)
    {
        java.util.regex.Matcher matcher = java.util.regex.Pattern.compile("from\\s+`([A-Za-z0-9_]+)`")
                .matcher(sql.toLowerCase());
        if (!matcher.find()) throw new IllegalStateException("假厂商库无法识别分表名：" + sql);
        return matcher.group(1);
    }

    private Map<String, Object> testColumns(TestRow row)
    {
        Map<String, Object> value = new LinkedHashMap<>();
        value.put("unit_id", row.unitId());
        value.put("chl_id", row.chlId());
        value.put("test_id", row.testId());
        value.put("main_first_table", row.firstTable());
        value.put("main_second_table", row.secondTable());
        return value;
    }

    private Map<String, Object> sampleColumns(SampleRow row)
    {
        Map<String, Object> value = new LinkedHashMap<>();
        value.put("test_id", row.testId());
        value.put("seq_id", row.seqId());
        value.put("cycle", row.cycle());
        value.put("step_id", row.stepId());
        value.put("step_type", row.stepType());
        value.put("test_time", row.testTimeMillis());
        value.put("test_vol", row.voltage());
        value.put("test_cur", row.current());
        value.put("test_capchg", row.capacityCharge());
        value.put("test_engchg", row.energyCharge());
        value.put("test_capdchg", row.capacityDischarge());
        value.put("test_engdchg", row.energyDischarge());
        value.put("test_atime", row.testAtime());
        return value;
    }

    private static ResultSet resultSet(List<Map<String, Object>> rows)
    {
        List<String> labels = rows.isEmpty() ? List.of() : List.copyOf(rows.get(0).keySet());
        int[] cursor = { -1 };
        return (ResultSet) Proxy.newProxyInstance(FakeVendorDatabase.class.getClassLoader(),
                new Class<?>[] { ResultSet.class }, (proxy, method, args) -> {
                    switch (method.getName())
                    {
                        case "next" -> { cursor[0]++; return cursor[0] < rows.size(); }
                        case "close" -> { return null; }
                        case "getLong", "getInt", "getString", "getBigDecimal" ->
                        {
                            Object value = value(rows.get(cursor[0]), labels, args[0]);
                            if (value == null) return null;
                            return switch (method.getName())
                            {
                                case "getLong" -> ((Number) value).longValue();
                                case "getInt" -> ((Number) value).intValue();
                                case "getString" -> String.valueOf(value);
                                default -> value instanceof BigDecimal decimal ? decimal
                                        : BigDecimal.valueOf(((Number) value).longValue());
                            };
                        }
                        default -> { return defaultValue(method); }
                    }
                });
    }

    private static Object value(Map<String, Object> row, List<String> labels, Object key)
    {
        if (key instanceof Integer index) return row.get(labels.get(index - 1));
        return row.get(((String) key).toLowerCase());
    }

    private static int intValue(List<Object> parameters, int index)
    {
        return ((Number) parameters.get(index)).intValue();
    }

    private static long longValue(List<Object> parameters, int index)
    {
        return ((Number) parameters.get(index)).longValue();
    }

    private static Object defaultValue(Method method)
    {
        Class<?> type = method.getReturnType();
        if (!type.isPrimitive() || type == void.class) return null;
        if (type == boolean.class) return false;
        if (type == long.class) return 0L;
        if (type == double.class) return 0d;
        if (type == float.class) return 0f;
        if (type == char.class) return (char) 0;
        return 0;
    }
}

 battery-device-neware/src/test/java/com/battery/device/neware/NewareMysqlAdapterContractTest.java

New file
@@ -0,0 +1,207 @@
package com.battery.device.neware;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.time.Clock;
import java.time.Instant;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.junit.jupiter.api.Test;
import com.battery.device.api.TestDataObservationAdapter.CabinetRealtime;
import com.battery.device.api.TestDataObservationAdapter.SampleWatermark;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.model.DetectedChannel;
import com.battery.device.api.model.DeviceConnectionStatus;
import com.battery.device.api.model.DeviceWorkStatus;
import com.battery.device.api.model.RealtimeSnapshot;
import com.battery.device.api.registry.AdapterInstanceConfig;

/**
 * 直连新威库的适配器契约测试：用可控的假厂商库验证发现清单与最近数据时间、
 * 最新测试按起始时间取、分表定位用测试记录给出的表名、活跃只看新采样行、单位换算、身份与水位。
 */
class NewareMysqlAdapterContractTest
{
    private static final String ADAPTER_CODE = "neware76";
    private static final long DEV_UID = 210036L;
    private static final ChannelIdentity CHANNEL = new ChannelIdentity(ADAPTER_CODE, 21, 36, 1, 2);
    private static final Clock CLOCK = Clock.fixed(Instant.parse("2026-09-28T02:00:00Z"), ZoneOffset.UTC);

    private static final String ACTIVE_TABLE = "data_210036_2026_09_18";

    private NewareMysqlAdapter adapter(FakeVendorDatabase database)
    {
        AdapterInstanceConfig config = new AdapterInstanceConfig(ADAPTER_CODE,
                NewareMysqlAdapter.ADAPTER_TYPE, "26.1.2.3", 3306, 60000, "Asia/Hong_Kong", true,
                AdapterCapabilities.observeOnly(), Map.of("database", "bts63", "user", "visitor"));
        // 缓存关闭，让每个断言都读到假厂商库的当前内容。
        return new NewareMysqlAdapter(config, database.dataSource(), () -> { }, CLOCK, 0);
    }

    /** 一台在用设备（dev_uid 210036）、一台只出现在历史数据里但不在登记表中的设备（240098）。 */
    private FakeVendorDatabase activeDeviceDatabase()
    {
        return new FakeVendorDatabase()
                .registeredChannel(DEV_UID, 1, 1)
                .registeredChannel(DEV_UID, 1, 2)
                .registeredChannel(DEV_UID, 2, 1)
                .test(new FakeVendorDatabase.TestRow(DEV_UID, 1, 2, 440L, "2026-09-18 09:00:00",
                        ACTIVE_TABLE, null))
                // 已退役设备的历史数据：登记表里没有它，不得成为候选。
                .test(new FakeVendorDatabase.TestRow(240098L, 1, 1, 900L, "2021-09-30 09:00:00",
                        "data_240098_2021_09_30", null))
                .sample(new FakeVendorDatabase.SampleRow(ACTIVE_TABLE, 440L, 1, 2, 7435L, 3, 12, "2", 300000L,
                        13733, -11889, 50350024, 1103331, 50350024, 1103331, "2026-09-28 09:54:56"))
                .sample(new FakeVendorDatabase.SampleRow(ACTIVE_TABLE, 440L, 1, 2, 7436L, 3, 9, "4", 0L,
                        13700, 0, 0, 0, 0, 0, "2026-09-28 09:54:56"));
    }

    @Test
    void discoveryListsRegisteredChannelsWithLastDataTimeOnlyFromSampleRows()
    {
        List<DetectedChannel> channels = adapter(activeDeviceDatabase()).discoverChannels();
        assertEquals(3, channels.size());
        // 候选集合来自通道登记表：只出现在历史数据里的退役设备不出现。
        assertTrue(channels.stream().allMatch(item -> item.identity().devid() == 36));
        DetectedChannel active = channels.stream()
                .filter(item -> item.identity().chlid() == 2).findFirst().orElseThrow();
        assertEquals(new ChannelIdentity(ADAPTER_CODE, 21, 36, 1, 2), active.identity());
        assertTrue(active.available());
        // 最近数据时间取该通道最新采样行的设备时间（按适配器时区解释）：Asia/Hong_Kong 09:54:56 = 01:54:56Z。
        assertEquals(Instant.parse("2026-09-28T01:54:56Z"), active.lastDataAt());
        // 没有采样行的通道不报最近数据时间：活跃只看有没有新采样行写入。
        DetectedChannel idle = channels.stream()
                .filter(item -> item.identity().chlid() == 1 && item.identity().subdevid() == 1)
                .findFirst().orElseThrow();
        assertNull(idle.lastDataAt());
    }

    @Test
    void latestTestIsPickedByStartTimeNotByTestId()
    {
        FakeVendorDatabase database = new FakeVendorDatabase()
                .registeredChannel(DEV_UID, 1, 2)
                // 测试编号更大的那条起始时间更早：按 test_id 排序会取错。
                .test(new FakeVendorDatabase.TestRow(DEV_UID, 1, 2, 2818575907L, "2026-02-20 09:00:00",
                        "data_210036_2026_02_20", null))
                .test(new FakeVendorDatabase.TestRow(DEV_UID, 1, 2, 2818575540L, "2026-09-28 09:00:00",
                        "data_210036_2026_09_28", null))
                .sample(new FakeVendorDatabase.SampleRow("data_210036_2026_02_20", 2818575907L, 1, 2, 10L, 1, 1, "1",
                        1000L, 5000, 1000, 10, 10, 10, 10, "2026-02-20 09:00:01"))
                .sample(new FakeVendorDatabase.SampleRow("data_210036_2026_09_28", 2818575540L, 1, 2, 20L, 4, 7, "2",
                        60000L, 31541, -12000, 700000, 900000, 800000, 1000000, "2026-09-28 09:01:00"));
        CabinetRealtime latest = adapter(database).latestCabinet(CHANNEL);
        assertEquals(20L, latest.sequenceId());
        assertEquals(Instant.parse("2026-09-28T01:01:00Z"), latest.deviceTime());
    }

    @Test
    void shardTableComesFromTheTestRecordInsteadOfTheDeviceTimeDate()
    {
        // 实测表名按建表日：9-28 启动的测试写进 9-18 的表。适配器必须用测试记录给出的表名。
        CabinetRealtime latest = adapter(activeDeviceDatabase()).latestCabinet(CHANNEL);
        assertEquals("2026-09-28 09:54:56", latest.deviceTimeRaw());
        assertEquals(Instant.parse("2026-09-28T01:54:56Z"), latest.deviceTime());
        assertEquals("Asia/Hong_Kong", latest.timezoneId());
        assertEquals(CLOCK.instant(), latest.serverReceivedAt());
    }

    @Test
    void sampledValuesAreConvertedWithTheFieldMeasuredFormula()
    {
        // 同秒两行的第一行承载本次实测的换算样本。
        List<CabinetRealtime> rows = adapter(activeDeviceDatabase()).samplesSince(CHANNEL,
                new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7434L));
        CabinetRealtime latest = rows.get(0);
        // test_vol 13733 → 1.3733 V；test_cur -11889 → -1.1889 A；放电容量与能量除以 36000000。
        assertEquals(1.3733, latest.voltageV(), 1e-9);
        assertEquals(-1.1889, latest.currentA(), 1e-9);
        assertEquals(50350024d / 36000000d, latest.capacityAh(), 1e-9);
        assertEquals(1103331d / 36000000d, latest.energyWh(), 1e-9);
        // test_time 单位为毫秒，作为工步内相对时长输出秒。
        assertEquals(300L, latest.relativeTimeSeconds());
        assertEquals(3, latest.cycleNo());
        assertEquals(12, latest.stepNo());
        assertEquals("2", latest.stepType());
        assertEquals("neware:440:7435", latest.sourceRecordId());
        // 工步类型保留设备给出的数字，不按电流符号猜方向；同一秒的另一行靠工步与序号区分。
        assertEquals(9, rows.get(1).stepNo());
    }

    @Test
    void onlineFactIsSourceReachabilityAndChannelWorkStatusStaysUnobservable()
    {
        NewareMysqlAdapter adapter = adapter(activeDeviceDatabase());
        Map<ChannelIdentity, RealtimeSnapshot> snapshots = adapter.inquire(Set.of(CHANNEL));
        RealtimeSnapshot snapshot = snapshots.get(CHANNEL);
        assertEquals(DeviceConnectionStatus.ONLINE, snapshot.connectionStatus());
        // 读库路线下通道工作状态不可观测，不按数据行推断。
        assertEquals(DeviceWorkStatus.UNOBSERVABLE, snapshot.workStatus());
        assertFalse(adapter.capabilities().control());
        // 辅助测量来自导入，不来自新威辅助表族。
        assertNull(adapter.latestCollector(ADAPTER_CODE, "anything"));
        assertEquals(NewareMysqlAdapter.SAMPLE_INTERVAL_MS, adapter.persistenceIntervalMillis(CHANNEL));
    }

    @Test
    void samplesSinceKeepsEveryRowInTheSameSecondAndDropsRowsAtOrBeforeTheWatermark()
    {
        NewareMysqlAdapter adapter = adapter(activeDeviceDatabase());
        SampleWatermark watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7434L);
        List<CabinetRealtime> rows = adapter.samplesSince(CHANNEL, watermark);
        // 同一绝对秒内的两行都保留，并且靠协议序号区分。
        assertEquals(2, rows.size());
        assertEquals(List.of(7435L, 7436L), rows.stream().map(CabinetRealtime::sequenceId).toList());
        assertEquals(List.of("neware:440:7435", "neware:440:7436"),
                rows.stream().map(CabinetRealtime::sourceRecordId).toList());
        // 严格大于水位：等于水位的行不返回。
        assertEquals(1, adapter.samplesSince(CHANNEL,
                new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7435L)).size());
        // 水位之后没有新行时返回空。
        assertTrue(adapter.samplesSince(CHANNEL,
                new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L)).isEmpty());
    }

    @Test
    void samplesWithoutWatermarkReturnNothingSoDeviceHistoryNeverEntersTheTest()
    {
        NewareMysqlAdapter adapter = adapter(activeDeviceDatabase());
        assertTrue(adapter.samplesSince(CHANNEL, null).isEmpty());
    }

    @Test
    void emptyWatermarkMeansTheChannelHadNoRowsSoEveryLaterRowBelongsToTheReception()
    {
        // 边界时该通道在来源里没有任何行（例如通道尚未开始测试），之后出现的行全部属于本次接收。
        List<CabinetRealtime> rows = adapter(activeDeviceDatabase())
                .samplesSince(CHANNEL, new SampleWatermark(null, 0L));
        assertEquals(2, rows.size());
    }

    @Test
    void watermarkIsTheLatestRowOfTheLatestTest()
    {
        SampleWatermark watermark = adapter(activeDeviceDatabase()).currentWatermark(CHANNEL);
        assertEquals(Instant.parse("2026-09-28T01:54:56Z"), watermark.deviceTime());
        assertEquals(7436L, watermark.sequenceId());
        // 通道没有测试记录时水位为空，接收路径不把设备历史并入本测试。
        NewareMysqlAdapter empty = adapter(new FakeVendorDatabase().registeredChannel(DEV_UID, 1, 2));
        assertNull(empty.currentWatermark(CHANNEL).deviceTime());
    }

    @Test
    void rejectsChannelIdentitiesThatBelongToAnotherInstance()
    {
        NewareMysqlAdapter adapter = adapter(activeDeviceDatabase());
        ChannelIdentity foreign = new ChannelIdentity("other-source", 21, 36, 1, 2);
        org.junit.jupiter.api.Assertions.assertThrows(RuntimeException.class, () -> adapter.latestCabinet(foreign));
        assertTrue(adapter.discoverChannels().stream()
                .allMatch(item -> ADAPTER_CODE.equals(item.identity().adapterCode())));
    }
}

 battery-device-neware/src/test/java/com/battery/device/neware/ReadOnlyVendorDataSourceTest.java

New file
@@ -0,0 +1,161 @@
package com.battery.device.neware;

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

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.Statement;
import java.time.Clock;
import java.time.Instant;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Map;
import java.util.Set;
import javax.sql.DataSource;
import org.junit.jupiter.api.Test;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.registry.AdapterInstanceConfig;

/**
 * 只读防线测试：非查询语句在到达驱动之前被拒绝，且该数据源在其生命周期内从未发出过非查询语句。
 */
class ReadOnlyVendorDataSourceTest
{
    private static final String ADAPTER_CODE = "neware76";
    private static final ChannelIdentity CHANNEL = new ChannelIdentity(ADAPTER_CODE, 21, 36, 1, 2);

    private ReadOnlyVendorDataSource dataSourceOver(FakeVendorDatabase database)
    {
        return new ReadOnlyVendorDataSource(database.dataSource(), ADAPTER_CODE);
    }

    @Test
    void queryStatementsAreForwardedAndRecorded() throws Exception
    {
        ReadOnlyVendorDataSource dataSource = new ReadOnlyVendorDataSource(new FakeVendorDatabase().dataSource(),
                ADAPTER_CODE);
        try (Connection connection = dataSource.getConnection();
                PreparedStatement statement = connection.prepareStatement("select 1 from channel"))
        {
            statement.executeQuery();
        }
        try (Connection connection = dataSource.getConnection();
                Statement statement = connection.createStatement())
        {
            statement.executeQuery("show tables");
            statement.executeQuery("describe `channel`");
        }
        assertTrue(dataSource.audit().rejectedStatements().isEmpty());
        assertEquals(3, dataSource.audit().forwardedStatements().size());
    }

    @Test
    void nonQueryStatementsAreRejectedBeforeTheyReachTheDriver() throws Exception
    {
        FakeVendorDatabase database = new FakeVendorDatabase();
        ReadOnlyVendorDataSource dataSource = dataSourceOver(database);
        List<String> nonQueries = List.of("update channel set chl_status = 1",
                "insert into channel(dev_uid) values(210036)", "delete from channel",
                "drop table data_210036_2026_09_18", "create index idx on channel(chl_status)",
                "truncate table channel", "set global read_only = 0", "grant select on *.* to x",
                "alter table channel add column x int", "call some_procedure()");
        for (String sql : nonQueries)
        {
            assertThrows(NonQueryStatementException.class, () -> {
                try (Connection connection = dataSource.getConnection();
                        PreparedStatement statement = connection.prepareStatement(sql))
                {
                    statement.execute();
                }
            }, "非查询语句必须被拒绝：" + sql);
        }
        assertEquals(nonQueries.size(), dataSource.audit().rejectedStatements().size());
        assertTrue(database.statements().isEmpty(), "被拒绝的语句不得到达驱动");
        // 语句级执行同样被拦下，并且没有任何非查询语句到达驱动。
        assertThrows(NonQueryStatementException.class, () -> {
            try (Connection connection = dataSource.getConnection(); Statement statement = connection.createStatement())
            {
                statement.executeUpdate("update channel set chl_status = 1");
            }
        });
        assertThrows(NonQueryStatementException.class,
                () -> dataSource.getConnection().prepareCall("{call anything()}"));
        assertTrue(database.statements().isEmpty(), "被拒绝的语句不得到达驱动");
        assertEquals(nonQueries.size() + 1, dataSource.audit().rejectedStatements().size());
    }

    /**
     * 这条断言对应验收要求：整个适配器生命周期内，该数据源没有发出过任何非查询语句。
     */
    @Test
    void adapterLifecycleNeverSendsANonQueryStatement()
    {
        FakeVendorDatabase database = new FakeVendorDatabase()
                .registeredChannel(210036L, 1, 2)
                .test(new FakeVendorDatabase.TestRow(210036L, 1, 2, 440L, "2026-09-18 09:00:00",
                        "data_210036_2026_09_18", null))
                .sample(new FakeVendorDatabase.SampleRow("data_210036_2026_09_18", 440L, 1, 2, 7435L, 3, 12, "2",
                        300000L, 13733, -11889, 0, 0, 50350024, 1103331, "2026-09-28 09:54:56"));
        ReadOnlyVendorDataSource guarded = dataSourceOver(database);
        AdapterInstanceConfig config = new AdapterInstanceConfig(ADAPTER_CODE, NewareMysqlAdapter.ADAPTER_TYPE,
                "26.1.2.3", 3306, 60000, "Asia/Hong_Kong", true, AdapterCapabilities.observeOnly(),
                Map.of("database", "bts63"));
        NewareMysqlAdapter adapter = new NewareMysqlAdapter(config, guarded, () -> { },
                Clock.fixed(Instant.parse("2026-09-28T02:00:00Z"), ZoneOffset.UTC), 0);

        adapter.discoverChannels();
        adapter.inquire(Set.of(CHANNEL));
        adapter.latestCabinet(CHANNEL);
        adapter.currentWatermark(CHANNEL);
        adapter.samplesSince(CHANNEL,
                new com.battery.device.api.TestDataObservationAdapter.SampleWatermark(
                        Instant.parse("2026-09-28T01:00:00Z"), 1L));

        assertTrue(guarded.audit().rejectedStatements().isEmpty(),
                "适配器不应尝试任何非查询语句：" + guarded.audit().rejectedStatements());
        assertTrue(!database.statements().isEmpty(), "适配器应当实际读取来源");
        for (String statement : database.statements())
        {
            assertTrue(ReadOnlyVendorDataSource.isQuery(statement), "到达驱动的语句必须是查询：" + statement);
        }
    }

    @Test
    void statementClassificationKeepsQueriesAndRejectsEverythingElse()
    {
        assertTrue(ReadOnlyVendorDataSource.isQuery("select dev_uid from channel"));
        assertTrue(ReadOnlyVendorDataSource.isQuery("\n  SELECT 1"));
        assertTrue(ReadOnlyVendorDataSource.isQuery("/* probe */ show grants"));
        assertTrue(ReadOnlyVendorDataSource.isQuery("-- probe\n describe channel"));
        assertTrue(ReadOnlyVendorDataSource.isQuery("desc `channel`"));
        assertTrue(!ReadOnlyVendorDataSource.isQuery("update channel set chl_status = 0"));
        assertTrue(!ReadOnlyVendorDataSource.isQuery(""));
        // 以 select 开头但紧跟字母的标识符不是查询语句。
        assertTrue(!ReadOnlyVendorDataSource.isQuery("selection_helper()"));
    }

    @Test
    void rejectedStatementsNeverReachTheDriverEvenWhenTheApplicationKeepsTrying()
    {
        // 连续尝试写操作后，来源侧一条语句都没收到，说明拦截发生在驱动之前。
        FakeVendorDatabase database = new FakeVendorDatabase();
        ReadOnlyVendorDataSource dataSource = dataSourceOver(database);
        for (String sql : List.of("update channel set chl_status = 1", "delete from channel",
                "insert into channel(dev_uid) values(210036)"))
        {
            try
            {
                dataSource.getConnection().prepareStatement(sql);
            }
            catch (Exception ignored)
            {
                // 每条都被白名单挡下。
            }
        }
        assertTrue(database.statements().isEmpty());
        assertEquals(3, dataSource.audit().rejectedStatements().size());
    }
}

 battery-device-simulator/src/main/java/com/battery/device/simulator/BtsDevelopmentSimulator.java

@@ -7,10 +7,14 @@
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import javax.xml.stream.XMLInputFactory;
import javax.xml.stream.XMLStreamConstants;
import javax.xml.stream.XMLStreamReader;
import com.battery.device.api.DeviceAdapter;
import com.battery.device.api.model.DeviceWorkStatus;
import com.battery.device.api.model.DeviceConnectionStatus;
import com.battery.device.api.model.DeviceIdentity;
@@ -19,6 +23,7 @@
import com.battery.device.api.model.DetectedDeviceMetadata;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.model.RealtimeSnapshot;
import com.battery.device.api.DeviceObservationAdapter;
import com.battery.device.api.TestDataObservationAdapter;
import com.battery.device.simulator.BtsSimulatorStateStore;
@@ -37,8 +42,12 @@
import com.battery.device.simulator.DevelopmentCollectorSimulation;
import com.battery.device.simulator.DevelopmentCollectorSimulation.CollectorRealtimeSnapshot;

/** 提供由开发后端托管的临时只读 BTS 模拟端点。 */
public final class BtsDevelopmentSimulator implements TestDataObservationAdapter
/**
 * 提供由开发后端托管的临时只读 BTS 模拟端点。
 * 同时实现发现与测试数据观测契约：模拟器直接暴露自身状态，因此发现与取数按同一实例解析，
 * 不再需要单独注入一个扩展观测 Bean。TCP 端点仍保留给协议层测试使用。
 */
public final class BtsDevelopmentSimulator implements DeviceAdapter, TestDataObservationAdapter
{
    public static final String HOST = "127.0.0.1";
    public static final int DEFAULT_PORT = 15020;
@@ -102,6 +111,34 @@
        for (ChannelIdentity requested : channels) result.put(requested, observation(requested));
        recordRequest("observe", channels.toString(), result.toString());
        return java.util.Map.copyOf(result);
    }

    /** 直接提供模拟器实时快照；不可观测通道不出现在结果里，与 BTS 协议解析后的语义一致。 */
    @Override
    public synchronized Map<ChannelIdentity, RealtimeSnapshot> inquire(Set<ChannelIdentity> channels)
    {
        if (channels == null || channels.isEmpty()) return Map.of();
        Map<ChannelIdentity, RealtimeSnapshot> result = new LinkedHashMap<>();
        for (ChannelIdentity requested : channels)
        {
            if (!canObserve(requested)) continue;
            CabinetRealtime measurement = cabinetMeasurement(requested);
            if (measurement == null) continue;
            DeviceObservation observed = observation(requested);
            result.put(requested, new RealtimeSnapshot(requested, observed.workStatus(), measurement.cycleNo(),
                    measurement.stepNo(), measurement.stepType(), null, BigDecimal.valueOf(measurement.currentA()),
                    BigDecimal.valueOf(measurement.voltageV()), BigDecimal.valueOf(measurement.capacityAh()),
                    BigDecimal.valueOf(measurement.energyWh()), totalTimeSeconds(requested),
                    measurement.relativeTimeSeconds(), null, measurement.serverReceivedAt(), Map.of(),
                    DeviceConnectionStatus.ONLINE, measurement.deviceTime(), List.of(), null));
        }
        recordRequest("inquire", channels.toString(), result.toString());
        return Map.copyOf(result);
    }

    private long totalTimeSeconds(ChannelIdentity identity)
    {
        return requireDevice(identity.devid()).channels().get(identity.chlid() - 1).configuration().totalTimeSeconds();
    }

    private DeviceObservation observation(ChannelIdentity requested)
@@ -368,7 +405,7 @@

    /** 请求采集器当前配置值；软件不可访问或设备离线时返回空测量。 */
    @Override
    public synchronized CollectorRealtime latestCollector(String collectorCode)
    public synchronized CollectorRealtime latestCollector(String adapterCode, String collectorCode)
    {
        CollectorState collector = state.collectors().stream().filter(item -> item.deviceCode().equals(collectorCode))
                .findFirst().orElse(null);

 battery-device-simulator/src/test/java/com/battery/device/simulator/SimulatorObservationAdapterContractTest.java

@@ -16,7 +16,7 @@
        {
            simulator.updateChannel(1, 1, true, DeviceWorkStatus.PROTECTED);
            simulator.updateCollectorChannelStatus(1, "S-01", DeviceWorkStatus.UNOBSERVABLE);
            var collector = simulator.latestCollector("dev-sim-collector-1");
            var collector = simulator.latestCollector(BtsDevelopmentSimulator.ADAPTER_CODE, "dev-sim-collector-1");
            assertTrue(collector.measurements().stream().noneMatch(item -> item.channelCode().equals("S-01")));
            firstSequence = collector.measurements().get(0).sequenceId();
            simulator.setSoftwareAccessible(false);
@@ -30,7 +30,7 @@
            assertEquals(DeviceWorkStatus.PROTECTED, restored.snapshot().devices().get(0).channels().get(0).workStatus());
            assertTrue(restored.snapshot().collectors().get(0).latestMeasurements().isEmpty());
            restored.setSoftwareAccessible(true);
            assertTrue(restored.latestCollector("dev-sim-collector-1").measurements().get(0).sequenceId() > firstSequence);
            assertTrue(restored.latestCollector(BtsDevelopmentSimulator.ADAPTER_CODE, "dev-sim-collector-1").measurements().get(0).sequenceId() > firstSequence);
            assertEquals(DeviceWorkStatus.UNOBSERVABLE, restored.snapshot().collectors().get(0).channels().get(0).workStatus());
        }
    }
@@ -63,14 +63,14 @@
            simulator.updateChannel(1, 1, true, DeviceWorkStatus.WORKING);
            var first = simulator.latestCabinet(channel);
            assertNotNull(first);
            assertFalse(simulator.latestCollector("dev-sim-collector-1").measurements().isEmpty());
            assertFalse(simulator.latestCollector(BtsDevelopmentSimulator.ADAPTER_CODE, "dev-sim-collector-1").measurements().isEmpty());
            simulator.setSoftwareAccessible(false);
            assertEquals(channel, simulator.discoverChannels().get(0).identity());
            assertTrue(simulator.snapshot().devices().get(0).online());
            assertEquals(DeviceWorkStatus.UNOBSERVABLE,
                    simulator.observe(java.util.Set.of(channel)).get(channel).workStatus());
            assertNull(simulator.latestCabinet(channel));
            assertTrue(simulator.latestCollector("dev-sim-collector-1").measurements().isEmpty());
            assertTrue(simulator.latestCollector(BtsDevelopmentSimulator.ADAPTER_CODE, "dev-sim-collector-1").measurements().isEmpty());
            simulator.setSoftwareAccessible(true);
            var next = simulator.latestCabinet(channel);
            assertTrue(next.sequenceId() > first.sequenceId());

 pom.xml

@@ -181,6 +181,11 @@
            </dependency>
            <dependency>
                <groupId>com.battery</groupId>
                <artifactId>battery-device-neware</artifactId>
                <version>${ruoyi.version}</version>
            </dependency>
            <dependency>
                <groupId>com.battery</groupId>
                <artifactId>battery-device-simulator</artifactId>
                <version>${ruoyi.version}</version>
            </dependency>
@@ -202,6 +207,7 @@
        <module>battery-business</module>
        <module>battery-device-api</module>
        <module>battery-device-bts</module>
        <module>battery-device-neware</module>
        <module>ruoyi-quartz</module>
        <module>ruoyi-generator</module>
        <module>ruoyi-common</module>

 ruoyi-admin/pom.xml

@@ -81,6 +81,12 @@
            <artifactId>battery-device-bts</artifactId>
        </dependency>

        <!-- 直连厂商库的只读适配器实现 -->
        <dependency>
            <groupId>com.battery</groupId>
            <artifactId>battery-device-neware</artifactId>
        </dependency>

        <!-- 定时任务-->
        <dependency>
            <groupId>com.ruoyi</groupId>

 ruoyi-admin/src/main/java/com/battery/business/runtime/DevelopmentRealtimeSnapshotScheduler.java

File was deleted

 ruoyi-admin/src/main/java/com/battery/business/runtime/DevelopmentSimulatorConfig.java

@@ -21,10 +21,12 @@
    @Bean
    @Profile("dev-simulator")
    public ApplicationRunner registerDevelopmentSimulatorAdapter(DeviceAdapterRegistry registry,
            BtsDevelopmentSimulator simulator,
            @Value("${battery.dev-simulator.port:15020}") int simulatorPort)
    {
        // 固定开发装配不进入 t_device_adapter，直接注册运行时实例。
        return arguments -> registry.synchronize(developmentSimulatorRuntimeConfig(simulatorPort));
        // 注册模拟器对象本身，使发现与取数落在同一实例上，不再单独注入扩展观测 Bean。
        return arguments -> registry.register(developmentSimulatorRuntimeConfig(simulatorPort), simulator);
    }

    private AdapterInstanceConfig developmentSimulatorRuntimeConfig(int simulatorPort)

 ruoyi-admin/src/main/java/com/battery/business/runtime/DeviceAdapterRuntimeConfig.java

@@ -3,29 +3,27 @@
import java.util.List;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
import org.springframework.scheduling.annotation.EnableScheduling;
import com.battery.device.api.NoopDeviceAdapterFactory;
import com.battery.device.bts.BtsDeviceAdapterFactory;
import com.battery.device.api.registry.DeviceAdapterRegistry;
import com.battery.device.neware.NewareMysqlAdapterFactory;

/** 装配单进程设备适配器运行时。 */
/**
 * 装配单进程设备适配器运行时。
 *
 * <p>注册表按存储形态持有全部可用实现；实例按 {@code t_device_adapter} 配置或代码装配注册。
 * 同一进程内可并存多个来源实例，取数按通道身份里的适配器编码解析到各自实例。
 * 某个类型是否真的有实例可用只由配置决定，因此这里不再按 profile 收窄可注册的类型。
 */
@Configuration
@EnableScheduling
public class DeviceAdapterRuntimeConfig
{
    @Bean(destroyMethod = "close")
    @Profile("!dev-simulator & !bts")
    public DeviceAdapterRegistry noopDeviceAdapterRegistry()
    {
        return new DeviceAdapterRegistry(List.of(new NoopDeviceAdapterFactory()));
    }

    @Bean(destroyMethod = "close")
    @Profile({ "dev-simulator", "bts" })
    public DeviceAdapterRegistry deviceAdapterRegistry()
    {
        return new DeviceAdapterRegistry(List.of(new BtsDeviceAdapterFactory()));
        return new DeviceAdapterRegistry(List.of(new NoopDeviceAdapterFactory(), new BtsDeviceAdapterFactory(),
                new NewareMysqlAdapterFactory()));
    }

}

 ruoyi-admin/src/main/java/com/battery/business/runtime/RedisDeviceDiscoveryCandidateStore.java

@@ -129,7 +129,8 @@
                object.getString("deviceCode"), object.getIntValue("devtype"), object.getIntValue("devid"),
                object.getString("ip"), object.getString("mac"), object.getString("name"),
                object.getString("manufacturer"), object.getString("model"), channels,
                requiredInstant(object, "observedAt"), requiredInstant(object, "expiresAt"));
                requiredInstant(object, "observedAt"), requiredInstant(object, "expiresAt"),
                optionalInstant(object, "lastDataAt"));
    }

    private ChannelCandidate channelCandidate(Object value)
@@ -141,7 +142,8 @@
                identity.getIntValue("subdevid"), identity.getIntValue("chlid"));
        RealtimeSnapshot latestSnapshot = object.getObject("latestSnapshot", RealtimeSnapshot.class);
        return new ChannelCandidate(channelIdentity, object.getBooleanValue("online"),
                DeviceWorkStatus.valueOf(object.getString("workStatus")), requiredInstant(object, "observedAt"), latestSnapshot);
                DeviceWorkStatus.valueOf(object.getString("workStatus")), requiredInstant(object, "observedAt"),
                latestSnapshot, optionalInstant(object, "lastDataAt"));
    }

    private CollectorCandidate collectorCandidate(Object value)
@@ -161,6 +163,12 @@
        return value;
    }

    /** 最近数据时间只在厂商存储来源上存在，缺失是正常取值。 */
    private Instant optionalInstant(JSONObject object, String field)
    {
        return object.getObject(field, Instant.class);
    }

    private void save(StoredState state)
    {
        redisTemplate.opsForValue().set(STATE_KEY, JSON.toJSONString(state));

 ruoyi-admin/src/main/java/com/battery/business/runtime/TestDataReceptionScheduler.java

New file
@@ -0,0 +1,22 @@
package com.battery.business.runtime;

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import com.battery.business.service.ITestDataReceptionService;

/**
 * 按固定周期触发数据接收，由业务服务按测试接收状态和采样来源保存成功观测。
 *
 * <p>它不限定 profile：真实来源的接收路径也必须在非开发 profile 下运行。
 * 开发模拟器专属的调度（模拟器发现轮询、采集器状态同步）仍只在 dev-simulator 下额外启用。
 */
@Component
public class TestDataReceptionScheduler
{
    private final ITestDataReceptionService reception;

    public TestDataReceptionScheduler(ITestDataReceptionService reception) { this.reception = reception; }

    @Scheduled(fixedDelay = 1000)
    public void capture() { reception.captureReceivedData(); }
}

 ruoyi-admin/src/main/resources/db/migration-final/V22__vendor_source_instance_config.sql

New file
@@ -0,0 +1,3 @@
-- Ticket01：适配器实例的类型按存储形态放宽到直连厂商库，来源形态配置改用 JSON 列承载（ADR-0065）。
-- 连接口令不进库：source_config 只放 host、port、database、user、时区这类非敏感项。
ALTER TABLE t_device_adapter ADD COLUMN source_config JSON NULL AFTER capabilities;

 ruoyi-admin/src/test/java/com/battery/business/runtime/DeviceAdapterRuntimeConfigTest.java

@@ -8,19 +8,22 @@
import org.mockito.ArgumentCaptor;
import com.battery.device.api.registry.AdapterInstanceConfig;
import com.battery.device.api.registry.DeviceAdapterRegistry;
import com.battery.device.simulator.BtsDevelopmentSimulator;

class DeviceAdapterRuntimeConfigTest
{
    @Test
    void registersDevelopmentSimulatorBeforeApplicationReady() throws Exception
    void registersDevelopmentSimulatorInstanceBeforeApplicationReady() throws Exception
    {
        DeviceAdapterRegistry registry = mock(DeviceAdapterRegistry.class);
        var runner = new DevelopmentSimulatorConfig().registerDevelopmentSimulatorAdapter(registry, 15021);
        BtsDevelopmentSimulator simulator = mock(BtsDevelopmentSimulator.class);
        var runner = new DevelopmentSimulatorConfig().registerDevelopmentSimulatorAdapter(registry, simulator, 15021);

        runner.run(new DefaultApplicationArguments(new String[0]));

        // 注册模拟器对象本身：发现与取数落在同一实例上，不再单独注入扩展观测 Bean。
        ArgumentCaptor<AdapterInstanceConfig> adapter = ArgumentCaptor.forClass(AdapterInstanceConfig.class);
        verify(registry).synchronize(adapter.capture());
        verify(registry).register(adapter.capture(), org.mockito.ArgumentMatchers.same(simulator));
        assertEquals("dev-sim", adapter.getValue().adapterCode());
        assertEquals("127.0.0.1", adapter.getValue().host());
        assertEquals(15021, adapter.getValue().port());

 ruoyi-admin/src/test/java/com/battery/business/runtime/VendorSourceInstanceMigrationContractTest.java

New file
@@ -0,0 +1,97 @@
package com.battery.business.runtime;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.apache.ibatis.datasource.unpooled.UnpooledDataSource;
import org.junit.jupiter.api.Test;
import org.springframework.core.io.ClassPathResource;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.init.ScriptUtils;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.registry.AdapterInstanceConfig;
import com.battery.device.api.registry.AdapterRegistryException;
import com.battery.device.api.registry.DeviceAdapterRegistry;

/**
 * 适配器实例按存储形态放宽与来源形态配置列的迁移契约：
 * 迁移把 {@code source_config} 加在能力列之后，运行库上该列确实存在且可空；
 * 类型白名单放宽到直连厂商库，未知类型仍然拒绝。
 */
class VendorSourceInstanceMigrationContractTest
{
    @Test
    void migrationAddsNullableSourceConfigColumnWithoutTouchingExistingRows() throws Exception
    {
        String path = "db/migration-final/V22__vendor_source_instance_config.sql";
        String migration = resource(path);
        assertTrue(migration.contains("ALTER TABLE t_device_adapter ADD COLUMN source_config JSON NULL"));
        assertTrue(migration.contains("AFTER capabilities"));
        // 迁移只加列：不改既有列、不写入任何数据，更不写入凭据。
        assertFalse(migration.matches("(?s).*\\b(INSERT|UPDATE|DELETE|DROP)\\b.*"));
        assertFalse(migration.toLowerCase().contains("password"));

        var source = new UnpooledDataSource("org.h2.Driver",
                "jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DATABASE_TO_LOWER=TRUE", "sa", "");
        try (var connection = source.getConnection())
        {
            JdbcTemplate jdbc = new JdbcTemplate(source);
            jdbc.execute("create table t_device_adapter(adapter_id bigint auto_increment primary key,"
                    + "adapter_code varchar(64), capabilities json)");
            ScriptUtils.executeSqlScript(connection, new ClassPathResource(path));

            // 来源形态配置只承载非敏感项；连接口令由环境变量提供，不入库。
            jdbc.update("insert into t_device_adapter(adapter_code, capabilities, source_config) values(?,?,?)",
                    "neware76", "{\"control\":false}", "{\"database\":\"bts63\",\"user\":\"visitor\"}");
            assertEquals(1, jdbc.queryForObject(
                    "select count(*) from t_device_adapter where source_config is not null", Integer.class));
            // 代码装配实例（例如开发模拟器）保持为空。
            jdbc.update("insert into t_device_adapter(adapter_code, capabilities) values(?,?)", "dev-sim",
                    "{\"control\":false}");
            assertEquals(1, jdbc.queryForObject(
                    "select count(*) from t_device_adapter where source_config is null", Integer.class));
        }
    }

    @Test
    void sourceConfigCarriesOnlyNonSensitiveOptionsAndUnknownTypesStayRejected()
    {
        AdapterInstanceConfig config = new AdapterInstanceConfig("neware76", "NEWARE_MYSQL", "26.1.2.3", 3306,
                60000, "Asia/Hong_Kong", true, AdapterCapabilities.observeOnly(),
                Map.of("database", "bts63", "user", "visitor", "passwordEnv", "BATTERY_VENDOR_DB_PASSWORD"));

        assertEquals("bts63", config.sourceOption("database"));
        assertTrue(config.capabilities().discovery());
        assertFalse(config.capabilities().control());

        try (DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of()))
        {
            // 放宽不等于接受任何类型：未注册的存储形态仍然明确失败。
            assertThrows(AdapterRegistryException.class,
                    () -> registry.synchronize(new AdapterInstanceConfig("mystery", "SOME_OTHER_VENDOR",
                            "127.0.0.1", 3306, 60000, "Asia/Hong_Kong", true,
                            AdapterCapabilities.observeOnly(), Map.of())));
        }
    }

    private static String resource(String path)
    {
        try (InputStream stream = VendorSourceInstanceMigrationContractTest.class.getClassLoader()
                .getResourceAsStream(path))
        {
            if (stream == null) throw new IllegalStateException("缺少迁移文件：" + path);
            return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
        }
        catch (java.io.IOException exception)
        {
            throw new IllegalStateException("无法读取迁移文件：" + path, exception);
        }
    }
}

 ruoyi-admin/src/test/java/com/battery/business/service/impl/DeviceDiscoveryPersistenceServiceTest.java

@@ -76,4 +76,40 @@
        assertEquals(1, candidateStore.snapshot(java.time.Instant.now()).cabinets().size());
        verify(adapter).discoverChannels();
    }

    /**
     * 直连厂商存储的来源：通道登记是长期事实，候选不按时间过期，
     * 每条候选附带最近一次采样行的设备时间，供操作员判断哪几台真的在产出数据。
     */
    @Test
    void vendorSourceCandidatesCarryLastDataTimeAndDoNotExpireByTime()
    {
        InMemoryDeviceDiscoveryCandidateStore candidateStore = new InMemoryDeviceDiscoveryCandidateStore();
        TDetectionCabinetMapper cabinetMapper = Mockito.mock(TDetectionCabinetMapper.class);
        TTestMapper testMapper = Mockito.mock(TTestMapper.class);
        DeviceAdapterRegistry registry = Mockito.mock(DeviceAdapterRegistry.class);
        DeviceAdapter adapter = Mockito.mock(DeviceAdapter.class);
        ChannelIdentity identity = new ChannelIdentity("neware76", 21, 36, 1, 2);
        java.time.Instant lastDataAt = java.time.Instant.parse("2026-09-28T01:54:56Z");
        when(testMapper.hasReceivingTestForAdapter("neware76")).thenReturn(0);
        when(adapter.discoverChannels()).thenReturn(List.of(new DetectedChannel(identity, "26.1.2.3", true,
                com.battery.device.api.model.DetectedDeviceMetadata.empty(), lastDataAt)));
        when(registry.execute(eq("neware76"), any())).thenAnswer(invocation -> {
            @SuppressWarnings("unchecked")
            Function<DeviceAdapter, Object> operation = (Function<DeviceAdapter, Object>) invocation.getArgument(1);
            return operation.apply(adapter);
        });

        DeviceDiscoveryPersistenceService service = new DeviceDiscoveryPersistenceService(candidateStore,
                cabinetMapper, testMapper);
        service.discoverForReception(registry, new DeviceDiscoveryPersistenceService.DiscoverySource(
                "neware76", 8L, "新威 7.6 那台", 60000, true));

        var candidate = candidateStore.snapshot(java.time.Instant.now()).cabinets().get(0);
        assertEquals(lastDataAt, candidate.lastDataAt());
        assertEquals(lastDataAt, candidate.channels().get(0).lastDataAt());
        // 一年后仍然是候选：候选只按"已登记且未删除"排除，不按时间过期。
        assertEquals(1, candidateStore.snapshot(java.time.Instant.now().plusSeconds(365L * 24 * 3600))
                .cabinets().size());
    }
}

 ruoyi-admin/src/test/java/com/battery/business/service/impl/RoutingAdapterWiringContractTest.java

New file
@@ -0,0 +1,33 @@
package com.battery.business.service.impl;

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

import org.junit.jupiter.api.Test;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Service;
import com.battery.device.api.TestDataObservationAdapter;

/**
 * 取数入口的装配契约：开发模拟器既是 Spring Bean，又实现了测试数据观测契约，
 * 所以业务侧注入的那个 Bean 必须显式声明为首选，否则上下文里出现两个候选会启动失败。
 * 这里断言的是"哪个 Bean 会被注入"这个结构性事实，而不是 Spring 内部行为。
 */
class RoutingAdapterWiringContractTest
{
    @Test
    void routingAdapterIsThePreferredTestDataObservationBean()
    {
        assertTrue(TestDataObservationAdapter.class.isAssignableFrom(RoutingTestDataObservationAdapter.class));
        assertTrue(RoutingTestDataObservationAdapter.class.isAnnotationPresent(Service.class),
                "取数入口必须是可注入的 Bean");
        assertTrue(RoutingTestDataObservationAdapter.class.isAnnotationPresent(Primary.class),
                "来源实例自己也实现该契约时必须显式首选路由入口，否则注入出现两个候选");
    }

    @Test
    void developmentSimulatorAlsoExposesTheContractWhichIsWhyPreferenceIsNeeded()
    {
        assertTrue(TestDataObservationAdapter.class
                .isAssignableFrom(com.battery.device.simulator.BtsDevelopmentSimulator.class));
    }
}

 ruoyi-admin/src/test/java/com/battery/business/service/impl/RoutingTestDataObservationAdapterTest.java

New file
@@ -0,0 +1,163 @@
package com.battery.business.service.impl;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.junit.jupiter.api.Test;
import com.battery.device.api.DeviceAdapter;
import com.battery.device.api.TestDataObservationAdapter;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.model.DetectedChannel;
import com.battery.device.api.model.RealtimeSnapshot;
import com.battery.device.api.registry.AdapterInstanceConfig;
import com.battery.device.api.registry.AdapterRegistryException;
import com.battery.device.api.registry.DeviceAdapterRegistry;

/**
 * 多来源实例：每个通道的取数请求只送到它所属的实例；实例缺失时明确失败，不回退到其他实例。
 */
class RoutingTestDataObservationAdapterTest
{
    @Test
    void sendsEachChannelRequestToTheInstanceNamedByItsAdapterCode()
    {
        DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of());
        try
        {
            RoutingTestDataObservationAdapter router = new RoutingTestDataObservationAdapter(registry);
            RecordingAdapter first = new RecordingAdapter("neware76", 1L);
            RecordingAdapter second = new RecordingAdapter("neware80", 2L);
            registry.register(config("neware76"), first);
            registry.register(config("neware80"), second);

            assertEquals(1L, router.latestCabinet(channel("neware76")).sequenceId());
            assertEquals(2L, router.latestCabinet(channel("neware80")).sequenceId());
            assertEquals(List.of("neware76"), first.calls());
            assertEquals(List.of("neware80"), second.calls());
            assertEquals(60000L, router.persistenceIntervalMillis(channel("neware76")));
            assertEquals(70000L, router.persistenceIntervalMillis(channel("neware80")));
        }
        finally
        {
            registry.close();
        }
    }

    @Test
    void missingInstanceFailsExplicitlyInsteadOfFallingBackToAnotherInstance()
    {
        DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of());
        try
        {
            RoutingTestDataObservationAdapter router = new RoutingTestDataObservationAdapter(registry);
            RecordingAdapter only = new RecordingAdapter("neware76", 1L);
            registry.register(config("neware76"), only);

            // 实例缺失：明确失败，且没有落到已注册的实例上。
            assertThrows(AdapterRegistryException.class, () -> router.latestCabinet(channel("neware80")));
            assertThrows(AdapterRegistryException.class, () -> router.samplesSince(channel("neware80"), null));
            assertThrows(AdapterRegistryException.class, () -> router.currentWatermark(channel("neware80")));
            assertTrue(only.calls().isEmpty(), "缺失实例的请求不得落到其他实例");
        }
        finally
        {
            registry.close();
        }
    }

    @Test
    void aggregatedAdapterDoesNotPretendToDiscoverDevices()
    {
        DeviceAdapterRegistry registry = new DeviceAdapterRegistry(List.of());
        try
        {
            RoutingTestDataObservationAdapter router = new RoutingTestDataObservationAdapter(registry);
            assertThrows(UnsupportedOperationException.class, router::discoverChannels);
            assertEquals(RoutingTestDataObservationAdapter.ADAPTER_CODE, router.adapterCode());
            assertFalse(router.capabilities().control());
        }
        finally
        {
            registry.close();
        }
    }

    private static ChannelIdentity channel(String adapterCode)
    {
        return new ChannelIdentity(adapterCode, 21, 36, 1, 2);
    }

    private static AdapterInstanceConfig config(String adapterCode)
    {
        return new AdapterInstanceConfig(adapterCode, "NEWARE_MYSQL", "127.0.0.1", 3306, 60000,
                "Asia/Hong_Kong", true, AdapterCapabilities.observeOnly());
    }

    /** 记录被调用次数的实例替身；返回的序号用来区分请求落到了哪一个实例。 */
    private static final class RecordingAdapter implements DeviceAdapter, TestDataObservationAdapter
    {
        private final String code;
        private final long sequenceId;
        private final List<String> calls = new ArrayList<>();

        private RecordingAdapter(String code, long sequenceId)
        {
            this.code = code;
            this.sequenceId = sequenceId;
        }

        private List<String> calls()
        {
            return List.copyOf(calls);
        }

        @Override public String adapterCode() { return code; }
        @Override public AdapterCapabilities capabilities() { return AdapterCapabilities.observeOnly(); }
        @Override public List<DetectedChannel> discoverChannels() { return List.of(); }
        @Override public Map<ChannelIdentity, RealtimeSnapshot> inquire(Set<ChannelIdentity> channels)
        {
            return Map.of();
        }

        @Override public CabinetRealtime latestCabinet(ChannelIdentity channel)
        {
            calls.add(code);
            return new CabinetRealtime(sequenceId, Instant.EPOCH, 0, 0, 0, 0, 0, 0, null, null, Instant.EPOCH,
                    null, "Asia/Hong_Kong", null);
        }

        @Override public CollectorRealtime latestCollector(String adapterCode, String collectorCode)
        {
            return null;
        }

        @Override public List<StepRecord> stepRecords(ChannelIdentity channel) { return List.of(); }
        @Override public List<CycleRecord> cycleRecords(ChannelIdentity channel) { return List.of(); }

        @Override public long persistenceIntervalMillis(ChannelIdentity channel)
        {
            calls.add(code);
            return "neware76".equals(code) ? 60000L : 70000L;
        }

        @Override public SampleWatermark currentWatermark(ChannelIdentity channel)
        {
            calls.add(code);
            return new SampleWatermark(Instant.EPOCH, 1L);
        }

        @Override public List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark watermark)
        {
            calls.add(code);
            return List.of();
        }
    }
}

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

@@ -66,7 +66,7 @@
        for (Instant deviceTime : List.of(now.minusSeconds(1), now)) {
            var measurement = new TestDataObservationAdapter.CollectorMeasurement("S-01", 1L, deviceTime, deviceTime.toString(),
                    now, "TEMPERATURE", new BigDecimal("38"), "℃");
            when(adapter.latestCollector("collector-7")).thenReturn(new TestDataObservationAdapter.CollectorRealtime(
            when(adapter.latestCollector(org.mockito.ArgumentMatchers.nullable(String.class), org.mockito.ArgumentMatchers.eq("collector-7"))).thenReturn(new TestDataObservationAdapter.CollectorRealtime(
                    "collector-7", true, 1000, List.of(measurement), null));
            assertEquals(deviceTime.equals(now) ? 1 : 0, service.current(10L).exceptions().size());
        }
@@ -391,7 +391,7 @@
        when(simulator.testPlan(any(ChannelIdentity.class))).thenReturn(plan);
        if (mapper != null && test.getBatteries().get(0).getCollectorId() != null) {
            when(mapper.selectCollectorCode(7L)).thenReturn("collector-7");
            when(simulator.latestCollector("collector-7")).thenReturn(
            when(simulator.latestCollector(org.mockito.ArgumentMatchers.nullable(String.class), org.mockito.ArgumentMatchers.eq("collector-7"))).thenReturn(
                    new TestDataObservationAdapter.CollectorRealtime("collector-7", online, 1000, List.of(), null));
        }


 ruoyi-admin/src/test/java/com/battery/business/service/impl/TestDataReceptionServiceImplTest.java

@@ -59,13 +59,13 @@
        Clock clock = mock(Clock.class);
        when(clock.instant()).thenReturn(CAPTURED_AT);
        var service = new TestDataReceptionServiceImpl(fixture.mapper, fixture.tests, fixture.simulator, clock);
        when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(true,
        when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(true,
                List.of(collector("S-01", 10L, "temperature", "25.1", "℃"))));
        when(fixture.mapper.insertRealtimeRecord(any())).thenReturn(1);
        service.captureReceivedData();

        when(clock.instant()).thenReturn(CAPTURED_AT.plusSeconds(5));
        when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(true,
        when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(true,
                List.of(collector("S-01", 11L, "temperature", "38", "℃"))));
        service.captureReceivedData();

@@ -87,7 +87,7 @@
        test.getBatteries().get(0).setCollectorChannels(List.of(
                mapping("PRIMARY_TEMPERATURE", "S-01"), mapping("SECONDARY_TEMPERATURE_2", "S-03"),
                mapping("SECONDARY_TEMPERATURE_3", "S-05")));
        when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(true, List.of(
        when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(true, List.of(
                collector("S-01", 10L, "temperature", "38", "℃"),
                collector("S-03", 10L, "temperature", "25.3", "℃"),
                collector("S-05", 10L, "temperature", "25.5", "℃"))));
@@ -125,7 +125,7 @@
    {
        Fixture fixture = fixture();
        when(fixture.simulator.latestCabinet(any(ChannelIdentity.class))).thenReturn(cabinetRecord());
        when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(true, List.of(
        when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(true, List.of(
                collector("S-01", 10L, "TEMPERATURE", "25.100", "℃"),
                collector("S-02", 10L, "POSITIVE_POTENTIAL", "1.620", "V"),
                collector("S-03", 10L, "NEGATIVE_POTENTIAL", "1.610", "V"),
@@ -159,7 +159,7 @@
    {
        Fixture fixture = fixture();
        when(fixture.simulator.latestCabinet(any(ChannelIdentity.class))).thenThrow(new IllegalArgumentException("offline"));
        when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(true,
        when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(true,
                List.of(collector("S-01", 11L, "TEMPERATURE", "26.2", "℃"))));
        when(fixture.mapper.insertRealtimeRecord(any())).thenReturn(1);

@@ -216,7 +216,7 @@
    {
        Fixture fixture = fixture();
        when(fixture.simulator.latestCabinet(any(ChannelIdentity.class))).thenReturn(cabinetRecord());
        when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(false, List.of()));
        when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(false, List.of()));
        when(fixture.mapper.insertRealtimeRecord(any())).thenReturn(1);

        fixture.service().captureReceivedData();
@@ -235,7 +235,7 @@
    {
        Fixture fixture = fixture();
        when(fixture.simulator.latestCabinet(any(ChannelIdentity.class))).thenReturn(cabinetRecord());
        when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(false, List.of()));
        when(fixture.simulator.latestCollector(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.eq("collector-9"))).thenReturn(collectorSnapshot(false, List.of()));
        when(fixture.mapper.selectRealtimeRecordContentHash(eq(1L), eq(4L), any(Date.class)))
                .thenReturn("different-content");


 ruoyi-admin/src/test/java/com/battery/business/service/impl/VendorSampleReceptionServiceTest.java

New file
@@ -0,0 +1,261 @@
package com.battery.business.service.impl;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.Date;
import java.util.HexFormat;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import com.battery.device.api.TestDataObservationAdapter;
import com.battery.device.api.TestDataObservationAdapter.CabinetRealtime;
import com.battery.device.api.TestDataObservationAdapter.CollectorRealtime;
import com.battery.device.api.TestDataObservationAdapter.CycleRecord;
import com.battery.device.api.TestDataObservationAdapter.SampleWatermark;
import com.battery.device.api.TestDataObservationAdapter.StepRecord;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.model.DetectedChannel;
import com.battery.device.api.model.DeviceObservation;
import com.battery.business.domain.TTest;
import com.battery.business.domain.TTestBattery;
import com.battery.business.domain.TTestChannelBinding;
import com.battery.business.domain.TTestRealtimeRecord;
import com.battery.business.mapper.TTestMapper;
import com.battery.business.service.ITTestService;

/**
 * 逐行接收的归属边界：水位由"开始接收"与"继续接收"这一刻来源里的最新一行决定，
 * 只取严格晚于水位的数据行；同秒多行都保留；重复同内容幂等；同身份不同内容不覆盖。
 */
class VendorSampleReceptionServiceTest
{
    private static final Instant START = Instant.parse("2026-09-28T02:00:00Z");
    private static final ChannelIdentity CHANNEL = new ChannelIdentity("neware76", 21, 36, 1, 2);
    private static final String IDENTITY = "neware76:21:36:1:2";

    @Test
    void firstPollReadsTheWatermarkWithoutWritingDeviceHistory()
    {
        Fixture fixture = fixture();
        fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);

        fixture.service.captureReceivedData();

        assertEquals(1, fixture.adapter.watermarkReads);
        assertTrue(fixture.adapter.sampleReads.isEmpty());
        verify(fixture.mapper, never()).insertRealtimeRecord(any());
    }

    @Test
    void everyRowStrictlyAfterTheWatermarkBecomesATestFactIncludingSameSecondRows()
    {
        Fixture fixture = fixture();
        fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);
        fixture.service.captureReceivedData();

        // 同秒两行都落库：槽位留空，来源行身份把它们区分开。
        fixture.adapter.samples = List.of(sample(7437L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56"),
                sample(7438L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56"));
        fixture.clock.advance(Duration.ofSeconds(60));
        fixture.service.captureReceivedData();

        ArgumentCaptor<TTestRealtimeRecord> rows = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
        verify(fixture.mapper, times(2)).insertRealtimeRecord(rows.capture());
        assertEquals(List.of("neware:440:7437", "neware:440:7438"),
                rows.getAllValues().stream().map(TTestRealtimeRecord::getSourceRecordId).toList());
        assertTrue(rows.getAllValues().stream().allMatch(row -> row.getCaptureSlotUtc() == null));
        assertEquals(Instant.parse("2026-09-28T01:55:56Z"),
                rows.getAllValues().get(0).getCabinetDeviceTimeUtc().toInstant());
        assertEquals("2026-09-28 09:55:56", rows.getAllValues().get(0).getCabinetDeviceTimeRaw());
        assertEquals(IDENTITY, rows.getAllValues().get(0).getSourceIdentity());
    }

    @Test
    void resumingReceptionReadsANewWatermarkSoThePauseGapIsNotBackfilled()
    {
        Fixture fixture = fixture();
        fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);
        fixture.service.captureReceivedData();

        fixture.adapter.samples = List.of(sample(7437L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56"));
        fixture.clock.advance(Duration.ofSeconds(60));
        fixture.service.captureReceivedData();
        assertEquals(List.of(new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L)),
                fixture.adapter.sampleReads);

        // 暂停后继续：接收边界推进，水位重新读取；暂停期间产生的行不再进本测试。
        fixture.test.setReceptionBoundaryAtUtc(Date.from(START.plusSeconds(600)));
        fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T02:05:56Z"), 7500L);
        fixture.adapter.samples = List.of();
        fixture.clock.advance(Duration.ofSeconds(60));
        fixture.service.captureReceivedData();

        assertEquals(2, fixture.adapter.watermarkReads);
        // 取到新水位后，只追加新水位之后的行；暂停期间的空档不会被补回来。
        fixture.clock.advance(Duration.ofSeconds(60));
        fixture.service.captureReceivedData();
        assertEquals(List.of(new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L),
                new SampleWatermark(Instant.parse("2026-09-28T02:05:56Z"), 7500L)), fixture.adapter.sampleReads);
    }

    @Test
    void repeatedIdenticalRowIsIgnoredAndConflictingContentIsNotOverwritten()
    {
        Fixture fixture = fixture();
        fixture.adapter.watermark = new SampleWatermark(Instant.parse("2026-09-28T01:54:56Z"), 7436L);
        fixture.service.captureReceivedData();

        CabinetRealtime sample = sample(7437L, "2026-09-28T01:55:56Z", "2026-09-28 09:55:56");
        fixture.adapter.samples = List.of(sample);
        fixture.clock.advance(Duration.ofSeconds(60));
        fixture.service.captureReceivedData();
        ArgumentCaptor<TTestRealtimeRecord> first = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
        verify(fixture.mapper).insertRealtimeRecord(first.capture());

        // 同身份同内容：幂等忽略，不再插入。
        when(fixture.mapper.selectRealtimeRecordHashBySourceRecord(4L, IDENTITY, "neware:440:7437"))
                .thenReturn(HexFormat.of().formatHex(first.getValue().getContentSha256()));
        fixture.adapter.watermark = new SampleWatermark(sample.deviceTime(), sample.sequenceId());
        fixture.clock.advance(Duration.ofSeconds(60));
        fixture.service.captureReceivedData();
        verify(fixture.mapper, times(1)).insertRealtimeRecord(any());

        // 同身份不同内容：保留原事实，拒绝覆盖。
        when(fixture.mapper.selectRealtimeRecordHashBySourceRecord(4L, IDENTITY, "neware:440:7437"))
                .thenReturn("0000000000000000000000000000000000000000000000000000000000000000");
        fixture.clock.advance(Duration.ofSeconds(60));
        fixture.service.captureReceivedData();
        verify(fixture.mapper, times(1)).insertRealtimeRecord(any());
    }

    @Test
    void sourceWithoutWatermarkKeepsThePollingSlotPath()
    {
        Fixture fixture = fixture();
        // 不提供水位的来源（例如开发模拟器）继续按轮询槽位保存柜端事实。
        fixture.adapter.watermark = null;
        fixture.adapter.latest = sample(9L, "2026-09-28T02:00:30Z", "2026-09-28 10:00:30");

        fixture.service.captureReceivedData();

        assertEquals(1, fixture.adapter.watermarkReads);
        assertTrue(fixture.adapter.sampleReads.isEmpty());
        ArgumentCaptor<TTestRealtimeRecord> rows = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
        verify(fixture.mapper).insertRealtimeRecord(rows.capture());
        assertTrue(rows.getValue().getCaptureSlotUtc() != null);
    }

    private static CabinetRealtime sample(long sequenceId, String deviceTime, String raw)
    {
        return new CabinetRealtime(sequenceId, Instant.parse(deviceTime), -1.1889, 1.3733, 1.398611, 0.030648,
                3, 12, null, "2", Instant.parse("2026-09-28T02:01:00Z"), raw, "Asia/Hong_Kong", 300L,
                "neware:440:" + sequenceId);
    }

    private static Fixture fixture()
    {
        TTestMapper mapper = mock(TTestMapper.class);
        ITTestService tests = mock(ITTestService.class);
        AdaptiveAdapter adapter = new AdaptiveAdapter();
        TTest test = receivingTest();
        when(mapper.selectReceivingTestIds()).thenReturn(List.of(1L));
        when(mapper.selectTestByIdForUpdate(1L)).thenReturn(test);
        when(mapper.selectCollectorCode(9L)).thenReturn("collector-9");
        when(tests.selectTestById(1L)).thenReturn(test);
        TestClock clock = new TestClock(START);
        return new Fixture(mapper, tests, adapter, test, clock,
                new TestDataReceptionServiceImpl(mapper, tests, adapter, clock));
    }

    private static TTest receivingTest()
    {
        TTestChannelBinding binding = new TTestChannelBinding();
        binding.setBindingId(2L);
        binding.setChannelId(3L);
        binding.setAdapterCode("neware76");
        binding.setAdapterTimezoneId("Asia/Hong_Kong");
        binding.setDevtype(21);
        binding.setDevid(36);
        binding.setSubdevid(1);
        binding.setChlid(2);
        TTestBattery battery = new TTestBattery();
        battery.setTestBatteryId(4L);
        battery.setCollectorId(9L);
        battery.setBinding(binding);
        battery.setCollectorChannels(List.of());
        TTest test = new TTest();
        test.setTestId(1L);
        test.setTestStatus("RUNNING");
        test.setReceptionStatus("RECEIVING");
        test.setStartedAtUtc(Date.from(START));
        test.setReceptionBoundaryAtUtc(Date.from(START));
        test.setBatteries(List.of(battery));
        return test;
    }

    private record Fixture(TTestMapper mapper, ITTestService tests, AdaptiveAdapter adapter, TTest test,
            TestClock clock, TestDataReceptionServiceImpl service) { }

    /** 只暴露水位与逐行读取的测试替身，并记录被问到的水位与逐行读取次数。 */
    private static final class AdaptiveAdapter implements TestDataObservationAdapter
    {
        private SampleWatermark watermark;
        private List<CabinetRealtime> samples = new ArrayList<>();
        private CabinetRealtime latest;
        private int watermarkReads;
        private final List<SampleWatermark> sampleReads = new ArrayList<>();

        @Override public String adapterCode() { return "neware76"; }
        @Override public AdapterCapabilities capabilities() { return AdapterCapabilities.observeOnly(); }
        @Override public List<DetectedChannel> discoverChannels() { return List.of(); }
        @Override public Map<ChannelIdentity, DeviceObservation> observe(Set<ChannelIdentity> channels)
        { return Map.of(); }
        @Override public CabinetRealtime latestCabinet(ChannelIdentity channel) { return latest; }
        @Override public CollectorRealtime latestCollector(String adapterCode, String collectorCode) { return null; }
        @Override public List<StepRecord> stepRecords(ChannelIdentity channel) { return List.of(); }
        @Override public List<CycleRecord> cycleRecords(ChannelIdentity channel) { return List.of(); }
        @Override public long persistenceIntervalMillis(ChannelIdentity channel) { return 60000L; }

        @Override public SampleWatermark currentWatermark(ChannelIdentity channel)
        {
            watermarkReads++;
            return watermark;
        }

        @Override public List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark since)
        {
            sampleReads.add(since);
            return List.copyOf(samples);
        }
    }

    /** 可推进的测试时钟：接收节奏与厂商写库节奏都按它推进。 */
    private static final class TestClock extends Clock
    {
        private Instant now;

        private TestClock(Instant now) { this.now = now; }

        private void advance(Duration duration) { now = now.plus(duration); }

        @Override public ZoneId getZone() { return ZoneOffset.UTC; }
        @Override public Clock withZone(ZoneId zone) { return this; }
        @Override public Instant instant() { return now; }
    }
}

 ruoyi-admin/src/test/java/com/battery/business/web/VendorSamplePersistenceDatabaseTest.java

New file
@@ -0,0 +1,156 @@
package com.battery.business.web;

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

import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.time.Instant;
import java.util.Date;
import java.util.HexFormat;
import java.util.Map;
import java.util.UUID;
import org.apache.ibatis.datasource.unpooled.UnpooledDataSource;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.SqlSessionTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
import com.battery.business.domain.TTestRealtimeRecord;
import com.battery.business.mapper.TTestMapper;

/**
 * 逐行接收的落库语义（生产 Mapper + H2 内存库）：
 * 设备数据行不占系统轮询槽位，所以同一绝对秒内的多行可以各自成立；
 * 来源行身份（厂商测试编号 + 协议序号）承担幂等与冲突判定。
 */
public class VendorSamplePersistenceDatabaseTest
{
    private static final String IDENTITY = "neware76:21:36:1:2";

    private java.sql.Connection keepAlive;
    private JdbcTemplate jdbc;
    private TTestMapper mapper;

    @BeforeEach void setUp() throws Exception
    {
        var source = new UnpooledDataSource("org.h2.Driver",
                "jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DATABASE_TO_LOWER=TRUE", "sa", "");
        keepAlive = source.getConnection();
        jdbc = new JdbcTemplate(source);
        // H2 无 MySQL 的 HEX()；测试中补一个等价别名，生产 SQL 仍在 MySQL 上验证。
        jdbc.execute("create alias HEX for 'com.battery.business.web.VendorSamplePersistenceDatabaseTest.hexOf'");
        jdbc.execute("create table t_test_realtime_record(record_id bigint auto_increment primary key,"
                + "test_id bigint, test_battery_id bigint, binding_id bigint, adapter_code varchar(64),"
                + "source_identity varchar(256) not null, source_record_id varchar(128),"
                + "collector_id bigint, collector_channel_code varchar(100), collector_sequence_id bigint,"
                + "collector_device_time_utc timestamp(3), collector_device_time_raw varchar(64),"
                + "collector_adapter_timezone_id varchar(64),"
                + "cabinet_devtype int, cabinet_devid int, cabinet_subdevid int, cabinet_chlid int,"
                + "cabinet_sequence_id bigint, cabinet_device_time_utc timestamp(3), cabinet_device_time_raw varchar(64),"
                + "cabinet_adapter_timezone_id varchar(64), cycle_no int, step_no int, step_type varchar(64),"
                + "step_name varchar(191), voltage_v decimal(30,12), current_a decimal(30,12),"
                + "capacity_ah decimal(30,12), energy_wh decimal(30,12), temperature decimal(30,12),"
                + "positive_potential decimal(30,12), negative_potential decimal(30,12),"
                + "extension_measurements_json varchar(2000), capture_slot_utc timestamp(3) null, "
                + "server_received_at_utc timestamp(3) not null, content_sha256 binary(32) not null,"
                // 轮询槽位唯一键允许多个空值：逐行接收因此可以把同一秒的多行都保存下来。
                + "unique(test_id, test_battery_id, capture_slot_utc))");

        SqlSessionFactoryBean factory = new SqlSessionFactoryBean();
        factory.setDataSource(source);
        factory.setTypeAliasesPackage("com.battery.business.domain");
        factory.setMapperLocations(new org.springframework.core.io.ClassPathResource("mapper/battery/TTestMapper.xml"));
        mapper = new SqlSessionTemplate(factory.getObject()).getMapper(TTestMapper.class);
    }

    @AfterEach void closeDatabase() throws Exception { if (keepAlive != null) keepAlive.close(); }

    @Test void sameSecondRowsAreKeptSeparatelyAndRepeatIsRecognisedBySourceRowIdentity()
    {
        TTestRealtimeRecord first = row(7437L, "2026-09-28 09:55:56", "2026-09-28T01:55:56Z");
        assertEquals(1, mapper.insertRealtimeRecord(first));

        // 同一绝对秒的第二行：槽位为空，唯一键不冲突，两行都成立。
        TTestRealtimeRecord second = row(7438L, "2026-09-28 09:55:56", "2026-09-28T01:55:56Z");
        assertEquals(1, mapper.insertRealtimeRecord(second));

        assertEquals(2, jdbc.queryForObject("select count(*) from t_test_realtime_record", Integer.class));
        // 幂等与冲突按来源行身份判定：命中同身份即视为重复，由调用方比较内容摘要。
        assertEquals(hex(first.getContentSha256()),
                mapper.selectRealtimeRecordHashBySourceRecord(4L, IDENTITY, "neware:440:7437"));
        assertNull(mapper.selectRealtimeRecordHashBySourceRecord(4L, IDENTITY, "neware:440:7439"));
        assertTrue(!hex(first.getContentSha256()).equals(hex(second.getContentSha256())));
    }

    @Test void deviceRowKeepsVendorRawTimeServerReceiveTimeAndNoPollingSlot()
    {
        mapper.insertRealtimeRecord(row(7437L, "2026-09-28 09:55:56", "2026-09-28T01:55:56Z"));

        Map<String, Object> stored = jdbc.queryForMap("select source_record_id, source_identity,"
                + " cabinet_device_time_raw, cabinet_device_time_utc, server_received_at_utc, capture_slot_utc,"
                + " cabinet_adapter_timezone_id, extension_measurements_json, voltage_v"
                + " from t_test_realtime_record");
        assertEquals("neware:440:7437", stored.get("source_record_id"));
        assertEquals(IDENTITY, stored.get("source_identity"));
        // 设备时间按适配器时区解释后以 UTC 保存，同时保留原始时间文本与服务器接收时间。
        assertEquals("2026-09-28 09:55:56", stored.get("cabinet_device_time_raw"));
        assertEquals("Asia/Hong_Kong", stored.get("cabinet_adapter_timezone_id"));
        assertEquals(Instant.parse("2026-09-28T01:55:56Z"),
                ((java.sql.Timestamp) stored.get("cabinet_device_time_utc")).toInstant());
        assertEquals(Instant.parse("2026-09-28T02:01:00Z"),
                ((java.sql.Timestamp) stored.get("server_received_at_utc")).toInstant());
        assertNull(stored.get("capture_slot_utc"));
        // H2 的 JSON 列会把 JSON 文本再引一层，这里只断言内容确实保存下来。
        assertTrue(String.valueOf(stored.get("extension_measurements_json")).contains("relativeTimeSeconds"));
        assertEquals(0, new BigDecimal("1.3733").compareTo((BigDecimal) stored.get("voltage_v")));
    }

    /** 供 H2 别名使用的 HEX 等价实现；H2 没有 MySQL 的 HEX()。 */
    public static String hexOf(byte[] value)
    {
        return value == null ? null : HexFormat.of().formatHex(value);
    }

    private TTestRealtimeRecord row(long sequenceId, String raw, String deviceTimeUtc)
    {
        TTestRealtimeRecord row = new TTestRealtimeRecord();
        row.setTestId(1L); row.setTestBatteryId(4L); row.setBindingId(2L); row.setAdapterCode("neware76");
        row.setSourceIdentity(IDENTITY); row.setSourceRecordId("neware:440:" + sequenceId);
        row.setCabinetDevtype(21); row.setCabinetDevid(36); row.setCabinetSubdevid(1); row.setCabinetChlid(2);
        row.setCabinetSequenceId(sequenceId);
        row.setCabinetDeviceTimeUtc(Date.from(Instant.parse(deviceTimeUtc)));
        row.setCabinetDeviceTimeRaw(raw);
        row.setCabinetAdapterTimezoneId("Asia/Hong_Kong");
        row.setCaptureSlotUtc(null);
        row.setServerReceivedAtUtc(Date.from(Instant.parse("2026-09-28T02:01:00Z")));
        row.setCycleNo(3); row.setStepNo(12); row.setStepType("2");
        row.setVoltageV(new BigDecimal("1.3733")); row.setCurrentA(new BigDecimal("-1.1889"));
        row.setCapacityAh(new BigDecimal("1.398612")); row.setEnergyWh(new BigDecimal("0.030648"));
        row.setExtensionMeasurementsJson("{\"relativeTimeSeconds\":300}");
        // 内容摘要把协议序号算进去，使同秒不同行的内容彼此不同。
        row.setContentSha256(sha256("neware:440:" + sequenceId));
        return row;
    }

    private static String hex(byte[] value)
    {
        return HexFormat.of().formatHex(value);
    }

    private static byte[] sha256(String value)
    {
        try
        {
            return MessageDigest.getInstance("SHA-256").digest(value.getBytes(StandardCharsets.UTF_8));
        }
        catch (Exception exception)
        {
            throw new IllegalStateException(exception);
        }
    }

}