feat(scada): require complete authorized investigation evidence

app/mcp/application.py

@@ -429,6 +429,9 @@ class McpQueryApplication:
                 output = display_query_result(result.model_dump(mode="json"))
                 content = []
                 from app.mcp.extended_service import ARTIFACT_TOOLS
+                # Complete SCADA inputs must reach the model intact, not as a first-page preview.
+                # The shared 8 MiB result ceiling above still applies before persistence and output.
+                complete_scada = name == "query_scada_readings" and args.device_ids is None
                 if (not include_full and name == "list_scada_devices"
                         and len(json.dumps(output, ensure_ascii=False).encode()) > 12000):
                     preview = compact_scada_catalog(output, budget=12000)
@@ -436,6 +439,7 @@ class McpQueryApplication:
                         output = preview
                 if (
                     not include_full
+                    and not complete_scada
                     and name not in ARTIFACT_TOOLS
                     and len(json.dumps(output, ensure_ascii=False).encode()) > 12000
                 ):

app/mcp/catalog.py

@@ -35,6 +35,12 @@ class QueryTool:
                 "。默认每页100个，最多500个；page.has_more=true时继续递增offset读取后续页面。"
                 "大目录内联保留本页全部设备的编号、类型和单位，模型映射及坐标用get_scada_device核验。"
             )
