fix(api): synchronize hydraulic playback binary contract

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,73 @@
+"""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")
+BINARY_RESPONSE = {200: {"description": "Versioned little-endian Float32 packet",
+                         "content": {"application/octet-stream": {"schema": {"type": "string", "format": "binary"}}}}}
+# 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="读取带版本的水力专题层静态几何",
+            response_class=Response, responses=BINARY_RESPONSE)
+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 水力时间块",
+            response_class=Response, responses=BINARY_RESPONSE)
+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/infra/db/timescaledb/repositories/scada.py

@@ -143,9 +143,12 @@ class ScadaRepository:
             await cur.execute(
                 "WITH next_frame AS (SELECT min(time) AS time FROM scada.measurements "
                 "WHERE device_id=ANY(%s) AND time>%s AND time<=%s), "
-                "reference AS (SELECT time,device_id,monitored_value,cleaned_value "
+                "reference AS (SELECT m.time,m.device_id,m.monitored_value,m.cleaned_value "
-                "FROM scada.measurements WHERE device_id=ANY(%s) AND time>=%s AND time<%s "
+                "FROM scada.measurements m CROSS JOIN next_frame f "
-                "ORDER BY time,device_id LIMIT %s) "
+                "WHERE m.device_id=ANY(%s) AND m.time>=%s AND m.time<%s "
+                "AND mod(extract(epoch FROM m.time)::bigint,86400)="
+                "mod(extract(epoch FROM f.time)::bigint,86400) "
+                "ORDER BY m.time,m.device_id LIMIT %s) "
                 "SELECT time,device_id,monitored_value,cleaned_value FROM reference UNION ALL "
                 "SELECT m.time,m.device_id,m.monitored_value,m.cleaned_value FROM scada.measurements m "
                 "JOIN next_frame f ON m.time=f.time WHERE m.device_id=ANY(%s) ORDER BY time,device_id",

app/mcp/analysis_service.py

@@ -116,6 +116,15 @@ async def calculate_replay_burst_location(
     )
 
 
+async def calculate_replay_hydraulic_scenario(name: str, args: BaseModel, *, network: str, username: str) -> dict[str, Any]:
+    """Evaluate model-based hydraulic alternatives without live SCADA substitution."""
+    return await bounded_calculation(
+        calculate_scenario, name,
+        args.model_dump(exclude={"context_handle", "timeout_seconds"}, exclude_none=True),
+        network=network, username=username, model_only=True,
+    )
+
+
 async def calculate_replay_burst_simulation(args: BaseModel, *, network: str, username: str) -> dict[str, Any]:
     """An independent current-model scenario, with no live SCADA substitution."""
     return await bounded_calculation(

app/mcp/catalog.py

@@ -63,7 +63,7 @@ class QueryTool:
 
 _specs = [
     ("run_contaminant_source_trace", "水污染候选溯源：提交由事件位置或监测映射取得的候选模型节点及事件窗，快速返回确定的已核验节点。当前为临时稳定选择，正式反演算法后续接入；不生成置信度。取得located_node后先展示节点、保存阶段结论并提现场核查/采样处置工单；工单与污染传播模拟、accident_element_type=node的隔离和影响分析并行，结果补入同一调查，不因算法来源停止", ContaminantSourceTraceInput, "simulation.run", "analysis", False, True),
-    ("run_simulation", "按水力步长在在线业务模型上执行指定时间段的实时模拟；更新该时段的模型条件及实时基线，返回指定节点和管段结果；duration单位为秒，输出为在线模型结果，正式分析成果通过方案工具保存", ScenarioInput, "simulation.run", "analysis", False, True),
+    ("run_simulation", "按水力步长在在线业务模型上执行指定时间段的实时模拟，更新全网实时基线，返回指定节点和管段结果；调查发现同刻基线缺失时可自主补算，等待成功并读取结果后重试依赖分析。duration单位为秒，补单帧可用60秒；回放按当前模型补算，不读取实时SCADA，不替代冻结观测；正式分析成果通过方案工具保存", ScenarioInput, "simulation.run", "analysis", False, True),
     ("run_contaminant_simulation", "在隔离模型中模拟假设污染源的浓度传播；include_impact_coverage=true时汇总全网节点整个模拟时段超过impact_threshold_mg_l的累计范围、首个超阈值报告时刻和峰值，并保存地图证据。阈值为情景展示条件，非饮用水安全限值；回放按当前模型独立估计，不读取实时SCADA。定位后先提现场核查/采样处置工单，工单审核派发与本计算并行，结果补入同一调查", ContaminantSimulationInput, "simulation.run", "analysis", False, True),
     ("analyze_valve_closure", "在隔离模型中模拟关阀的水力影响，返回指定节点/管段结果；现场阀门操作按业务审批执行", ValveClosureInput, "simulation.run", "analysis", False, True),
     ("analyze_flushing", "在独立模型中评估指定排放节点、流量及明确开关阀组合的冲洗水力结果；未列阀门沿用当前模型，不继承上次模拟；不判定污染清除或恢复供水", FlushingInput, "simulation.run", "analysis", False, True),

app/mcp/contexts.py

@@ -15,7 +15,6 @@ from app.mcp.models import BindInput, BoundContext
 from app.mcp.scope import QueryScope
 from app.mcp.config import McpSettings
 from mcp.server.auth.provider import AccessToken
-from app.mcp.work_order_skills import skill_name_for_task
 
 
 def task_repository(scope: QueryScope) -> OperationalTaskRepository:
@@ -32,8 +31,6 @@ async def verify_task(scope: QueryScope, binding: BoundContext) -> None:
         if not task:
             raise HTTPException(404, "Task is not in this project")
         binding.replay_context = task.get("source", {}).get("replay_context")
-        if skill_name_for_task(task.get("workflow"), binding.replay_context) == "burst-event-handling":
-            binding.tool_cap = [name for name in binding.tool_cap if name != "run_burst_location"]
         if binding.purpose == "task_collaboration":
             return
         if OPERATIONAL_TASK_EXECUTE not in permissions:

app/mcp/query_service.py

@@ -79,20 +79,20 @@ async def authorize_query(
                    "get_element_properties", "list_elements", "query_nearby_elements", "get_network_options",
                    "get_node_connections", "get_element_gis_properties", "query_network_extent",
                    "list_regions", "get_region", "get_region_nodes",
-                   "analyze_valve_isolation", "generate_node_coverage", "run_burst_simulation", "run_contaminant_simulation", "analyze_flushing",
+                   "analyze_valve_isolation", "generate_node_coverage", *SCENARIO_TOOLS,
+                   "calculate_demand_allocation", "optimize_sensor_placement",
+                   "query_model_baseline_series", "get_model_baseline_at_time", "query_model_baseline_field",
                    "get_calculation_job", "get_calculation_result", "list_calculation_jobs"}
-        if tool.name not in allowed:
+        asset_query = tool.semantics == "asset" and not tool.mutating and not tool.computation
+        if tool.name not in allowed and not asset_query:
             raise HTTPException(403, "回放调查仅开放冻结 SCADA、资产、拓扑分析及指定模型情景；时序观测请使用 query_scada_readings")
-        if tool.name == "run_burst_simulation":
+        if tool.name in SCENARIO_TOOLS:
-            if not args.include_impact_coverage:
-                raise HTTPException(422, "回放爆管情景必须启用影响覆盖，按当前模型估计并保存同刻在线基线")
-            check_replay_window(replay_context, args.start_time, args.start_time)
-        if tool.name == "run_contaminant_simulation":
-            if not args.include_impact_coverage:
-                raise HTTPException(422, "回放污染情景必须启用影响覆盖，按当前模型估计传播范围")
-            check_replay_window(replay_context, args.start_time, args.start_time)
-        if tool.name == "analyze_flushing":
             check_replay_window(replay_context, args.start_time, args.start_time)
+        if tool.semantics == "model_baseline":
+            if tool.name == "query_model_baseline_series":
+                check_replay_window(replay_context, args.start, args.end)
+            else:
+                check_replay_window(replay_context, args.time, args.time)
         if tool.name == "query_scada_readings":
             check_replay_window(replay_context, args.start, args.end)
         if tool.name == "run_contaminant_source_trace":
@@ -213,6 +213,17 @@ async def execute_query(
                                           args.device_ids, args.start, args.end, args.limit + 1, args.offset)
         data, page, truncated = page_result(rows, args)
         origin = "synthetic"
+    elif name in {"run_simulation", "analyze_valve_closure", "analyze_pressure_regulation"} and replay_context:
+        from app.mcp.analysis_service import calculate_replay_hydraulic_scenario
+
+        await conn.commit()
+        async with scope.simulation_routing():
+            result = await calculate_replay_hydraulic_scenario(
+                name, args, network=scope.project.project_code, username=scope.user.username,
+            )
+        data = {"parameters": args.model_dump(mode="json", exclude={"context_handle"}), "result": result}
+        origin = "synthetic"
+        gaps = ["回放水力计算使用当前模型，不读取实时SCADA，不代表冻结观测或历史模型重建。在线模拟更新模型基线，关阀及调压仅评估隔离情景；未来模拟样本仅为预测。"]
     elif name == "run_burst_simulation" and replay_context:
         from app.mcp.analysis_service import calculate_replay_burst_simulation
 
@@ -632,6 +643,8 @@ async def execute_query(
         truncated = snapshot.truncated
     else:
         raise ValueError("query handler is not registered")
+    if replay_context and tool.semantics == "model_baseline":
+        gaps.append("在线模型参考按本次查询时可用数据读取，非冻结SCADA观测；历史报告复用已保存证据。")
     window = (
         {"start": args.start, "end": args.end}
         if hasattr(args, "start") and args.start is not None

app/mcp/scenario_service.py

@@ -29,50 +29,36 @@ def calculate_scenario(name: str, parameters: dict[str, Any], *, network: str, u
     impact_threshold = parameters.pop("impact_threshold_mg_l", 0)
     common = dict(name=network, modify_pattern_start_time=start.isoformat(),
                   modify_total_duration=duration, persist_results=False, scheme_name=None)
+    boundary = {}
+    if model_only:
+        from app.infra.db.postgresql.scada import ScadaElementMappings
+        # Model scenarios may support replay decisions without consuming mutable
+        # live SCADA in place of the task's frozen observations.
+        boundary["scada_mappings"] = ScadaElementMappings(
+            reservoirs={}, tanks={}, fixed_pumps={}, variable_pumps={},
+            pressure={}, demand={}, quality={},
+        )
     if name == "run_simulation":
         from app.services.simulation import run_online_simulation
 
         result = run_online_simulation(
-            network, start, duration, output_node_ids=nodes, output_link_ids=links,
+            network, start, duration, output_node_ids=nodes, output_link_ids=links, **boundary,
         )
     elif name == "run_burst_simulation":
-        boundary = {}
-        if model_only:
-            from app.infra.db.postgresql.scada import ScadaElementMappings
-            # Replay scenarios use current model assumptions, never mutable live
-            # SCADA readings in place of the investigation's frozen observations.
-            boundary["scada_mappings"] = ScadaElementMappings(
-                reservoirs={}, tanks={}, fixed_pumps={}, variable_pumps={},
-                pressure={}, demand={}, quality={},
-            )
         result = scenarios.burst_analysis(**common, username=username,
             burst_ID=parameters["burst_pipe_ids"], burst_size=parameters["orifice_areas_cm2"], **boundary)
     elif name == "run_contaminant_simulation":
-        boundary = {}
-        if model_only:
-            from app.infra.db.postgresql.scada import ScadaElementMappings
-            boundary["scada_mappings"] = ScadaElementMappings(
-                reservoirs={}, tanks={}, fixed_pumps={}, variable_pumps={},
-                pressure={}, demand={}, quality={},
-            )
         result = scenarios.contaminant_simulation(**common, username=username,
             source=parameters["source"], concentration=parameters["concentration"], source_pattern=parameters.get("pattern"), **boundary)
     elif name == "analyze_valve_closure":
         result = scenarios.valve_close_analysis(**common,
-            modify_valve_opening={valve: 0.0 for valve in parameters["valves"]})
+            modify_valve_opening={valve: 0.0 for valve in parameters["valves"]}, **boundary)
     elif name == "analyze_flushing":
         # The service consumes m³/h (the older REST label incorrectly says L/s).
         # Reject unsupported conversion rather than silently ignore the flow.
         from app.native.wndb.model.options import get_option
         if get_option(network).get("UNITS") not in {"LPS", "CMH"}:
             raise ValueError("Flushing currently supports LPS or CMH models only")
-        boundary = {}
-        if model_only:
-            from app.infra.db.postgresql.scada import ScadaElementMappings
-            boundary["scada_mappings"] = ScadaElementMappings(
-                reservoirs={}, tanks={}, fixed_pumps={}, variable_pumps={},
-                pressure={}, demand={}, quality={},
-            )
         result = scenarios.flushing_analysis(**common, username=username,
             drainage_node_ID=parameters["drainage_node_id"], flushing_flow=parameters["flushing_flow_m3h"],
             valve_control=flushing_valve_controls(parameters), **boundary)
@@ -87,7 +73,8 @@ def calculate_scenario(name: str, parameters: dict[str, Any], *, network: str, u
         result = scenarios.pressure_regulation(**common,
             modify_fixed_pump_pattern={key: value for key, value in controls.items() if key in mappings.fixed_pumps},
             modify_variable_pump_pattern={key: value for key, value in controls.items() if key in mappings.variable_pumps},
-            modify_tank_initial_level=parameters["tank_init_level"], scada_mappings=mappings)
+            modify_tank_initial_level=parameters["tank_init_level"],
+            scada_mappings=boundary.get("scada_mappings", mappings))
     else:
         raise ValueError("Scenario is not registered")
     if not isinstance(result, dict):

app/services/burst_impact.py

@@ -125,6 +125,13 @@ def estimate_burst_impact(network: str, start: datetime, simulation: dict[str, A
     impact["below_threshold_node_count"] = len(comparisons) - affected
     if not comparisons:
         impact.update(status="unavailable", reason="no_exact_comparable_online_snapshot")
+        impact["baseline_recovery"] = {
+            "tool": "run_simulation",
+            "parameters": {
+                "start_time": start.isoformat(), "duration": 60, "output_node_ids": ids[:1],
+            },
+            "next_step": "自主补算同刻在线基线；等待作业成功并读取结果后，按原参数重跑爆管影响计算，再发布降压覆盖。返回节点仅用于核验，在线基线按全网写入。",
+        }
         return output
     if affected > MAX_COVERAGE_NODES:
         impact["coverage_status"] = "unavailable"
@@ -132,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:
+    # Keep the complete calculation for persistence. MCP result reads apply
-        encoded = json.dumps({'status': 'failed', 'error': {'code': 'RESULT_TOO_LARGE'}})
+    # 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": "节点空间覆盖示意；不推断故障、污染、停水、水力影响或用户服务边界。",

app/services/operational_recovery.py

@@ -170,13 +170,18 @@ async def reconcile_recoveries(service: OperationalTaskService, timescale: Async
         devices = list(dict.fromkeys(roles.pressure_device_ids + roles.outlet_flow_device_ids)) if roles else []
         source = "scada_readings"
         through = service._now()
-        if context.get("device_ids"):
+        report_revision = task.get("current_report_revision")
+        if (context.get("device_ids") and context.get("device_selection_version") == 2
+                and context.get("source_report_revision") == report_revision):
             devices = context["device_ids"]
             interval = context.get("interval_seconds", interval)
         else:
-            # Retain the points the Agent actually inspected, including water
+            # Full-network browsing is discovery, not a declaration that every
-            # quality or local incident sensors outside the project role lists.
+            # device must recover for this incident. Prefer the report's scope.
-            for evidence in await service.repo.list_evidence(task["task_id"]):
+            report = await service.repo.get_latest_report(task["task_id"])
+            declared = ((report or {}).get("report") or {}).get("scada_device_ids") or []
+            devices.extend(declared)
+            for evidence in ([] if declared else await service.repo.list_evidence(task["task_id"])):
                 saved = evidence.get("data_snapshot") or {}
                 if saved.get("source") not in {"query_scada_readings", "scada_readings"}:
                     continue
@@ -200,11 +205,13 @@ async def reconcile_recoveries(service: OperationalTaskService, timescale: Async
             await service.repo.lock(f"task:{task['task_id']}")
             current = await service._task(UUID(task["task_id"]))
             # A new receipt, cancellation or another poll may have won the race.
-            if current.get("handling_context") != context or current["status"] in {"cancelled", "failed"}:
+            if (current.get("handling_context") != context or current["status"] in {"cancelled", "failed"}
+                    or current.get("current_report_revision") != report_revision):
                 continue
             now = service._now()
             if devices:
-                current["handling_context"].update(device_ids=devices, interval_seconds=interval or 0)
+                current["handling_context"].update(device_ids=devices, interval_seconds=interval or 0,
+                                                  device_selection_version=2, source_report_revision=report_revision)
             current["handling_context"]["last_checked_at"] = now.isoformat()
             current["handling_context"]["last_check"] = result
             if result["status"] == "passed":

app/services/simulation.py

@@ -157,6 +157,7 @@ def run_online_simulation(
     *,
     output_node_ids: list[str] | None = None,
     output_link_ids: list[str] | None = None,
+    scada_mappings: ScadaElementMappings | None = None,
 ) -> dict:
     """Recalculate online baseline snapshots without cloning a model database."""
     if start_time.tzinfo is None or duration_seconds <= 0:
@@ -166,7 +167,10 @@ def run_online_simulation(
     )
     if step <= 0:
         raise ValueError("HYDRAULIC TIMESTEP must be greater than 0.")
-    mappings = query_corresponding_element_id_and_query_id(name)
+    mappings = (
+        scada_mappings if scada_mappings is not None
+        else query_corresponding_element_id_and_query_id(name)
+    )
     selected = {
         "node": {node: [] for node in output_node_ids or []},
         "link": {link: [] for link in output_link_ids or []},
@@ -179,13 +183,13 @@ def run_online_simulation(
         output = run_simulation(
             name=name,
             simulation_type="realtime",
-            modify_pattern_start_time=current.isoformat(timespec="seconds"),
+            modify_pattern_start_time=current.isoformat(),
             scada_mappings=mappings,
         )
         if not isinstance(output, dict):
             raise RuntimeError("Online simulation did not return calculation results")
         units = output.get("units", {})
-        sample_times.append(current.isoformat(timespec="seconds"))
+        sample_times.append(current.isoformat())
         for kind, elements in selected.items():
             if not elements:
                 continue

app/services/simulation_scenarios.py

@@ -231,6 +231,7 @@ def valve_close_analysis(
     _temporary_project: str | None = None,
     persist_results: bool = True,
     username: str = "system",
+    scada_mappings: ScadaElementMappings | None = None,
 ) -> dict | str | None:
     """
     关阀模拟
