diff --git a/app/mcp/query_service.py b/app/mcp/query_service.py index 192c2c4..e5878a8 100644 --- a/app/mcp/query_service.py +++ b/app/mcp/query_service.py @@ -394,6 +394,9 @@ async def execute_query( if hasattr(args, "field") and args.field not in units: gaps.append(f"Unit metadata is unavailable for field {args.field}") else: + # Keep the selected field in durable evidence; `value` alone cannot + # distinguish raw observations from cleaned observations. + data = [{**row, args.field: row["value"]} if "value" in row else row for row in data] metadata = [ r for device in ids diff --git a/app/services/scada_presentation_coverage.py b/app/services/scada_presentation_coverage.py index 5bc9d28..ea1d817 100644 --- a/app/services/scada_presentation_coverage.py +++ b/app/services/scada_presentation_coverage.py @@ -77,7 +77,8 @@ class ScadaObservations: if "time" in row and "device_id" in row: at = datetime.fromisoformat(row["time"]) if start <= at <= end: - self.values[(str(row["device_id"]), at)] = row + observed = self.values.setdefault((str(row["device_id"]), at), {}) + observed.update({field: row[field] for field in ("monitored_value", "cleaned_value") if field in row}) def absent_reason(self, device: str, at: datetime) -> str | None: for ids, outside, start, end in self.windows: @@ -206,6 +207,8 @@ def verify_chart(task: dict, chart: dict, evidence: list[dict], refs: list[str]) "历史散点只保留每日末三个对应时刻") continue record = observations.values.get((device, at)) + require(record is None or field in record, "SCADA_CHART_EVIDENCE_REQUIRED", + f"设备 {device} 的证据未标明 {field},请重新读取明确的监测字段", refs=[device]) actual = record.get(field) if record is not None else None if actual is None: absent = observations.absent_reason(device, at) diff --git a/docs/investigation-progress-and-presentation.md b/docs/investigation-progress-and-presentation.md index 2c8c267..f113b19 100644 --- a/docs/investigation-progress-and-presentation.md +++ b/docs/investigation-progress-and-presentation.md @@ -178,8 +178,10 @@ JSON 包含累计 `targets`、按 `chart_id` 保存的最新验证 `charts` 及 历史读取不要求新字段,已完成报告不回写;旧 Agent 的无绑定图表仍可读,但不能通过新轮次的覆盖门禁。 部署按[工作台数据库变更门禁](workbench-database-rollout.md)核实目标 BizDB、备份和回滚演练。 -先停止目标项目的任务写入并安排配套切换,在维护窗口应用 -[035 迁移](../resources/sql/035_scada_presentation_coverage.sql),随后发布 Server、Agent 和前端。 +先部署已兼容内部清单读取的 Server,再暂停目标项目的任务写入并在维护窗口应用 +[035 迁移](../resources/sql/035_scada_presentation_coverage.sql),核对读取与存储就绪后发布 Agent 和前端。 +不要在旧 Server 仍处理请求时新增该列:旧版整行 Run 投影会将内部字段传入严格响应模型。 +新 Server 在迁移前可读取历史任务,SCADA 写入明确返回 503;完成迁移后再恢复目标项目写入。 迁移只为 Run 新增通用 JSON 产物清单,默认空对象;不会自动回填旧图或自动迁移其他项目。 未应用迁移的 SCADA 写入与收尾返回明确的 503,历史读取保持可用。 Run 读取投影显式排除内部清单,REST 与 MCP 原有 Run 结构保持一致。 diff --git a/tests/unit/test_mcp_query.py b/tests/unit/test_mcp_query.py index a37d1c4..ecaeea1 100644 --- a/tests/unit/test_mcp_query.py +++ b/tests/unit/test_mcp_query.py @@ -1116,6 +1116,31 @@ def test_time_series_sql_is_bounded_and_field_names_cannot_inject_sql(): asyncio.run(check()) +@pytest.mark.parametrize("field", ["monitored_value", "cleaned_value"]) +def test_scada_detail_preserves_the_requested_field_with_its_value_alias(monkeypatch, field): + at = datetime.now(timezone.utc) + monkeypatch.setattr(query_service, "permissions_for_context", lambda ctx: PERMISSIONS) + monkeypatch.setattr(query_service.reads, "series_rows", AsyncMock(return_value=[{ + "device_id": "D1", "time": at, "value": 32.0, + }])) + monkeypatch.setattr(query_service.reads, "scada_devices", AsyncMock(return_value=[{ + "device_id": "D1", "measurement_unit": "m", + }])) + + class Scope: + project = object() + connection = Connection() + + @asynccontextmanager + async def timescale(self): + yield self.connection + + args = ScadaInput(context_handle="a" * 43, device_ids=["D1"], field=field, + start=at - timedelta(hours=3), end=at) + result = asyncio.run(query_service.execute_query(Scope(), binding(), TOOLS["query_scada_readings"], args)) + assert result.data[0][field] == result.data[0]["value"] == 32.0 + + def test_analysis_preserves_run_units_and_reports_inference(monkeypatch): from app.mcp.models import AnalysisValuesInput from unittest.mock import AsyncMock diff --git a/tests/unit/test_scada_presentation_coverage.py b/tests/unit/test_scada_presentation_coverage.py index 2014324..4854b03 100644 --- a/tests/unit/test_scada_presentation_coverage.py +++ b/tests/unit/test_scada_presentation_coverage.py @@ -1,5 +1,6 @@ """SCADA display coverage is independent of whether a device supports a conclusion.""" from datetime import datetime, timedelta, timezone +from copy import deepcopy from uuid import uuid4 import pytest @@ -16,6 +17,44 @@ END = datetime(2026, 7, 16, tzinfo=timezone.utc) START = END - timedelta(hours=3) +@pytest.mark.anyio +async def test_value_only_detail_evidence_cannot_claim_missing_observations(): + from app.services.scada_presentation_coverage import verify_chart + task, repository, _, evidence_id = await investigation() + detail = deepcopy(repository.evidence[evidence_id]) + detail["evidence_id"] = "detail-values" + rows = detail["data_snapshot"]["data"]["data"]["readings"] + detail["data_snapshot"]["data"]["data"] = [ + {"device_id": row["device_id"], "time": row["time"], "value": row["monitored_value"]} + for row in rows + ] + value = chart(devices=["524529"]) + value["series"][0].update(data=[None] * 13, sample_times=[None] * 13) + with pytest.raises(DomainError) as failure: + verify_chart(task, value, [repository.evidence[evidence_id], detail], ["detail-values"]) + assert failure.value.code == "SCADA_CHART_EVIDENCE_REQUIRED" + + +@pytest.mark.anyio +@pytest.mark.parametrize("field", ["monitored_value", "cleaned_value"]) +async def test_detail_fields_do_not_discard_other_observed_fields(field): + from app.services.scada_presentation_coverage import verify_chart + task, repository, _, evidence_id = await investigation() + detail = deepcopy(repository.evidence[evidence_id]) + detail["evidence_id"] = "detail-cleaned" + rows = detail["data_snapshot"]["data"]["data"]["readings"] + detail["data_snapshot"]["data"]["data"] = [ + {"device_id": row["device_id"], "time": row["time"], "value": row["cleaned_value"] + 10, + "cleaned_value": row["cleaned_value"] + 10} for row in rows + ] + value = chart(devices=["524529"]) + value["scada"]["value_field"] = field + if field == "cleaned_value": + value["series"][0]["data"] = [point + 10 for point in value["series"][0]["data"]] + result = verify_chart(task, value, [repository.evidence[evidence_id], detail], [evidence_id, "detail-cleaned"]) + assert result["gaps"] == [] + + @pytest.fixture def anyio_backend(): return "asyncio"