fix(isolation): distinguish accident links from same-numbered nodes

Prefer link endpoints for legacy accident IDs and allow explicit node selection. Read authoritative initial statuses and sources, verify source isolation, and separate direct isolation from newly disconnected supply nodes. Return validation errors as HTTP 422.

The node-first lookup misrouted adjacent Fengyang pipes 19025 and 19026 into unrelated regions. Regression cases fail against the previous algorithm and pass after this change.

app/algorithms/valve_isolation/topology_search.py

@@ -1,10 +1,11 @@
 """Topology-only valve isolation search."""
 
 from collections import defaultdict, deque
-from typing import Any, Iterable
+from typing import Any, Iterable, Literal
 
 
 VALVE_LINK_TYPE = "valve"
+AccidentElementType = Literal["auto", "link", "node"]
 
 
 def _parse_link_entry(link_entry: str) -> tuple[str, str, str, str]:
@@ -14,12 +15,37 @@ def _parse_link_entry(link_entry: str) -> tuple[str, str, str, str]:
     return parts[0], parts[1], parts[2], parts[3]
 
 
+def _reachable_nodes(
+    adjacency: dict[str, list[tuple[str, str]]],
+    start_nodes: Iterable[str],
+    blocked_links: set[str],
+) -> set[str]:
+    visited = set(start_nodes)
+    queue = deque(visited)
+    while queue:
+        for neighbor, link_id in adjacency.get(queue.popleft(), ()):
+            if link_id not in blocked_links and neighbor not in visited:
+                visited.add(neighbor)
+                queue.append(neighbor)
+    return visited
+
+
 def valve_isolation_analysis(
     link_entries: Iterable[str],
     accident_elements: str | list[str],
     disabled_valves: list[str] | None = None,
+    *,
+    accident_element_type: AccidentElementType = "auto",
+    water_source_nodes: Iterable[str] | None = None,
+    closed_links: Iterable[str] = (),
 ) -> dict[str, Any]:
