feat(api): add compact hydraulic playback endpoints

Full-network pressure comparisons exceeded the worker limit before saved-result previews could run. Preserve complete calculation results and keep read-time summaries and pagination.

Incremental graph publication accepted isolated nodes after flow-authoring rules were lost. Require new steps to connect explicitly while retaining historical snapshots and status-only patches. Regression tests reproduce both failures on the previous code.

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

@@ -0,0 +1,69 @@
+"""Project-authorized compact hydraulic playback transport."""
+
+from collections import OrderedDict
+import asyncio
+import time
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
+from psycopg import AsyncConnection
+
+from app.auth.project_dependencies import ProjectContext, get_project_context, get_project_pg_connection
+from app.services.hydraulic_playback import MAX_FRAME_COUNT, PlaybackGeometry, read_frames, read_geometry
+from app.domain.time import parse_utc_time
+from .dependencies import get_timescale_connection
+
+router = APIRouter(prefix="/timeseries/realtime/playback")
+# Bounded per-process cache, private to the authorized project and database route.
+_cache: OrderedDict[tuple, tuple[float, PlaybackGeometry]] = OrderedDict()
+_load_lock = asyncio.Lock()
+
+
+async def geometry_for_project(ctx: ProjectContext, conn: AsyncConnection) -> PlaybackGeometry:
+    key = (str(ctx.project_id), conn.info.host, conn.info.port, conn.info.dbname)
+    cached = _cache.get(key)
+    if cached and time.monotonic() - cached[0] < 60:
+        _cache.move_to_end(key)
+        return cached[1]
+    async with _load_lock:
+        cached = _cache.get(key)
+        if cached and time.monotonic() - cached[0] < 60:
+            return cached[1]
+        geometry = await read_geometry(conn)
+        _cache[key] = (time.monotonic(), geometry)
+        _cache.move_to_end(key)
+        while len(_cache) > 4:
+            _cache.popitem(last=False)
+        return geometry
+
+
+@router.get("/geometry", summary="读取带版本的水力专题层静态几何")
+async def playback_geometry(ctx: ProjectContext = Depends(get_project_context),
+                            conn: AsyncConnection = Depends(get_project_pg_connection)):
+    try:
+        geometry = await geometry_for_project(ctx, conn)
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
+    return Response(geometry.content, media_type="application/octet-stream",
+                    headers={"Cache-Control": "private, no-store", "ETag": f'"{geometry.version}"'})
+
+
+@router.get("/frames", summary="读取索引一致的 Float32 水力时间块")
+async def playback_frames(
+    start_time: str,
+    model_version: str = Query(min_length=24, max_length=24),
+    frame_count: int = Query(default=4, ge=1, le=MAX_FRAME_COUNT),
+    step_seconds: int = Query(default=900, ge=60, le=3600),
+    ctx: ProjectContext = Depends(get_project_context),
+    pg_conn: AsyncConnection = Depends(get_project_pg_connection),
+    ts_conn: AsyncConnection = Depends(get_timescale_connection),
+):
+    try:
+        start = parse_utc_time(start_time, field_name="start_time")
+        geometry = await geometry_for_project(ctx, pg_conn)
+        if geometry.version != model_version:
+            raise HTTPException(409, "管网模型已更新，请重新加载专题层")
+        content = await read_frames(ts_conn, geometry, start, frame_count, step_seconds)
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
+    return Response(content, media_type="application/octet-stream",
+                    headers={"Cache-Control": "private, no-store"})

app/api/v1/router.py