+        elif self.name == "query_scada_readings":
+            suffix = (
+                "。首次调查省略device_ids，Server读取完整设备属性并自动翻页读取窗口内全部授权观测，"
+                "返回devices、readings和coverage，包括缺测及冻结范围外设备；全量结果完整内联。"
+                "显式device_ids仅用于后续局部复核，不能替代全量核查。"
+            )
         else:
             suffix = (
                 "。先限定对象和时间范围，再读取明细；集合默认20条，按需逐页扩大，"

app/mcp/models.py

@@ -106,8 +106,20 @@ class WindowInput(PageInput):
 
 
 class ScadaInput(WindowInput):
-    device_ids: list[Id] = Field(min_length=1, max_length=20)
+    device_ids: list[Id] | None = Field(
-    field: Literal["monitored_value", "cleaned_value"] = "monitored_value"
+        default=None, min_length=1, max_length=20,
+        description="Omit to read the complete SCADA catalogue and all authorized readings in the window; explicit IDs select a paginated detail query.",
+    )
+    field: Literal["monitored_value", "cleaned_value"] = Field(
+        default="monitored_value",
+        description="Field for explicit-device detail queries; complete all-device reads return both raw and cleaned values.",
+    )
+
+    @model_validator(mode="after")
+    def complete_query_starts_at_zero(self):
+        if self.device_ids is None and self.offset:
+            raise ValueError("All-device SCADA queries read every page internally; offset must be zero")
+        return self
 
 
 class BaselineSeriesInput(WindowInput):

app/mcp/query_service.py

@@ -129,6 +129,11 @@ async def authorize_query(
     if not granted and binding.task_id and source in binding.task_scope_cap:
         async with scope.connection.transaction():
             granted = await task_service(scope).can_read_source(binding.task_id, source)
+            if granted and tool.name == "query_scada_readings" and args.device_ids is None:
+                granted = (
+                    "scada_devices" in binding.task_scope_cap
+                    and await task_service(scope).can_read_source(binding.task_id, "scada_devices")
+                )
     if not granted:
         raise HTTPException(403, "Data source permission is required")
     if tool.name == "run_burst_location" and args.burst_leakage is None:
@@ -175,7 +180,10 @@ async def execute_query(
     source_ref = "mcp:" + str(uuid4())
     evidence_id = None
     replay_context = getattr(binding, "replay_context", None)
-    if name == "query_scada_readings" and replay_context:
+    if name == "query_scada_readings" and args.device_ids is None:
+        from app.mcp.scada_readings import read_complete_scada
+        data, units, gaps, origin = await read_complete_scada(scope, binding, args)
+    elif name == "query_scada_readings" and replay_context:
         from app.services.scada_replay_queries import replay_snapshot_rows
         rows = await replay_snapshot_rows(conn, scope.project.project_id, replay_context,
                                           args.device_ids, args.start, args.end, args.limit + 1, args.offset)

app/mcp/scada_readings.py

@@ -0,0 +1,121 @@
+"""Complete project SCADA reads with explicit frozen-scope and missing-data accounting."""
+
+import json
+from collections import Counter
+
+from fastapi import HTTPException
+from fastapi.encoders import jsonable_encoder
+
+from app.mcp.models import BoundContext, ScadaInput
+from app.mcp.scope import QueryScope
+from app.infra.db.postgresql import read_queries as reads
+from app.infra.db.postgresql.scada_replays import ScadaReplayRepository
+from app.services.scada_replay_queries import check_replay_window, check_snapshot_start
+
+
+MAX_COMPLETE_SCADA_BYTES = 8 * 1024 * 1024
+
+
+async def read_complete_scada(
+    scope: QueryScope, binding: BoundContext, args: ScadaInput,
+) -> tuple[dict, dict, list[str], str]:
+    """Exhaust source pages; never treat a first page or selected device subset as a full scan."""
+    devices = []
+    offset = 0
+    size = 0
+
+    def account(items):
+        nonlocal size
+        size += len(json.dumps(jsonable_encoder(items), ensure_ascii=False).encode())
+        if size > MAX_COMPLETE_SCADA_BYTES:
+            raise HTTPException(413, "Complete SCADA result exceeds 8 MiB; split the time window and retain every device")
+
+    while True:
+        page = await reads.scada_devices(scope.connection, limit=501, offset=offset)
+        batch = page[:500]
+        account(batch)
+        devices.extend(batch)
+        if len(page) <= 500:
+            break
+        offset += len(batch)
+
+    project_device_count = len(devices)
+    context = getattr(binding, "replay_context", None)
+    catalog = {str(device["device_id"]): device for device in devices}
+    outside_scope = []
+    if context:
+        check_replay_window(context, args.start, args.end)
+        repository = ScadaReplayRepository(scope.connection, scope.project.project_id)
+        run = await repository.get(context["replay_id"])
+        if not run:
+            raise HTTPException(404, "SCADA replay was not found in the current project")
+        check_snapshot_start(run, args.start)
+        frozen = {str(device["device_id"]): device for device in run.get("devices", [])}
+        selected = list(dict.fromkeys(run["config"]["device_ids"]))
+        outside_scope = sorted(set(catalog) - set(selected))
+        # Retain devices removed from the current catalogue but present in the frozen source.
+        for device in selected:
+            if device not in catalog:
+                catalog[device] = frozen.get(device, {"device_id": device})
+        devices = list(catalog.values())
+    else:
+        selected = list(catalog)
+
+    rows = []
+    offset = 0
+    # The public limit controls source page size, not completeness of this operation.
+    page_size = min(max(args.limit, 1000), 10000)
+
+    async def read_pages(connection=None):
+        nonlocal offset
+        while selected:
+            if context:
+                page = await repository.detected_outputs(
+                    context["replay_id"], selected, args.start, args.end, context["batch_sequence"],
+                    limit=page_size + 1, offset=offset,
+                )
+            else:
+                page = await reads.series_rows(
+                    connection, source="scada", element_type="junction", ids=selected,
+                    start=args.start, end=args.end, field=None,
+                    limit=page_size + 1, offset=offset,
+                )
+            batch = page[:page_size]
+            if context:
+                # Match the existing frozen-reading contract; internal source hashes are not observations.
+                fields = ("time", "device_id", "monitored_value", "cleaned_value", "quality", "reason")
+                batch = [{key: row[key] for key in fields if key in row} for row in batch]
+            account(batch)
+            rows.extend(batch)
+            if len(page) <= page_size:
+                break
+            offset += len(batch)
+
+    if context:
+        await read_pages()
+    elif selected:
+        async with scope.timescale() as connection:
+            await read_pages(connection)
+
+    counts = Counter(str(row["device_id"]) for row in rows)
+    missing = sorted(set(selected) - counts.keys())
+    coverage = {
+        "mode": "all_scada", "query_complete": True,
+        "project_device_count": project_device_count, "queried_device_count": len(selected),
+        "devices_with_readings": len(counts), "reading_count": len(rows),
+        "queried_device_ids": selected, "missing_device_ids": missing,
+        "outside_frozen_scope_device_ids": outside_scope,
+        "readings_per_device": {device: counts[device] for device in selected},
+        "replay_id": context["replay_id"] if context else None,
+        "batch_sequence": context["batch_sequence"] if context else None,
+    }
+    data = {"devices": devices, "readings": rows, "coverage": coverage}
+    account([coverage])
+    gaps = []
+    if missing:
+        gaps.append(f"{len(missing)} queried SCADA devices have no readings in this window; see coverage.missing_device_ids")
+    if outside_scope:
+        gaps.append(f"{len(outside_scope)} project devices are outside this frozen replay; see coverage.outside_frozen_scope_device_ids")
+    units = {device: catalog[device]["measurement_unit"] for device in selected
+             if catalog[device].get("measurement_unit")}
+    return data, units, gaps, "synthetic" if context else "system_data"

app/services/operational_scada_coverage.py

@@ -0,0 +1,40 @@
+"""Require persisted complete SCADA input before final investigation conclusions."""
+
+from datetime import datetime
+
+from app.domain.domain_errors import require
+
+
+def require_complete_scada_evidence(task: dict, evidence: list[dict], refs: list[str]) -> None:
+    replay = task.get("source", {}).get("replay_context")
+    if task.get("workflow") != "scada_anomaly" and not replay:
+        return
+    target = datetime.fromisoformat(
+        replay["knowledge_cutoff"] if replay else task["trigger"]["occurred_at"]
+    )
+    for item in evidence:
+        if item["evidence_id"] not in refs:
+            continue
+        snapshot = item.get("data_snapshot") or {}
+        if snapshot.get("source") != "query_scada_readings" or snapshot.get("truncated"):
+            continue
+        result = snapshot.get("data") or {}
+        data = result.get("data") if isinstance(result, dict) else None
+        coverage = data.get("coverage") if isinstance(data, dict) else None
+        if not coverage or coverage.get("mode") != "all_scada" or coverage.get("query_complete") is not True:
+            continue
+        window = result.get("window") or {}
+        if not window.get("start") or not window.get("end"):
+            continue
+        start, end = (datetime.fromisoformat(window[key]) for key in ("start", "end"))
+        if not start <= target <= end:
+            continue
+        if replay and (coverage.get("replay_id") != replay["replay_id"]
+                       or coverage.get("batch_sequence") != replay["batch_sequence"]
+                       or end != target):
+            continue
+        return
+    require(False, "SCADA_COVERAGE_REQUIRED",
+            "最终SCADA报告必须引用本轮全量读取证据。调用query_scada_readings并省略device_ids，"
+            "在任务时间窗内读取全部授权设备、完整属性与观测，核对coverage和缺测清单后再发布。",
+            status=409)

app/services/operational_tasks.py

@@ -73,6 +73,7 @@ from app.domain.schemas.operational_data import TaskDataSnapshot, TaskDataSource
 from app.services.content_digest import digest
 from app.services.operational_presentation import resolve_overlay
 from app.services.operational_recovery import accept_completion, track_work_order
+from app.services.operational_scada_coverage import require_complete_scada_evidence
 
 
 LEASE_DURATION = timedelta(seconds=60)
@@ -929,6 +930,10 @@ class OperationalTaskService:
                 current_plan = snapshot
             if command.report:
                 self._require_evidence_refs(command.report.evidence_refs, known_evidence)
+                if command.report.stage == "final":
+                    require_complete_scada_evidence(
+                        task, await self.repo.list_evidence(str(task_id)), command.report.evidence_refs,
+                    )
                 for gap in command.report.data_gaps:
                     require(set(gap.evidence_refs) <= set(command.report.evidence_refs),
                             "REPORT_GAP_EVIDENCE_INVALID",
@@ -1132,7 +1137,7 @@ class OperationalTaskService:
             await self.repo.lock(f"task:{task_id}")
             task = await self._task(task_id)
             self._require_actionable_task(task)
-            self._require_real_action(task)
+            self._require_real_action(task, command.kind)
             self._validate_action_payload(command)
             if task["task_type"] == "event_investigation":
                 report = await self.repo.get_latest_report(str(task_id))
@@ -1203,8 +1208,8 @@ class OperationalTaskService:
             await self.repo.lock(f"action:{action_id}")
             task = await self._task(task_id)
             self._require_actionable_task(task)
-            self._require_real_action(task)
             action = await self._action(task_id, action_id)
+            self._require_real_action(task, action.kind)
             require(action.kind == "create_work_order", "INVALID_ACTION_KIND",
                     "Only work-order drafts can be edited", status=422)
             require(action.version == command.expected_version, "VERSION_CONFLICT",
@@ -1259,8 +1264,8 @@ class OperationalTaskService:
             await self.repo.lock(f"action:{action_id}")
             task = await self._task(task_id)
             self._require_actionable_task(task)
-            self._require_real_action(task)
             action = await self._action(task_id, action_id)
+            self._require_real_action(task, action.kind)
             require(action.version == command.expected_version, "VERSION_CONFLICT",
                     "Action version changed", status=409)
             require(action.approval_status == "pending", "INVALID_TRANSITION",
@@ -1306,8 +1311,8 @@ class OperationalTaskService:
             await self.repo.lock(f"action:{action_id}")
             task = await self._task(task_id)
             self._require_actionable_task(task)
-            self._require_real_action(task)
             action = await self._action(task_id, action_id)
+            self._require_real_action(task, action.kind)
             require(action.version == command.expected_version, "VERSION_CONFLICT",
                     "Action version changed", status=409)
             require(action.approval_status == "approved", "INVALID_TRANSITION",
@@ -2151,9 +2156,10 @@ class OperationalTaskService:
         return task
 
     @staticmethod
-    def _require_real_action(task: dict) -> None:
+    def _require_real_action(task: dict, kind: str) -> None:
-        require(not (task.get("source") or {}).get("replay_context"), "REPLAY_ACTION_DISABLED",
+        require(kind == "create_work_order" or not (task.get("source") or {}).get("replay_context"),
-                "当前任务的工单和设备动作未开放；请继续完成已授权的定位、地图展示、隔离影响分析与报告", status=409)
+                "REPLAY_ACTION_DISABLED",
+                "回放任务不支持设备命令；可生成工单草稿，由有权限的用户确认发布", status=409)
 
     @staticmethod
     def _require_actionable_task(task: dict) -> None:

docs/mcp-query-implementation.md

@@ -14,12 +14,16 @@ Agent 根据401 challenge 或路径/根 well-known 发现受保护资源元数
 
 调查数据返回前，在短事务内追加现有任务证据并保存调用回执。同一主体、会话、任务、run 和原生 call ID 的重试复用证据；不同参数或失效范围明确报错。七类原有访问申请 scope 保持原契约，MCP 的新增来源只扩充快照来源枚举。没有新增报告发布、审批、控制、Shell 或递归委派权限。
 
-普通查询默认20条；目录类最多500条，时序最多10000条、20个标识、24小时；offset 最大10000。Agent 首次优先单对象、必要的短时间窗和小页，仅在证据不足时按需扩大，不自动遍历全网。SQL 有 limit/offset 和 statement_timeout。历史对比在同一条SQL中按序列、时间对数据点分页，返回 page.has_more；非零offset也标明覆盖不完整。历史曲线展示调用初始每页500点，`RESULT_TOO_LARGE` 时原位减半重试，最多读取1000条、发出16次请求，共用120秒超时（单次请求同时受该剩余总预算约束），图表最多100个有效点；这不改变模型首次查询默认值或时序接口上限。
+普通查询默认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 并记录数据缺口。
 
 `list_scada_devices` 默认每页100个、最多500个，其他集合仍沿用各自的分页默认值。设备目录超过12,000字节时，先内联本页全部设备的 `device_id`、`device_type` 和 `measurement_unit`，省略其他属性并明确标注；模型映射、坐标等通过 `get_scada_device` 核验，任务证据和幂等回执保留完整属性。此时 `truncated` 表示属性省略，设备是否还有后续页以 `page.has_more` 为准，按本页实际设备数递增 `offset`。精简目录仍超预算时沿用证据引用或缩页路径。
 
+全量查询同时要求设备目录与读数权限；通过任务授权读取时，捕获范围及当前批准状态都必须同时包含 `scada_devices` 和 `scada_readings`，显式设备读数查询仍沿用读数授权。
+
+全量 `query_scada_readings` 返回 `data.devices` 的完整属性、`data.readings` 的原始和清洗值，以及 `data.coverage` 的查询设备、读数数量和缺测清单；该模式保留完整内联数据，不经过12KB缩略，仍执行8MiB整份结果上限。回放仅读取指定批次之前的冻结观测，项目中未纳入该回放的设备单列 `outside_frozen_scope_device_ids`。`query_complete=true` 表示授权范围内的查询已读完，不表示设备均有有效数据。SCADA及历史回放任务发布final时，必须引用窗口覆盖本轮时刻、且回放批次匹配的全量证据；单设备、局部集合、未引用或旧批次证据不能通过此校验。既有报告不回写，后续重跑适用新约束。
+
 `get_task_context` 默认 section=summary，只查固定元数据、160字符任务标题及200字符结论预览，并返回 available_sections；preview_only 和 truncated 明确说明不是完整报告。按 section=task/report/evidence/actions/progress/access_requests/plan/run 读取详情，report_revision 保留选择旧报告的语义。data_path 选择嵌套值，limit 默认20、最多500，offset 最大10000；证据、动作和授权申请集合以稳定 ID 为键，按 ID 字典序分页。字符串按字符分页。详情在数据库返回前检查8KB预算，超限缩页或选更深路径。缺少可选进度或授权申请表时返回 metadata.available=false，不冒充空事实。绑定权限范围只读取七类既有 scope，不再加载整个任务上下文。不增加工具数量、数据库表或迁移，也不改变原 REST 完整上下文契约。
 
 外围 MCP 直接读取所选项目 BizDB 的工单事实和既有时序。外围工单统一返回 `system_data`；测试数据和 DEMO-* 观测仍按各自数据来源处理。DEMO-* 观测在数据库端拒绝超过8MiB的完整样本，预算内仍需读全样本、核验摘要后筛选；此限制不能通过缩小查询窗口解决。监测点列表沿用设备目录，不把未登记的测试点冒充实测设备。

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

@@ -2,22 +2,22 @@
 name: scada-anomaly
 description: 从 SCADA 指标异常开展数据核验、爆管诊断、候选定位与影响分析。
 metadata:
-  version: "1.0.9"
+  version: "1.0.12"
 ---
 
 # SCADA 数据异常调查
 
 读取任务与 replay_context，核对设备、单位、异常时刻、knowledge_cutoff 和 batch_sequence。使用本轮已提交的 SCADA 观测开展调查，同一持续事件沿用已有调查。证据来源字段用于记录数据取得方式，不构成停止分析的条件。
 
-报告与图表的阅读时区统一为Asia/Shanghai（UTC+8）。工具返回的+08:00时间按原值使用；Z时间换算后同步更新日期、历史系列名和横轴标签，例如2026-07-16T00:00:00Z对应7月16日08:00。sample_times记录相同真实时间点并使用+08:00偏移，查询与冻结边界按带时区的时间点比较。
+报告与图表的阅读时区统一为Asia/Shanghai（UTC+8）。工具返回的+08:00时间按原值使用；Z时间换算后同步更新日期、历史系列名和横轴标签，例如2026-07-16T00:00:00Z对应7月16日08:00。sample_times记录相同真实时间点并使用+08:00偏移，查询与冻结边界按带时区的时间点比较。图表标题、横轴和系列名只写日期、时刻与单位，不写“上海时间”“Asia/Shanghai”或“UTC+8”；横轴命名为“时间”。
 
 ## 数据核验与诊断
 
 任一已配置指标异常即进入调查，包含压力、流量、水质规则和压力模型信号。先核验数据质量与测点，再判别原因；模型或组合指标有缺口时，继续核查已发现的单项异常。
 
-通过 query_scada_readings 读取冻结异常窗和正常对照，按需核查开始前15天基线；查询结束时间止于 knowledge_cutoff，数据批次止于 batch_sequence。复用已有监测回执。结合 monitored_value、cleaned_value 和质量标记判断变化，可靠采集的低压与水质变化保留为业务异常。
+通过 query_scada_readings 省略device_ids读取冻结异常窗内全部SCADA设备的完整属性和授权观测，覆盖压力、流量、水质等全部类型；正常对照也采用全量读取。Server自动读完所有设备页和观测页，完整返回devices、readings、coverage，结果仍受8MiB整份上限约束，超限明确报错后按时间分段，不能丢弃设备。query_complete表示查询已读完，缺测与无有效值另行判断。查询结束时间止于knowledge_cutoff，数据批次止于batch_sequence；冻结范围外设备单列，不回退实时数据。按需核查开始前15天基线，复用已有完整监测回执。结合monitored_value、cleaned_value和质量标记判断变化，可靠采集的低压与水质变化保留为业务异常。
 
-设备目录用 list_scada_devices，默认每页100个；page.has_more=true时递增offset继续读取，直到覆盖本轮目标设备。大目录内联可能仅包含本页全部设备的编号、类型和单位，模型映射与坐标用get_scada_device核验。读数查询的limit按设备数乘时间点数安排，检查page.has_more；缩略响应通过evidence_read指定的工具和data_path分小页读取已保存证据，不能把首屏、首个设备或属性缩略当作完整覆盖。
+全量响应中的devices包含完整目录属性，readings包含实际读数，coverage列出全部查询设备、各设备读数数量、缺测和冻结范围外设备。不能只读取已标记、异常或代表设备后推断全网状态。单独读取list_scada_devices时，默认每页100个，持续翻页到page.has_more=false；属性缩略不等于已读取完整设备信息。全量读取后可显式指定device_ids做局部复核，局部读数仍按page.has_more接续，不能替代全量输入。最终报告引用覆盖本轮knowledge_cutoff和batch_sequence的全量证据，Server会拒绝仅有局部或旧批次证据的final。
 
 分别说明压力异常模型得分、压力绝对下降及变化率、已配置出厂边界流量变化，再说明联合证据支持程度。使用实际权重、嫌疑门槛、动态阈值、参与测点与各自基线。综合分用于爆管嫌疑判断；得分越低越异常，目标告警与参考日分别解释。水质等其他异常按实际证据选择原因分支。
 
@@ -31,9 +31,9 @@ metadata:
 
 ## 对比展示与影响
 
-先发布覆盖本轮全部已标识SCADA设备的当前趋势总览，包括地图定位、等级标记及表格列出的目标设备。按指标和单位分图，同组各设备各占一条当前序列，系列名带device_id；每图最多8组，超过时分组发布，各组使用不同且稳定的chart_id。缺测填null，完全无读数的设备明确列为缺测并说明查询窗口，不能静默省略。收尾前逐项核对目标设备、查询覆盖、已发布图表和缺测清单，不用少量代表设备替代全量总览。
+先查询并发布截至本轮knowledge_cutoff或异常时刻t的[t-3小时,t]当前趋势总览，覆盖本轮全部已标识SCADA设备，包括地图定位、等级标记及表格列出的目标设备。按指标和单位分图，同组各设备各占一条当前序列，系列名带device_id；每图最多8组，超过时分组发布，各组使用不同且稳定的chart_id。缺测填null，完全无读数的设备明确列为缺测并说明查询窗口，不能静默省略。压力和流量分别发布图表，包含正常与异常测点；某指标无设备或无有效读数时明确说明。每条当前序列保留三小时窗口内全部已取得采样点，不能用用于判断突变的[t-采样周期,t]相邻两点代替三小时趋势。15分钟数据正常应有13个时刻；缺测保留对应空槽并说明实际覆盖。收尾前逐项核对目标设备、查询覆盖、已发布图表和缺测清单，不用少量代表设备替代全量总览。图表“数据”视图沿用同一完整序列，报告按指标附测点、单位、窗口、有效点数和缺测说明表格。
 
-定位输入具备后继续推进定位；七天对照作为总览之外的补充，按设备分图，不能因历史散点占用序列额度而遗漏其他设备的当前趋势。复合图采用 [t-3小时,t] 当前折线和前7天每天 t-30分钟、t-15分钟、t 的21个参考位置。15分钟数据用13槽，当前一个line、历史七个scatter；历史散点只填末3槽，其他为null。sample_times保留真实带时区时刻，坐标轴注明单位。各采样周期保留实际缺口，按需翻页并报告实际参考点数。
+定位输入具备后继续推进定位。压力、流量均需发布按设备分图的七天历史复合对照，作为全量当前趋势总览之外的必要展示；不能因历史散点占用序列额度而遗漏其他设备的当前趋势。复合图采用 [t-3小时,t] 当前折线和前7天每天 t-30分钟、t-15分钟、t 的21个参考位置。15分钟数据用13槽，当前一个line、历史七个scatter；历史散点只填末3槽，其他为null。sample_times保留真实带时区时刻，坐标轴注明单位。前7天按t转换后的当地日期逐日回退，不包含当天；按实际采样周期保留缺口，按需翻页并报告实际参考点数。基线未冻结、读取失败或缺测时明确说明并保留已有三小时趋势，不能静默退化为前一天或相邻两个时刻。
 
 影响说明列出异常前参考、异常期间和已有恢复观测中的压力、流量或水质变化、持续时间、涉及测点及已核实资产。可比对象、单位、时段与运行条件下给出变化量，分别标明实际观测、模型预测和拓扑隔离。停水范围与漏量引用相应拓扑、水力、流量平衡或现场依据；恢复观测、用户关联或现场证据缺失时列为具体待核实项。
 
@@ -42,3 +42,5 @@ metadata:
 本轮时序分析保持任务冻结窗口，后续播放形成的观测由相应轮次处理。资产属性、空间查询和拓扑隔离使用项目当前模型，注明模型时态；水力工具按已开放的冻结输入能力调用。定位调用执行权限、设备与冻结观测校验；基线或覆盖缺口按实际返回记录。
 
 报告给出观测事实、规则信号、原因候选、定位与隔离结果、影响说明和建议。摘要以150字以内为目标，测点明细用表格、完整依据留在证据。候选、相似度和拓扑结果具备时，完成相应分析、展示与报告后再收尾；现场核验、审批和维修列为后续待办。现场工单与设备操作遵守服务端动作权限，本轮分析完成与现场事件闭环分别说明。
+
+需要现场核验时，先发布本轮outcome=concluded的调查报告，再调用propose_task_action提出kind=create_work_order、work_order_type=general的关联预约工单草稿，引用候选与影响证据，写明核验对象、内容和回执要求。带replay_context的任务同样可提出草稿，描述中注明历史异常时刻及回放来源；草稿须由有权限的用户编辑、确认后发布，后续派发与反馈进入预约任务。草稿待确认时说明等待人工处理，不将调查报告完成写成现场处置完成。回放任务不支持设备命令。

tests/unit/test_complete_scada.py

@@ -0,0 +1,179 @@
+import asyncio
+from contextlib import asynccontextmanager
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+from uuid import uuid4
+
+import pytest
+from fastapi import HTTPException
+
+from app.mcp import query_service, scada_readings
+from app.mcp.catalog import TOOLS
+from app.mcp.models import ScadaInput
+from app.domain.domain_errors import DomainError
+from app.services.operational_scada_coverage import require_complete_scada_evidence
+
+
+START = datetime(2026, 7, 16, tzinfo=timezone.utc)
+
+
+def _args(**extra):
+    return ScadaInput(context_handle="a" * 43, start=START, end=START + timedelta(hours=1), **extra)
+
+
+def test_all_scada_reads_every_catalogue_and_observation_page_with_full_attributes(monkeypatch):
+    devices = [{"device_id": f"D{i}", "device_type": ["pressure", "pipe_flow", "quality"][i % 3],
+                "measurement_unit": ["m", "m3/h", "mg/L"][i % 3], "node_id": f"J{i}",
+                "longitude": 121.5, "latitude": 31.0, "reliability": 1} for i in range(503)]
+    rows = [{"device_id": f"D{i % 502}", "time": START.isoformat(),
+             "monitored_value": float(i), "cleaned_value": None} for i in range(1205)]
+
+    async def catalog(connection, *, limit, offset):
+        return devices[offset:offset + limit]
+
+    async def series(connection, **args):
+        assert args["ids"] == [device["device_id"] for device in devices]
+        assert args["field"] is None
+        return rows[args["offset"]:args["offset"] + args["limit"]]
+
+    catalog_reader, series_reader = AsyncMock(side_effect=catalog), AsyncMock(side_effect=series)
+    monkeypatch.setattr(scada_readings.reads, "scada_devices", catalog_reader)
+    monkeypatch.setattr(scada_readings.reads, "series_rows", series_reader)
+    monkeypatch.setattr(query_service, "authorize_query", AsyncMock())
+
+    @asynccontextmanager
+    async def timescale(): yield object()
+
+    scope = SimpleNamespace(connection=object(), timescale=timescale)
+    result = asyncio.run(query_service.execute_query(scope, SimpleNamespace(replay_context=None),
+                                                    TOOLS["query_scada_readings"], _args()))
+    assert result.data["devices"] == devices and result.data["readings"] == rows
+    assert [call.kwargs["offset"] for call in catalog_reader.await_args_list] == [0, 500]
+    assert [call.kwargs["offset"] for call in series_reader.await_args_list] == [0, 1000]
+    coverage = result.data["coverage"]
+    assert coverage["query_complete"] is True and coverage["queried_device_count"] == 503
+    assert coverage["devices_with_readings"] == 502 and coverage["missing_device_ids"] == ["D502"]
+    assert coverage["reading_count"] == 1205 and len(result.units) == 503
+    assert result.origin == "system_data" and result.page is None and not result.truncated
+
+
+def test_all_scada_replay_reads_only_frozen_devices_batches_and_window(monkeypatch):
+    devices = [{"device_id": key, "device_type": kind, "measurement_unit": unit, "node_id": key}
+               for key, kind, unit in [("P1", "pressure", "m"), ("Q1", "quality", "mg/L"), ("NEW", "pressure", "m")]]
+    rows = [{"device_id": "P1", "time": START.isoformat(), "monitored_value": 20.,
+             "cleaned_value": 20., "quality": "valid", "reason": "冻结观测", "source_hash": "internal-hash"}]
+    context = {"replay_id": str(uuid4()), "knowledge_cutoff": (START + timedelta(hours=1)).isoformat(),
+               "batch_sequence": 9}
+    repository = SimpleNamespace(
+        get=AsyncMock(return_value={"start_time": START.isoformat(), "config": {"device_ids": ["P1", "Q1"]},
+                                   "devices": devices[:2]}),
+        detected_outputs=AsyncMock(return_value=rows),
+    )
+    monkeypatch.setattr(scada_readings.reads, "scada_devices", AsyncMock(return_value=devices))
+    monkeypatch.setattr(scada_readings, "ScadaReplayRepository", lambda *args: repository)
+    timescale = AsyncMock(side_effect=AssertionError("Frozen reads must never use live TSDB"))
+    scope = SimpleNamespace(connection=object(), project=SimpleNamespace(project_id=uuid4()), timescale=timescale)
+    data, units, gaps, origin = asyncio.run(scada_readings.read_complete_scada(
+        scope, SimpleNamespace(replay_context=context), _args(),
+    ))
+    assert data["devices"] == devices
+    assert data["readings"] == [{key: value for key, value in rows[0].items() if key != "source_hash"}]
+    assert data["coverage"]["queried_device_ids"] == ["P1", "Q1"]
+    assert data["coverage"]["missing_device_ids"] == ["Q1"]
+    assert data["coverage"]["outside_frozen_scope_device_ids"] == ["NEW"]
+    repository.detected_outputs.assert_awaited_once_with(
+        context["replay_id"], ["P1", "Q1"], START, START + timedelta(hours=1), 9, limit=1001, offset=0,
+    )
+    assert origin == "synthetic" and len(gaps) == 2 and units == {"P1": "m", "Q1": "mg/L"}
+    timescale.assert_not_called()
+
+
+def test_all_scada_empty_catalogue_does_not_query_unfiltered_timeseries(monkeypatch):
+    monkeypatch.setattr(scada_readings.reads, "scada_devices", AsyncMock(return_value=[]))
+    timescale = AsyncMock(side_effect=AssertionError("An empty catalogue must not select every TSDB row"))
+    data, units, _, _ = asyncio.run(scada_readings.read_complete_scada(
+        SimpleNamespace(connection=object(), timescale=timescale), SimpleNamespace(replay_context=None), _args(),
+    ))
+    assert data["coverage"]["queried_device_count"] == 0 and data["readings"] == [] and units == {}
+    timescale.assert_not_called()
+
+
+def test_all_scada_rejects_offset_and_empty_explicit_selection():
+    for values in ({"offset": 1}, {"device_ids": []}):
+        with pytest.raises(ValueError):
+            _args(**values)
+    assert _args(device_ids=["P1"], offset=100).offset == 100
+
+
+@pytest.mark.parametrize("captured,approved,explicit,direct,allowed", [
+    ({"scada_readings", "scada_devices"}, {"scada_readings"}, False, False, False),
+    ({"scada_readings"}, {"scada_readings", "scada_devices"}, False, False, False),
+    ({"scada_readings", "scada_devices"}, {"scada_devices"}, False, False, False),
+    ({"scada_readings", "scada_devices"}, {"scada_readings", "scada_devices"}, False, False, True),
+    ({"scada_readings"}, {"scada_readings"}, True, False, True),
+    (set(), set(), False, True, True),
+])
+def test_complete_scada_requires_both_captured_and_current_source_grants(
+    monkeypatch, captured, approved, explicit, direct, allowed,
+):
+    @asynccontextmanager
+    async def transaction():
+        yield
+
+    can_read = AsyncMock(side_effect=lambda task_id, source: source in approved)
+    monkeypatch.setattr(query_service, "permissions_for_context", lambda _: {"scada.view"} if direct else set())
+    monkeypatch.setattr(query_service, "task_service", lambda _: SimpleNamespace(can_read_source=can_read))
+    binding = SimpleNamespace(
+        replay_context=None, tool_cap=["query_scada_readings"],
+        permission_cap=["scada.view"] if direct else [], task_id=uuid4(), task_scope_cap=captured,
+    )
+    scope = SimpleNamespace(project=object(), connection=SimpleNamespace(transaction=transaction))
+    args = _args(**({"device_ids": ["P1"]} if explicit else {}))
+    call = query_service.authorize_query(scope, binding, TOOLS["query_scada_readings"], args)
+    if allowed:
+        asyncio.run(call)
+    else:
+        with pytest.raises(HTTPException) as error:
+            asyncio.run(call)
+        assert error.value.status_code == 403
+    if direct:
+        can_read.assert_not_awaited()
+    elif allowed:
+        expected = {"scada_readings"} if explicit else {"scada_readings", "scada_devices"}
+        assert {call.args[1] for call in can_read.await_args_list} == expected
+
+
+def test_all_scada_reports_oversize_instead_of_successfully_returning_a_partial_scan(monkeypatch):
+    monkeypatch.setattr(scada_readings.reads, "scada_devices", AsyncMock(return_value=[
+        {"device_id": "P1", "description": "x" * 200},
+    ]))
+    monkeypatch.setattr(scada_readings, "MAX_COMPLETE_SCADA_BYTES", 100)
+    with pytest.raises(HTTPException) as error:
+        asyncio.run(scada_readings.read_complete_scada(
+            SimpleNamespace(connection=object()), SimpleNamespace(replay_context=None), _args(),
+        ))
+    assert error.value.status_code == 413
+
+
+@pytest.mark.parametrize("defect", [None, "old_batch", "other_replay", "old_window", "unreferenced", "truncated"])
+def test_final_scada_evidence_must_match_the_current_frozen_scope(defect):
+    replay_id = str(uuid4())
+    replay = {"replay_id": replay_id, "knowledge_cutoff": START.isoformat(), "batch_sequence": 9}
+    task = {"workflow": "burst_report", "source": {"replay_context": replay}}
+    coverage = {"mode": "all_scada", "query_complete": True, "replay_id": replay_id, "batch_sequence": 9}
+    window = {"start": (START - timedelta(hours=1)).isoformat(), "end": START.isoformat()}
+    if defect == "old_batch": coverage["batch_sequence"] = 8
+    if defect == "other_replay": coverage["replay_id"] = str(uuid4())
+    if defect == "old_window": window["end"] = (START - timedelta(minutes=15)).isoformat()
+    evidence = [{"evidence_id": "mcp:all", "data_snapshot": {
+        "source": "query_scada_readings", "truncated": defect == "truncated",
+        "data": {"data": {"coverage": coverage}, "window": window},
+    }}]
+    refs = [] if defect == "unreferenced" else ["mcp:all"]
+    if defect is None:
+        require_complete_scada_evidence(task, evidence, refs)
+    else:
+        with pytest.raises(DomainError) as error:
+            require_complete_scada_evidence(task, evidence, refs)
+        assert error.value.code == "SCADA_COVERAGE_REQUIRED"

tests/unit/test_mcp_query.py

@@ -774,6 +774,34 @@ def test_scada_catalog_default_100_paginates_without_changing_other_queries(runt
     assert TOOLS["list_elements"].model(context_handle=value.context_handle, element_type="pipe").limit == 20
 
 
+@pytest.mark.parametrize("complete", [False, True])
+def test_complete_scada_input_is_not_replaced_by_12kb_preview_but_details_stay_bounded(runtime, complete):
+    app, client, _, _ = runtime
+    value = binding()
+    data = {"devices": [{"device_id": f"D{index}", "node_id": f"J{index}", "description": "x" * 100}
+                        for index in range(168)],
+            "readings": [{"device_id": f"D{index}", "monitored_value": 20., "cleaned_value": 19.}
+                         for index in range(168)],
+            "coverage": {"mode": "all_scada", "query_complete": True, "queried_device_count": 168}}
+    assert len(json.dumps(data).encode()) > 12000
+    app.executor = AsyncMock(return_value=QueryResult(
+        source="query_scada_readings", source_ref="mcp:all-scada", queried_at=datetime.now(timezone.utc),
+        semantics="observed", data=data,
+    ))
+    arguments = {"context_handle": value.context_handle, "start": "2026-07-16T05:00:00+08:00",
+                 "end": "2026-07-16T08:00:00+08:00"}
+    if not complete:
+        arguments["device_ids"] = ["D1"]
+    response = request(client, "tools/call", {"name": "query_scada_readings", "arguments": arguments}).json()["result"]
+    if complete:
+        assert not response.get("isError"), response
+        assert response["structuredContent"]["data"] == data
+        assert response["structuredContent"]["truncated"] is False
+    else:
+        assert response["isError"] is True
+        assert response["structuredContent"]["error"]["code"] == "RESULT_TOO_LARGE"
+
+
 def test_business_skill_is_delivered_over_mcp_without_recording_event_evidence(runtime, monkeypatch):
     app, client, _, _ = runtime
     value = binding(task=True)

tests/unit/test_operational_tasks.py

@@ -577,7 +577,7 @@ async def test_presentation_cursor_and_event_are_read_only_and_idempotent():
 
 
 @pytest.mark.anyio
-async def test_replay_investigation_can_publish_evidenced_composite_chart_but_cannot_propose_real_action():
+async def test_replay_investigation_can_publish_chart_and_propose_work_order():
     now = datetime(2026, 10, 10, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
@@ -602,35 +602,48 @@ async def test_replay_investigation_can_publish_evidenced_composite_chart_but_ca
     ))
     assert result.recorded
     assert repository.events[-1]["metadata"]["command"]["chart"]["series"][1]["data"] == [None, 28.]
-    with pytest.raises(DomainError) as caught:
+    _seed_concluded_report(repository, now)
-        await service.propose_action(task["task_id"], ActionProposalCommand(
+    action = await service.propose_action(task["task_id"], ActionProposalCommand(
-            request_id=uuid4(), kind="create_work_order", reason="演示不得创建真实工单",
+        request_id=uuid4(), kind="create_work_order", reason="候选管段待现场核验",
-            payload={"title": "现场核查", "description": "核查", "work_order_type": "general"},
+        evidence_refs=[evidence.evidence_id],
-        ))
+        payload={"title": "现场核查", "description": "核查", "work_order_type": "general"},
-    assert caught.value.code == "REPLAY_ACTION_DISABLED"
+    ))
+    assert action.approval_status == "pending"
+    assert action.execution_status == "not_submitted"
+    assert repository.action["action_id"] == str(action.action_id)
 
 
 @pytest.mark.anyio
-@pytest.mark.parametrize("operation", ["draft", "approve", "submit"])
+@pytest.mark.parametrize("operation", ["propose", "approve", "submit"])
-async def test_replay_cannot_edit_approve_or_submit_preexisting_real_actions(operation):
+async def test_replay_cannot_propose_approve_or_submit_device_commands(operation):
     now = datetime(2026, 10, 10, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
     task["source"]["replay_context"] = {
         "replay_id": str(uuid4()), "knowledge_cutoff": now.isoformat(), "batch_sequence": 1,
     }
+    action = HumanApprovalAction(
+        action_id=uuid4(), task_id=task["task_id"], version=1, kind="issue_command",
+        reason="关阀", payload={"command_type": "close_valve", "target": {"kind": "valve", "id": "V1"},
+            "parameters": {}, "impact_summary": "隔离候选管段", "preconditions": ["现场核验"],
+            "expires_at": now + timedelta(minutes=5)},
+        evidence_refs=[], required_approval_permission="command.approve",
+        required_execution_permission="command.execute", approval_status="pending",
+        execution_status="not_submitted", proposed_by="principal:test", proposed_at=now,
+    )
     repository = RunTaskRepository(task, _active_run(task, now))
+    repository.action = action.model_dump(mode="json")
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     dispatcher = SimpleNamespace(dispatch=AsyncMock())
     args = {"request_id": uuid4(), "expected_version": 1}
     with pytest.raises(DomainError) as caught:
-        if operation == "draft":
+        if operation == "propose":
-            await service.update_work_order_draft(task["task_id"], uuid4(), WorkOrderDraftUpdateCommand(
+            await service.propose_action(task["task_id"], ActionProposalCommand(
-                **args, payload={"title": "现场核查", "description": "核查", "work_order_type": "general"}))
+                request_id=uuid4(), kind="issue_command", reason=action.reason, payload=action.payload))
         elif operation == "approve":
-            await service.decide_action(task["task_id"], uuid4(), ActionDecisionCommand(**args, decision="approved"))
+            await service.decide_action(task["task_id"], action.action_id, ActionDecisionCommand(**args, decision="approved"))
         else:
-            await service.submit_action(task["task_id"], uuid4(), ActionExecuteCommand(**args), dispatcher, None)
+            await service.submit_action(task["task_id"], action.action_id, ActionExecuteCommand(**args), dispatcher, None)
     assert caught.value.code == "REPLAY_ACTION_DISABLED"
     dispatcher.dispatch.assert_not_awaited()
 
@@ -965,6 +978,36 @@ def _current_report():
     }
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize("complete", [False, True])
+async def test_scada_final_report_requires_complete_current_input_not_selected_device_evidence(complete):
+    now = datetime(2026, 10, 10, tzinfo=timezone.utc)
+    context = _context()
+    task = _task_document(context, now, lease=now + timedelta(minutes=1))
+    task["workflow"] = "scada_anomaly"
+    repository = RunTaskRepository(task, _active_run(task, now))
+    service = OperationalTaskService(repository, context, set(), clock=lambda: now)
+    data = {"devices": [{"device_id": "P1"}], "readings": [], "coverage": {
+        "mode": "all_scada", "query_complete": True,
+    }} if complete else [{"device_id": "P1", "monitored_value": 20.}]
+    evidence, _ = await service.record_data_snapshot(task["task_id"], TaskDataSnapshot(
+        source="query_scada_readings", source_ref="mcp:scada-coverage", queried_at=now,
+        origin="system_data", truncated=False,
+        data={"data": data, "window": {"start": (now - timedelta(hours=1)).isoformat(), "end": now.isoformat()}},
+    ))
+    command = AgentArtifactsCommand(
+        request_id=uuid4(), source_session_id="session-1",
+        report={**_current_report(), "evidence_refs": [evidence.evidence_id]},
+    )
+    if complete:
+        result = await service.publish_artifacts(task["task_id"], command)
+        assert result.task.latest_report.report.stage == "final"
+    else:
+        with pytest.raises(DomainError) as error:
+            await service.publish_artifacts(task["task_id"], command)
+        assert error.value.code == "SCADA_COVERAGE_REQUIRED" and not repository.reports
+
+
 def _insufficient_report():
     return {
         **_current_report(), "outcome": "insufficient_data",
@@ -1684,10 +1727,15 @@ async def test_completed_gate_requires_access_request_decision():
 
 
 @pytest.mark.anyio
-async def test_work_order_draft_edit_creates_new_review_version():
+@pytest.mark.parametrize("replay", [False, True])
+async def test_work_order_draft_edit_creates_new_review_version(replay):
     now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
+    if replay:
+        task["source"]["replay_context"] = {
+            "replay_id": str(uuid4()), "knowledge_cutoff": now.isoformat(), "batch_sequence": 1,
+        }
     repository = RunTaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, {OPERATIONAL_ACTION_WORK_ORDER_APPROVE}, clock=lambda: now)
     _seed_concluded_report(repository, now)
@@ -1710,10 +1758,15 @@ async def test_work_order_draft_edit_creates_new_review_version():
 
 
 @pytest.mark.anyio
-async def test_one_click_work_order_confirmation_replays_without_duplicate_dispatch():
+@pytest.mark.parametrize("is_replay", [False, True])
+async def test_one_click_work_order_confirmation_replays_without_duplicate_dispatch(is_replay):
     now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
+    if is_replay:
+        task["source"]["replay_context"] = {
+            "replay_id": str(uuid4()), "knowledge_cutoff": now.isoformat(), "batch_sequence": 1,
+        }
     repository = RunTaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, {
         OPERATIONAL_ACTION_WORK_ORDER_APPROVE, OPERATIONAL_ACTION_WORK_ORDER_EXECUTE,
@@ -1732,6 +1785,11 @@ async def test_one_click_work_order_confirmation_replays_without_duplicate_dispa
             return ActionDispatchOutcome("unsupported", "测试适配器未配置")
 
     dispatcher = Dispatcher()
+    with pytest.raises(DomainError) as caught:
+        await service.submit_action(task["task_id"], action.action_id,
+            ActionExecuteCommand(request_id=uuid4(), expected_version=action.version), dispatcher, None)
+    assert caught.value.code == "INVALID_TRANSITION"
+    assert dispatcher.calls == 0
     command = WorkOrderConfirmDispatchCommand(request_id=uuid4(), expected_version=action.version)
     first = await service.confirm_work_order_dispatch(task["task_id"], action.action_id,
                                                        command, dispatcher, None)