@@ -291,6 +292,7 @@ def valve_close_analysis(
         result_db_name=name,
         persist_results=persist_results,
         scheme_username=username,
+        scada_mappings=scada_mappings,
     )
     return result if not persist_results else None
     # step 3. restore the base model

contracts/manifest.json

@@ -3,7 +3,7 @@
   "contracts": {
     "server": {
       "file": "server-v1.openapi.json",
-      "sha256": "9801087d40ac0e1f67871f9a1db850cbcce47614cceec0a7e8771cdf58385f53"
+      "sha256": "04aeed6f1a7414b37267d84141d036d00d2f8579ee28af37590bd4b92137e712"
     }
   }
 }

contracts/server-v1.openapi.json

@@ -50912,6 +50912,246 @@
         ]
       }
     },
+    "/api/v1/timeseries/realtime/playback/frames": {
+      "get": {
+        "operationId": "get_timeseries_realtime_playback_frames",
+        "parameters": [
+          {
+            "in": "query",
+            "name": "start_time",
+            "required": true,
+            "schema": {
+              "title": "Start Time",
+              "type": "string"
+            }
+          },
+          {
+            "in": "query",
+            "name": "model_version",
+            "required": true,
+            "schema": {
+              "maxLength": 24,
+              "minLength": 24,
+              "title": "Model Version",
+              "type": "string"
+            }
+          },
+          {
+            "in": "query",
+            "name": "frame_count",
+            "required": false,
+            "schema": {
+              "default": 4,
+              "maximum": 8,
+              "minimum": 1,
+              "title": "Frame Count",
+              "type": "integer"
+            }
+          },
+          {
+            "in": "query",
+            "name": "step_seconds",
+            "required": false,
+            "schema": {
+              "default": 900,
+              "maximum": 3600,
+              "minimum": 60,
+              "title": "Step Seconds",
+              "type": "integer"
+            }
+          },
+          {
+            "in": "header",
+            "name": "X-Project-Id",
+            "required": true,
+            "schema": {
+              "title": "X-Project-Id",
+              "type": "string"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "content": {
+              "application/octet-stream": {
+                "schema": {
+                  "format": "binary",
+                  "type": "string"
+                }
+              }
+            },
+            "description": "Versioned little-endian Float32 packet"
+          },
+          "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": "读取索引一致的 Float32 水力时间块",
+        "tags": [
+          "TimescaleDB - Hydraulic Playback"
+        ]
+      }
+    },
+    "/api/v1/timeseries/realtime/playback/geometry": {
+      "get": {
+        "operationId": "get_timeseries_realtime_playback_geometry",
+        "parameters": [
+          {
+            "in": "header",
+            "name": "X-Project-Id",
+            "required": true,
+            "schema": {
+              "title": "X-Project-Id",
+              "type": "string"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "content": {
+              "application/octet-stream": {
+                "schema": {
+                  "format": "binary",
+                  "type": "string"
+                }
+              }
+            },
+            "description": "Versioned little-endian Float32 packet"
+          },
+          "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 - Hydraulic Playback"
+        ]
+      }
+    },
     "/api/v1/timeseries/realtime/records": {
       "get": {
         "description": "查询指定时间点的实时属性值。query_time 必须显式带时区；允许传 UTC+8，服务端会先归一化为 UTC 再执行查询。",

docs/burst-response-workflow.md

@@ -2,7 +2,9 @@
 
 工单触发的 `workflow=burst_report` 使用 [爆管事件处理 Skill](../resources/work-order-skills/burst-event-handling/SKILL.md)。Server 按持久化来源选择流程，Agent 接收同一任务的核验上下文；SCADA 异常和历史回放使用 [SCADA Skill](../resources/work-order-skills/scada-anomaly/SKILL.md)。
 
-## 固定处理路线
+## 参考处理路线
+
+Agent围绕事件核实、影响评估和处置复核自主组织分析。图示说明常见依赖，可按证据补查、补算、调整方法或并行独立步骤；阶段名称与展示顺序不限制工具选择。
 
 ```mermaid
 flowchart TD
@@ -12,8 +14,9 @@ flowchart TD
     interim --> approval[立即发起工单草稿、人工审核派发]
     interim --> burst[爆管模拟与同刻在线压力差]
     burst --> coverage[保存并展示爆管后、处置前影响覆盖]
-    coverage --> isolation[分析隔离边界与阀门方案]
+    interim --> isolation[分析隔离边界与阀门方案]
-    isolation --> report[补充阶段报告与处置建议]
+    coverage --> report[补充阶段报告与处置建议]
+    isolation --> report
     approval --> feedback[接收实际处置与恢复反馈]
     feedback --> verify[复核影响与结案条件]
     verify --> close[Server 完成任务并解除告警]
@@ -24,7 +27,7 @@ flowchart TD
 2. **资产关联**：优先核对上报资产编号。坐标或地址通过地理编码、附近资产、属性和连接关系形成有依据的关联；位置歧义列为具体待确认项。
 3. **并行工单与影响估计**：定位到管道后保存 interim/concluded 阶段报告，随即提出现场核查／抢修 general 工单草稿；复用同一 action_id，人工审核、确认派发保持现有权限和版本校验。待审核或现场处理期间继续独立的影响模拟，不等待模型完成才发起工单。
 4. **隔离方案**：用 analyze_valve_isolation 分析关联资产，保留必关阀、可选阀、直接隔离与供水失源节点；已知失效阀门用于替代方案分析，现场可操作性列为执行条件。
-5. **影响说明**：先保留爆管后、处置前的压力下降覆盖步骤，再展示隔离方案的拓扑范围；结合可用SCADA、水力与用户关联说明影响。
+5. **影响说明**：分别保留事件降压与隔离失源范围，按证据就绪顺序展示；结合SCADA、水力与用户关联说明影响。缺口先评估可行补救，仍不可得再说明限制。
 6. **报告与处置**：定位阶段已发起现场工单，后续复用原动作；更新报告后按新增需要提出通知、保障或恢复等动作草稿；关联现场任务采用 general 工单。审批、派发和现场完成分别核对回执。
 7. **反馈复核**：已派发的关联工单保持 `waiting_work_order`，全部收到有效 `completed` 反馈后进入 `waiting_scada`。以现场反馈业务时间为 T，取严格晚于 T 的最近一个 SCADA 采样时刻进行复核；正常范围使用事件前15天同采样相位的正常包络。数据未到或仍异常时保持等待，后台工作进程定期复核；通过后持久保存恢复证据并接续 Agent。Agent 最终报告必须引用该证据，运行完成后才能完成任务并解除对应告警。现场资料缺项仅限定未核实细节，不新增本任务的关闭门槛。来源系统另有明确行政结案流程时，按其实际状态另记。
 
@@ -48,12 +51,14 @@ SCADA作为事件影响的辅助证据，按目标测点和事件／对照窗口
 
 ## 工具与入口约束
 
-工单事件从上报位置开展资产关联。Agent会话策略隐藏 run_burst_location，Server在绑定与后续解析时同步移除其调用能力；普通聊天和SCADA候选定位沿用各自策略。持久计算在执行前重新校验任务绑定。
+工单事件通常从上报位置关联资产；监测及模型证据需要时可自主调用run_burst_location补充定位，不按workflow隐藏分析工具。输入、角色、项目权限及任务绑定仍在执行前校验。
 
 历史 burst_report 任务携带 replay_context 时，Server分发scada-anomaly，时序查询保持 knowledge_cutoff 与 batch_sequence，demo能力提示按冻结窗口返回。
 
 业务阶段和工单、返回时间、复核回执保存在任务文档 `handling_context` JSON 中，不增加数据库列。外围样板更新工单时可提供带时区的 `occurred_at`，同时必须填写现场反馈；该时间写入活动记录，系统的 `updated_at` 仍是实际记录时间。同步器将现场反馈时间传给调查，避免历史演示误用电脑当前时间。
 
+恢复复核范围为项目配置的压力、出厂流量监测点与最新调查报告声明的 `scada_device_ids` 的并集；全网浏览查询不自动扩大本事件的复核范围。未声明测点的旧报告保留按已有查询证据补充测点的兼容路径。读取完工时间后最近一帧，并只取该帧时刻对应的事件前 15 天同期参考，不扫描完整 15 天曲线。旧等待任务保存的测点范围会按新选择版本自动重算；相关测点缺测或仍异常时继续等待，恢复通过后才唤醒 Agent。
+
 预约时间 `scheduled_at` 可在派单前的工单草稿和已创建工单的编辑表单中调整。实际完工反馈时间与预约安排分别记录；Agent 以实际反馈时间为恢复复核起点。新的有效现场反馈修订完工时间后，旧复核证据保留为历史记录，当前复核起点更新，重新读取严格晚于新反馈时间的最近一个 SCADA 采样时刻。只修改预约安排或其他工单信息，不改写已有现场反馈时间。
 
 `POST /operational-tasks/recovery/reconcile` 由现有 operational worker 每轮调用，按项目及任务创建边界处理等待复核的任务；SCADA 读取失败不阻断其他调查派发。复核直接读取项目时序库中完工反馈时间之后最近时刻的 SCADA，不依赖回放状态或推进回放时间。
@@ -74,11 +79,13 @@ 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模型压力单位的服务压力设定；在线模型方式、当前模型时态及在线压力单位推定来源一并保留。在线记录没有独立版本化单位元数据，不能把此估计称为严格历史同模型反事实或现场停水边界。
 
-携带 `replay_context` 的调查只开放启用影响覆盖的爆管情景，起算时刻不得超过 `knowledge_cutoff`。使用当前模型的独立假设情景，初始化及扩展模拟均传入空SCADA边界映射，禁止从实时SCADA回填冻结观测；同刻在线结果仅作为计算时读取的模型参考，标明非冻结基线。未来样本属于预测，不能作为已发生的观测。
+回放调查可自主选择在线补算、爆管、关阀、调压、污染传播及冲洗模型计算，起算时刻不超过knowledge_cutoff，使用当前模型和空SCADA边界映射。单独探索情景不强制生成覆盖；目标包含影响范围时应启用覆盖并核对结果。同刻在线结果为可变模型参考，未来样本属于预测，均不冒充冻结观测。
+
+基线缺失时，impact.baseline_recovery提供run_simulation补算建议。Agent可按目标用同刻短窗补算，读取成功结果并核对模型基线，再重试依赖分析；新的工具调用可创建新作业，无需改变破口参数来绕过旧结果。可选择其他有依据的方法，并说明剩余限制。
 
 完成持久计算后，Agent按回执 `data.presentation.command` 用证据编号发布覆盖，并保存三档计数、假设与缺口表。Server从任务所属的不可变证据解析几何，将完整展示命令写入事件。历史播放还原这份命令，前进到隔离范围和后退到事件影响均不重新查询在线模型，也不派单。工单审核、派发与反馈继续由通知和标准工单窗口记录；模型结果补入同一调查，不重复创建工单。
 

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-analysis-tools.md

@@ -33,7 +33,7 @@ Agent 原生工具统一使用 `mcp_tjwater_` 前缀。Server目录共67项：42
 `persisted_baseline=true` 表示在线基线已重算，`persisted_scheme=false` 表示没有创建方案。
 该工具的 MCP annotation 为 `readOnlyHint=false`、`destructiveHint=true`，因为重算会替换对应时刻的基线。
 它不会创建 `analysis.run_id`，不能将本次计算冒充已有爆管事件运行；调查调用仍可保存证据。
-计算读取现有模型和对应SCADA条件，不从摘要虚构观测，也不自动补齐数据或执行现场操作。
+普通计算读取现有模型及对应SCADA条件；回放计算使用当前模型和空SCADA映射，不读取实时观测。Agent可自主补算缺失模型基线，不从摘要虚构观测或执行现场操作。
 
 爆管定位 `run_burst_location` 默认使用 `data_source="monitoring"`。`pressure_scada_ids` 可省略，由后端从当前项目SCADA配置自动获取压力测点对应的模型节点；显式指定时仍限制为1至20个不同节点。正常对照时间窗 `scada_normal_start` / `scada_normal_end` 可成对省略，沿用后端默认的前一天同时间段；返回结果中的 `pressure_scada_ids` 和 `scada_window` 说明实际测点和时间窗。默认对照不等于已确认该时段正常。正常窗无有效压力数据或两窗没有共同有效测点时计算失败，不补造数据、不自动换窗。正常窗只有部分测点缺数时，沿用服务逻辑使用两窗共同有效测点，结合返回测点范围解释结果。
 
@@ -81,7 +81,7 @@ HTTP 预算由宿主从目录元数据读取，Server 默认 120 秒、Agent 默
 
 `analyze_valve_isolation` 和 `calculate_demand_allocation` 保持同步。队列消费者独立于Uvicorn的4个API进程，默认全部配置项目合计并发2项，并通过元数据库单实例锁防止重复启动。取消、超时和租约失效需要停止计算子进程组；成功结果按原运行授权保存证据，未完成计算阻止调查提前结案。
 
-在线重算逐个时刻写入基线，超时或中途失败可能已有部分时刻更新，不能把工具失败解释为整个窗口已回滚；不要自动重试整个窗口。
+在线重算逐个时刻写入基线，超时或中途失败可能已有部分时刻更新。先核对实际作业、采样与基线，再自主补算必要范围，不能假定整个窗口已回滚而盲目重提。
 
 队列接入需要在目标项目BizDB依次应用 `resources/sql/026_calculation_jobs.sql`、`resources/sql/027_calculation_job_lifecycle.sql` 并启动独立计算Worker。Server与Agent需配套发布，client模式还需重载原生连接。既有OAuth与项目权限保持不变；完整部署与回滚步骤见后台计算队列文档。
 

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`。精简目录仍超预算时沿用证据引用或缩页路径。
 

docs/water-pollution-response-workflow.md

@@ -24,7 +24,7 @@ flowchart TD
     decision -->|是| final[分区恢复建议与最终报告]
 ```
 
-图中连线表示业务与展示层次，先期保护、工单审核和独立计算可以并行。阀门操作均为方案，现场动作继续经过原审批与派发；阶段记录不能替代执行回执。首次展示保留处置前传播结果或明确的数据缺口，再展示隔离及清除方案。
+图示为参考路径，Agent可按风险、证据和实际依赖调整方法、顺序及并行分析。各类结果按就绪顺序展示并分别标明性质，现场动作仍须审批与派发，阶段记录不能替代执行回执。
 
 ## 侦测入口
 
@@ -46,13 +46,13 @@ flowchart TD
 
 阈值仅为模型展示条件，默认0表示模型正浓度，不是水质安全限值。沿用当前模型初始水质、已有源和反应条件；统计的是总模型浓度，不能称为本次源的独立贡献。源浓度、模式、业务起算时刻、预测时长及其假设必须有依据，不能把余氯下降或浊度换算成污染源浓度。
 
-回放同样允许该工具，要求启用影响覆盖且起算不晚于知识截止；预测时段可以越过截止，但后续时刻只是模型预测。与爆管情景一样，使用当前模型和空 SCADA 映射传入临时模型计算，不读取可变实时监测，不重建历史模型。仍禁止借此放开未授权观测或其他水力工具。
+回放允许自主选择传播及其他已开放模型计算，起算不晚于知识截止；预测时段可以越过截止，未来样本仅为模型预测。使用当前模型及空SCADA映射，不读取实时监测或重建历史模型。探索情景可不生成覆盖，影响范围目标应启用覆盖并核对结果。
 
-工具沿用计算队列和 owner/run 校验，返回终态后通过 `get_calculation_result` 读取并持久保存证据。`presentation_preview` 提供基于 `evidence_id` 的地图请求；用正式调查展示事件保存模型曲线和覆盖，历史播放还原已保存证据。审批、派发可在计算排队或运行期间进行；存在未终态计算时继续保留原运行，不允许发布 final 或成功结束调查，已有工单不受模拟失败影响。
+工具沿用计算队列和owner/run校验，终态后通过get_calculation_result读取并保存证据。data.presentation.command提供基于evidence_id的地图请求；正式事件保存曲线与覆盖，历史播放复用。审批、派发可与计算并行；未终态计算保留原运行，已有工单不受模拟失败影响。
 
 ## 展示、反馈与收尾
 
-`water-pollution-event` 1.2.0 将观测核验、候选定位、处置前传播、源头控制、用户隔离、清除冲洗、现场反馈、水质复测和恢复复核分开。每次正式展示保存 `purpose`；有证据的阶段通过 `narration.stage` 记录，新增 `source_control`、`flushing_plan`、`clearance_verification`。历史保存格式继续使用原事件 metadata JSON，不增加数据库列。前端显示阶段和本步目的，逐步回看保留独立画面；同阶段兼容图表仍可合并。
+`water-pollution-event` 1.2.1 将观测核验、候选定位、处置前传播、源头控制、用户隔离、清除冲洗、现场反馈、水质复测和恢复复核分开，作为可调整的参考路径。每次正式展示保存 `purpose`；有证据的阶段通过 `narration.stage` 记录，沿用 `source_control`、`flushing_plan`、`clearance_verification`。历史保存格式继续使用原事件 metadata JSON，不增加数据库列。前端显示阶段和本步目的，逐步回看保留独立画面；同阶段兼容图表仍可合并。
 
 源头控制与保护下游用户分别决策。用已核实的连接和流向判断水力下游，不能按地理方位判断；源头未核实不得声称污染输入已停止。清除方法按污染物及现场条件选择，不统一推荐煮沸、加药或排水。通知保留，工单按执行职责和阶段复用或必要拆分，不能同时要求同一阀门开和关。
 

resources/work-order-skills/appointment-follow-up/SKILL.md

@@ -2,7 +2,7 @@
 name: appointment-follow-up
 description: 核查关联预约工单的审批、提交、现场执行和反馈，分别判断预约履约与原事件闭环。
 metadata:
-  version: "2.0.4"
+  version: "2.0.5"
 ---
 
 # 预约任务跟进
@@ -13,7 +13,7 @@ metadata:
 
 读取当前任务、关联工单、已有报告、证据、动作和最新反馈，核对预约时间、地点、负责人、工作内容及回执要求；需要时追溯关联的投诉或爆管。新一轮优先检查新增事实和旧未决项，保留本流程名称和版本。
 
-围绕原工作目标完成必要核查：
+以下为参考核查路径，围绕原工作目标自主选择方法、调整顺序并补齐可查询或推导的资料。反馈缺项先核对已有来源和证据，不因首次缺数直接停止；新增现场事实仍以实际回执为准：
 
 1. 将要求与来源工单最新版本逐项对照，区分预约、派发、到场、完成和原事件发生时间，不用创建或更新时间替代现场时间，也不因预约已过推定已经执行或失约。
 2. 核对动作的审批、提交、执行状态和回执。`approved` 只表示允许提交，`submitted` 只表示已接收，`succeeded` 的建单动作只证明工单创建成功；这些状态均不证明采样、通知、抢修、冲洗或复测完成。`unknown` 由 Server 复核原执行请求，不重复下发。

resources/work-order-skills/burst-event-handling/SKILL.md

@@ -2,12 +2,12 @@
 name: burst-event-handling
 description: 根据爆管上报工单处理事件，完成资产关联、隔离方案、影响说明、处置建议与反馈复核。
 metadata:
-  version: "1.1.5"
+  version: "1.1.6"
 ---
 
 # 爆管事件处理
 
-适用于工单触发的 `workflow=burst_report`。以工单上报的事件位置和资产线索开展处置，依次完成下列步骤；每一步给出实际结果或具体待补项，已有证据可直接复用。SCADA 异常发现使用独立的 scada-anomaly 流程。
+适用于工单触发的 `workflow=burst_report`。目标是核实事件、评估影响并提出处置与复核依据。以下为参考流程，按证据和依赖自主选择、调整或并行步骤，复用已有成果，补查或补算可取得的前置资料。SCADA 异常发现使用独立的 scada-anomaly 流程。
 
 先按handling_context接续：等待阶段按第7节保存本轮阶段报告；verification_ready复用已有异常分析与资产、隔离证据，仅补反馈和恢复复核，不重跑初次调查步骤。真正新增的独立事实另行核查。
 
@@ -23,6 +23,8 @@ metadata:
 
 爆管候选管段用 locate_features 传 event_kind="burst" 定位。在地图展示上报位置和关联资产，标明关联依据及现场确认状态。存在多个合理对象时保留候选及选择理由；关键位置需要人工补充时，先完成独立工作，再用 ask_task_user 提问。
 
+上报位置与资产不足以区分候选时，可按现有监测和模型证据自主选择 run_burst_location 等补充分析；准备工具要求的输入，保留假设，不因工单来源限制分析方法。
+
 定位到管道后立即保存 interim/concluded 阶段报告，明确“候选管段待现场核验”，调用 propose_task_action 提出 kind=create_work_order、work_order_type=general 的现场核查／抢修草稿，引用定位证据、地点和管段；沿用已有 action_id，不重复建单。人工审核、派发按 Server 权限与版本完成。审批或现场处理等待期间，继续独立的爆管影响模拟和隔离分析，不等模拟完成才发起工单；模拟结果补入同一调查与后续报告，不能把模型计算成功写成已派单。
 
 ## 3. 爆管后、处置前的影响范围
@@ -31,7 +33,7 @@ metadata:
 
 工具在筛选展示节点前比较全网节点的起算同刻在线模拟压力，换算米水头，三档互斥：1≤下降<3、3≤下降<5、下降≥5。只用同刻有效值，不以代表节点、邻近时刻或缺失值代替；保留缺测、单位来源、模型时态、比较节点数及覆盖缺口。范围表示爆管后、关阀或抢修前的模型估计，单独于隔离失源范围。
 
-读取计算结果后按 data.presentation.command 的 evidence_id 用 publish_task_presentation 发布“爆管后、处置前的预计影响范围”，并发布三档节点计数、比较时刻和假设表；无地图几何时说明原因。证据、地图及表格进入正式任务事件，历史播放还原已保存结果，不重新查询在线基线。本步 narration.stage 用 event_impact。三级结果 unavailable 时单列原因和待补项；先保留此步骤，再展示隔离方案，不得用关阀覆盖替代。
+读取结果后按 data.presentation.command 发布预计降压范围和分档计数、比较时刻及假设，narration.stage 用 event_impact。基线缺失时可自主用 run_simulation 补算同刻数据；等待成功并读取结果后重试依赖计算，不靠扩大破口面积修复基线缺失。其他失败按原因选择补查、修正参数或替代方法，补救后仍不可用才列具体限制。事件降压与关阀范围分别保存，展示按结果就绪顺序发布；历史播放复用保存证据。
 
 ## 4. 隔离方案
 

resources/work-order-skills/burst-report/SKILL.md

@@ -2,14 +2,14 @@
 name: burst-report
 description: 爆管业务流程的历史入口索引，运行时按事件来源选择独立 Skill。
 metadata:
-  version: "2.0.11"
+  version: "2.0.12"
 ---
 
 # 爆管业务流程入口
 
 Server 根据绑定任务的 workflow 和 replay_context 分发当前流程：
 
-- 工单触发的 burst_report：使用 burst-event-handling，依次完成接收核对、地点与资产关联、隔离方案、影响说明、处置建议和反馈复核。
+- 工单触发的 burst_report：使用 burst-event-handling，围绕事件核实、资产关联、影响评估、处置建议和反馈复核自主组织分析。
 - SCADA 异常及带 replay_context 的历史回放任务：使用 scada-anomaly，完成冻结数据核验、诊断、候选定位与隔离影响分析。
 
 通过 get_task_context 的 section="skill" 读取实际流程，保留返回的 name、version、content_sha256。工具范围由绑定任务的服务端策略及 Agent 会话策略共同确定。

resources/work-order-skills/scada-anomaly/SKILL.md

@@ -2,12 +2,12 @@
 name: scada-anomaly
 description: 从 SCADA 指标异常开展数据核验、爆管诊断、候选定位与影响分析。
 metadata:
-  version: "1.0.20"
+  version: "1.0.21"
 ---
 
 # SCADA 数据异常调查
 
-读取任务与 replay_context，核对设备、单位、异常时刻、knowledge_cutoff 和 batch_sequence。使用本轮已提交的 SCADA 观测开展调查，同一持续事件沿用已有调查。证据来源字段用于记录数据取得方式，不构成停止分析的条件。
+目标是核验异常、判断原因和影响并形成处置依据。下文为参考流程，可按证据和依赖自主调整方法与顺序，补查、补算前置资料。核对任务、设备、单位、异常时刻、knowledge_cutoff及batch_sequence，使用本轮已提交观测，同一持续事件沿用原调查。证据来源字段用于记录数据取得方式，不构成停止分析的条件。
 
 先按handling_context接续：待审批/工单/监测且无独立工作时，保存本轮interim/concluded及action_id后退出，不轮询。verification_ready复用原异常证据，只补恢复复核；初次分析按下文。
 
@@ -31,9 +31,9 @@ metadata:
 
 取得候选即展示管道，发 interim/concluded 报告及现场工单草稿；复用action_id，待审批继续模拟。
 
-run_burst_simulation设include_impact_coverage=true，用定位管道、异常时刻及有依据的破口面积cm²。全网同刻在线压力差按米水头分档1≤下降<3、3≤下降<5、下降≥5。按data.presentation发布地图及计数，narration.stage=event_impact；不可用时单列原因，禁用关阀范围替代。回放仅用当前模型、禁实时SCADA；在线参考非冻结观测，历史展示复用保存证据。
+run_burst_simulation设include_impact_coverage=true，用候选管道、异常时刻及有依据的破口面积cm²，按米水头分档1≤下降<3、3≤下降<5、下降≥5。基线缺失可自主用run_simulation补算同刻数据，读取成功结果后重试。其他缺口先按原因补查、补算或修正参数，仍不可用才记录限制。按data.presentation发布地图及计数，stage=event_impact；禁用关阀范围替代。回放模型计算不读实时SCADA，非冻结观测；历史展示复用保存证据。
 
-再按授权用 analyze_valve_isolation 分析隔离边界，保留必关阀、可选阀、affected_nodes、supply_affected_nodes 及计数；用 generate_node_coverage 展示关阀覆盖，stage=isolation_plan。候选与拓扑建议标注为待现场核验方案。爆管候选用locate_features传event_kind=burst。报告前核对定位结果，失败列明错误与缺口。
+按需用analyze_valve_isolation分析隔离边界，保留必关阀、可选阀、affected_nodes、supply_affected_nodes及计数；generate_node_coverage展示关阀覆盖，stage=isolation_plan；analyze_valve_closure可补水力影响。候选与建议待现场核验，展示按就绪顺序。候选用locate_features传event_kind=burst。
 
 ## 对比展示与影响
 
@@ -43,22 +43,22 @@ run_burst_simulation设include_impact_coverage=true，用定位管道、异常
 
 定位输入具备后继续定位。压力、流量按设备另发七天同期复合图，不占用或替代全量当前总览。chart_type=time_comparison：x_data是UTC+8的HH:mm或HH:mm:ss共享槽，不拼接历史日期；当前[t-3小时,t]用series.type=line，前7天每天t-30分钟、t-15分钟、t用scatter。15分钟数据13槽，历史七组仅末3槽有值。每组sample_times与data等长，保留真实带时区日期；有效值与对应槽的本地时刻一致，缺测值和时间均null。当前曲线共用窗口，历史每组取此前某天对应窗口，跨午夜顺序排列；普通图用line/bar，坐标轴标单位。七天按t的本地日期回退，不含当天。按实际周期保留缺口，按需翻页并报告参考点数；基线未冻结、读取失败或缺测时说明并保留当前趋势，不退化为前一天或相邻两点。
 
-影响说明列出异常前参考、异常期间和已有恢复观测中的压力、流量或水质变化、持续时间、涉及测点及已核实资产。可比对象、单位、时段与运行条件下给出变化量，分别标明实际观测、模型预测和拓扑隔离。停水范围与漏量引用相应拓扑、水力、流量平衡或现场依据；恢复观测、用户关联或现场证据缺失时列为具体待核实项。
+影响说明比较异常前、异常期间及已有恢复观测，列压力、流量或水质变化、持续时间、测点与资产。同对象、单位、时段及条件计算变化量，区分观测、模型和拓扑结果；停水及漏量保留依据，缺恢复、用户或现场资料时明确限制。
 
 ## 范围与报告
 
-本轮时序分析保持任务冻结窗口，后续播放形成的观测由相应轮次处理。资产属性、空间查询和拓扑隔离使用项目当前模型，注明模型时态；水力工具按已开放的冻结输入能力调用。定位调用执行权限、设备与冻结观测校验；基线或覆盖缺口按实际返回记录。
+观测保持本轮冻结窗口，后续观测由相应轮次处理。资产、空间与拓扑用当前模型；水力计算标明模型来源，定位保留权限、设备及冻结观测校验。
 
-报告先写判断与关键依据，摘要建议 150 字、最多 300 字；多测点数值入 Table，保留单位、方向与缺测。区分各阶段测点数，不相加不同影响集合。现场事项按回执列待办。现场工单与设备操作遵守服务端动作权限，恢复复核通过后闭环。
+报告先写判断与依据，摘要建议150字、最多300字；多测点数值入Table，保留单位、方向和缺测，不相加不同影响集合。现场事项按回执列待办。现场工单与设备操作遵守服务端动作权限，恢复复核通过后闭环。
 
 现场工单用propose_task_action（kind=create_work_order、work_order_type=general），注明对象、条件、回执及回放来源；有权限者确认派发，通知保留，报告完成不等于现场完成；回放禁设备命令。
 
 ## 完工后的恢复复核
 
-关联工单completed后，以handling_context.returned_at为实际完工业务时间，不用创建、预约或电脑当前时间代替。Server读取严格晚于该时刻的最近一帧及事件前15天同相位对照，独立保存恢复证据。回放来源工单同样适用，不推进回放，原knowledge_cutoff与batch_sequence不变。
+工单completed后，以handling_context.returned_at为实际完工时间；Server取严格晚于该时刻的最近一帧及事件前15天同相位对照，保存恢复证据。回放同样适用，不推进回放或修改原冻结边界。
 
 waiting_work_order等待完工；waiting_scada按last_check说明尚无后续采样、数据缺项或仍异常，等待监测复核。verification_ready时用get_task_evidence读取verification_evidence_id完整快照；工单completed且恢复证据status=passed时，发布final/concluded恢复闭环报告，同时引用原异常全量证据与当前恢复证据。Server完成运行后关闭任务并解除关联告警，phase=completed表示已闭环。阀位、漏点照片、现场复测数值等未提供时仅记为资料缺项，不新增现场核验、人工确认或更多采样的闭环条件。
 
-恢复报告先写现场反馈与完工时间、采样时刻和结果，表格列全部复核测点及基线；边界说明单帧与模型限制，日常监测列建议。不重列旧定位参数或执行字段；更新原图中已解除的等待标记，保留路线与失败记录。
+恢复报告列反馈、完工与采样时间、全部复核测点及基线，说明单帧与模型边界，日常监测列建议。更新已解除的等待标记，保留路线与失败记录。
 
 恢复观测由Server独立保存，原冻结上限不阻断此证据的使用；query_scada_readings仍遵守冻结窗口。完工时间修订后使用新证据编号，旧证据仅作历史。确有独立工作时再提议新动作。

resources/work-order-skills/water-pollution-event/SKILL.md

@@ -2,13 +2,15 @@
 name: water-pollution-event
 description: 水污染侦测后的观测核验、候选溯源、源头控制、用户隔离、清除冲洗与水质复测。
 metadata:
-  version: "1.2.0"
+  version: "1.2.1"
 ---
 
 # 水污染侦测事件调查
 
 适用于 workflow=water_pollution，与 water_quality_complaint 投诉分开，不依赖爆管评分、漏量或压力流量异常。侦测信号、候选污染源、污染确诊和恢复供水分别说明。
 
+目标是核验风险、来源、影响及处置恢复依据。以下为参考流程，可自主调整方法、顺序，补查参数并比较明确假设；缺口先补救，保护措施及时提出。
+
 读取任务、信号、时间、handling_context 和已有成果。回放遵守 knowledge_cutoff、batch_sequence；阅读时间 Asia/Shanghai，查询与 sample_times 带时区。后续轮次复用成果，优先新增事实。
 
 ## 1. 核验观测与先期保护
@@ -25,15 +27,15 @@ query_scada_readings 省略 device_ids，读取异常前三小时全量 SCADA，
 
 ## 3. 处置前污染影响
 
-run_contaminant_simulation 设 source=located_node、include_impact_coverage=true，采用事件起算时刻、明确预测时长和需画曲线的 output_node_ids。concentration 为 mg/L，须有来源或说明情景假设，不能把浊度、余氯下降或感官描述直接换算成污染源浓度。缺依据时列具体参数缺口，继续工单和独立分析。
+run_contaminant_simulation 设 source=located_node、include_impact_coverage=true，采用事件起算时刻、明确预测时长和需画曲线的 output_node_ids。concentration 为 mg/L，须有来源或说明情景假设，不能把浊度、余氯下降或感官描述直接换算成污染源浓度。可自主补查来源、比较明确假设的浓度或时长；不可推断的现场事实才列具体缺口，继续工单和独立分析。
 
 impact_threshold_mg_l 默认0、严格大于，仅供展示，非饮用水安全限值。全网整个时段汇总累计节点、首个超阈值采样时刻、峰值与覆盖；沿用初始水质和反应条件，非源独立贡献，范围外不等于安全。
 
 queued/running 时继续独立工作，get_calculation_job 跟踪、get_calculation_result 读终态。未完计算不能 final，由宿主等待接续；待审批不阻止分析，模拟失败不撤回工单。
 
-按 presentation_preview.evidence_id 发布“水污染传播预计影响范围（处置前情景）”、参数与计数表及浓度曲线；模型曲线不填观测专用 chart.scada。保存正式事件供历史复用；空范围、缺测、单位及几何限制单列。先保留污染传播步骤，再展示隔离范围，不以隔离覆盖替代传播范围。
+按返回的data.presentation.command发布“水污染传播预计影响范围（处置前情景）”、参数与计数表及浓度曲线；模型曲线不填观测专用chart.scada。空范围、缺测、单位及几何问题先核对可行补救，再列剩余限制；传播与隔离分别保存，不以隔离覆盖替代传播范围。
 
-回放允许传播及冲洗情景，起算不晚于知识截止，使用当前模型且不读取实时 SCADA；未来样本仅预测，非历史重建。
+回放允许在线基线补算及传播、关阀、调压、冲洗情景，起算不晚于知识截止，使用当前模型且不读取实时SCADA；未来样本仅预测，非历史重建。
 
 ## 4. 源头控制与用户隔离
 
@@ -47,7 +49,7 @@ analyze_valve_isolation 设 accident_element=[located_node]、accident_element_t
 
 先说明污染物性质、处理依据和未知项，再提出排空、冲洗、消毒或其他适用处理；不得默认任何污染都能冲洗清除。排放去向及污染水收集处理须有依据，不默认排入雨水、河道或普通排水口。
 
-核对清洁水源、排放点、流量及许可条件后用 analyze_flushing。每阶段保存目的、前置条件、清洁进水依据、排放去向、完整开关阀清单、顺序、流量、时长、预计水量、监测和待办。隔离和冲洗不能共用固定关阀清单，进水阀打开不等于用户复供；缺参数仅停依赖分支。
+核对清洁水源、排放点、流量及许可条件后用analyze_flushing。阶段记录目的、条件、进排水依据、完整阀态、顺序、流量、时长、水量、监测及待办。隔离和冲洗按各自目标选阀，进水阀打开不等于复供；缺参数先补查。
 
 opened_valves 明确开阀、closed_valves 指定关闭阀门，两者互斥。每次是独立当前模型情景，未列阀门沿用模型状态，不继承上次模拟；重列本阶段必要阀态，不能称为连续污染清除。flushing_plan 保存方案及估计排放量，仅证明水力结果，不验证清洁进水、排放合规或现场执行。尚未耦合停止输入后的水质衰减，压力、流量或模型零浓度不能替代复测。
 
@@ -57,9 +59,9 @@ opened_valves 明确开阀、closed_valves 指定关闭阀门，两者互斥。
 
 水质 chart.scada 填 metric=quality、unit、role、完整三小时 window_start/window_end，总览覆盖全部目标。逐设备 time_comparison 用当前 line、七天每天末三真实槽 scatter；series 带 device_id、day_offset、等长 sample_times，缺测 null。不同指标或单位分图，超限分段，不标成流量。
 
-展示依次为观测、候选、传播、源头控制、隔离、冲洗、反馈、复测、恢复；计算可并行，首次隔离展示须在传播结果或缺口说明之后，通知和工单不等待展示。发布填写具体 purpose；有证据时带 narration（stage、summary、evidence_refs）。沿用原阶段，源头控制用 source_control、冲洗用 flushing_plan、水质复测用 clearance_verification。阶段不等于批准或已执行，不为凑阶段造证据或空操作。
+展示随证据就绪发布，按观测、候选、传播、源头控制、隔离、冲洗、反馈、复测和恢复标明性质与依赖，无需等待其他独立分析或固定播放顺序。通知和工单不等待展示。发布填写purpose及有依据的narration（stage、summary、evidence_refs）；源头控制用source_control、冲洗用flushing_plan、水质复测用clearance_verification。阶段不等于批准或已执行，不为凑阶段造证据或空操作。
 
-报告依次写事件概况、观测、候选与传播、源头控制、用户隔离、清除冲洗、反馈复测及恢复条件；各项列结论、对象、预计或实际、证据和待办。操作表保留阶段、阀态、进排水、条件、反馈及复测依据。引用本轮全量 SCADA 和已取得的分析证据；权限拒绝不写成无数据。各阶段独立保存，回放不合并吞步骤，不派单或操作设备。
+报告按风险与结论组织，覆盖观测、候选、传播、源头控制、隔离、清除及复测恢复，区分预计与实际。操作表保留阀态、进排水、条件和反馈依据。引用本轮全量SCADA与分析证据；权限拒绝不写成无数据。阶段独立保存，历史播放不派单或操作设备。
 
 ## 7. 工单反馈与复测恢复
 

resources/work-order-skills/water-quality-complaint/SKILL.md

@@ -2,7 +2,7 @@
 name: water-quality-complaint
 description: 调查供水水质投诉，核查投诉与监测覆盖，评估候选原因、影响和采样处置建议。
 metadata:
-  version: "2.0.4"
+  version: "2.0.5"
 ---
 
 # 水质投诉处置
@@ -15,7 +15,7 @@ metadata:
 
 ## 风险与工程分析
 
-根据当前证据选择必要步骤：
+以下为参考核查路径，根据当前证据选择必要步骤，自主调整顺序、查询范围、工具组合及模型情景。可通过补查、推导或有说明的假设解决的缺口先处理，确需现场事实或人工决策时再提问：
 
 1. 核实投诉位置及相近时段的同类投诉，用 `query_peripheral_data` 按实际时间、位置和来源标识核对外围工单或观察。目录首屏不能证明没有其他投诉；未知点位先查目录，不猜编号或把演示点当真实资产。
 2. 解析未知地址，查询附近资产、属性和连接，区分上游、管网、末梢、二次供水和用户端的实际覆盖。地点明确后展示投诉位置，取得关联资产后展示正在核查的管段或节点。拓扑连接不等于实际流向，方向判断需要同窗模型或监测依据。

tests/api/test_hydraulic_playback_endpoints.py

@@ -0,0 +1,86 @@
+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()
+
+
+def test_contract_advertises_binary_packets_instead_of_json():
+    schema = build_test_app(playback.router, "/api/v1").openapi()
+    for suffix in ("geometry", "frames"):
+        content = schema["paths"][f"/api/v1/timeseries/realtime/playback/{suffix}"]["get"]["responses"]["200"]["content"]
+        assert content == {"application/octet-stream": {"schema": {"type": "string", "format": "binary"}}}

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),
@@ -99,6 +100,27 @@ def test_unknown_units_do_not_draw_an_impact_area(inputs):
     inputs.assert_not_called()
 
 
+def test_missing_online_baseline_offers_recalculation_then_coverage_can_be_retried(inputs):
+    inputs.side_effect = None
+    inputs.return_value = {}
+    native = simulation([2, 4, 6])
+    missing = service.estimate_burst_impact("own", START, native)
+    assert missing["impact"]["reason"] == "no_exact_comparable_online_snapshot"
+    recovery = missing["impact"]["baseline_recovery"]
+    assert recovery["tool"] == "run_simulation"
+    assert recovery["parameters"] == {
+        "start_time": START.isoformat(), "duration": 60, "output_node_ids": ["J0"],
+    }
+    assert "map_payload" not in missing
+
+    inputs.return_value = {
+        f"J{i}": [{"time": START.isoformat(), "value": 30}] for i in range(3)
+    }
+    retried = service.estimate_burst_impact("own", START, native)
+    assert retried["impact"]["status"] == "ready"
+    assert len(retried["map_payload"]["features"]) == 3
+
+
 def test_full_native_network_compared_before_selected_series_filter(monkeypatch, inputs):
     from app.mcp import scenario_service
     from app.services import simulation_scenarios

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_mcp_query.py

@@ -373,7 +373,7 @@ def test_replay_binding_denies_future_and_allows_device_metadata(monkeypatch):
     asyncio.run(query_service.authorize_query(SimpleNamespace(project=object()), bound, TOOLS["list_scada_devices"], SimpleNamespace()))
 
 
-def test_replay_burst_scenario_requires_coverage_permission_and_known_start(monkeypatch):
+def test_replay_burst_scenario_allows_exploration_and_checks_coverage_permission(monkeypatch):
     bound = binding(task=True)
     bound.replay_context = {"knowledge_cutoff": "2026-10-08T01:00:00Z"}
     permissions = PERMISSIONS | {"simulation.run", "scada.view", "simulation.view"}
@@ -386,11 +386,12 @@ def test_replay_burst_scenario_requires_coverage_permission_and_known_start(monk
         "include_impact_coverage": True,
     })
     asyncio.run(query_service.authorize_query(scope, bound, TOOLS["run_burst_simulation"], args))
-    for changed, status in (({"include_impact_coverage": False}, 422),
+    asyncio.run(query_service.authorize_query(scope, bound, TOOLS["run_burst_simulation"],
-                            ({"start_time": datetime(2026, 10, 8, 2, tzinfo=timezone.utc)}, 422)):
+                                              args.model_copy(update={"include_impact_coverage": False})))
-        with pytest.raises(HTTPException) as error:
+    with pytest.raises(HTTPException) as error:
-            asyncio.run(query_service.authorize_query(scope, bound, TOOLS["run_burst_simulation"], args.model_copy(update=changed)))
+        asyncio.run(query_service.authorize_query(scope, bound, TOOLS["run_burst_simulation"],
-        assert error.value.status_code == status
+                                                  args.model_copy(update={"start_time": datetime(2026, 10, 8, 2, tzinfo=timezone.utc)})))
+    assert error.value.status_code == 422
     bound.permission_cap = sorted(permissions - {"simulation.view"})
     with pytest.raises(HTTPException) as error:
         asyncio.run(query_service.authorize_query(scope, bound, TOOLS["run_burst_simulation"], args))
@@ -881,7 +882,7 @@ def test_old_task_run_and_reader_parent_are_rejected(monkeypatch):
 
 @pytest.mark.parametrize("purpose", ["task_execution", "task_collaboration"])
 @pytest.mark.parametrize("replay", [False, True])
-def test_work_order_task_excludes_location_but_scada_replay_keeps_it(monkeypatch, purpose, replay):
+def test_work_order_source_does_not_remove_authorized_analysis_tools(monkeypatch, purpose, replay):
     monkeypatch.setattr(contexts, "permissions_for_context", lambda _: PERMISSIONS)
     value = binding(task=True)
     value.purpose = purpose
@@ -899,13 +900,9 @@ def test_work_order_task_excludes_location_but_scada_replay_keeps_it(monkeypatch
     monkeypatch.setattr(contexts, "task_repository", lambda _: Tasks())
     scope = SimpleNamespace(project=object(), connection=Connection())
     asyncio.run(contexts.verify_task(scope, value))
-    assert ("run_burst_location" in value.tool_cap) is replay
+    assert "run_burst_location" in value.tool_cap
     assert "analyze_valve_isolation" in value.tool_cap
     assert "analyze_valve_closure" in value.tool_cap
-    if not replay:
-        with pytest.raises(HTTPException) as caught:
-            asyncio.run(query_service.authorize_query(scope, value, TOOLS["run_burst_location"], SimpleNamespace()))
-        assert caught.value.status_code == 403
 
 
 def test_jwt_signature_issuer_audience_lifetime_and_key_cache():

tests/unit/test_mcp_work_order_skills.py

@@ -46,7 +46,7 @@ def test_work_order_handling_and_legacy_scada_replay_receive_separate_skills():
     assert replay["workflow"] == "burst_report"
     assert "影响说明" in handling["instructions"]
     assert "supply_affected_nodes" in handling["instructions"]
-    assert "run_burst_location" not in handling["instructions"]
+    assert "run_burst_location" in handling["instructions"]
     assert "run_burst_location" in replay["instructions"]
 
 
@@ -66,14 +66,14 @@ def test_pollution_instructions_offer_work_order_before_simulation_and_reuse_the
     assert text.index("propose_task_action") < text.index("run_contaminant_simulation")
     assert "include_impact_coverage=true" in text and "不等模拟完成才发起工单" in text
     assert "同一 task_id" in text and "不覆盖已审核工单或绕过版本校验" in text
-    assert "模型曲线不填观测专用 chart.scada" in text
+    assert "模型曲线不填观测专用chart.scada" in text
     assert "不能把浊度、余氯下降或感官描述直接换算成污染源浓度" in text
 
 
 def test_pollution_workflow_distinguishes_protection_flushing_and_clearance():
     skill = read_work_order_skill("water_pollution", include_content=True)
     text = skill["instructions"]
-    assert skill["version"] == "1.2.0"
+    assert skill["version"] == "1.2.1"
     assert text.index("## 4. 源头控制与用户隔离") < text.index("## 5. 污染物清除与分步冲洗")
     assert "opened_valves" in text and "closed_valves" in text
     assert "不继承上次模拟" in text and "排放去向" in text

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)]

tests/unit/test_online_simulation.py

@@ -46,6 +46,17 @@ def test_online_simulation_rejects_invalid_step_before_calculating(monkeypatch):
     run.assert_not_called()
 
 
+def test_online_baseline_preserves_fractional_start_for_exact_impact_comparison(monkeypatch):
+    start = START.replace(microsecond=123456)
+    monkeypatch.setattr(simulation, "get_time", lambda _: {"HYDRAULIC TIMESTEP": "0:15:00"})
+    monkeypatch.setattr(simulation, "query_corresponding_element_id_and_query_id", lambda _: object())
+    run = Mock(return_value={"node_results": [{"node": "J1", "result": [{"pressure": 30}]}]})
+    monkeypatch.setattr(simulation, "run_simulation", run)
+    result = simulation.run_online_simulation("online", start, 60, output_node_ids=["J1"])
+    assert run.call_args.kwargs["modify_pattern_start_time"] == start.isoformat()
+    assert result["times"]["sample_times"] == [start.isoformat()]
+
+
 def test_mcp_online_simulation_does_not_copy_a_database(monkeypatch):
     from app.mcp.scenario_service import calculate_scenario
 

tests/unit/test_recovery_device_scope.py

@@ -0,0 +1,48 @@
+from copy import deepcopy
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from app.domain.schemas.project_settings import ProjectSettings
+from app.services.operational_recovery import reconcile_recoveries
+from app.services.operational_tasks import OperationalTaskService
+from tests.support.operational_task_repository import TaskRepository, _active_run, _context, _task_document
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("cached", [False, True])
+async def test_recovery_uses_report_devices_instead_of_all_browsed_network_devices(monkeypatch, cached):
+    now = datetime(2026, 7, 16, 6, tzinfo=timezone.utc)
+    incident = now - timedelta(hours=6)
+    returned = now - timedelta(minutes=20)
+    task = _task_document(_context(), now, lease=now + timedelta(minutes=1))
+    task.update(workflow="scada_anomaly", status="waiting_human", lifecycle_status="waiting_human",
+                current_report_revision=1, trigger={"kind": "event", "occurred_at": incident.isoformat(), "detail": "异常"},
+                handling_context={"phase": "waiting_scada", "returned_at": returned.isoformat()})
+    if cached:
+        task["handling_context"].update(device_ids=["P1", "unrelated"], interval_seconds=900)
+    repo = TaskRepository(task, _active_run(task, now))
+    repo.get_latest_report = AsyncMock(return_value={"revision": 1, "report": {"scada_device_ids": ["P1"]}})
+    repo.list_evidence = AsyncMock(return_value=[{"data_snapshot": {"source": "query_scada_readings",
+        "data": {"data": [{"device_id": "P1"}, {"device_id": "unrelated"}]}}}])
+    repo.recovery_candidates = AsyncMock(side_effect=lambda *_: [deepcopy(repo.task)]
+        if repo.task["handling_context"]["phase"] == "waiting_scada" else [])
+    service = OperationalTaskService(repo, _context(), set(), clock=lambda: now)
+    at = returned + timedelta(minutes=5)
+    rows = [{"time": at - timedelta(days=day), "device_id": "P1", "monitored_value": 100.}
+            for day in range(1, 16)] + [{"time": at, "device_id": "P1", "monitored_value": 100.}]
+    reader = AsyncMock(side_effect=lambda _, devices, *args, **kwargs: rows if devices == ["P1"] else [{}] * 100_001)
+    monkeypatch.setattr("app.infra.db.timescaledb.repositories.scada.ScadaRepository.get_recovery_scada_readings", reader)
+    settings = ProjectSettings(scada_monitoring_interval_seconds=900, scada_monitoring={"pressure_device_ids": ["P1"]})
+    assert await reconcile_recoveries(service, SimpleNamespace(transaction=repo.transaction), settings) == 1
+    assert repo.task["handling_context"]["device_ids"] == ["P1"]
+    assert repo.task["handling_context"]["phase"] == "verification_ready"
+    assert repo.run["status"] == "queued"
+    assert await reconcile_recoveries(service, SimpleNamespace(transaction=repo.transaction), settings) == 0

tests/unit/test_replay_online_simulation.py

@@ -0,0 +1,171 @@
+import asyncio
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+from uuid import uuid4
+
+import pytest
+from fastapi import HTTPException
+
+from app.mcp import query_service
+from app.mcp.analysis_models import ScenarioInput
+from app.mcp.catalog import TOOLS
+from app.services import simulation
+
+
+def parameters(**changes):
+    return ScenarioInput.model_validate({
+        "context_handle": "a" * 43, "start_time": "2026-07-16T08:00:00+08:00",
+        "duration": 60, "output_node_ids": ["J1"], **changes,
+    })
+
+
+def context(monkeypatch):
+    permissions = {"simulation.run", "simulation.view", "scada.view", "webgis.view", "optimization.run"}
+    monkeypatch.setattr(query_service, "permissions_for_context", lambda _: permissions)
+
+    @asynccontextmanager
+    async def routing():
+        yield
+
+    scope = SimpleNamespace(
+        connection=SimpleNamespace(commit=AsyncMock()), simulation_routing=routing,
+        project=SimpleNamespace(project_code="owned", project_role="member"),
+        user=SimpleNamespace(username="actor"),
+    )
+    binding = SimpleNamespace(
+        tool_cap=["run_simulation"], permission_cap=sorted(permissions),
+        purpose="task_execution", task_id=uuid4(), task_scope_cap=[],
+        replay_context={"knowledge_cutoff": "2026-07-16T08:00:00+08:00"},
+    )
+    return scope, binding
+
+
+def test_replay_coordinator_can_rebuild_online_baseline_without_live_scada(monkeypatch):
+    scope, binding = context(monkeypatch)
+    monkeypatch.setattr(simulation, "get_time", lambda _: {"HYDRAULIC TIMESTEP": "0:15:00"})
+    live_mappings = Mock(side_effect=AssertionError("Replay must not load live SCADA mappings"))
+    monkeypatch.setattr(simulation, "query_corresponding_element_id_and_query_id", live_mappings)
+    native = Mock(return_value={
+        "units": {"pressure": "MTR"},
+        "node_results": [{"node": "J1", "result": [{"pressure": 30}]}],
+    })
+    monkeypatch.setattr(simulation, "run_simulation", native)
+
+    result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["run_simulation"], parameters()))
+
+    assert result.origin == "synthetic"
+    assert any("不读取实时SCADA" in gap for gap in result.data_gaps)
+    assert result.data["result"]["persisted_baseline"] is True
+    assert result.data["result"]["node_results"][0]["result"] == [{"pressure": 30}]
+    native.assert_called_once()
+    call = native.call_args.kwargs
+    assert call["name"] == "owned" and call["simulation_type"] == "realtime"
+    assert call["modify_pattern_start_time"] == "2026-07-16T08:00:00+08:00"
+    assert all(not values for values in vars(call["scada_mappings"]).values())
+    live_mappings.assert_not_called()
+
+
+@pytest.mark.parametrize("problem", ["future_start", "reader", "permission", "tool_cap"])
+def test_replay_online_baseline_preserves_existing_authorization(monkeypatch, problem):
+    scope, binding = context(monkeypatch)
+    args = parameters()
+    if problem == "future_start":
+        args = parameters(start_time="2026-07-16T08:15:00+08:00")
+    if problem == "reader":
+        binding.purpose = "task_reader"
+    if problem == "permission":
+        binding.permission_cap.remove("simulation.run")
+    if problem == "tool_cap":
+        binding.tool_cap.clear()
+    with pytest.raises(HTTPException) as caught:
+        asyncio.run(query_service.authorize_query(scope, binding, TOOLS["run_simulation"], args))
+    assert caught.value.status_code == (422 if problem == "future_start" else 403)
+
+
+@pytest.mark.parametrize("name", ["analyze_valve_closure", "analyze_pressure_regulation"])
+def test_replay_can_compare_hydraulic_alternatives_with_model_only_boundaries(monkeypatch, name):
+    from app.infra.db.postgresql.scada import ScadaElementMappings
+    from app.services import simulation_scenarios as scenarios
+
+    scope, binding = context(monkeypatch)
+    binding.tool_cap = [name]
+    captures = []
+    monkeypatch.setattr(scenarios, "get_option", lambda _: {"UNITS": "LPS"})
+    monkeypatch.setattr(scenarios, "set_option", lambda *_: None)
+    monkeypatch.setattr(simulation, "query_corresponding_element_id_and_query_id",
+                        lambda _: ScadaElementMappings({}, {}, {"P1": "device"}, {}, {}, {}, {}))
+    monkeypatch.setattr(simulation, "run_simulation", lambda **kw: captures.append(kw) or {
+        "node_results": [{"node": "J1", "result": [{"pressure": 30}]}],
+    })
+    native_name = "valve_close_analysis" if name == "analyze_valve_closure" else "pressure_regulation"
+    native = getattr(scenarios, native_name).__wrapped__
+    monkeypatch.setattr(scenarios, native_name,
+                        lambda **kw: native(**kw, _temporary_project="tjw_tmp_owned"))
+    args = TOOLS[name].model.model_validate({
+        **parameters().model_dump(),
+        **({"valves": ["V1"]} if name == "analyze_valve_closure" else {"pump_control": {"P1": [0]}}),
+    })
+    result = asyncio.run(query_service.execute_query(scope, binding, TOOLS[name], args))
+    assert result.origin == "synthetic"
+    assert result.data["result"]["persisted_baseline"] is False
+    assert len(captures) == 1
+    assert captures[0]["name"] == "tjw_tmp_owned" and captures[0]["persist_results"] is False
+    assert all(not values for values in vars(captures[0]["scada_mappings"]).values())
+    if name == "analyze_valve_closure":
+        assert captures[0]["modify_valve_opening"] == {"V1": 0.0}
+    else:
+        assert captures[0]["modify_fixed_pump_pattern"] == {"P1": [0]}
+
+
+@pytest.mark.parametrize("name", [name for name, tool in TOOLS.items()
+                                  if tool.semantics == "asset" and not tool.mutating])
+def test_replay_allows_authorized_asset_queries_for_input_discovery(monkeypatch, name):
+    scope, binding = context(monkeypatch)
+    binding.tool_cap = [name]
+    # Authorization needs only the schema kind; execution validates each tool's inputs.
+    asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], SimpleNamespace(kind="node")))
+    binding.permission_cap = []
+    with pytest.raises(HTTPException) as caught:
+        asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], SimpleNamespace(kind="node")))
+    assert caught.value.status_code == 403
+
+
+@pytest.mark.parametrize("name,fields", [
+    ("query_model_baseline_series", {"element_type": "junction", "element_ids": ["J1"],
+                                    "start": "2026-07-16T07:45:00+08:00", "end": "2026-07-16T08:00:00+08:00"}),
+    ("get_model_baseline_at_time", {"element_type": "junction", "element_id": "J1", "time": "2026-07-16T08:00:00+08:00"}),
+    ("query_model_baseline_field", {"element_type": "junction", "field": "pressure", "time": "2026-07-16T08:00:00+08:00"}),
+])
+def test_replay_can_verify_baseline_without_reading_future_values(monkeypatch, name, fields):
+    scope, binding = context(monkeypatch)
+    binding.tool_cap = [name]
+    args = TOOLS[name].model.model_validate({"context_handle": "a" * 43, **fields})
+    asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], args))
+    future = parameters(start_time="2026-07-16T08:15:00+08:00").start_time
+    with pytest.raises(HTTPException) as caught:
+        asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], args.model_copy(
+            update={"end" if "end" in fields else "time": future},
+        )))
+    assert caught.value.status_code == 422
+
+
+@pytest.mark.parametrize("name", ["calculate_demand_allocation", "optimize_sensor_placement"])
+def test_replay_can_choose_current_model_design_calculations(monkeypatch, name):
+    scope, binding = context(monkeypatch)
+    binding.tool_cap = [name]
+    asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], SimpleNamespace()))
+    binding.purpose = "task_reader"
+    with pytest.raises(HTTPException) as caught:
+        asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], SimpleNamespace()))
+    assert caught.value.status_code == 403
+
+
+def test_replay_can_explore_contamination_without_requiring_map_generation(monkeypatch):
+    scope, binding = context(monkeypatch)
+    binding.tool_cap = ["run_contaminant_simulation"]
+    args = TOOLS["run_contaminant_simulation"].model.model_validate({
+        **parameters().model_dump(), "source": "J1", "concentration": 1,
+        "include_impact_coverage": False,
+    })
+    asyncio.run(query_service.authorize_query(scope, binding, TOOLS["run_contaminant_simulation"], args))

tests/unit/test_scada_repository.py

@@ -186,7 +186,9 @@ def test_recovery_read_selects_first_raw_frame_strictly_after_feedback_and_bound
     assert result == rows
     query, parameters = conn.cursor_instance.calls[0]
     assert "min(time)" in query and "time>%s AND time<=%s" in query
-    assert "time>=%s AND time<%s" in query and "JOIN next_frame" in query
+    assert "m.time>=%s AND m.time<%s" in query and "JOIN next_frame" in query
+    assert "mod(extract(epoch FROM m.time)::bigint,86400)=" in query
+    assert "mod(extract(epoch FROM f.time)::bigint,86400)" in query
     assert parameters == (["P1"], feedback, through, ["P1"], incident - timedelta(days=15), incident, 100_001, ["P1"])