@@ -47,6 +47,7 @@ from app.api.v1.endpoints.timeseries import (
     analysis as ts_analysis,
     composite as ts_composite,
     realtime as ts_realtime,
+    playback as ts_playback,
     scada as ts_scada,
 )
 from app.auth.permissions import (
@@ -191,6 +192,7 @@ api_router.include_router(
 # TimescaleDB data
 for endpoint_router, tag in (
     (ts_realtime.router, "TimescaleDB - Realtime"),
+    (ts_playback.router, "TimescaleDB - Hydraulic Playback"),
     (ts_analysis.router, "TimescaleDB - Analysis"),
 ):
     api_router.include_router(

app/services/burst_impact.py

@@ -139,9 +139,12 @@ def estimate_burst_impact(network: str, start: datetime, simulation: dict[str, A
         return output
     groups = [{key: group[key] for key in ("group_id", "title", "node_ids")}
               for group in impact["grades"]]
-    geometry = generate_node_coverage(network, groups)
+    # GRADES are ordered from low to high. Keep the numerical classes intact
+    # while removing overlap between their independently derived envelopes.
+    geometry = generate_node_coverage(network, groups, exclusive_groups=True)
     impact["coverage_status"] = geometry["coverage"]["status"]
     geometry["coverage"]["boundary"] = impact["boundary"]
+    geometry["coverage"]["boundary"] += "分档包络互斥展示，重叠区域优先显示较高降压档，不改变节点分档或表示精确水力等值边界。"
     for feature in geometry["map_payload"]["features"]:
         feature["properties"].update(
             impact_phase=impact["phase"], pressure_drop_unit="m",

app/services/calculation_worker.py

@@ -93,8 +93,8 @@ async def execute_child(input_path: str, output_path: str):
             category = 'EXECUTION_CONTEXT_UNAVAILABLE'
         payload = {'status': 'failed', 'error': {'code': category}}
     encoded = json.dumps(payload, allow_nan=False)
-    if len(encoded.encode()) > 8 * 1024 * 1024:
-        encoded = json.dumps({'status': 'failed', 'error': {'code': 'RESULT_TOO_LARGE'}})
+    # Keep the complete calculation for persistence. MCP result reads apply
+    # their own summary/page budgets after the result has been saved.
     Path(output_path).write_text(encoded)
 
 
@@ -212,8 +212,6 @@ class CalculationWorker:
                         terminal = {'status': 'timed_out', 'error': {'code': 'EXECUTION_TIMEOUT'}}
                     elif process.returncode is not None:
                         try:
-                            if output_path.stat().st_size > 8 * 1024 * 1024:
-                                raise ValueError('Result exceeds persistence limit')
                             terminal = json.loads(output_path.read_text())
                         except (OSError, ValueError):
                             terminal = {'status': 'failed', 'error': {'code': 'CHILD_PROCESS_FAILED'}}

app/services/hydraulic_playback.py

@@ -0,0 +1,184 @@
+"""Versioned static geometry and compact, indexed hydraulic playback frames."""
+
+from array import array
+import asyncio
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+import hashlib
+import json
+import math
+import struct
+import sys
+from typing import Any
+
+from psycopg import AsyncConnection
+
+from app.infra.db.postgresql.network_settings import NetworkSettingsRepository
+
+WORLD_METERS = 40075016.68557849
+MAX_FRAME_COUNT = 8
+
+
+def packet(header: dict[str, Any], values: array) -> bytes:
+    """Four-byte aligned JSON header followed by little-endian Float32 values."""
+    encoded = json.dumps(header, separators=(",", ":"), ensure_ascii=False).encode()
+    encoded += b" " * (-len(encoded) % 4)
+    if sys.byteorder != "little":
+        values = array("f", values)
+        values.byteswap()
+    return struct.pack("<I", len(encoded)) + encoded + values.tobytes()
+
+
+def mercator(longitude: float, latitude: float) -> tuple[float, float]:
+    latitude = max(-85.05112878, min(85.05112878, latitude))
+    return ((longitude + 180) / 360,
+            (1 - math.asinh(math.tan(math.radians(latitude))) / math.pi) / 2)
+
+
+@dataclass(frozen=True)
+class PlaybackGeometry:
+    version: str
+    node_ids: tuple[str, ...]
+    pipe_ids: tuple[str, ...]
+    node_index: dict[str, int]
+    pipe_index: dict[str, int]
+    content: bytes
+    flow_factor: float
+    pressure_factor: float
+    velocity_factor: float
+
+
+def result_factors(units: dict[str, str]) -> tuple[float, float, float]:
+    """Match EPANET outputs to m³/h, metres of water, and m/s."""
+    flow = units["flow"].upper()
+    flow_factors = {"LPS": 3.6, "LPM": .06, "MLD": 1000 / 24,
+                    "CMH": 1, "CMD": 1 / 24, "CFS": .028316846592 * 3600,
+                    "GPM": .003785411784 * 60, "MGD": 3785.411784 / 24,
+                    "IMGD": 4546.09 / 24, "AFD": 1233.48183754752 / 24}
+    pressure = units["pressure"].upper()
+    pressure_factors = {"M": 1, "METERS": 1, "PSI": .7032496149, "KPA": .1019716213}
+    if flow not in flow_factors or pressure not in pressure_factors:
+        raise ValueError("不支持的模拟结果单位")
+    return (flow_factors[flow], pressure_factors[pressure],
+            .3048 if flow in {"CFS", "GPM", "MGD", "IMGD", "AFD"} else 1)
+
+
+def line_paths(geometry: dict) -> list:
+    if geometry["type"] == "LineString":
+        return [geometry["coordinates"]]
+    if geometry["type"] == "MultiLineString":
+        return geometry["coordinates"]
+    if geometry["type"] == "GeometryCollection":
+        # PostGIS LineMerge returns an empty collection for zero-length lines.
+        return [path for child in geometry["geometries"] for path in line_paths(child)]
+    raise ValueError("管线几何必须为线类型")
+
+
+def build_geometry(nodes: list[dict], pipes: list[dict], units: dict[str, str]) -> PlaybackGeometry:
+    if not nodes or not pipes:
+        raise ValueError("管网没有可发布的节点或管线几何")
+    node_ids = tuple(str(row["id"]) for row in nodes)
+    pipe_ids = tuple(str(row["id"]) for row in pipes)
+    node_positions = [mercator(float(row["longitude"]), float(row["latitude"])) for row in nodes]
+    positions_by_id = dict(zip(node_ids, node_positions))
+    origin = [(min(p[i] for p in node_positions) + max(p[i] for p in node_positions)) / 2 for i in (0, 1)]
+    values = array("f", (coordinate - origin[axis] for position in node_positions for axis, coordinate in enumerate(position)))
+    segment_count = 0
+    for index, row in enumerate(pipes):
+        geometry = row["geometry"]
+        if isinstance(geometry, str):
+            geometry = json.loads(geometry)
+        paths = line_paths(geometry)
+        start_node = positions_by_id.get(str(row.get("start_node_id", "")))
+        if start_node and paths and paths[0] and paths[-1]:
+            first = mercator(*paths[0][0][:2])
+            last = mercator(*paths[-1][-1][:2])
+            if math.dist(last, start_node) < math.dist(first, start_node):
+                paths = [list(reversed(path)) for path in reversed(paths)]
+        distance = 0.0
+        for path in paths:
+            positions = [mercator(float(p[0]), float(p[1])) for p in path]
+            for start, end in zip(positions, positions[1:]):
+                length = math.hypot(end[0] - start[0], end[1] - start[1])
+                if length <= 0:
+                    continue
+                # Instanced segment: endpoints relative to origin, pipe index,
+                # cumulative distance, segment length, physical diameter (mm).
+                values.extend((start[0] - origin[0], start[1] - origin[1],
+                               end[0] - origin[0], end[1] - origin[1],
+                               index, distance, length, float(row["diameter"] or 0)))
+                distance += length
+                segment_count += 1
+    factors = result_factors(units)
+    descriptor = json.dumps([node_ids, pipe_ids, origin, units], separators=(",", ":")).encode()
+    version = hashlib.sha256(descriptor + values.tobytes()).hexdigest()[:24]
+    header = {"kind": "geometry", "protocol": 1, "modelVersion": version,
+              "nodeIds": node_ids, "pipeIds": pipe_ids, "origin": origin,
+              "segmentCount": segment_count, "units": {"pressure": "m", "flow": "m³/h", "velocity": "m/s"}}
+    return PlaybackGeometry(version, node_ids, pipe_ids,
+                            {key: i for i, key in enumerate(node_ids)},
+                            {key: i for i, key in enumerate(pipe_ids)},
+                            packet(header, values), *factors)
+
+
+async def read_geometry(conn: AsyncConnection) -> PlaybackGeometry:
+    async with conn.cursor() as cursor:
+        await cursor.execute("""
+            SELECT n.id, ST_X(ST_Transform(g.geom,4326)) AS longitude,
+                   ST_Y(ST_Transform(g.geom,4326)) AS latitude
+            FROM gis.network_nodes n JOIN gis.node_geometries g ON g.node_id=n.id
+            WHERE g.geom IS NOT NULL AND NOT ST_IsEmpty(g.geom) ORDER BY n.id
+        """)
+        nodes = await cursor.fetchall()
+        await cursor.execute("""
+            SELECT id, start_node_id, diameter, ST_AsGeoJSON(ST_Transform(ST_LineMerge(geom),4326))::json AS geometry
+            FROM gis.pipes WHERE geom IS NOT NULL AND NOT ST_IsEmpty(geom) ORDER BY id
+        """)
+        pipes = await cursor.fetchall()
+    units = await NetworkSettingsRepository.get_result_units(conn)
+    return await asyncio.to_thread(build_geometry, nodes, pipes, units)
+
+
+def finite(value: Any, factor: float) -> float:
+    if value is None:
+        return math.nan
+    result = float(value) * factor
+    return result if math.isfinite(result) and abs(result) <= 3.402823466e38 else math.nan
+
+
+async def read_frames(conn: AsyncConnection, geometry: PlaybackGeometry, start: datetime,
+                      frame_count: int, step_seconds: int) -> bytes:
+    times = [start + timedelta(seconds=step_seconds * i) for i in range(frame_count)]
+    node_count, pipe_count = len(geometry.node_ids), len(geometry.pipe_ids)
+    stride = node_count + 2 * pipe_count
+    values = array("f", [math.nan]) * (stride * frame_count)
+    time_index = {value: i for i, value in enumerate(times)}
+    valid_counts = {metric: [0] * frame_count for metric in ("pressure", "flow", "velocity")}
+    async with conn.cursor() as cursor:
+        await cursor.execute("""
+            SELECT time, node_id, pressure FROM realtime.node_results
+            WHERE time = ANY(%s) ORDER BY time, node_id
+        """, (times,))
+        async for row in cursor:
+            index = geometry.node_index.get(str(row["node_id"]))
+            if index is not None:
+                frame = time_index[row["time"]]
+                values[frame * stride + index] = finite(row["pressure"], geometry.pressure_factor)
+                valid_counts["pressure"][frame] += math.isfinite(values[frame * stride + index])
+        await cursor.execute("""
+            SELECT time, link_id, flow, velocity FROM realtime.link_results
+            WHERE time = ANY(%s) ORDER BY time, link_id
+        """, (times,))
+        async for row in cursor:
+            index = geometry.pipe_index.get(str(row["link_id"]))
+            if index is not None:
+                frame = time_index[row["time"]]
+                offset = frame * stride + node_count
+                values[offset + index] = finite(row["flow"], geometry.flow_factor)
+                values[offset + pipe_count + index] = finite(row["velocity"], geometry.velocity_factor)
+                valid_counts["flow"][frame] += math.isfinite(values[offset + index])
+                valid_counts["velocity"][frame] += math.isfinite(values[offset + pipe_count + index])
+    return packet({"kind": "frames", "protocol": 1, "modelVersion": geometry.version,
+                   "nodeCount": node_count, "pipeCount": pipe_count,
+                   "times": [int(t.timestamp() * 1000) for t in times], "validCounts": valid_counts,
+                   "available": [valid_counts["pressure"][i] > 0 and valid_counts["flow"][i] > 0 for i in range(frame_count)]}, values)

app/services/investigation_graph.py

@@ -39,6 +39,7 @@ def update_investigation_graph(
         ))
 
     nodes = {node.node_id: node.model_copy(deep=True) for node in previous.node_refs} if previous else {}
+    saved_ids = set(nodes)
     require(len({node.node_id for node in update.nodes}) == len(update.nodes),
             "GRAPH_NODE_DUPLICATE", "Graph update contains duplicate node ids", status=422)
     for node in update.nodes:
@@ -81,6 +82,26 @@ def update_investigation_graph(
             edges.append(edge.model_copy(deep=True))
     require(len(nodes) <= 100 and len(edges) <= 200, "GRAPH_CAPACITY_EXCEEDED",
             "Investigation graph exceeds 100 nodes or 200 edges; existing paths were retained", status=422)
+    # Validate only newly added steps. Historical snapshots and status-only
+    # patches stay valid, even if an earlier publisher omitted relationships.
+    new_ids = set(nodes) - saved_ids
+    if new_ids and len(nodes) > 1:
+        neighbours = {node_id: set() for node_id in nodes}
+        for edge in edges:
+            neighbours[edge.source].add(edge.target)
+            neighbours[edge.target].add(edge.source)
+        connected = set(saved_ids) or {next(iter(nodes))}
+        pending = list(connected)
+        while pending:
+            for node_id in neighbours[pending.pop()] - connected:
+                connected.add(node_id)
+                pending.append(node_id)
+        missing = sorted(new_ids - connected)
+        require(not missing, "GRAPH_RELATION_REQUIRED",
+                "Add explicit edges connecting these new steps to the investigation narrative: "
+                + ", ".join(missing)
+                + ". Describe branches, parallel findings or retries; edges do not schedule execution.",
+                status=422)
     lines = [source or "flowchart TD"]
     for node in nodes.values():
         if in_source(node.node_id):

app/services/node_coverage.py

@@ -6,7 +6,7 @@ from typing import Any
 
 from pyproj import CRS, Transformer
 from shapely import MultiPoint, concave_hull, get_num_coordinates
-from shapely.geometry import mapping
+from shapely.geometry import GeometryCollection, mapping, shape
 from shapely.ops import transform, unary_union
 
 from app.native.wndb.gis.network_views import get_node_coverage_links, get_node_geographic_positions
@@ -17,6 +17,47 @@ SIMPLIFY_METERS = 2.0
 MAX_COVERAGE_NODES = 10000
 
 
+def exclusive_coverage_features(features: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Clip display envelopes, with later groups taking precedence; retain input order.
+
+    This changes only derived geometry, never node membership or model values.
+    No post-clipping buffer or simplification may reintroduce an overlap.
+    """
+    occupied = GeometryCollection()
+    result = []
+    summaries = {}
+    for feature in features:
+        properties = feature["properties"]
+        summaries.update(properties.get("coverage_group_summary", {}))
+    for feature in features:
+        properties = feature["properties"]
+        summaries[properties["coverage_group"]] = {
+            key: properties.get(key) for key in ("node_count", "mapped_node_count", "missing_node_count")
+        }
+    for feature in reversed(features):
+        original = shape(feature["geometry"])
+        visible = original.difference(occupied)
+        occupied = unary_union([occupied, original])
+        if visible.is_empty:
+            continue
+        if (visible.geom_type not in {"Polygon", "MultiPolygon"} or not visible.is_valid
+                or get_num_coordinates(visible) > 20000):
+            raise ValueError("Invalid exclusive derived coverage")
+        result.append({
+            **feature,
+            "properties": {**feature["properties"], "overlap_policy": "higher_priority_group_first"},
+            "geometry": mapping(visible),
+        })
+    visible_groups = {feature["properties"]["coverage_group"] for feature in result}
+    summaries = {key: {**value, "display_area_available": key in visible_groups}
+                 for key, value in summaries.items()}
+    for feature in result:
+        # Retain counts even when that group's entire display envelope was
+        # clipped away. Overlay transport deliberately sends only GeoJSON.
+        feature["properties"]["coverage_group_summary"] = summaries
+    return list(reversed(result))
+
+
 def _components(node_ids: list[str], links: list[dict[str, Any]], closed: set[str]) -> list[list[str]]:
     parent = {node_id: node_id for node_id in node_ids}
 
@@ -41,6 +82,7 @@ def build_node_coverage(
     links: list[dict[str, Any]], *, split_by_connectivity: bool = False,
     excluded_link_ids: list[str] | None = None,
     concave_ratio: float = CONCAVE_RATIO, buffer_m: float = BUFFER_METERS,
+    exclusive_groups: bool = False,
 ) -> dict[str, Any]:
     """Keep caller-defined comparison groups and derive geometry without assigning business meaning."""
     coordinates = {}
@@ -99,6 +141,12 @@ def build_node_coverage(
         }
         features.append(feature)
 
+    if exclusive_groups:
+        features = exclusive_coverage_features(features)
+        visible_groups = {feature["properties"]["coverage_group"] for feature in features}
+        for group_id, metadata in groups_metadata.items():
+            metadata["display_area_available"] = group_id in visible_groups
+
     return {
         "status": "partial" if any(group["missing_node_ids"] for group in groups_metadata.values())
         else "ready" if features else "empty",
@@ -106,7 +154,8 @@ def build_node_coverage(
         "parameters": {"concave_ratio": concave_ratio, "buffer_m": buffer_m,
                        "split_by_connectivity": split_by_connectivity,
                        "excluded_link_ids": sorted(closed),
-                       "simplify_m": simplify_m, "projection": "local_azimuthal_equidistant"},
+                       "simplify_m": simplify_m, "projection": "local_azimuthal_equidistant",
+                       "exclusive_groups": exclusive_groups},
         "output_crs": "EPSG:4326", "display_only": True,
         "position_source": "gis.node_geometries",
         "boundary": "节点空间覆盖示意；不推断故障、污染、停水、水力影响或用户服务边界。",

docs/burst-response-workflow.md

@@ -79,7 +79,7 @@ SCADA作为事件影响的辅助证据，按目标测点和事件／对照窗口
 
 在筛选最多20个展示节点之前，以原生模拟全网节点的起算首帧为计算对象，严格读取同一时刻已有在线模拟压力。不同压力单位换算为米水头，下降量为在线压力减爆管压力，互斥划分 `1≤Δ<3 m`、`3≤Δ<5 m`、`Δ≥5 m`。小于1米与缺少可比数据分别计数；错时、缺测、无效值或未知单位不补零、不取邻近值，不从代表测点外推全网。需要 `simulation.run`、`scada.view`，影响比较另需 `simulation.view`。
 
-输出 `impact` 保留每个可比节点的在线压力、爆管压力、下降量、分组、缺测清单、比较时刻和读取来源；`coverage` 和 `map_payload` 使用实际GIS节点坐标生成三档空间包络。缺坐标、覆盖节点超过10000或几何超预算时明确报告，不截取前若干节点冒充完整范围。地图采用黄／橙／红，图例保留每档已绘制／分组节点数。
+输出 `impact` 保留每个可比节点的在线压力、爆管压力、下降量、分组、缺测清单、比较时刻和读取来源；`coverage` 和 `map_payload` 使用实际GIS节点坐标生成三档空间包络。缺坐标、覆盖节点超过10000或几何超预算时明确报告，不截取前若干节点冒充完整范围。爆管分档包络按高降压档优先裁剪为互斥展示面，保留整体覆盖并集、原始节点分组和计数；低档包络被完全覆盖时保留其节点元数据，不输出空几何。该裁剪只处理空间包络的展示重叠，不把节点包络变为精确水力等值区。通用节点分组及关阀覆盖仍保持原有生成语义。地图三档采用蓝／橙／洋红，避免与关阀覆盖使用的红色混淆，图例保留每档已绘制／分组节点数。
 
 爆管模型使用 PDA 需水方式和原有10模型压力单位的服务压力设定；在线模型方式、当前模型时态及在线压力单位推定来源一并保留。在线记录没有独立版本化单位元数据，不能把此估计称为严格历史同模型反事实或现场停水边界。
 

docs/calculation-queue.md

@@ -43,6 +43,9 @@ API 保留 `uvicorn --workers 4`。长计算在项目 BizDB 持久化入队，
 
 调查的排队回执不写成分析证据。成功结果读取时，按原计算来源、任务及运行验证，并幂等地保存证据。未完成的计算阻止提交最终报告、`insufficient_data` 或将运行标记成功；中间进度仍可保存。
 
+计算子进程的完整结果文件及 supervisor 读取不设 8MiB 整份体积上限，全网分析结果按原值保存到 `calculation_jobs.result`。
+Agent 读取仍使用 `get_calculation_result` 的摘要及 `data_path/limit/offset` 分页预算；完整结果保存与单次工具消息大小分别处理。
+
 Agent 调查执行在原生回合空闲后由宿主短轮询等待，继续维持调查心跳。新作业进入终态即触发模型读取结果，其余作业继续后台运行。状态读取遇到网络、限流或临时服务错误，最多重试 3 次，采用指数退避和随机抖动；权限拒绝、上下文失效立即停止，重试耗尽明确报错，不重新提交或取消计算。普通聊天展示任务编号，可在当前会话后续查询。异常流结束时若无法安全继续处理新结果，会明确失败而不是虚报完成；Server 中的计算结果仍保留。
 
 ## 排队期限与结果保留

docs/investigation-progress-and-presentation.md

@@ -57,6 +57,12 @@ Server 不验证依赖执行顺序或推断节点完工。
 `progress.graph_update` 包含可选 `title`、`nodes` 和 `edges`。节点沿用 `node_id/label/status/evidence_refs/action_refs/map_targets`，
 可用 `detail` 记录实际结果或失败原因；连线为 `{source,target,label?}`。仅提交新增或变化的内容，
 `update_investigation_graph` 合并节点、追加明确连线并生成 Mermaid，保留已有路径和引用。
+新建调查图使用 `flowchart TD`，按自上而下的阅读顺序展示。多步骤图必须显式连接流程关系：
+独立核查从共同问题分叉并在判断处汇总，条件分支标注条件，失败尝试连接重试或替代路线。
+新增节点须通过连线接入已有调查脉络；首次发布多个节点时须形成相连的图。
+未接入的新增节点返回 `GRAPH_RELATION_REQUIRED`（422），错误中列出需补关系的节点。
+根据真实关系修正后重试，不按节点列表顺序自动补线；关系仅解释调查，不约束执行先后。
+单个起点、仅更新已有节点、纯文字更新和已保存的历史图仍可不传连线，不重排或改写历史报告。
 “保留路径”约束图的结构，不冻结已有节点的标记。`nodes` 按 `node_id` 合并：新增节点必须提供 `label`；
 修改已保存节点时可只提交 `node_id` 与变化字段，无需再提供 `label`、全部节点或连线。
 省略字段保留原值，`label: null` 也保留名称；`status: null` 显式恢复为参考步骤。

docs/mcp-query-implementation.md

@@ -16,7 +16,7 @@ Agent 根据401 challenge 或路径/根 well-known 发现受保护资源元数
 
 普通查询默认20条；目录类最多500条，显式指定设备的时序查询最多10000条、20个标识、24小时；offset 最大10000。SCADA调查首次省略 `query_scada_readings.device_ids`，由Server读取项目完整设备目录及窗口内全部授权观测，内部自动翻页，`offset` 必须为0。其余查询按问题选定对象和窗口。SQL 有 limit/offset 和 statement_timeout。历史对比在同一条SQL中按序列、时间对数据点分页，返回 page.has_more；非零offset也标明覆盖不完整。历史曲线展示调用初始每页500点，`RESULT_TOO_LARGE` 时原位减半重试，最多读取1000条、发出16次请求，共用120秒超时（单次请求同时受该剩余总预算约束），图表最多100个有效点；这不改变模型首次查询默认值或时序接口上限。
 
-单份结果超过8MiB报错；超过12KB的调查工具消息默认返回证据资源引用，设备目录的精简路径见下段。完整快照仍保存，Server 的 `resources/read` 可以读取它。Agent 本地 MCP 只提供工具，将上游资源引用转为 `evidence_read` 中的 `mcp_tjwater_get_task_evidence` 参数，不向 OpenCode 泄露不可在本地读取的链接。`get_task_evidence` 在数据库中按 data_path、limit、offset 选页，嵌套集合和超过256字符的字符串返回路径描述；字符串按 Unicode 字符分页。页和完整资源均在数据库返回前检查8MiB上限，保留原证据编号和合成来源；普通查询要求缩小范围。其他结果仍保留执行后的整体8MiB检查，不能把它理解为所有查询均有数据库端字节限制。原始/清洗实测、模型基线、分析结果、比较和 synthetic 均明确标记。非有限数值转为 null 并记录数据缺口。
+同步查询单份结果超过8MiB报错；后台计算作业的完整结果保存不设此上限，读取时仍执行工具消息摘要和分页预算；超过12KB的调查工具消息默认返回证据资源引用，设备目录的精简路径见下段。完整快照仍保存，Server 的 `resources/read` 可以读取它。Agent 本地 MCP 只提供工具，将上游资源引用转为 `evidence_read` 中的 `mcp_tjwater_get_task_evidence` 参数，不向 OpenCode 泄露不可在本地读取的链接。`get_task_evidence` 在数据库中按 data_path、limit、offset 选页，嵌套集合和超过256字符的字符串返回路径描述；字符串按 Unicode 字符分页。页和完整资源均在数据库返回前检查8MiB上限，保留原证据编号和合成来源；普通查询要求缩小范围。其他结果仍保留执行后的整体8MiB检查，不能把它理解为所有查询均有数据库端字节限制。原始/清洗实测、模型基线、分析结果、比较和 synthetic 均明确标记。非有限数值转为 null 并记录数据缺口。
 
 `list_scada_devices` 默认每页100个、最多500个，其他集合仍沿用各自的分页默认值。设备目录超过12,000字节时，先内联本页全部设备的 `device_id`、`device_type` 和 `measurement_unit`，省略其他属性并明确标注；模型映射、坐标等通过 `get_scada_device` 核验，任务证据和幂等回执保留完整属性。此时 `truncated` 表示属性省略，设备是否还有后续页以 `page.has_more` 为准，按本页实际设备数递增 `offset`。精简目录仍超预算时沿用证据引用或缩页路径。
 

tests/api/test_hydraulic_playback_endpoints.py

@@ -0,0 +1,79 @@
+from types import SimpleNamespace
+from uuid import uuid4
+import asyncio
+
+from fastapi.testclient import TestClient
+import pytest
+
+from app.api.v1.endpoints.timeseries import playback
+from app.auth.project_dependencies import ProjectContext
+from tests.conftest import build_test_app
+
+
+def client(monkeypatch):
+    context = ProjectContext(project_id=uuid4(), project_code="test", user_id=uuid4(), project_role="viewer")
+    app = build_test_app(playback.router, "/api/v1")
+    app.dependency_overrides[playback.get_project_context] = lambda: context
+    app.dependency_overrides[playback.get_project_pg_connection] = lambda: object()
+    app.dependency_overrides[playback.get_timescale_connection] = lambda: object()
+    async def geometry(*_):
+        return SimpleNamespace(version="a" * 24, content=b"geometry")
+    monkeypatch.setattr(playback, "geometry_for_project", geometry)
+    return TestClient(app)
+
+
+def test_geometry_is_private_and_frames_reject_a_different_model_version(monkeypatch):
+    http = client(monkeypatch)
+    response = http.get("/api/v1/timeseries/realtime/playback/geometry")
+    assert response.status_code == 200
+    assert response.headers["cache-control"] == "private, no-store"
+    assert response.content == b"geometry"
+    response = http.get("/api/v1/timeseries/realtime/playback/frames", params={
+        "start_time": "2026-07-16T00:00:00+08:00", "model_version": "b" * 24,
+    })
+    assert response.status_code == 409
+
+
+@pytest.mark.parametrize("params", [
+    {"start_time": "2026-07-16T00:00:00"},
+    {"start_time": "2026-07-16T00:00:00Z", "frame_count": 9},
+    {"start_time": "2026-07-16T00:00:00Z", "step_seconds": 0},
+])
+def test_frames_require_timezone_and_bound_the_block_size(monkeypatch, params):
+    http = client(monkeypatch)
+    response = http.get("/api/v1/timeseries/realtime/playback/frames", params={"model_version": "a" * 24, **params})
+    assert response.status_code == 422
+
+
+def test_frames_pass_exact_utc_time_and_counts_to_reader(monkeypatch):
+    http = client(monkeypatch)
+    async def frames(_connection, geometry, start, count, step):
+        assert geometry.version == "a" * 24
+        assert start.isoformat() == "2026-07-15T16:00:00+00:00"
+        assert count == 4 and step == 900
+        return b"frames"
+    monkeypatch.setattr(playback, "read_frames", frames)
+    response = http.get("/api/v1/timeseries/realtime/playback/frames", params={
+        "start_time": "2026-07-16T00:00:00+08:00", "model_version": "a" * 24,
+    })
+    assert response.status_code == 200 and response.content == b"frames"
+
+
+def test_cache_separates_projects_and_database_routes(monkeypatch):
+    playback._cache.clear()
+    calls = []
+    async def read(conn):
+        calls.append(conn.info.dbname)
+        return SimpleNamespace(version=conn.info.dbname)
+    monkeypatch.setattr(playback, "read_geometry", read)
+    ctx = SimpleNamespace(project_id=uuid4())
+    async def run():
+        for name in ["one", "one", "two"]:
+            connection = SimpleNamespace(info=SimpleNamespace(host="host", port=5432, dbname=name))
+            result = await playback.geometry_for_project(ctx, connection)
+            assert result.version == name
+        other = SimpleNamespace(project_id=uuid4())
+        await playback.geometry_for_project(other, SimpleNamespace(info=SimpleNamespace(host="host", port=5432, dbname="one")))
+    asyncio.run(run())
+    assert calls == ["one", "two", "one"]
+    playback._cache.clear()

tests/unit/test_burst_impact.py

@@ -22,11 +22,11 @@ def inputs(monkeypatch):
     read = Mock(side_effect=lambda ids, *a, **kw: {
         node: [{"time": START.isoformat(), "value": 30}] for node in ids})
     monkeypatch.setattr(service.InternalQueries, "query_realtime_simulation_by_ids_timerange", read)
-    def coverage(network, groups):
+    def coverage(network, groups, **parameters):
         ids = sorted({node for group in groups for node in group["node_ids"]})
         data = build_node_coverage(groups, [
             {"id": node, "longitude": 121.5 + i * .001, "latitude": 30.9}
-            for i, node in enumerate(ids)], [])
+            for i, node in enumerate(ids)], [], **parameters)
         payload = data.pop("map_payload")
         return {"coverage": data, "map_payload": payload}
     monkeypatch.setattr(service, "generate_node_coverage", coverage)
@@ -44,6 +44,7 @@ def test_thresholds_are_exclusive_with_no_rounding_and_no_closure(inputs):
     inputs.assert_called_once_with([f"J{i}" for i in range(8)], START, START, "node", "pressure", db_name="own")
     assert len(result["map_payload"]["features"]) == 3
     assert all(feature["properties"]["model_estimate"] for feature in result["map_payload"]["features"])
+    assert result["coverage"]["parameters"]["exclusive_groups"] is True
 
 
 @pytest.mark.parametrize("unit,factor", [("kPa", 1 / 9.80665), ("PSI", 6894.757293168 / 9806.65),

tests/unit/test_calculation_worker.py

@@ -1,5 +1,6 @@
 """Worker boundaries use real managed subprocesses, without external databases."""
 import asyncio
+import json
 from functools import wraps
 from types import SimpleNamespace
 from contextlib import asynccontextmanager
@@ -27,6 +28,66 @@ def run_async(function):
     return wrapper
 
 
+@run_async
+async def test_child_keeps_complete_result_above_eight_mib(monkeypatch, tmp_path):
+    from app.mcp import calculation_jobs
+    from app.services import calculation_worker
+
+    data = {"comparisons": "x" * (8 * 1024 * 1024), "map_payload": {"features": ["coverage"]}}
+
+    @asynccontextmanager
+    async def scope(row):
+        yield object()
+
+    async def calculate(scope, row):
+        return SimpleNamespace(model_dump=lambda **kwargs: data)
+
+    monkeypatch.setattr(calculation_worker, "execution_scope", scope)
+    monkeypatch.setattr(calculation_jobs, "execute_calculation_job", calculate)
+    input_path, output_path = tmp_path / "input.json", tmp_path / "result.json"
+    input_path.write_text("{}")
+    await calculation_worker.execute_child(str(input_path), str(output_path))
+    assert output_path.stat().st_size > 8 * 1024 * 1024
+    assert json.loads(output_path.read_text()) == {"status": "succeeded", "result": data}
+
+
+@run_async
+async def test_supervisor_persists_complete_result_above_eight_mib(monkeypatch, tmp_path):
+    monkeypatch.setenv("CALCULATION_WORKER_STATE_DIR", str(tmp_path))
+    original = asyncio.create_subprocess_exec
+
+    async def spawn(*args, **kwargs):
+        # A real child writes the same file protocol as the calculation process.
+        code = ('import json,pathlib,sys; pathlib.Path(sys.argv[1]).write_text('
+                'json.dumps({"status":"succeeded","result":{"comparisons":"x"*(8*1024*1024),'
+                '"map_payload":{"features":["coverage"]}}}))')
+        return await original(sys.executable, "-c", code, args[-1], start_new_session=True)
+
+    monkeypatch.setattr(asyncio, "create_subprocess_exec", spawn)
+    results = []
+
+    class Repository:
+        async def heartbeat(self, *args):
+            return "running"
+
+        async def finish(self, row, owner, **terminal):
+            results.append(terminal)
+            return True
+
+    @asynccontextmanager
+    async def repository(project):
+        yield Repository()
+
+    worker = CalculationWorker(WorkerSettings(heartbeat_seconds=.02), repository=repository)
+    await asyncio.wait_for(worker.run_job({"project_id": "project", "job_id": "large-result",
+                                          "timeout_seconds": 10}), 15)
+    assert len(results) == 1
+    assert results[0]["status"] == "succeeded"
+    assert results[0]["result"] == {
+        "comparisons": "x" * (8 * 1024 * 1024), "map_payload": {"features": ["coverage"]},
+    }
+
+
 def test_worker_settings_require_valid_concurrency(monkeypatch):
     settings = WorkerSettings.from_environment()
     assert settings.max_concurrency == 2

tests/unit/test_contaminant_simulation_mcp.py

@@ -42,27 +42,29 @@ def context(monkeypatch):
     return scope, binding
 
 
-def test_replay_prediction_can_extend_past_cutoff_without_reading_future_observations(monkeypatch):
+@pytest.mark.parametrize("include_impact", [True, False])
+def test_replay_prediction_can_extend_past_cutoff_without_reading_future_observations(monkeypatch, include_impact):
     scope, binding = context(monkeypatch)
     worker = AsyncMock(return_value={"impact": {"affected_node_count": 2}})
     monkeypatch.setattr(analysis_service, "calculate_replay_contaminant_simulation", worker)
-    result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["run_contaminant_simulation"], args()))
+    parameters = args(include_impact_coverage=include_impact)
+    result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["run_contaminant_simulation"], parameters))
     assert result.origin == "synthetic" and result.data["result"]["impact"]["affected_node_count"] == 2
     assert "预测" in result.data_gaps[0]
     scope.connection.commit.assert_awaited_once()
     worker.assert_awaited_once()
+    assert worker.await_args.args[0].include_impact_coverage is include_impact
     assert worker.await_args.kwargs == {"network": "project", "username": "actor"}
     assert should_queue(TOOLS["run_contaminant_simulation"], args())
 
 
-@pytest.mark.parametrize("problem", ["future_start", "no_impact", "reader", "permission"])
+@pytest.mark.parametrize("problem", ["future_start", "reader", "permission"])
 def test_replay_admission_keeps_time_permission_and_reader_boundaries(monkeypatch, problem):
     scope, binding = context(monkeypatch)
     parameters = args()
     if problem == "future_start": parameters = args(start_time="2026-07-16T08:15:00+08:00")
-    if problem == "no_impact": parameters = args(include_impact_coverage=False)
     if problem == "reader": binding.purpose = "task_reader"
     if problem == "permission": binding.permission_cap.remove("scada.view")
     with pytest.raises(HTTPException) as caught:
         asyncio.run(query_service.authorize_query(scope, binding, TOOLS["run_contaminant_simulation"], parameters))
-    assert caught.value.status_code == (422 if problem in {"future_start", "no_impact"} else 403)
+    assert caught.value.status_code == (422 if problem == "future_start" else 403)

tests/unit/test_hydraulic_playback.py

@@ -0,0 +1,93 @@
+import asyncio
+from datetime import datetime, timezone
+import json
+import math
+import struct
+
+import pytest
+
+from app.services.hydraulic_playback import build_geometry, read_frames, result_factors
+
+
+def fixture_geometry(longitude=121.5):
+    return build_geometry([
+        {"id": "J1", "longitude": longitude, "latitude": 30.8},
+        {"id": "J2", "longitude": longitude + .001, "latitude": 30.8},
+    ], [{"id": "P1", "start_node_id": "J1", "diameter": 600,
+         "geometry": {"type": "LineString", "coordinates": [[longitude + .001, 30.8], [longitude, 30.8]]}}],
+        {"flow": "LPS", "pressure": "METERS"})
+
+
+def unpack(content):
+    length = struct.unpack_from("<I", content)[0]
+    header = json.loads(content[4:4 + length])
+    count = (len(content) - 4 - length) // 4
+    return header, struct.unpack_from(f"<{count}f", content, 4 + length)
+
+
+def test_geometry_version_tracks_absolute_location_and_orients_geometry_with_start_node():
+    geometry = fixture_geometry()
+    header, values = unpack(geometry.content)
+    assert header["nodeIds"] == ["J1", "J2"]
+    assert header["pipeIds"] == ["P1"]
+    assert header["segmentCount"] == 1
+    assert values[4:6] == values[:2]  # reversed GIS geometry is oriented J1 -> J2
+    assert values[6:8] == values[2:4]
+    assert fixture_geometry().version == geometry.version
+    assert fixture_geometry(121.6).version != geometry.version
+
+
+def test_unit_factors_preserve_signed_flow_and_reject_unknown_units():
+    assert result_factors({"flow": "LPS", "pressure": "METERS"}) == (3.6, 1, 1)
+    flow, pressure, velocity = result_factors({"flow": "GPM", "pressure": "PSI"})
+    assert flow == pytest.approx(.003785411784 * 60)
+    assert pressure == pytest.approx(.7032496149)
+    assert velocity == .3048
+    with pytest.raises(ValueError):
+        result_factors({"flow": "unknown", "pressure": "METERS"})
+
+
+class Cursor:
+    def __init__(self, time):
+        self.time = time
+        self.rows = []
+    async def __aenter__(self):
+        return self
+    async def __aexit__(self, *_):
+        pass
+    async def execute(self, query, params):
+        assert len(params[0]) == 2
+        if "node_results" in query:
+            self.rows = [{"time": self.time, "node_id": "J1", "pressure": 25},
+                         {"time": self.time, "node_id": "J2", "pressure": None}]
+        else:
+            self.rows = [{"time": self.time, "link_id": "P1", "flow": -10, "velocity": .7}]
+    def __aiter__(self):
+        self.iterator = iter(self.rows)
+        return self
+    async def __anext__(self):
+        try:
+            return next(self.iterator)
+        except StopIteration:
+            raise StopAsyncIteration
+
+
+class Connection:
+    def __init__(self, time):
+        self.time = time
+    def cursor(self):
+        return Cursor(self.time)
+
+
+def test_frames_pack_exact_sample_times_signed_values_and_missing_values_without_extrapolation():
+    time = datetime(2026, 7, 15, 16, tzinfo=timezone.utc)
+    content = asyncio.run(read_frames(Connection(time), fixture_geometry(), time, 2, 900))
+    header, values = unpack(content)
+    assert header["times"] == [1784131200000, 1784132100000]
+    assert header["available"] == [True, False]
+    assert header["validCounts"]["pressure"] == [1, 0]
+    assert values[0] == 25
+    assert math.isnan(values[1])
+    assert values[2] == -36
+    assert values[3] == pytest.approx(.7)
+    assert all(math.isnan(value) for value in values[4:])

tests/unit/test_investigation_graph.py

@@ -239,6 +239,85 @@ def test_graph_renderer_escapes_labels_and_keeps_explicit_parallel_and_failed_pa
     assert graph.node_refs[2].status == "reference"
 
 
+@pytest.mark.parametrize("edges", [[], [{"source": "check", "target": "check"}]])
+def test_new_multistep_graph_requires_relationships_instead_of_isolated_nodes(edges):
+    with pytest.raises(DomainError) as error:
+        update_investigation_graph(None, InvestigationGraphUpdate.model_validate({
+            "nodes": [{"node_id": "read", "label": "观测核验"},
+                      {"node_id": "check", "label": "异常判断"}],
+            "edges": edges,
+        }))
+    assert error.value.code == "GRAPH_RELATION_REQUIRED"
+    assert "check" in str(error.value)
+
+
+def test_new_branch_must_connect_to_the_saved_narrative_without_inventing_edges():
+    original = update_investigation_graph(None, InvestigationGraphUpdate.model_validate({
+        "nodes": [{"node_id": "root", "label": "调查起点"}],
+    }))
+    with pytest.raises(DomainError) as error:
+        update_investigation_graph(original, InvestigationGraphUpdate.model_validate({
+            "nodes": [{"node_id": "left", "label": "独立核查"},
+                      {"node_id": "right", "label": "核查发现"}],
+            "edges": [{"source": "left", "target": "right"}],
+        }))
+    assert error.value.code == "GRAPH_RELATION_REQUIRED"
+    assert original.edges == []
+    assert len(original.node_refs) == 1
+
+
+def test_connected_parallel_branches_and_join_render_top_down():
+    graph = update_investigation_graph(None, InvestigationGraphUpdate.model_validate({
+        "nodes": [{"node_id": key, "label": label} for key, label in (
+            ("start", "异常核验"), ("isolation", "隔离分析"),
+            ("impact", "爆管影响"), ("summary", "汇总判断"))],
+        "edges": [{"source": source, "target": target, "label": label}
+                  for source, target, label in (
+                      ("start", "isolation", "并行核查"), ("start", "impact", "并行核查"),
+                      ("isolation", "summary", "汇总依据"), ("impact", "summary", "汇总依据"))],
+    }))
+    assert graph.mermaid.startswith("flowchart TD\n")
+    assert len(graph.edges) == 4
+    assert 'start -->|"并行核查"| impact' in graph.mermaid
+    patched = update_investigation_graph(graph, InvestigationGraphUpdate.model_validate({
+        "nodes": [{"node_id": "impact", "status": "recorded"}],
+    }))
+    assert patched.edges == graph.edges
+
+
+def test_existing_disconnected_snapshot_remains_readable_and_patchable():
+    original = InvestigationDiagram.model_validate({
+        "title": "已保存的旧图", "mermaid": 'flowchart TD\n  read["观测"]\n  check["判断"]',
+        "edges": [], "node_refs": [{"node_id": "read", "label": "观测"},
+                                   {"node_id": "check", "label": "判断"}],
+    })
+    patched = update_investigation_graph(original, InvestigationGraphUpdate.model_validate({
+        "nodes": [{"node_id": "check", "status": "recorded"}],
+    }))
+    assert patched.mermaid == original.mermaid
+    assert patched.edges == original.edges == []
+    assert original.node_refs[1].status is None
+
+
+@pytest.mark.anyio
+async def test_publication_rejects_missing_links_and_accepts_a_corrected_graph():
+    task, repo, service = setup_service()
+    first = await publish(service, repo, task["task_id"], nodes=[
+        {"node_id": "read", "label": "观测核验"},
+    ])
+    original = first.task.latest_progress.progress.diagram.model_dump(mode="json")
+    with pytest.raises(DomainError) as error:
+        await publish(service, repo, task["task_id"], nodes=[
+            {"node_id": "check", "label": "异常判断"},
+        ])
+    assert error.value.code == "GRAPH_RELATION_REQUIRED"
+    assert (await service.detail(task["task_id"])).latest_progress.progress.diagram.model_dump(mode="json") == original
+    corrected = await publish(service, repo, task["task_id"], nodes=[
+        {"node_id": "check", "label": "异常判断"},
+    ], edges=[{"source": "read", "target": "check", "label": "核验依据"}])
+    assert 'read -->|"核验依据"| check' in corrected.task.latest_progress.progress.diagram.mermaid
+
+
 def test_first_structured_update_preserves_an_existing_mermaid_without_reinterpreting_it():
     original = InvestigationDiagram.model_validate({
         "title": "原调查路线", "mermaid": 'flowchart LR\nA{核查条件} -->|具备| B[原路径]\nA --> C[另一分支]',

tests/unit/test_node_coverage.py

@@ -3,11 +3,12 @@ from datetime import datetime, timezone
 import pytest
 from fastapi.encoders import jsonable_encoder
 from pydantic import ValidationError
-from shapely.geometry import Point, shape
+from shapely.geometry import Point, box, mapping, shape
+from shapely.ops import unary_union
 
 from app.domain.schemas.operational_data import TaskDataSnapshot
 from app.mcp.analysis_models import NodeCoverageInput
-from app.services.node_coverage import build_node_coverage
+from app.services.node_coverage import build_node_coverage, exclusive_coverage_features
 from app.services.operational_presentation import resolve_overlay
 
 
@@ -32,6 +33,44 @@ def test_coverage_can_compare_arbitrary_groups_without_assigning_isolation_seman
     assert coverage["parameters"]["split_by_connectivity"] is False
 
 
+def test_exclusive_envelopes_keep_union_and_counts_with_higher_groups_taking_precedence():
+    originals = [box(0, 0, 4, 4), box(2, 0, 6, 4), box(3, 1, 5, 3)]
+    features = [{"type": "Feature", "id": str(i), "properties": {
+        "coverage_group": str(i), "node_count": 10 + i, "mapped_node_count": 10 + i,
+    }, "geometry": mapping(geometry)} for i, geometry in enumerate(originals)]
+    result = exclusive_coverage_features(features)
+    visible = [shape(feature["geometry"]) for feature in result]
+    assert len(result) == 3
+    assert all(a.intersection(b).area == 0 for i, a in enumerate(visible) for b in visible[i + 1:])
+    assert unary_union(visible).equals(unary_union(originals))
+    assert visible[-1].equals(originals[-1])
+    assert [feature["properties"]["node_count"] for feature in result] == [10, 11, 12]
+    assert [feature["properties"]["mapped_node_count"] for feature in result] == [10, 11, 12]
+    assert all(feature["properties"]["overlap_policy"] == "higher_priority_group_first" for feature in result)
+    assert shape(features[0]["geometry"]).equals(originals[0])  # Never rewrite saved inputs.
+
+
+def test_covered_grade_keeps_node_metadata_without_invalid_empty_geometry():
+    positions = [position("A", 117.24), position("B", 117.24)]
+    groups = [group(["A"], "low"), group(["B"], "high")]
+    ordinary = build_node_coverage(groups, positions, [])
+    assert len(ordinary["map_payload"]["features"]) == 2
+    exclusive = build_node_coverage(groups, positions, [], exclusive_groups=True)
+    assert [feature["id"] for feature in exclusive["map_payload"]["features"]] == ["high"]
+    assert exclusive["groups"]["low"]["node_ids"] == ["A"]
+    assert exclusive["groups"]["low"]["mapped_node_count"] == 1
+    assert exclusive["groups"]["low"]["display_area_available"] is False
+    assert exclusive["groups"]["high"]["display_area_available"] is True
+    summaries = exclusive["map_payload"]["features"][0]["properties"]["coverage_group_summary"]
+    assert summaries["low"]["node_count"] == 1
+    assert summaries["low"]["mapped_node_count"] == 1
+    assert summaries["low"]["display_area_available"] is False
+    # A second application must not drop the summary for the clipped group.
+    repeated = exclusive_coverage_features(exclusive["map_payload"]["features"])
+    assert repeated[0]["properties"]["coverage_group_summary"] == summaries
+    assert exclusive["status"] == "ready"
+
+
 def test_connectivity_option_does_not_bridge_disconnected_groups():
     positions = [position("A", 117.24), position("B", 117.241),
                  position("C", 117.26), position("D", 117.261)]