From 06d8539aa82ecd18ab7ed427c9799cf2ef7dd475 Mon Sep 17 00:00:00 2001 From: Huarch Date: Sat, 7 Nov 2026 21:15:01 +0800 Subject: [PATCH] feat(scada): require complete authorized investigation evidence --- app/mcp/application.py | 4 + app/mcp/catalog.py | 6 + app/mcp/models.py | 16 +- app/mcp/query_service.py | 10 +- app/mcp/scada_readings.py | 121 ++++++++++++ app/services/operational_scada_coverage.py | 40 ++++ app/services/operational_tasks.py | 20 +- docs/mcp-query-implementation.md | 6 +- .../work-order-skills/scada-anomaly/SKILL.md | 8 +- tests/unit/test_complete_scada.py | 179 ++++++++++++++++++ tests/unit/test_mcp_query.py | 28 +++ tests/unit/test_operational_tasks.py | 90 +++++++-- 12 files changed, 498 insertions(+), 30 deletions(-) create mode 100644 app/mcp/scada_readings.py create mode 100644 app/services/operational_scada_coverage.py create mode 100644 tests/unit/test_complete_scada.py diff --git a/app/mcp/application.py b/app/mcp/application.py index bcffd68..2f3430d 100644 --- a/app/mcp/application.py +++ b/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 ): diff --git a/app/mcp/catalog.py b/app/mcp/catalog.py index af1be5a..dadd1d8 100644 --- a/app/mcp/catalog.py +++ b/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条，按需逐页扩大，" diff --git a/app/mcp/models.py b/app/mcp/models.py index 37ac779..49497c3 100644 --- a/app/mcp/models.py +++ b/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) - field: Literal["monitored_value", "cleaned_value"] = "monitored_value" + device_ids: list[Id] | None = Field( + 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): diff --git a/app/mcp/query_service.py b/app/mcp/query_service.py index 624dcf7..151fbcc 100644 --- a/app/mcp/query_service.py +++ b/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) diff --git a/app/mcp/scada_readings.py b/app/mcp/scada_readings.py new file mode 100644 index 0000000..8b7ad2a --- /dev/null +++ b/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" diff --git a/app/services/operational_scada_coverage.py b/app/services/operational_scada_coverage.py new file mode 100644 index 0000000..d3d99c4 --- /dev/null +++ b/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) diff --git a/app/services/operational_tasks.py b/app/services/operational_tasks.py index e76a05b..6b0590b 100644 --- a/app/services/operational_tasks.py +++ b/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: - require(not (task.get("source") or {}).get("replay_context"), "REPLAY_ACTION_DISABLED", - "当前任务的工单和设备动作未开放；请继续完成已授权的定位、地图展示、隔离影响分析与报告", status=409) + def _require_real_action(task: dict, kind: str) -> None: + require(kind == "create_work_order" or not (task.get("source") or {}).get("replay_context"), + "REPLAY_ACTION_DISABLED", + "回放任务不支持设备命令；可生成工单草稿，由有权限的用户确认发布", status=409) @staticmethod def _require_actionable_task(task: dict) -> None: diff --git a/docs/mcp-query-implementation.md b/docs/mcp-query-implementation.md index 2f84ebf..df12fef 100644 --- a/docs/mcp-query-implementation.md +++ b/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的完整样本，预算内仍需读全样本、核验摘要后筛选；此限制不能通过缩小查询窗口解决。监测点列表沿用设备目录，不把未登记的测试点冒充实测设备。 diff --git a/resources/work-order-skills/scada-anomaly/SKILL.md b/resources/work-order-skills/scada-anomaly/SKILL.md index 7c6f4d4..27deaf0 100644 --- a/resources/work-order-skills/scada-anomaly/SKILL.md +++ b/resources/work-order-skills/scada-anomaly/SKILL.md @@ -2,7 +2,7 @@ name: scada-anomaly description: 从 SCADA 指标异常开展数据核验、爆管诊断、候选定位与影响分析。 metadata: - version: "1.0.11" + version: "1.0.12" --- # SCADA 数据异常调查 @@ -15,9 +15,9 @@ metadata: 任一已配置指标异常即进入调查，包含压力、流量、水质规则和压力模型信号。先核验数据质量与测点，再判别原因；模型或组合指标有缺口时，继续核查已发现的单项异常。 -通过 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。 分别说明压力异常模型得分、压力绝对下降及变化率、已配置出厂边界流量变化，再说明联合证据支持程度。使用实际权重、嫌疑门槛、动态阈值、参与测点与各自基线。综合分用于爆管嫌疑判断；得分越低越异常，目标告警与参考日分别解释。水质等其他异常按实际证据选择原因分支。 @@ -42,3 +42,5 @@ metadata: 本轮时序分析保持任务冻结窗口，后续播放形成的观测由相应轮次处理。资产属性、空间查询和拓扑隔离使用项目当前模型，注明模型时态；水力工具按已开放的冻结输入能力调用。定位调用执行权限、设备与冻结观测校验；基线或覆盖缺口按实际返回记录。 报告给出观测事实、规则信号、原因候选、定位与隔离结果、影响说明和建议。摘要以150字以内为目标，测点明细用表格、完整依据留在证据。候选、相似度和拓扑结果具备时，完成相应分析、展示与报告后再收尾；现场核验、审批和维修列为后续待办。现场工单与设备操作遵守服务端动作权限，本轮分析完成与现场事件闭环分别说明。 + +需要现场核验时，先发布本轮outcome=concluded的调查报告，再调用propose_task_action提出kind=create_work_order、work_order_type=general的关联预约工单草稿，引用候选与影响证据，写明核验对象、内容和回执要求。带replay_context的任务同样可提出草稿，描述中注明历史异常时刻及回放来源；草稿须由有权限的用户编辑、确认后发布，后续派发与反馈进入预约任务。草稿待确认时说明等待人工处理，不将调查报告完成写成现场处置完成。回放任务不支持设备命令。 diff --git a/tests/unit/test_complete_scada.py b/tests/unit/test_complete_scada.py new file mode 100644 index 0000000..afb2d05 --- /dev/null +++ b/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" diff --git a/tests/unit/test_mcp_query.py b/tests/unit/test_mcp_query.py index 215a9c9..a37d1c4 100644 --- a/tests/unit/test_mcp_query.py +++ b/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) diff --git a/tests/unit/test_operational_tasks.py b/tests/unit/test_operational_tasks.py index a1a0ac8..e59fb66 100644 --- a/tests/unit/test_operational_tasks.py +++ b/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: - await service.propose_action(task["task_id"], ActionProposalCommand( - request_id=uuid4(), kind="create_work_order", reason="演示不得创建真实工单", - payload={"title": "现场核查", "description": "核查", "work_order_type": "general"}, - )) - assert caught.value.code == "REPLAY_ACTION_DISABLED" + _seed_concluded_report(repository, now) + action = await service.propose_action(task["task_id"], ActionProposalCommand( + request_id=uuid4(), kind="create_work_order", reason="候选管段待现场核验", + evidence_refs=[evidence.evidence_id], + payload={"title": "现场核查", "description": "核查", "work_order_type": "general"}, + )) + 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"]) -async def test_replay_cannot_edit_approve_or_submit_preexisting_real_actions(operation): +@pytest.mark.parametrize("operation", ["propose", "approve", "submit"]) +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": - await service.update_work_order_draft(task["task_id"], uuid4(), WorkOrderDraftUpdateCommand( - **args, payload={"title": "现场核查", "description": "核查", "work_order_type": "general"})) + if operation == "propose": + await service.propose_action(task["task_id"], ActionProposalCommand( + 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)