-    """Determine boundary valves and affected nodes from a topology snapshot."""
+    """Find nearest valve boundaries, then check source connectivity.
+
+    Legacy untyped IDs prefer links; explicit node mode handles cross-type ID
+    collisions. Supply impact is a before/after connectivity difference, not
+    a hydraulic pressure or demand calculation. Without source metadata only
+    the provisional boundary search is available.
+    """
 
     disabled_valves_set = set(disabled_valves or [])
     target_elements = (
@@ -27,8 +53,13 @@ def valve_isolation_analysis(
         if isinstance(accident_elements, str)
         else accident_elements
     )
+    if not target_elements:
+        raise ValueError("At least one accident element is required")
+    if accident_element_type not in {"auto", "link", "node"}:
+        raise ValueError(f"Invalid accident element type: {accident_element_type}")
+    closed_links_set = set(closed_links)
 
-    pipe_adj: dict[str, set[str]] = defaultdict(set)
+    adjacency: dict[str, list[tuple[str, str]]] = defaultdict(list)
     all_valves: dict[str, tuple[str, str]] = {}
     link_lookup: dict[str, tuple[str, str, str]] = {}
     node_set: set[str] = set()
@@ -39,46 +70,31 @@ def valve_isolation_analysis(
         node_set.update((node1, node2))
         if link_type_name == VALVE_LINK_TYPE:
             all_valves[link_id] = (node1, node2)
-        else:
+        if link_id not in closed_links_set:
-            pipe_adj[node1].add(node2)
+            adjacency[node1].append((node2, link_id))
-            pipe_adj[node2].add(node1)
+            adjacency[node2].append((node1, link_id))
+
+    unknown_disabled = disabled_valves_set - all_valves.keys()
+    if unknown_disabled:
+        raise ValueError(f"Disabled valve was not found in topology: {sorted(unknown_disabled)}")
 
     start_nodes: set[str] = set()
     for element in target_elements:
-        if element in node_set:
+        if accident_element_type != "node" and element in link_lookup:
-            start_nodes.add(element)
-        elif element in link_lookup:
             node1, node2, _ = link_lookup[element]
             start_nodes.update((node1, node2))
+        elif accident_element_type != "link" and element in node_set:
+            start_nodes.add(element)
         else:
             raise ValueError(f"Accident element {element} was not found in topology")
 
-    extra_adj: dict[str, list[str]] = defaultdict(list)
+    closable_valves = set(all_valves) - disabled_valves_set - closed_links_set
-    boundary_valves: dict[str, tuple[str, str]] = {}
+    affected_nodes = _reachable_nodes(adjacency, start_nodes, closable_valves)
-    for valve_id, (node1, node2) in all_valves.items():
-        if valve_id in disabled_valves_set:
-            extra_adj[node1].append(node2)
-            extra_adj[node2].append(node1)
-        else:
-            boundary_valves[valve_id] = (node1, node2)
-
-    affected_nodes: set[str] = set()
-    queue = deque(start_nodes)
-    while queue:
-        node = queue.popleft()
-        if node in affected_nodes:
-            continue
-        affected_nodes.add(node)
-        queue.extend(pipe_adj.get(node, set()) - affected_nodes)
-        queue.extend(
-            neighbor
-            for neighbor in extra_adj.get(node, ())
-            if neighbor not in affected_nodes
-        )
 
     must_close_valves: list[str] = []
     optional_valves: list[str] = []
-    for valve_id, (node1, node2) in boundary_valves.items():
+    for valve_id in closable_valves:
+        node1, node2 = all_valves[valve_id]
         node1_affected = node1 in affected_nodes
         node2_affected = node2 in affected_nodes
         if node1_affected and node2_affected:
@@ -89,14 +105,37 @@ def valve_isolation_analysis(
     must_close_valves.sort()
     optional_valves.sort()
     isolatable = bool(must_close_valves)
+    failure_reason = None if isolatable else "no_boundary_valves"
+    sources = set(water_source_nodes or ())
+    source_check_performed = bool(sources)
+    supply_affected_nodes = None
+    if water_source_nodes is not None:
+        if not sources:
+            isolatable = False
+            failure_reason = "water_sources_unavailable"
+        else:
+            after_closure = _reachable_nodes(adjacency, sources, set(must_close_valves))
+            if affected_nodes & after_closure:
+                isolatable = False
+                failure_reason = "connected_to_water_source"
+            if isolatable:
+                before_closure = _reachable_nodes(adjacency, sources, set())
+                supply_affected_nodes = sorted(before_closure - after_closure)
     result: dict[str, Any] = {
         "accident_elements": target_elements,
+        "accident_element_type": accident_element_type,
         "disabled_valves": disabled_valves,
         "affected_nodes": sorted(affected_nodes) if isolatable else [],
         "affected_node_count": len(affected_nodes),
         "must_close_valves": must_close_valves,
         "optional_valves": optional_valves,
         "isolatable": isolatable,
+        "source_check_performed": source_check_performed,
+        "isolation_failure_reason": failure_reason,
+        "supply_affected_nodes": supply_affected_nodes,
+        "supply_affected_node_count": (
+            len(supply_affected_nodes) if supply_affected_nodes is not None else None
+        ),
     }
     if len(target_elements) == 1:
         result["accident_element"] = target_elements[0]

app/api/v1/endpoints/admin_metadata.py

@@ -7,6 +7,7 @@ from sqlalchemy.engine.url import make_url
 from sqlalchemy.exc import IntegrityError, SQLAlchemyError
 from sqlalchemy.ext.asyncio import create_async_engine
 
+from app.api.pagination import PaginatedList
 from app.auth.metadata_dependencies import (
     get_current_metadata_admin,
     get_metadata_repository,
@@ -236,7 +237,9 @@ async def list_metadata_users(
     metadata_repo: MetadataRepository = Depends(get_metadata_repository),
 ) -> List[MetadataUserResponse]:
     users = await metadata_repo.list_users(skip=skip, limit=limit)
-    return [MetadataUserResponse.model_validate(user) for user in users]
+    total = await metadata_repo.count_users()
+    items = [MetadataUserResponse.model_validate(user) for user in users]
+    return PaginatedList(items, total=total)
 
 
 @router.get("/admin/projects", response_model=List[AdminProjectResponse])

app/api/v1/endpoints/simulation.py

@@ -180,6 +180,9 @@ def valve_isolation_endpoint(
     network: str = Query(..., description="管网名称（或数据库名称）"),
     accident_element: List[str] = Query(..., description="发生事故的管段/节点ID列表"),
     disabled_valves: List[str] = Query(None, description="已故障的阀门ID列表（可选）"),
+    accident_element_type: Literal["auto", "link", "node"] = Query(
+        "auto", description="事故要素类型；auto优先管段再节点；分析同名节点须指定node，纯管段分析可指定link",
+    ),
 ):
     """
     阀门隔离分析
@@ -187,13 +190,19 @@ def valve_isolation_endpoint(
     - **network**: 管网名称（或数据库名称）
     - **accident_element**: 发生事故的管段/节点ID列表
     - **disabled_valves**: 已故障的阀门ID列表（可选）
+    - **accident_element_type**: auto优先管段；link仅匹配管段；node仅匹配节点
     
     返回隔离方案，包括：
     - must_close_valves: 必须关闭的阀门列表
     - optional_valves: 可选关闭的阀门列表
-    - affected_nodes: 受影响的节点列表；不可隔离时为空列表
+    - affected_nodes / affected_node_count: 最近阀门围成的直接隔离区；不可隔离时列表为空，计数保留
-    - affected_node_count: 受影响的节点总数
+    - supply_affected_nodes / supply_affected_node_count: 按建议关阀后新增失去水源连接的节点；未验证或不可隔离时为null
-    - isolatable: 是否可以有效隔离
+    - source_check_performed: 是否已执行水源连通检查
+    - isolation_failure_reason: 不可隔离原因；可隔离时为null
+    - isolatable: 存在关阀边界，且直接隔离区不再连通任何水源
+
+    使用模型初始开关状态进行无向拓扑分析，水池按潜在水源处理。
+    不包含时变控制、水池蓄水量或压力流量计算，不是现场停水范围的水力证明。
     """
     # result = {
     #     "accident_element": "P461309",
@@ -212,7 +221,13 @@ def valve_isolation_endpoint(
     #     "must_close_valves": ["210521658", "V12974", "V12986", "V12993"],
     #     "optional_valves": [],
     # }
-    result = analyze_valve_isolation(network, accident_element, disabled_valves)
+    try:
+        result = analyze_valve_isolation(
+            network, accident_element, disabled_valves,
+            accident_element_type=accident_element_type,
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
     return result
 
 

app/api/v1/endpoints/timeseries/composite.py

@@ -1,183 +1,34 @@
 from fastapi import APIRouter, Depends, HTTPException, Query
 from datetime import datetime
 from psycopg import AsyncConnection
-from uuid import UUID
 
 from app.services.timeseries_analysis import TimeseriesAnalysisService
+from app.domain.schemas.timeseries_history import (
+    ElementHistoryQuery,
+    ElementHistoryResponse,
+)
+from app.services.timeseries_history import TimeseriesHistoryService
 from .dependencies import get_timescale_connection, get_postgres_connection
 
 router = APIRouter()
 
 
-@router.get("/timeseries/views/scada-simulations", summary="获取SCADA关联的模拟数据")
+@router.post(
-async def get_scada_associated_simulation_data(
+    "/timeseries/views/element-history/query",
-    start_time: datetime = Query(..., description="查询开始时间"),
+    summary="批量查询管网元素历史数据",
-    end_time: datetime = Query(..., description="查询结束时间"),
+    response_model=ElementHistoryResponse,
-    device_ids: str = Query(..., description="SCADA设备ID列表，逗号分隔"),
+)
-    run_id: UUID | None = Query(None, description="分析运行 ID；为空时查询实时数据"),
+async def query_element_history(
+    payload: ElementHistoryQuery,
     timescale_conn: AsyncConnection = Depends(get_timescale_connection),
     postgres_conn: AsyncConnection = Depends(get_postgres_connection),
-):
+) -> ElementHistoryResponse:
-    """
-    获取SCADA关联的link/node模拟值
-    
-    根据传入的SCADA device_ids，找到关联的link/node，
-    并根据对应的type，查询对应的模拟数据。支持查询实时或分析运行数据。
-
-    Args:
-        start_time: 查询开始时间
-        end_time: 查询结束时间
-        device_ids: SCADA设备ID列表，用逗号分隔
-        run_id: 分析运行 ID，若为空则查询实时数据
-        timescale_conn: TimescaleDB连接
-        postgres_conn: PostgreSQL连接
-        
-    Returns:
-        SCADA关联的模拟数据
-        
-    Raises:
-        HTTPException: 当查询参数无效时返回400错误，未找到数据时返回404错误
-    """
     try:
-        device_ids_list = (
+        return await TimeseriesHistoryService.query(
-            [id.strip() for id in device_ids.split(",") if id.strip()]
+            timescale_conn, postgres_conn, payload
-            if device_ids
-            else []
         )
-
+    except ValueError as exc:
-        if run_id is not None:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
-            result = await TimeseriesAnalysisService.get_scada_associated_analysis_simulation_data(
-                timescale_conn,
-                postgres_conn,
-                device_ids_list,
-                start_time,
-                end_time,
-                run_id,
-            )
-        else:
-            result = (
-                await TimeseriesAnalysisService.get_scada_associated_realtime_simulation_data(
-                    timescale_conn,
-                    postgres_conn,
-                    device_ids_list,
-                    start_time,
-                    end_time,
-                )
-            )
-        if result is None:
-            raise HTTPException(status_code=404, detail="No simulation data found")
-        return result
-    except ValueError as e:
-        raise HTTPException(status_code=400, detail=str(e))
-
-
-@router.get("/timeseries/views/element-simulations", summary="获取管网元素的模拟数据")
-async def get_feature_simulation_data(
-    start_time: datetime = Query(..., description="查询开始时间"),
-    end_time: datetime = Query(..., description="查询结束时间"),
-    feature_infos: str = Query(
-        ..., description="特征信息，格式: id1:type1,id2:type2，type为pipe（管道）或junction（节点）"
-    ),
-    run_id: UUID | None = Query(None, description="分析运行 ID；为空时查询实时数据"),
-    timescale_conn: AsyncConnection = Depends(get_timescale_connection),
-):
-    """
-    获取link/node模拟值
-    
-    根据传入的featureInfos，找到关联的link/node，
-    并根据对应的type，查询对应的模拟数据。支持查询实时或分析运行数据。
-
-    Args:
-        start_time: 查询开始时间
-        end_time: 查询结束时间
-        feature_infos: 格式为 "element_id1:type1,element_id2:type2"
-                      例如: "P1:pipe,J1:junction"
-        run_id: 分析运行 ID，若为空则查询实时数据
-        timescale_conn: TimescaleDB连接
-        
-    Returns:
-        管网元素的模拟数据
-        
-    Raises:
-        HTTPException: 当feature_infos为空返回400错误，未找到数据返回404错误，其他错误返回400错误
-    """
-    try:
-        feature_infos_list = []
-        if feature_infos:
-            for item in feature_infos.split(","):
-                item = item.strip()
-                if ":" in item:
-                    element_id, element_type = item.split(":", 1)
-                    feature_infos_list.append(
-                        (element_id.strip(), element_type.strip())
-                    )
-
-        if not feature_infos_list:
-            raise HTTPException(status_code=400, detail="feature_infos cannot be empty")
-
-        if run_id is not None:
-            result = await TimeseriesAnalysisService.get_analysis_simulation_data(
-                timescale_conn,
-                feature_infos_list,
-                start_time,
-                end_time,
-                run_id,
-            )
-        else:
-            result = await TimeseriesAnalysisService.get_realtime_simulation_data(
-                timescale_conn,
-                feature_infos_list,
-                start_time,
-                end_time,
-            )
-
-        if result is None:
-            raise HTTPException(status_code=404, detail="No simulation data found")
-        return result
-    except ValueError as e:
-        raise HTTPException(status_code=400, detail=str(e))
-
-
-@router.get("/timeseries/views/element-scada-readings", summary="获取管网元素关联的SCADA监测数据")
-async def get_element_associated_scada_data(
-    element_id: str = Query(..., description="管网元素ID（管道或节点）"),
-    start_time: datetime = Query(..., description="查询开始时间"),
-    end_time: datetime = Query(..., description="查询结束时间"),
-    use_cleaned: bool = Query(False, description="是否使用清洗后的数据"),
-    timescale_conn: AsyncConnection = Depends(get_timescale_connection),
-    postgres_conn: AsyncConnection = Depends(get_postgres_connection),
-):
-    """
-    获取link/node关联的SCADA监测值
-    
-    根据传入的link/node id，匹配SCADA信息，
-    如果存在关联的SCADA device_id，获取实际的监测数据。
-
-    Args:
-        element_id: 管网元素ID
-        start_time: 查询开始时间
-        end_time: 查询结束时间
-        use_cleaned: 是否使用清洗后的数据，默认为False使用原始数据
-        timescale_conn: TimescaleDB连接
-        postgres_conn: PostgreSQL连接
-        
-    Returns:
-        管网元素关联的SCADA监测数据
-        
-    Raises:
-        HTTPException: 当查询参数无效时返回400错误，未找到关联数据返回404错误
-    """
-    try:
-        result = await TimeseriesAnalysisService.get_element_associated_scada_data(
-            timescale_conn, postgres_conn, element_id, start_time, end_time, use_cleaned
-        )
-        if result is None:
-            raise HTTPException(
-                status_code=404, detail="No associated SCADA data found"
-            )
-        return result
-    except ValueError as e:
-        raise HTTPException(status_code=400, detail=str(e))
 
 
 @router.post("/timeseries/scada-cleaning-runs", summary="清洗SCADA监测数据")

app/domain/measurement_units.py

@@ -0,0 +1,13 @@
+from typing import Literal
+
+
+Metric = Literal["flow", "pressure", "velocity"]
+
+DISPLAY_UNITS: dict[Metric, str] = {
+    "flow": "m³/h",
+    "pressure": "m",
+    "velocity": "m/s",
+}
+
+def display_unit_for_metric(metric: Metric) -> str:
+    return DISPLAY_UNITS[metric]

app/domain/schemas/scada.py

@@ -9,6 +9,7 @@ class ScadaDeviceResponse(BaseModel):
     node_id: str | None = None
     link_id: str | None = None
     api_query_id: str | None = None
+    measurement_unit: str
     transmission_mode: str
     transmission_frequency: str
     reliability: int | None = None

app/domain/schemas/timeseries_history.py

@@ -0,0 +1,88 @@
+from datetime import datetime
+from enum import StrEnum
+from uuid import UUID
+
+from pydantic import BaseModel, Field, field_validator, model_validator
+
+
+class HistoryElementType(StrEnum):
+    PIPE = "pipe"
+    JUNCTION = "junction"
+
+
+class HistoryMode(StrEnum):
+    OBSERVED = "observed"
+    REALTIME_COMPARISON = "realtime_comparison"
+    ANALYSIS_COMPARISON = "analysis_comparison"
+
+
+class HistoryMetric(StrEnum):
+    FLOW = "flow"
+    PRESSURE = "pressure"
+
+
+class HistorySource(StrEnum):
+    SCADA_RAW = "scada_raw"
+    SCADA_CLEANED = "scada_cleaned"
+    REALTIME_SIMULATION = "realtime_simulation"
+    ANALYSIS_SIMULATION = "analysis_simulation"
+
+
+class ElementHistoryTarget(BaseModel):
+    element_id: str = Field(min_length=1, max_length=128)
+    element_type: HistoryElementType
+    device_ids: list[str] | None = Field(default=None, max_length=100)
+
+    @field_validator("element_id")
+    @classmethod
+    def normalize_element_id(cls, value: str) -> str:
+        return value.strip()
+
+    @field_validator("device_ids")
+    @classmethod
+    def normalize_device_ids(cls, value: list[str] | None) -> list[str] | None:
+        if value is None:
+            return None
+        normalized = list(dict.fromkeys(item.strip() for item in value if item.strip()))
+        if not normalized:
+            raise ValueError("device_ids must contain at least one device")
+        return normalized
+
+
+class ElementHistoryQuery(BaseModel):
+    start_time: datetime
+    end_time: datetime
+    mode: HistoryMode = HistoryMode.OBSERVED
+    run_id: UUID | None = None
+    elements: list[ElementHistoryTarget] = Field(min_length=1, max_length=200)
+
+    @model_validator(mode="after")
+    def validate_query(self) -> "ElementHistoryQuery":
+        if self.start_time >= self.end_time:
+            raise ValueError("start_time must be earlier than end_time")
+        if self.mode == HistoryMode.ANALYSIS_COMPARISON and self.run_id is None:
+            raise ValueError("run_id is required for analysis_comparison")
+        if self.mode != HistoryMode.ANALYSIS_COMPARISON and self.run_id is not None:
+            raise ValueError("run_id is only valid for analysis_comparison")
+        return self
+
+
+class HistoryPoint(BaseModel):
+    time: datetime
+    value: float | None
+
+
+class ElementHistorySeries(BaseModel):
+    element_id: str
+    element_type: HistoryElementType
+    device_id: str | None = None
+    metric: HistoryMetric
+    source: HistorySource
+    source_unit: str = Field(description="Unit used by the returned point values")
+    display_unit: str = Field(description="Recommended UI display unit")
+    unit_inferred: bool = False
+    points: list[HistoryPoint]
+
+
+class ElementHistoryResponse(BaseModel):
+    series: list[ElementHistorySeries]

app/infra/db/metadb/repositories/metadata_repository.py

@@ -4,7 +4,7 @@ from typing import Optional, List
 from uuid import UUID, uuid4
 
 from cryptography.fernet import InvalidToken
-from sqlalchemy import delete, select
+from sqlalchemy import delete, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.encryption import (
@@ -122,6 +122,12 @@ class MetadataRepository:
         )
         return list(result.scalars().all())
 
+    async def count_users(self) -> int:
+        result = await self.session.execute(
+            select(func.count()).select_from(models.User)
+        )
+        return int(result.scalar() or 0)
+
     async def upsert_user_from_keycloak(
         self,
         *,

app/infra/db/postgresql/network_settings.py

@@ -0,0 +1,32 @@
+from typing import Any
+
+from psycopg import AsyncConnection
+
+
+class NetworkSettingsRepository:
+    @staticmethod
+    async def get_result_units(conn: AsyncConnection) -> dict[str, str]:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                """
+                SELECT engine_version, key, value
+                FROM network.simulation_settings
+                WHERE (engine_version = 'v3' AND key IN ('FLOW_UNITS', 'PRESSURE_UNITS'))
+                   OR (engine_version = 'legacy' AND key IN ('UNITS', 'PRESSURE'))
+                ORDER BY CASE engine_version WHEN 'v3' THEN 0 ELSE 1 END
+                """
+            )
+            rows: list[dict[str, Any]] = await cur.fetchall()
+
+        units: dict[str, str] = {}
+        for row in rows:
+            key = str(row["key"])
+            metric = "flow" if key in {"FLOW_UNITS", "UNITS"} else "pressure"
+            units.setdefault(metric, str(row["value"]).strip())
+        missing = {"flow", "pressure"} - units.keys()
+        if missing:
+            raise ValueError(
+                "network simulation settings missing result units: "
+                + ", ".join(sorted(missing))
+            )
+        return units

app/infra/db/postgresql/scada.py

@@ -9,7 +9,7 @@ from app.native.wndb.core.database import read_all, try_read
 
 _SCADA_VIEW_SELECT = """
     SELECT id AS device_id, device_type, node_id, link_id, api_query_id,
-           transmission_mode, transmission_frequency, reliability, x, y,
+           measurement_unit, transmission_mode, transmission_frequency, reliability, x, y,
            ST_X(ST_Transform(geom, 4326)) AS longitude,
            ST_Y(ST_Transform(geom, 4326)) AS latitude
     FROM gis.scada_devices
@@ -57,6 +57,7 @@ def _device(record: dict[str, Any]) -> dict[str, Any]:
         "node_id": _optional_text(record["node_id"]),
         "link_id": _optional_text(record["link_id"]),
         "api_query_id": _optional_text(record["api_query_id"]),
+        "measurement_unit": str(record["measurement_unit"]).strip(),
         "transmission_mode": record["transmission_mode"],
         "transmission_frequency": record["transmission_frequency"],
         "reliability": _optional_int(record["reliability"]),
@@ -105,6 +106,22 @@ class ScadaInfoRepository:
             )
             return {str(row["device_id"]).strip() for row in await cur.fetchall()}
 
+    @staticmethod
+    async def get_scadas_for_elements(
+        conn: AsyncConnection,
+        node_ids: list[str],
+        link_ids: list[str],
+    ) -> list[dict[str, Any]]:
+        if not node_ids and not link_ids:
+            return []
+        async with conn.cursor() as cur:
+            await cur.execute(
+                _SCADA_VIEW_SELECT
+                + " WHERE node_id = ANY(%s) OR link_id = ANY(%s) ORDER BY device_id",
+                (node_ids, link_ids),
+            )
+            return [_device(record) for record in await cur.fetchall()]
+
 
 def get_scada_info_schema(name: str) -> dict[str, dict[str, Any]]:
     return {
@@ -113,6 +130,7 @@ def get_scada_info_schema(name: str) -> dict[str, dict[str, Any]]:
         "node_id": {"type": "str", "optional": True, "readonly": True},
         "link_id": {"type": "str", "optional": True, "readonly": True},
         "api_query_id": {"type": "str", "optional": True, "readonly": True},
+        "measurement_unit": {"type": "str", "optional": False, "readonly": True},
         "transmission_mode": {"type": "str", "optional": False, "readonly": True},
         "transmission_frequency": {"type": "str", "optional": False, "readonly": True},
         "reliability": {"type": "int", "optional": False, "readonly": True},

app/native/wndb/gis/network_views.py

@@ -1,4 +1,4 @@
-"""Read-only projections backed by the GIS materialized-view query layer."""
+"""Read-only GIS projections and authoritative network topology snapshots."""
 
 from typing import Any
 
@@ -69,6 +69,46 @@ def get_network_link_nodes(name: str) -> list[str]:
     ]
 
 
+def get_valve_isolation_topology(name: str) -> tuple[list[str], list[str], list[str]]:
+    """Read links, closed statuses and sources from one consistent SQL snapshot.
+
+    Isolation must include model links without published geometries, and use
+    initial statuses rather than treating every link as open. UNION ALL keeps
+    source nodes in the same statement snapshot without repeating them for
+    every link or relying on materialized view refresh timing.
+    """
+    rows = read_all(
+        name,
+        """
+        SELECT l.id, l.link_type::text AS element_type,
+               l.start_node_id, l.end_node_id,
+               COALESCE(s.status::text, p.status::text) AS status
+        FROM network.links AS l
+        LEFT JOIN network.pipes AS p ON p.link_id = l.id
+        LEFT JOIN network.link_initial_settings AS s ON s.link_id = l.id
+        UNION ALL
+        SELECT n.id, n.node_type::text AS element_type,
+               NULL, NULL, NULL
+        FROM network.nodes AS n
+        WHERE n.node_type::text IN ('reservoir', 'tank')
+        """,
+    )
+    link_entries: list[str] = []
+    source_nodes: list[str] = []
+    closed_links: list[str] = []
+    for row in rows:
+        element_id = str(row["id"])
+        if row["element_type"] in {"reservoir", "tank"}:
+            source_nodes.append(element_id)
+        else:
+            link_entries.append(
+                f"{element_id}:{row['element_type']}:{row['start_node_id']}:{row['end_node_id']}"
+            )
+            if row["status"] == "CLOSED":
+                closed_links.append(element_id)
+    return link_entries, source_nodes, closed_links
+
+
 def get_major_pipe_nodes(name: str, diameter: int) -> list[str]:
     """Return large pipes in the established API wire format."""
     rows = read_all(

app/services/simulation.py

@@ -532,7 +532,14 @@ def run_simulation(
             raise RuntimeError("run_project output missing times.report_step")
         if not scheme_type or not scheme_name:
             raise ValueError("extended simulation requires analysis run type and name")
-        detail = scheme_detail or {}
+        detail = dict(scheme_detail or {})
+        result_units = output_data.get("units")
+        if isinstance(result_units, dict):
+            detail["result_units"] = {
+                key: str(result_units[key]).strip()
+                for key in ("flow", "pressure")
+                if result_units.get(key)
+            }
         run_id = create_analysis_run(
             name=db_name,
             scheme_name=scheme_name,

app/services/timeseries_history.py

@@ -0,0 +1,237 @@
+from collections import defaultdict
+from typing import Any
+
+from psycopg import AsyncConnection
+
+from app.domain.measurement_units import display_unit_for_metric
+from app.domain.schemas.timeseries_history import (
+    ElementHistoryQuery,
+    ElementHistoryResponse,
+    ElementHistorySeries,
+    HistoryElementType,
+    HistoryMetric,
+    HistoryPoint,
+    HistorySource,
+)
+from app.infra.db.postgresql.analysis import AnalysisRepository
+from app.infra.db.postgresql.network_settings import NetworkSettingsRepository
+from app.infra.db.postgresql.scada import ScadaInfoRepository
+from app.infra.db.timescaledb.repositories.analysis import AnalysisResultsRepository
+from app.infra.db.timescaledb.repositories.realtime import RealtimeRepository
+from app.infra.db.timescaledb.repositories.scada import ScadaRepository
+
+
+def _target_metric(element_type: HistoryElementType) -> HistoryMetric:
+    return (
+        HistoryMetric.FLOW
+        if element_type == HistoryElementType.PIPE
+        else HistoryMetric.PRESSURE
+    )
+
+
+def _device_metric(device_type: str) -> HistoryMetric | None:
+    normalized = device_type.strip().lower()
+    if normalized in {"pipe_flow", "flow"}:
+        return HistoryMetric.FLOW
+    if normalized == "pressure":
+        return HistoryMetric.PRESSURE
+    return None
+
+
+def _points(
+    rows: list[dict[str, Any]],
+    *,
+    value_field: str,
+) -> list[HistoryPoint]:
+    return [
+        HistoryPoint(
+            time=row["time"],
+            value=(
+                float(row[value_field])
+                if row.get(value_field) is not None
+                else None
+            ),
+        )
+        for row in rows
+    ]
+
+
+class TimeseriesHistoryService:
+    @staticmethod
+    async def query(
+        timescale_conn: AsyncConnection,
+        postgres_conn: AsyncConnection,
+        query: ElementHistoryQuery,
+    ) -> ElementHistoryResponse:
+        pipe_ids = list(
+            dict.fromkeys(
+                target.element_id
+                for target in query.elements
+                if target.element_type == HistoryElementType.PIPE
+            )
+        )
+        junction_ids = list(
+            dict.fromkeys(
+                target.element_id
+                for target in query.elements
+                if target.element_type == HistoryElementType.JUNCTION
+            )
+        )
+        devices = await ScadaInfoRepository.get_scadas_for_elements(
+            postgres_conn, junction_ids, pipe_ids
+        )
+        devices_by_element: dict[tuple[HistoryElementType, str], list[dict[str, Any]]] = (
+            defaultdict(list)
+        )
+        for device in devices:
+            if device.get("link_id") in pipe_ids:
+                devices_by_element[(HistoryElementType.PIPE, device["link_id"])].append(
+                    device
+                )
+            if device.get("node_id") in junction_ids:
+                devices_by_element[
+                    (HistoryElementType.JUNCTION, device["node_id"])
+                ].append(device)
+
+        selected_devices: dict[str, tuple[dict[str, Any], HistoryElementType, str]] = {}
+        for target in query.elements:
+            target_devices = devices_by_element.get(
+                (target.element_type, target.element_id), []
+            )
+            if target.device_ids is not None:
+                target_device_ids = {device["device_id"] for device in target_devices}
+                unknown = set(target.device_ids) - target_device_ids
+                if unknown:
+                    raise ValueError(
+                        f"SCADA devices do not belong to {target.element_type.value} "
+                        f"{target.element_id}: {', '.join(sorted(unknown))}"
+                    )
+                requested = set(target.device_ids)
+                target_devices = [
+                    device
+                    for device in target_devices
+                    if device["device_id"] in requested
+                ]
+            expected_metric = _target_metric(target.element_type)
+            for device in target_devices:
+                if _device_metric(device["device_type"]) != expected_metric:
+                    continue
+                selected_devices[device["device_id"]] = (
+                    device,
+                    target.element_type,
+                    target.element_id,
+                )
+
+        scada_rows = await ScadaRepository.get_scada_by_ids_time_range(
+            timescale_conn,
+            list(selected_devices),
+            query.start_time,
+            query.end_time,
+        )
+        scada_by_device: dict[str, list[dict[str, Any]]] = defaultdict(list)
+        for row in scada_rows:
+            scada_by_device[str(row["device_id"])].append(row)
+
+        series: list[ElementHistorySeries] = []
+        for device_id, (device, element_type, element_id) in selected_devices.items():
+            rows = scada_by_device.get(device_id, [])
+            if not rows:
+                continue
+            metric = _target_metric(element_type)
+            unit = device["measurement_unit"]
+            for source, field in (
+                (HistorySource.SCADA_RAW, "monitored_value"),
+                (HistorySource.SCADA_CLEANED, "cleaned_value"),
+            ):
+                series.append(
+                    ElementHistorySeries(
+                        element_id=element_id,
+                        element_type=element_type,
+                        device_id=device_id,
+                        metric=metric,
+                        source=source,
+                        source_unit=unit,
+                        display_unit=display_unit_for_metric(metric.value),
+                        points=_points(
+                            rows,
+                            value_field=field,
+                        ),
+                    )
+                )
+
+        if query.mode.value == "observed":
+            return ElementHistoryResponse(series=series)
+
+        current_units = await NetworkSettingsRepository.get_result_units(postgres_conn)
+        if query.mode.value == "analysis_comparison":
+            run = await AnalysisRepository.get_run(postgres_conn, query.run_id)
+            if run is None:
+                raise ValueError(f"analysis run does not exist: {query.run_id}")
+            parameters = run.get("parameters") or {}
+            recorded_units = parameters.get("result_units")
+            unit_inferred = not isinstance(recorded_units, dict)
+            result_units = recorded_units if not unit_inferred else current_units
+            link_values = await AnalysisResultsRepository.get_series_by_ids(
+                timescale_conn,
+                query.run_id,
+                "link",
+                pipe_ids,
+                query.start_time,
+                query.end_time,
+                "flow",
+            )
+            node_values = await AnalysisResultsRepository.get_series_by_ids(
+                timescale_conn,
+                query.run_id,
+                "node",
+                junction_ids,
+                query.start_time,
+                query.end_time,
+                "pressure",
+            )
+            source = HistorySource.ANALYSIS_SIMULATION
+        else:
+            unit_inferred = False
+            result_units = current_units
+            link_values = await RealtimeRepository.get_link_fields_by_ids_time_range(
+                timescale_conn,
+                query.start_time,
+                query.end_time,
+                pipe_ids,
+                "flow",
+            )
+            node_values = await RealtimeRepository.get_node_fields_by_ids_time_range(
+                timescale_conn,
+                query.start_time,
+                query.end_time,
+                junction_ids,
+                "pressure",
+            )
+            source = HistorySource.REALTIME_SIMULATION
+
+        for target in query.elements:
+            metric = _target_metric(target.element_type)
+            rows = (
+                link_values.get(target.element_id, [])
+                if target.element_type == HistoryElementType.PIPE
+                else node_values.get(target.element_id, [])
+            )
+            if not rows:
+                continue
+            source_unit = str(result_units.get(metric.value, "")).strip()
+            series.append(
+                ElementHistorySeries(
+                    element_id=target.element_id,
+                    element_type=target.element_type,
+                    metric=metric,
+                    source=source,
+                    source_unit=source_unit,
+                    display_unit=display_unit_for_metric(metric.value),
+                    unit_inferred=unit_inferred,
+                    points=_points(
+                        rows,
+                        value_field="value",
+                    ),
+                )
+            )
+        return ElementHistoryResponse(series=series)

app/services/valve_isolation.py

@@ -1,13 +1,22 @@
 from typing import Any
 
-from app.algorithms.valve_isolation.topology_search import valve_isolation_analysis
+from app.algorithms.valve_isolation.topology_search import (
-from app.native.wndb.gis.network_views import get_network_link_nodes
+    AccidentElementType,
+    valve_isolation_analysis,
+)
+from app.native.wndb.gis.network_views import get_valve_isolation_topology
 
 
 def analyze_valve_isolation(
     network: str,
     accident_element: str | list[str],
-    disabled_valves: list[str] = None,
+    disabled_valves: list[str] | None = None,
+    accident_element_type: AccidentElementType = "auto",
 ) -> dict[str, Any]:
-    link_entries = get_network_link_nodes(network)
+    link_entries, source_nodes, closed_links = get_valve_isolation_topology(network)
-    return valve_isolation_analysis(link_entries, accident_element, disabled_valves)
+    return valve_isolation_analysis(
+        link_entries, accident_element, disabled_valves,
+        accident_element_type=accident_element_type,
+        water_source_nodes=source_nodes,
+        closed_links=closed_links,
+    )

contracts/manifest.json

@@ -3,7 +3,7 @@
   "contracts": {
     "server": {
       "file": "server-v1.openapi.json",
-      "sha256": "fb720e3009948c3cb2f37674d0bbdbb7b6136973223eccca89880475697435bc"
+      "sha256": "5b8ec0b469f1194e72214003a9cf427489f48de19ae31ee6492821cf64916b6d"
     }
   }
 }

contracts/server-v1.openapi.json

@@ -982,6 +982,224 @@
         "title": "BurstLocationRequestRest",
         "type": "object"
       },
+      "ElementHistoryQuery": {
+        "properties": {
+          "elements": {
+            "items": {
+              "$ref": "#/components/schemas/ElementHistoryTarget"
+            },
+            "maxItems": 200,
+            "minItems": 1,
+            "title": "Elements",
+            "type": "array"
+          },
+          "end_time": {
+            "format": "date-time",
+            "title": "End Time",
+            "type": "string"
+          },
+          "mode": {
+            "$ref": "#/components/schemas/HistoryMode",
+            "default": "observed"
+          },
+          "run_id": {
+            "anyOf": [
+              {
+                "format": "uuid",
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Run Id"
+          },
+          "start_time": {
+            "format": "date-time",
+            "title": "Start Time",
+            "type": "string"
+          }
+        },
+        "required": [
+          "start_time",
+          "end_time",
+          "elements"
+        ],
+        "title": "ElementHistoryQuery",
+        "type": "object"
+      },
+      "ElementHistoryResponse": {
+        "properties": {
+          "series": {
+            "items": {
+              "$ref": "#/components/schemas/ElementHistorySeries"
+            },
+            "title": "Series",
+            "type": "array"
+          }
+        },
+        "required": [
+          "series"
+        ],
+        "title": "ElementHistoryResponse",
+        "type": "object"
+      },
+      "ElementHistorySeries": {
+        "properties": {
+          "device_id": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Device Id"
+          },
+          "display_unit": {
+            "description": "Recommended UI display unit",
+            "title": "Display Unit",
+            "type": "string"
+          },
+          "element_id": {
+            "title": "Element Id",
+            "type": "string"
+          },
+          "element_type": {
+            "$ref": "#/components/schemas/HistoryElementType"
+          },
+          "metric": {
+            "$ref": "#/components/schemas/HistoryMetric"
+          },
+          "points": {
+            "items": {
+              "$ref": "#/components/schemas/HistoryPoint"
+            },
+            "title": "Points",
+            "type": "array"
+          },
+          "source": {
+            "$ref": "#/components/schemas/HistorySource"
+          },
+          "source_unit": {
+            "description": "Unit used by the returned point values",
+            "title": "Source Unit",
+            "type": "string"
+          },
+          "unit_inferred": {
+            "default": false,
+            "title": "Unit Inferred",
+            "type": "boolean"
+          }
+        },
+        "required": [
+          "element_id",
+          "element_type",
+          "metric",
+          "source",
+          "source_unit",
+          "display_unit",
+          "points"
+        ],
+        "title": "ElementHistorySeries",
+        "type": "object"
+      },
+      "ElementHistoryTarget": {
+        "properties": {
+          "device_ids": {
+            "anyOf": [
+              {
+                "items": {
+                  "type": "string"
+                },
+                "maxItems": 100,
+                "type": "array"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Device Ids"
+          },
+          "element_id": {
+            "maxLength": 128,
+            "minLength": 1,
+            "title": "Element Id",
+            "type": "string"
+          },
+          "element_type": {
+            "$ref": "#/components/schemas/HistoryElementType"
+          }
+        },
+        "required": [
+          "element_id",
+          "element_type"
+        ],
+        "title": "ElementHistoryTarget",
+        "type": "object"
+      },
+      "HistoryElementType": {
+        "enum": [
+          "pipe",
+          "junction"
+        ],
+        "title": "HistoryElementType",
+        "type": "string"
+      },
+      "HistoryMetric": {
+        "enum": [
+          "flow",
+          "pressure"
+        ],
+        "title": "HistoryMetric",
+        "type": "string"
+      },
+      "HistoryMode": {
+        "enum": [
+          "observed",
+          "realtime_comparison",
+          "analysis_comparison"
+        ],
+        "title": "HistoryMode",
+        "type": "string"
+      },
+      "HistoryPoint": {
+        "properties": {
+          "time": {
+            "format": "date-time",
+            "title": "Time",
+            "type": "string"
+          },
+          "value": {
+            "anyOf": [
+              {
+                "type": "number"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Value"
+          }
+        },
+        "required": [
+          "time",
+          "value"
+        ],
+        "title": "HistoryPoint",
+        "type": "object"
+      },
+      "HistorySource": {
+        "enum": [
+          "scada_raw",
+          "scada_cleaned",
+          "realtime_simulation",
+          "analysis_simulation"
+        ],
+        "title": "HistorySource",
+        "type": "string"
+      },
       "JsonValue": {},
       "LeakageIdentifyRequestRest": {
         "properties": {
@@ -2409,6 +2627,10 @@
             ],
             "title": "Longitude"
           },
+          "measurement_unit": {
+            "title": "Measurement Unit",
+            "type": "string"
+          },
           "node_id": {
             "anyOf": [
               {
@@ -2465,6 +2687,7 @@
         "required": [
           "device_id",
           "device_type",
+          "measurement_unit",
           "transmission_mode",
           "transmission_frequency"
         ],
@@ -35146,58 +35369,10 @@
         ]
       }
     },
-    "/api/v1/timeseries/views/element-scada-readings": {
+    "/api/v1/timeseries/views/element-history/query": {
-      "get": {
+      "post": {
-        "description": "获取link/node关联的SCADA监测值\n\n根据传入的link/node id，匹配SCADA信息，\n如果存在关联的SCADA device_id，获取实际的监测数据。\n\nArgs:\n    element_id: 管网元素ID\n    start_time: 查询开始时间\n    end_time: 查询结束时间\n    use_cleaned: 是否使用清洗后的数据，默认为False使用原始数据\n    timescale_conn: TimescaleDB连接\n    postgres_conn: PostgreSQL连接\n    \nReturns:\n    管网元素关联的SCADA监测数据\n    \nRaises:\n    HTTPException: 当查询参数无效时返回400错误，未找到关联数据返回404错误",
+        "operationId": "post_timeseries_views_element_history_query",
-        "operationId": "get_timeseries_views_element_scada_readings",
         "parameters": [
-          {
-            "description": "管网元素ID（管道或节点）",
-            "in": "query",
-            "name": "element_id",
-            "required": true,
-            "schema": {
-              "description": "管网元素ID（管道或节点）",
-              "title": "Element Id",
-              "type": "string"
-            }
-          },
-          {
-            "description": "查询开始时间",
-            "in": "query",
-            "name": "start_time",
-            "required": true,
-            "schema": {
-              "description": "查询开始时间",
-              "format": "date-time",
-              "title": "Start Time",
-              "type": "string"
-            }
-          },
-          {
-            "description": "查询结束时间",
-            "in": "query",
-            "name": "end_time",
-            "required": true,
-            "schema": {
-              "description": "查询结束时间",
-              "format": "date-time",
-              "title": "End Time",
-              "type": "string"
-            }
-          },
-          {
-            "description": "是否使用清洗后的数据",
-            "in": "query",
-            "name": "use_cleaned",
-            "required": false,
-            "schema": {
-              "default": false,
-              "description": "是否使用清洗后的数据",
-              "title": "Use Cleaned",
-              "type": "boolean"
-            }
-          },
           {
             "in": "header",
             "name": "X-Project-Id",
@@ -35208,12 +35383,22 @@
             }
           }
         ],
+        "requestBody": {
+          "content": {
+            "application/json": {
+              "schema": {
+                "$ref": "#/components/schemas/ElementHistoryQuery"
+              }
+            }
+          },
+          "required": true
+        },
         "responses": {
           "200": {
             "content": {
               "application/json": {
                 "schema": {
-                  "$ref": "#/components/schemas/JsonValue"
+                  "$ref": "#/components/schemas/ElementHistoryResponse"
                 }
               }
             },
@@ -35285,311 +35470,7 @@
             "OAuth2PasswordBearer": []
           }
         ],
-        "summary": "获取管网元素关联的SCADA监测数据",
+        "summary": "批量查询管网元素历史数据",
-        "tags": [
-          "TimescaleDB - Composite"
-        ]
-      }
-    },
-    "/api/v1/timeseries/views/element-simulations": {
-      "get": {
-        "description": "获取link/node模拟值\n\n根据传入的featureInfos，找到关联的link/node，\n并根据对应的type，查询对应的模拟数据。支持查询实时或分析运行数据。\n\nArgs:\n    start_time: 查询开始时间\n    end_time: 查询结束时间\n    feature_infos: 格式为 \"element_id1:type1,element_id2:type2\"\n                  例如: \"P1:pipe,J1:junction\"\n    run_id: 分析运行 ID，若为空则查询实时数据\n    timescale_conn: TimescaleDB连接\n    \nReturns:\n    管网元素的模拟数据\n    \nRaises:\n    HTTPException: 当feature_infos为空返回400错误，未找到数据返回404错误，其他错误返回400错误",
-        "operationId": "get_timeseries_views_element_simulations",
-        "parameters": [
-          {
-            "description": "查询开始时间",
-            "in": "query",
-            "name": "start_time",
-            "required": true,
-            "schema": {
-              "description": "查询开始时间",
-              "format": "date-time",
-              "title": "Start Time",
-              "type": "string"
-            }
-          },
-          {
-            "description": "查询结束时间",
-            "in": "query",
-            "name": "end_time",
-            "required": true,
-            "schema": {
-              "description": "查询结束时间",
-              "format": "date-time",
-              "title": "End Time",
-              "type": "string"
-            }
-          },
-          {
-            "description": "特征信息，格式: id1:type1,id2:type2，type为pipe（管道）或junction（节点）",
-            "in": "query",
-            "name": "feature_infos",
-            "required": true,
-            "schema": {
-              "description": "特征信息，格式: id1:type1,id2:type2，type为pipe（管道）或junction（节点）",
-              "title": "Feature Infos",
-              "type": "string"
-            }
-          },
-          {
-            "description": "分析运行 ID；为空时查询实时数据",
-            "in": "query",
-            "name": "run_id",
-            "required": false,
-            "schema": {
-              "anyOf": [
-                {
-                  "format": "uuid",
-                  "type": "string"
-                },
-                {
-                  "type": "null"
-                }
-              ],
-              "description": "分析运行 ID；为空时查询实时数据",
-              "title": "Run Id"
-            }
-          },
-          {
-            "in": "header",
-            "name": "X-Project-Id",
-            "required": true,
-            "schema": {
-              "title": "X-Project-Id",
-              "type": "string"
-            }
-          }
-        ],
-        "responses": {
-          "200": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/JsonValue"
-                }
-              }
-            },
-            "description": "Successful Response"
-          },
-          "401": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Authentication required"
-          },
-          "403": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Insufficient permission"
-          },
-          "404": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Resource not found"
-          },
-          "409": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Resource conflict"
-          },
-          "422": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Validation error"
-          },
-          "503": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Dependency unavailable"
-          }
-        },
-        "security": [
-          {
-            "OAuth2PasswordBearer": []
-          }
-        ],
-        "summary": "获取管网元素的模拟数据",
-        "tags": [
-          "TimescaleDB - Composite"
-        ]
-      }
-    },
-    "/api/v1/timeseries/views/scada-simulations": {
-      "get": {
-        "description": "获取SCADA关联的link/node模拟值\n\n根据传入的SCADA device_ids，找到关联的link/node，\n并根据对应的type，查询对应的模拟数据。支持查询实时或分析运行数据。\n\nArgs:\n    start_time: 查询开始时间\n    end_time: 查询结束时间\n    device_ids: SCADA设备ID列表，用逗号分隔\n    run_id: 分析运行 ID，若为空则查询实时数据\n    timescale_conn: TimescaleDB连接\n    postgres_conn: PostgreSQL连接\n    \nReturns:\n    SCADA关联的模拟数据\n    \nRaises:\n    HTTPException: 当查询参数无效时返回400错误，未找到数据时返回404错误",
-        "operationId": "get_timeseries_views_scada_simulations",
-        "parameters": [
-          {
-            "description": "查询开始时间",
-            "in": "query",
-            "name": "start_time",
-            "required": true,
-            "schema": {
-              "description": "查询开始时间",
-              "format": "date-time",
-              "title": "Start Time",
-              "type": "string"
-            }
-          },
-          {
-            "description": "查询结束时间",
-            "in": "query",
-            "name": "end_time",
-            "required": true,
-            "schema": {
-              "description": "查询结束时间",
-              "format": "date-time",
-              "title": "End Time",
-              "type": "string"
-            }
-          },
-          {
-            "description": "SCADA设备ID列表，逗号分隔",
-            "in": "query",
-            "name": "device_ids",
-            "required": true,
-            "schema": {
-              "description": "SCADA设备ID列表，逗号分隔",
-              "title": "Device Ids",
-              "type": "string"
-            }
-          },
-          {
-            "description": "分析运行 ID；为空时查询实时数据",
-            "in": "query",
-            "name": "run_id",
-            "required": false,
-            "schema": {
-              "anyOf": [
-                {
-                  "format": "uuid",
-                  "type": "string"
-                },
-                {
-                  "type": "null"
-                }
-              ],
-              "description": "分析运行 ID；为空时查询实时数据",
-              "title": "Run Id"
-            }
-          },
-          {
-            "in": "header",
-            "name": "X-Project-Id",
-            "required": true,
-            "schema": {
-              "title": "X-Project-Id",
-              "type": "string"
-            }
-          }
-        ],
-        "responses": {
-          "200": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/JsonValue"
-                }
-              }
-            },
-            "description": "Successful Response"
-          },
-          "401": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Authentication required"
-          },
-          "403": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Insufficient permission"
-          },
-          "404": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Resource not found"
-          },
-          "409": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Resource conflict"
-          },
-          "422": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Validation error"
-          },
-          "503": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Dependency unavailable"
-          }
-        },
-        "security": [
-          {
-            "OAuth2PasswordBearer": []
-          }
-        ],
-        "summary": "获取SCADA关联的模拟数据",
         "tags": [
           "TimescaleDB - Composite"
         ]
@@ -36088,6 +35969,23 @@
               "type": "array"
             }
           },
+          {
+            "description": "事故要素类型；auto优先管段再节点；分析同名节点须指定node，纯管段分析可指定link",
+            "in": "query",
+            "name": "accident_element_type",
+            "required": false,
+            "schema": {
+              "default": "auto",
+              "description": "事故要素类型；auto优先管段再节点；分析同名节点须指定node，纯管段分析可指定link",
+              "enum": [
+                "auto",
+                "link",
+                "node"
+              ],
+              "title": "Accident Element Type",
+              "type": "string"
+            }
+          },
           {
             "in": "header",
             "name": "X-Project-Id",

resources/db_v2/DATABASE_ARCHITECTURE.md

@@ -1,6 +1,6 @@
 # TJWater 数据库改造说明与当前结构
 
-> 本文记录截至 2026-09-10 的数据库实际状态。结构、约束、行数、TimescaleDB chunk 和策略均直接读取数据库，不以仓库中的 SQL 脚本为依据。文中不包含主机、端口、账号、密码或 DSN。
+> 本文记录截至 2026-09-14 的数据库实际状态。结构、约束、行数、TimescaleDB chunk 和策略均直接读取数据库，不以仓库中的 SQL 脚本为依据。文中不包含主机、端口、账号、密码或 DSN。
 
 ## 改造范围与当前状态
 
@@ -20,10 +20,11 @@
 - `system_hub.public` 补充了项目数据库外键、数据库路由约束、连接池约束、必要的非空约束，以及 5 张表和 44 个字段的中文数据库注释。
 - 用户角色和项目角色仍是可扩展字符串，没有增加枚举检查约束。
 - `audit_logs.user_id` 和 `audit_logs.project_id` 仍为逻辑关联，没有增加外键。
-- 业务库 48 个表或物化视图、192 个字段，以及时序库 7 张表、47 个字段均已写入中文数据库注释。
+- 业务库 48 个表或物化视图、194 个字段，以及时序库 7 张表、47 个字段均已写入中文数据库注释。
 - 后端已对接新 schema。WNDB、PostgreSQL 管理连接和同步 TimescaleDB 访问均使用有界连接池，闲置项目按最近使用顺序回收；动态异步池采用代际切换，配置更新不会中断旧借用或阻塞新请求。
 - 后端批量元素查询读取 GIS 物化视图，模型增删改和 INP 导入提交后执行并发刷新；批量事务只刷新一次。
 - `pattern_values`、`pattern_flow_samples`、`curve_points`、`demands` 和 `link_vertices` 的顺序号按所属父对象编号，主键已改为父对象 ID 与 `sequence_no` 的复合键。
+- 5 个活跃业务库、5 个对应管网模板库和 `tjwater_v2_schema_template` 均已增加 `asset.scada_devices.measurement_unit`；压力设备回填为 `m`，流量设备回填为 `m3/h`，`gis.scada_devices` 物化视图同步暴露该字段。
 
 逻辑项目 `tjwater_v2` 当前为 `active`。
 
@@ -359,15 +360,27 @@ TimescaleDB 的空结构模板为 `tjwater_v2_timescale_template`，普通连接
 
 方案查询统一读取 `/api/v1/analysis/runs`，详情和时序结果按 UUID `run_id` 关联。时间轴读取 `/api/v1/timeseries/analysis/runs/{run_id}/values`，爆管定位的模拟数据源也传递 `simulation_run_id`。监测点优化使用 `/api/v1/sensor-placement-runs`。前端不再调用旧的 `/schemes`、`/timeseries/schemes` 和 `/sensor-placement-schemes` 接口。
 
+管网元素历史查询统一使用 `POST /api/v1/timeseries/views/element-history/query`。单次请求批量返回元素关联的原始/清洗 SCADA、实时模拟或指定分析运行时序，并显式返回 `source_unit`、`display_unit` 和 `unit_inferred`。点值仍使用 `source_unit`，前端由共享单位模块统一换算为流量 `m³/h`、压力 `m` 和流速 `m/s`。旧的 `element-simulations`、`element-scada-readings` 和 `scada-simulations` 组合接口已移除。
+
+当前业务库 `network.simulation_settings` 中的模型结果源单位如下。`LPS` 表示 L/s，`MLD` 表示百万升/日；公制模型的流速源单位为 m/s。前端不得再假定所有项目都是 `LPS`，而应读取项目设置并通过共享单位模块换算。
+
+| 业务库 | 流量源单位 | 压力源单位 | 前端显示单位 |
+| --- | --- | --- | --- |
+| `fengyang_v2` | `LPS` | `METERS` | m³/h、m、m/s |
+| `md_v2` | `LPS` | `METERS` | m³/h、m、m/s |
+| `szh_v2` | `MLD` | `METERS` | m³/h、m、m/s |
+| `tjwater_v2` | `LPS` | `METERS` | m³/h、m、m/s |
+| `zjb` | `LPS` | `METERS` | m³/h、m、m/s |
+
 模型修改提交后，后端先调用 `gis.refresh_all_materialized_views(boolean)` 刷新数据库查询层。GeoWebCache 不会感知 PostgreSQL 物化视图刷新，因此 7 个新图层配置了 300 秒的服务端和客户端缓存有效期，前端最迟在 5 分钟后读到新瓦片。部署或批量迁移完成后仍可执行一次图层缓存清空，避免等待已有瓦片自然过期。
 
 ### asset：SCADA 设备配置
 
-`asset.scada_devices` 保存设备类型、采集接口标识、传输模式、频率、可靠性和可选几何。每台设备必须关联一个节点或一条连接，不能同时关联两者。设备的历史测量值不放在业务库，保存在时序库 `scada.measurements`。
+`asset.scada_devices` 保存设备类型、采集接口标识、传输模式、频率、可靠性、测量源单位 `measurement_unit` 和可选几何。单位字段非空且不允许空字符串。每台设备必须关联一个节点或一条连接，不能同时关联两者。设备的历史测量值不放在业务库，保存在时序库 `scada.measurements`。
 
 ### analysis：分析运行和非时序结果
 
-`analysis.runs` 表示一次实际执行，保存运行名称、类型、创建人、开始时间、状态和参数。当前没有单独的 `scenarios` 模板表。
+`analysis.runs` 表示一次实际执行，保存运行名称、类型、创建人、开始时间、状态和参数。新执行会在 `parameters.result_units` 中保存 EPANET 输出的流量和压力源单位，避免日后调整模型设置时误读历史值。旧运行没有该元数据时，API 使用当前模型单位并返回 `unit_inferred=true`。当前没有单独的 `scenarios` 模板表。
 
 每次执行都会创建新的 `run_id`，名称和业务时间相同也不会覆盖旧结果。扩展仿真开始写结果前，业务库先记录 `running`；时序库写入完成后更新为 `completed`，写入失败则保留为 `failed`。业务库和时序库据此共用同一个执行标识。
 

tests/api/test_admin_metadata_endpoints.py

@@ -7,8 +7,9 @@ import pytest
 from fastapi import HTTPException
 from fastapi import Response
 
-from app.auth.metadata_dependencies import get_current_metadata_admin
 from app.api.v1.endpoints import admin_metadata
+from app.api.pagination import PaginatedList
+from app.auth.metadata_dependencies import get_current_metadata_admin
 from app.domain.schemas.admin_metadata import (
     AdminProjectCreateRequest,
     MetadataUsersBatchSyncRequest,
@@ -83,6 +84,28 @@ def test_to_async_sqlalchemy_url_preserves_password():
     assert "***" not in url
 
 
+@pytest.mark.anyio
+async def test_list_metadata_users_returns_page_with_real_total():
+    users = [_user(username="alice"), _user(username="bob")]
+    repo = SimpleNamespace(
+        list_users=AsyncMock(return_value=users),
+        count_users=AsyncMock(return_value=7),
+    )
+
+    response = await admin_metadata.list_metadata_users(
+        skip=2,
+        limit=2,
+        current_user=_user(role="admin", is_superuser=True),
+        metadata_repo=repo,
+    )
+
+    assert isinstance(response, PaginatedList)
+    assert response.total == 7
+    assert [item.username for item in response] == ["alice", "bob"]
+    repo.list_users.assert_awaited_once_with(skip=2, limit=2)
+    repo.count_users.assert_awaited_once_with()
+
+
 @pytest.mark.anyio
 async def test_sync_metadata_user_upserts_without_password(monkeypatch):
     keycloak_id = uuid4()

tests/api/test_openapi_contract.py

@@ -254,14 +254,16 @@ def test_committed_write_refresh_failure_has_explicit_http_contract() -> None:
     assert response.json()["code"] == "materialized_view_refresh_failed_after_commit"
 
 
-def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch) -> None:
+@pytest.mark.parametrize("element_type", ["auto", "link", "node"])
+def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch, element_type) -> None:
     captured: dict[str, object] = {}
 
-    def fake_analyze_valve_isolation(network, accident_element, disabled_valves):
+    def fake_analyze_valve_isolation(network, accident_element, disabled_valves, accident_element_type):
         captured.update(
             network=network,
             accident_element=accident_element,
             disabled_valves=disabled_valves,
+            accident_element_type=accident_element_type,
             business_dsn=get_project_pgconn_string(network),
             timescale_dsn=get_project_timescale_pgconn_string(network),
         )
@@ -288,6 +290,7 @@ def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch) -> None:
             ("accident_element", "P-1"),
             ("accident_element", "P-2"),
             ("disabled_valves", "V-9"),
+            *([] if element_type == "auto" else [("accident_element_type", element_type)]),
         ],
     )
 
@@ -297,11 +300,39 @@ def test_valve_isolation_runtime_accepts_frontend_query(monkeypatch) -> None:
         "network": "fengyang",
         "accident_element": ["P-1", "P-2"],
         "disabled_valves": ["V-9"],
+        "accident_element_type": element_type,
         "business_dsn": "postgresql://user:password@biz/fengyang",
         "timescale_dsn": "postgresql://user:password@timescale/fengyang",
     }
 
 
+@pytest.mark.parametrize(
+    "target,disabled,detail",
+    [("P", "typo", "Disabled valve"), ("missing", "V", "Accident element")],
+)
+def test_valve_isolation_invalid_ids_return_validation_error(monkeypatch, target, disabled, detail):
+    from app.services import valve_isolation as service
+
+    monkeypatch.setattr(
+        service, "get_valve_isolation_topology",
+        lambda _network: (["P:pipe:A:B", "V:valve:B:R"], ["R"], []),
+    )
+    monkeypatch.setattr(simulation_endpoint, "analyze_valve_isolation", service.analyze_valve_isolation)
+    app = FastAPI()
+    app.include_router(api_router, prefix="/api/v1")
+    _override_project_routing(app, ProjectContext(
+        project_id=uuid4(), project_code="project", user_id=uuid4(), project_role="member",
+    ))
+
+    response = TestClient(app, raise_server_exceptions=False).post(
+        "/api/v1/valve-isolation-analyses",
+        params={"accident_element": target, "disabled_valves": disabled},
+    )
+
+    assert response.status_code == 422
+    assert detail in response.json()["detail"]
+
+
 def test_scada_cleaning_runs_are_post() -> None:
     methods_by_path = {
         route.path: route.methods

tests/unit/test_analysis_simulation.py

@@ -159,6 +159,7 @@ def test_extended_simulation_stores_results_by_run_id(monkeypatch):
         lambda name: json.dumps(
             {
                 "output": {
+                    "units": {"flow": "LPS", "pressure": "MTR"},
                     "times": {"num_periods": 2, "report_step": 900},
                     "node_results": [{"node": "J1", "result": [{}, {}]}],
                     "link_results": [{"link": "P1", "result": [{}, {}]}],
@@ -167,7 +168,12 @@ def test_extended_simulation_stores_results_by_run_id(monkeypatch):
         ),
     )
     lifecycle_calls: list[tuple] = []
-    monkeypatch.setattr(simulation, "create_analysis_run", lambda **kwargs: run_id)
+    create_calls: list[dict] = []
+    monkeypatch.setattr(
+        simulation,
+        "create_analysis_run",
+        lambda **kwargs: (create_calls.append(kwargs), run_id)[1],
+    )
     monkeypatch.setattr(
         simulation,
         "update_analysis_run",
@@ -195,6 +201,10 @@ def test_extended_simulation_stores_results_by_run_id(monkeypatch):
     assert kwargs["db_name"] == "demo"
     assert returned_run_id == run_id
     assert lifecycle_calls[-1][1]["status"] == "completed"
+    assert create_calls[0]["scheme_detail"]["result_units"] == {
+        "flow": "LPS",
+        "pressure": "MTR",
+    }
     assert transaction_calls == [("begin", "demo"), ("end", "demo")]
     refresh_mock.assert_called_once_with("demo")
 

tests/unit/test_measurement_units.py

@@ -0,0 +1,7 @@
+from app.domain.measurement_units import display_unit_for_metric
+
+
+def test_display_units_are_stable_ui_units():
+    assert display_unit_for_metric("flow") == "m³/h"
+    assert display_unit_for_metric("pressure") == "m"
+    assert display_unit_for_metric("velocity") == "m/s"

tests/unit/test_postgres_scada_repository.py

@@ -29,6 +29,7 @@ class _FakeCursor:
                 "node_id": " J1 ",
                 "link_id": None,
                 "api_query_id": "query-1",
+                "measurement_unit": "m",
                 "transmission_mode": "realtime",
                 "transmission_frequency": None,
                 "reliability": "95",
@@ -63,6 +64,7 @@ def test_get_scadas_normalizes_id_and_type():
             "node_id": "J1",
             "link_id": None,
             "api_query_id": "query-1",
+            "measurement_unit": "m",
             "transmission_mode": "realtime",
             "transmission_frequency": None,
             "reliability": 95,

tests/unit/test_project_scada_metadata.py

@@ -13,6 +13,7 @@ PROJECT_SCADA = {
     "node_id": "J1",
     "link_id": None,
     "api_query_id": "query-1",
+    "measurement_unit": "m",
     "transmission_mode": "realtime",
     "transmission_frequency": None,
     "reliability": 1.0,

tests/unit/test_timeseries_history.py

@@ -0,0 +1,146 @@
+import asyncio
+from datetime import UTC, datetime, timedelta
+from unittest.mock import AsyncMock
+
+import pytest
+
+from app.domain.schemas.timeseries_history import ElementHistoryQuery
+from app.services import timeseries_history
+
+
+START = datetime(2026, 9, 1, tzinfo=UTC)
+END = START + timedelta(hours=1)
+
+
+def test_realtime_history_batches_devices_and_converts_units(monkeypatch):
+    monkeypatch.setattr(
+        timeseries_history.ScadaInfoRepository,
+        "get_scadas_for_elements",
+        AsyncMock(
+            return_value=[
+                {
+                    "device_id": "F-1",
+                    "device_type": "pipe_flow",
+                    "link_id": "P-1",
+                    "node_id": None,
+                    "measurement_unit": "m3/h",
+                },
+                {
+                    "device_id": "P-1A",
+                    "device_type": "pressure",
+                    "link_id": None,
+                    "node_id": "J-1",
+                    "measurement_unit": "m",
+                },
+                {
+                    "device_id": "P-1B",
+                    "device_type": "pressure",
+                    "link_id": None,
+                    "node_id": "J-1",
+                    "measurement_unit": "m",
+                },
+            ]
+        ),
+    )
+    scada_query = AsyncMock(
+        return_value=[
+            {
+                "time": START,
+                "device_id": "F-1",
+                "monitored_value": 36.0,
+                "cleaned_value": 35.0,
+            },
+            {
+                "time": START,
+                "device_id": "P-1A",
+                "monitored_value": 20.0,
+                "cleaned_value": 21.0,
+            },
+            {
+                "time": START,
+                "device_id": "P-1B",
+                "monitored_value": 22.0,
+                "cleaned_value": 23.0,
+            },
+        ]
+    )
+    monkeypatch.setattr(
+        timeseries_history.ScadaRepository,
+        "get_scada_by_ids_time_range",
+        scada_query,
+    )
+    monkeypatch.setattr(
+        timeseries_history.NetworkSettingsRepository,
+        "get_result_units",
+        AsyncMock(return_value={"flow": "MLD", "pressure": "METERS"}),
+    )
+    monkeypatch.setattr(
+        timeseries_history.RealtimeRepository,
+        "get_link_fields_by_ids_time_range",
+        AsyncMock(return_value={"P-1": [{"time": START, "value": 2.0}]}),
+    )
+    monkeypatch.setattr(
+        timeseries_history.RealtimeRepository,
+        "get_node_fields_by_ids_time_range",
+        AsyncMock(return_value={"J-1": [{"time": START, "value": 24.0}]}),
+    )
+
+    result = asyncio.run(
+        timeseries_history.TimeseriesHistoryService.query(
+            object(),
+            object(),
+            ElementHistoryQuery(
+                start_time=START,
+                end_time=END,
+                mode="realtime_comparison",
+                elements=[
+                    {"element_id": "P-1", "element_type": "pipe"},
+                    {"element_id": "J-1", "element_type": "junction"},
+                ],
+            ),
+        )
+    )
+
+    assert scada_query.await_count == 1
+    assert set(scada_query.await_args.args[1]) == {"F-1", "P-1A", "P-1B"}
+    assert {item.device_id for item in result.series if item.device_id} == {
+        "F-1",
+        "P-1A",
+        "P-1B",
+    }
+    simulation = {
+        (item.element_id, item.metric.value): item
+        for item in result.series
+        if item.source.value == "realtime_simulation"
+    }
+    assert simulation[("P-1", "flow")].points[0].value == 2.0
+    assert simulation[("P-1", "flow")].source_unit == "MLD"
+    assert simulation[("J-1", "pressure")].points[0].value == 24.0
+    assert all(item.display_unit in {"m³/h", "m"} for item in result.series)
+
+
+def test_requested_device_must_belong_to_element(monkeypatch):
+    monkeypatch.setattr(
+        timeseries_history.ScadaInfoRepository,
+        "get_scadas_for_elements",
+        AsyncMock(return_value=[]),
+    )
+
+    with pytest.raises(ValueError, match="do not belong"):
+        asyncio.run(
+            timeseries_history.TimeseriesHistoryService.query(
+                object(),
+                object(),
+                ElementHistoryQuery(
+                    start_time=START,
+                    end_time=END,
+                    elements=[
+                        {
+                            "element_id": "J-1",
+                            "element_type": "junction",
+                            "device_ids": ["missing"],
+                        }
+                    ],
+                ),
+            )
+        )

tests/unit/test_timeseries_history_schema.py

@@ -0,0 +1,50 @@
+from datetime import UTC, datetime, timedelta
+from uuid import uuid4
+
+import pytest
+from pydantic import ValidationError
+
+from app.domain.schemas.timeseries_history import ElementHistoryQuery
+
+
+def _query(**overrides):
+    start = datetime(2026, 1, 1, tzinfo=UTC)
+    values = {
+        "start_time": start,
+        "end_time": start + timedelta(hours=1),
+        "mode": "observed",
+        "elements": [{"element_id": " P1 ", "element_type": "pipe"}],
+    }
+    values.update(overrides)
+    return ElementHistoryQuery(**values)
+
+
+def test_history_query_normalizes_targets_and_device_ids():
+    query = _query(
+        elements=[
+            {
+                "element_id": " P1 ",
+                "element_type": "pipe",
+                "device_ids": [" D1 ", "D1", "D2"],
+            }
+        ]
+    )
+
+    assert query.elements[0].element_id == "P1"
+    assert query.elements[0].device_ids == ["D1", "D2"]
+
+
+def test_analysis_comparison_requires_run_id():
+    with pytest.raises(ValidationError, match="run_id is required"):
+        _query(mode="analysis_comparison")
+
+
+def test_other_modes_reject_run_id():
+    with pytest.raises(ValidationError, match="only valid"):
+        _query(run_id=uuid4())
+
+
+def test_history_query_rejects_reversed_time_range():
+    start = datetime(2026, 1, 1, tzinfo=UTC)
+    with pytest.raises(ValidationError, match="start_time must be earlier"):
+        _query(start_time=start, end_time=start)

tests/unit/test_valve_isolation.py

@@ -1,3 +1,5 @@
+import pytest
+
 from app.algorithms.valve_isolation import topology_search as valve
 
 
@@ -23,6 +25,183 @@ def test_isolatable_keeps_affected_node_ids_and_count():
     assert result["must_close_valves"] == ["V-close"]
 
 
+@pytest.fixture
+def colliding_topology():
+    # The reported adjacent pipes and their same-numbered nodes are in
+    # different valve-bounded regions. Keep this fixture independent of a DB.
+    return [
+        "19025:pipe:17038:17037",
+        "19026:pipe:17038:17039",
+        "AUTO_TCV_4:valve:17037:D",
+        "AUTO_TCV_5:valve:17039:E",
+        "AUTO_TCV_546:valve:17038:R",
+        "32252:pipe:19025:X",
+        "15383:pipe:19026:Y",
+        "V-wrong-25:valve:X:R",
+        "V-wrong-26:valve:Y:R",
+    ]
+
+
+@pytest.mark.parametrize("target", ["19025", "19026", ["19025", "19026"]])
+def test_adjacent_pipes_take_priority_over_same_numbered_nodes(colliding_topology, target):
+    result = valve.valve_isolation_analysis(colliding_topology, target)
+
+    assert result["affected_nodes"] == ["17037", "17038", "17039"]
+    assert result["must_close_valves"] == ["AUTO_TCV_4", "AUTO_TCV_5", "AUTO_TCV_546"]
+    assert result["optional_valves"] == []
+
+
+def test_explicit_node_type_resolves_the_other_side_of_an_id_collision(colliding_topology):
+    result = valve.valve_isolation_analysis(
+        colliding_topology, "19025", accident_element_type="node",
+    )
+
+    assert result["affected_nodes"] == ["19025", "X"]
+    assert result["must_close_valves"] == ["V-wrong-25"]
+
+
+def test_unambiguous_legacy_node_input_still_works():
+    result = valve.valve_isolation_analysis(["P:pipe:A:B", "V:valve:B:R"], "A")
+    assert result["affected_nodes"] == ["A", "B"]
+
+
+def test_explicit_link_type_does_not_fall_back_to_a_node():
+    with pytest.raises(ValueError, match="not found"):
+        valve.valve_isolation_analysis(
+            ["P:pipe:A:B"], "A", accident_element_type="link",
+        )
+
+
+def test_a_boundary_valve_does_not_prove_source_isolation():
+    result = valve.valve_isolation_analysis(
+        ["P:pipe:R:A", "V:valve:A:B"], "P", water_source_nodes=["R"],
+    )
+
+    assert result["isolatable"] is False
+    assert result["source_check_performed"] is True
+    assert result["isolation_failure_reason"] == "connected_to_water_source"
+    assert result["affected_nodes"] == []
+    assert result["affected_node_count"] == 2
+    assert result["supply_affected_nodes"] is None
+
+
+def test_supply_impact_includes_downstream_regions_but_not_preexisting_outages():
+    links = [
+        "P:pipe:A:B", "V-source:valve:A:R", "V-downstream:valve:B:C",
+        "P-downstream:pipe:C:D", "P-already-off:pipe:X:Y",
+    ]
+    result = valve.valve_isolation_analysis(links, "P", water_source_nodes=["R"])
+
+    assert result["isolatable"] is True
+    assert result["source_check_performed"] is True
+    assert result["affected_nodes"] == ["A", "B"]
+    assert result["affected_node_count"] == 2
+    assert result["supply_affected_nodes"] == ["A", "B", "C", "D"]
+    assert result["supply_affected_node_count"] == 4
+
+
+def test_closed_links_are_not_traversed_or_recommended_for_closure():
+    result = valve.valve_isolation_analysis(
+        ["P:pipe:A:B", "P-closed:pipe:B:R", "V:valve:A:R", "V-closed:valve:B:C"],
+        "P", water_source_nodes=["R"], closed_links=["P-closed", "V-closed"],
+    )
+
+    assert result["isolatable"] is True
+    assert result["affected_nodes"] == ["A", "B"]
+    assert result["must_close_valves"] == ["V"]
+    assert result["supply_affected_nodes"] == ["A", "B"]
+
+
+def test_disabled_valve_cannot_cut_the_source_path():
+    result = valve.valve_isolation_analysis(
+        ["P:pipe:A:B", "V-disabled:valve:A:R", "V-other:valve:B:C"],
+        "P", disabled_valves=["V-disabled"], water_source_nodes=["R"],
+    )
+
+    assert result["isolatable"] is False
+    assert result["isolation_failure_reason"] == "connected_to_water_source"
+
+
+def test_joint_accidents_fail_if_any_region_remains_connected_to_a_source():
+    result = valve.valve_isolation_analysis(
+        ["P1:pipe:A:B", "V1:valve:B:R", "P2:pipe:R:C", "V2:valve:C:D"],
+        ["P1", "P2"], water_source_nodes=["R"],
+    )
+    assert result["isolatable"] is False
+
+
+def test_missing_sources_do_not_claim_verified_isolation():
+    result = valve.valve_isolation_analysis(
+        ["P:pipe:A:B", "V:valve:B:C"], "P", water_source_nodes=[],
+    )
+    assert result["isolatable"] is False
+    assert result["source_check_performed"] is False
+    assert result["isolation_failure_reason"] == "water_sources_unavailable"
+    assert result["supply_affected_node_count"] is None
+
+
+@pytest.mark.parametrize("sources", [["R1"], ["R1", "R2"]])
+def test_all_water_sources_are_considered(sources):
+    result = valve.valve_isolation_analysis(
+        ["P:pipe:A:B", "V:valve:A:R1", "P2:pipe:B:R2"],
+        "P", water_source_nodes=sources,
+    )
+    assert result["isolatable"] is (len(sources) == 1)
+
+
+def test_empty_targets_are_rejected():
+    with pytest.raises(ValueError, match="At least one"):
+        valve.valve_isolation_analysis(["P:pipe:A:B"], [])
+
+
+def test_unknown_disabled_valve_is_rejected():
+    with pytest.raises(ValueError, match="Disabled valve"):
+        valve.valve_isolation_analysis(["P:pipe:A:B"], "P", disabled_valves=["typo"])
+
+
+def test_service_passes_authoritative_sources_statuses_and_element_type(monkeypatch):
+    from app.services import valve_isolation as service
+
+    projects = []
+
+    def snapshot(network):
+        projects.append(network)
+        return (
+            ["P:pipe:A:B", "P-closed:pipe:B:R", "V:valve:A:R"],
+            ["R"], ["P-closed"],
+        )
+
+    monkeypatch.setattr(service, "get_valve_isolation_topology", snapshot)
+    result = service.analyze_valve_isolation(
+        "trusted-project", ["A"], accident_element_type="node",
+    )
+
+    assert projects == ["trusted-project"]
+    assert result["isolatable"] is True
+    assert result["source_check_performed"] is True
+    assert result["must_close_valves"] == ["V"]
+    assert result["supply_affected_nodes"] == ["A", "B"]
+
+
+def test_parallel_links_preserve_an_open_bypass():
+    result = valve.valve_isolation_analysis(
+        ["P:pipe:A:B", "V:valve:B:R", "P-bypass:pipe:B:R"],
+        "P", water_source_nodes=["R"],
+    )
+    assert result["isolatable"] is False
+    assert result["isolation_failure_reason"] == "connected_to_water_source"
+
+
+def test_closed_disabled_valve_does_not_reopen_a_known_closed_link():
+    result = valve.valve_isolation_analysis(
+        ["P:pipe:A:B", "V:valve:B:R", "V-disabled:valve:A:R"],
+        "P", disabled_valves=["V-disabled"], closed_links=["V-disabled"],
+        water_source_nodes=["R"],
+    )
+    assert result["isolatable"] is True
+    assert result["must_close_valves"] == ["V"]
+
+
 def test_disabled_valve_expands_affected_area_before_counting():
     links = [
         "P-1:pipe:A:B",

tests/unit/test_wndb_network_views.py

@@ -52,6 +52,36 @@ def test_network_link_nodes_use_one_unified_view_query(monkeypatch) -> None:
     assert "FROM gis.network_links" in calls[0][1]
 
 
+def test_isolation_snapshot_reads_authoritative_links_statuses_and_sources_once(monkeypatch):
+    calls = []
+
+    def fake_read_all(name, statement, params=None):
+        calls.append((name, statement, params))
+        return [
+            {"id": "P", "element_type": "pipe", "start_node_id": "A", "end_node_id": "B", "status": None},
+            {"id": "P-closed", "element_type": "pipe", "start_node_id": "A", "end_node_id": "R", "status": "CLOSED"},
+            {"id": "V", "element_type": "valve", "start_node_id": "B", "end_node_id": "R", "status": "CLOSED"},
+            {"id": "PU", "element_type": "pump", "start_node_id": "T", "end_node_id": "A", "status": "CLOSED"},
+            {"id": "R", "element_type": "reservoir", "start_node_id": None, "end_node_id": None, "status": None},
+            {"id": "T", "element_type": "tank", "start_node_id": None, "end_node_id": None, "status": None},
+        ]
+
+    monkeypatch.setattr(network_views, "read_all", fake_read_all)
+
+    assert network_views.get_valve_isolation_topology("project_a") == (
+        ["P:pipe:A:B", "P-closed:pipe:A:R", "V:valve:B:R", "PU:pump:T:A"],
+        ["R", "T"], ["P-closed", "V", "PU"],
+    )
+    assert len(calls) == 1
+    name, statement, _params = calls[0]
+    assert name == "project_a"
+    assert "FROM network.links" in statement
+    assert "network.link_initial_settings" in statement
+    assert "COALESCE(s.status::text, p.status::text)" in statement
+    assert "FROM network.nodes" in statement
+    assert "UNION ALL" in statement
+
+
 def test_topology_rows_are_loaded_in_two_batch_queries(monkeypatch) -> None:
     calls = []
     responses = [