From 2e8fe86f3fb864bbacf6ce0a0ba4dfb182ab5c56 Mon Sep 17 00:00:00 2001 From: Huarch Date: Thu, 12 Nov 2026 10:38:55 +0800 Subject: [PATCH] fix: enforce SCADA chart device coverage --- app/domain/schemas/operational_task.py | 21 + app/infra/db/postgresql/operational_tasks.py | 34 +- app/infra/db/postgresql/read_queries.py | 2 +- app/mcp/scada_readings.py | 2 + app/services/operational_tasks.py | 65 ++- app/services/scada_presentation_coverage.py | 265 +++++++++++ app/services/time_comparison_chart.py | 15 +- contracts/manifest.json | 2 +- contracts/server-v1.openapi.json | 174 +++++++ contracts/view-command.schema.json | 101 ++++ ...investigation-progress-and-presentation.md | 36 ++ .../sql/035_scada_presentation_coverage.sql | 14 + .../work-order-skills/scada-anomaly/SKILL.md | 8 +- tests/support/operational_task_repository.py | 7 + tests/unit/test_complete_scada.py | 1 + ...test_operational_task_storage_readiness.py | 54 +++ .../unit/test_scada_presentation_coverage.py | 432 ++++++++++++++++++ 17 files changed, 1220 insertions(+), 13 deletions(-) create mode 100644 app/services/scada_presentation_coverage.py create mode 100644 resources/sql/035_scada_presentation_coverage.sql create mode 100644 tests/unit/test_scada_presentation_coverage.py diff --git a/app/domain/schemas/operational_task.py b/app/domain/schemas/operational_task.py index 66830be..927e248 100644 --- a/app/domain/schemas/operational_task.py +++ b/app/domain/schemas/operational_task.py @@ -320,6 +320,7 @@ class InvestigationReport(StrictModel): data_gaps: list[InvestigationDataGap] | None = None diagram: InvestigationDiagram | None = None ui_blocks: list["ReportUiBlock"] = Field(default_factory=list, max_length=8) + scada_device_ids: list[TaskFeatureId] | None = Field(default=None, max_length=1000) @model_validator(mode="after") def validate_result(self): @@ -742,8 +743,27 @@ class AgentArtifactsResult(StrictModel): progress_revision: PositiveInt | None = None +class ScadaChartGap(StrictModel): + device_id: TaskFeatureId + day_offset: Annotated[int, Field(ge=0, le=7)] + missing_count: Annotated[int, Field(ge=1, le=100)] + reason: Literal["no_readings", "outside_frozen_scope", "outside_frozen_window", "no_scheduled_samples"] + + +class ScadaChartContext(StrictModel): + role: Literal["current", "comparison"] + metric: Literal["pressure", "flow"] + unit: Annotated[str, Field(min_length=1, max_length=128)] + window_start: WorkbenchDateTime + window_end: WorkbenchDateTime + value_field: Literal["monitored_value", "cleaned_value"] = "monitored_value" + gaps: list[ScadaChartGap] = Field(default_factory=list, max_length=8) + + class TaskSafeChartSeries(StrictModel): name: Annotated[str, Field(max_length=128)] + device_id: TaskFeatureId | None = None + day_offset: Annotated[int, Field(ge=0, le=7)] | None = None type: Literal["line", "bar", "scatter"] | None = None data: Annotated[list[Annotated[float, Field(strict=True, allow_inf_nan=False)] | None], Field(min_length=1, max_length=100)] sample_times: Annotated[list[Annotated[str, Field( @@ -770,6 +790,7 @@ class TaskSafeChartSeries(StrictModel): class TaskSafeChart(StrictModel): title: Annotated[str, Field(max_length=200)] + scada: ScadaChartContext | None = None chart_type: Literal["line", "bar", "time_comparison"] x_axis_name: Annotated[str, Field(max_length=128)] | None = None y_axis_name: Annotated[str, Field(max_length=128)] | None = None diff --git a/app/infra/db/postgresql/operational_tasks.py b/app/infra/db/postgresql/operational_tasks.py index 3a48cc9..d000482 100644 --- a/app/infra/db/postgresql/operational_tasks.py +++ b/app/infra/db/postgresql/operational_tasks.py @@ -164,7 +164,7 @@ class OperationalTaskRepository: ORDER BY generation DESC LIMIT %s OFFSET %s""", (self.project_id, task_id, limit, offset), ) - return [{key: value for key, value in row.items() if key != "project_id"} + return [{key: value for key, value in row.items() if key not in {"project_id", "artifact_manifest"}} for row in await cursor.fetchall()] async def get_latest_run(self, task_id: str) -> dict | None: @@ -371,6 +371,34 @@ class OperationalTaskRepository: ) return [dict(row) for row in await cursor.fetchall()] + async def get_scada_presentation(self, task_id: str, run_id: str) -> dict | None: + cursor = await self.connection.execute( + """SELECT EXISTS (SELECT 1 FROM pg_attribute + WHERE attrelid=to_regclass('workbench.operational_task_runs') + AND attname='artifact_manifest' AND NOT attisdropped) AS available""" + ) + if not (await cursor.fetchone())["available"]: + raise DomainError("DEPENDENCY_UNAVAILABLE", + "SCADA display coverage requires resources/sql/035_scada_presentation_coverage.sql in this project database", + status=503) + cursor = await self.connection.execute( + """SELECT artifact_manifest->'scada_presentation' AS document FROM workbench.operational_task_runs + WHERE project_id=%s AND task_id=%s AND run_id=%s""", + (self.project_id, task_id, run_id), + ) + row = await cursor.fetchone() + return row["document"] if row else None + + async def save_scada_presentation(self, task_id: str, run_id: str, document: dict, now: datetime) -> None: + cursor = await self.connection.execute( + """UPDATE workbench.operational_task_runs + SET artifact_manifest=jsonb_set(artifact_manifest,'{scada_presentation}',%s::jsonb,true) + WHERE project_id=%s AND task_id=%s AND run_id=%s""", + (Jsonb({**document, "updated_at": now.isoformat()}), self.project_id, task_id, run_id), + ) + if cursor.rowcount != 1: + raise DomainError("RUN_NOT_FOUND", "SCADA display coverage targets an unknown run", status=409) + async def get_task_by_create_request(self, request_id: str) -> dict | None: cursor = await self.connection.execute( """SELECT document FROM workbench.operational_tasks @@ -405,9 +433,9 @@ class OperationalTaskRepository: cursor = await self.connection.execute( """SELECT task.document, CASE WHEN active_run.run_id IS NULL THEN NULL - ELSE to_jsonb(active_run) - 'project_id' END AS active_run, + ELSE to_jsonb(active_run) - 'project_id' - 'artifact_manifest' END AS active_run, CASE WHEN latest_run.run_id IS NULL THEN NULL - ELSE to_jsonb(latest_run) - 'project_id' END AS latest_run + ELSE to_jsonb(latest_run) - 'project_id' - 'artifact_manifest' END AS latest_run FROM workbench.operational_tasks AS task LEFT JOIN LATERAL ( SELECT run.* FROM workbench.operational_task_runs AS run diff --git a/app/infra/db/postgresql/read_queries.py b/app/infra/db/postgresql/read_queries.py index 3e71499..16b657f 100644 --- a/app/infra/db/postgresql/read_queries.py +++ b/app/infra/db/postgresql/read_queries.py @@ -404,7 +404,7 @@ async def task_context_page( where = " ORDER BY revision DESC LIMIT 1" stored = f"SELECT jsonb_build_object('section',%s::text,'data',document) AS snapshot FROM {relation} WHERE project_id=%s AND task_id=%s{where}" elif section == "run": - stored = """SELECT jsonb_build_object('section','run','data',to_jsonb(r)-'project_id') AS snapshot + stored = """SELECT jsonb_build_object('section','run','data',to_jsonb(r)-'project_id'-'artifact_manifest') AS snapshot FROM workbench.operational_task_runs r WHERE project_id=%s AND task_id=%s ORDER BY generation DESC LIMIT 1""" params = (project_id, task_id) elif section == "progress": diff --git a/app/mcp/scada_readings.py b/app/mcp/scada_readings.py index 8b7ad2a..fbc5d98 100644 --- a/app/mcp/scada_readings.py +++ b/app/mcp/scada_readings.py @@ -109,6 +109,8 @@ async def read_complete_scada( "replay_id": context["replay_id"] if context else None, "batch_sequence": context["batch_sequence"] if context else None, } + if context: + coverage["frozen_start"] = (run.get("baseline") or {}).get("start_time") or run["start_time"] data = {"devices": devices, "readings": rows, "coverage": coverage} account([coverage]) gaps = [] diff --git a/app/services/operational_tasks.py b/app/services/operational_tasks.py index a8e4e93..3a23a5d 100644 --- a/app/services/operational_tasks.py +++ b/app/services/operational_tasks.py @@ -77,6 +77,7 @@ from app.services.investigation_graph import adopt_legacy_investigation_graph, u 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 +from app.services import scada_presentation_coverage from app.services.time_comparison_chart import validate_time_comparison_chart @@ -955,6 +956,7 @@ class OperationalTaskService: require_complete_scada_evidence( task, await self.repo.list_evidence(str(task_id)), report_input.evidence_refs, ) + await self._verify_scada_report_presentation(task, active_run, _json(report_input), now) for gap in report_input.data_gaps: require(set(gap.evidence_refs) <= set(report_input.evidence_refs), "REPORT_GAP_EVIDENCE_INVALID", @@ -994,7 +996,7 @@ class OperationalTaskService: or previous_result.get("stage") != "final" or any(previous_result.get(key, [] if key == "ui_blocks" else None) != effective_report.get(key) for key in ("current_conclusion", "conclusion_boundary", "outcome", "stop_reason", - "evidence_refs", "recommendations", "pending_items", "data_gaps", "ui_blocks")) + "evidence_refs", "recommendations", "pending_items", "data_gaps", "ui_blocks", "scada_device_ids")) ): await self._resolve_notifications( task_id, kinds=["report_updated"], subject_type="task", subject_id=task_id, @@ -1092,8 +1094,11 @@ class OperationalTaskService: now = self._now() self._require_active_run_lease(run, now) payload = _json(command.command) + await self._record_scada_presentation(task, run, payload, now) if command.command.action == "open_chart": - validate_time_comparison_chart(payload["chart"]) + validate_time_comparison_chart(payload["chart"], allow_missing_current=bool( + payload["chart"].get("scada") and scada_presentation_coverage.applies(task), + )) known = {item["evidence_id"] for item in await self.repo.list_evidence(str(task_id))} self._require_evidence_refs(command.command.evidence_refs, known) elif command.command.action == "set_overlay": @@ -1131,6 +1136,61 @@ class OperationalTaskService: await self._save_request(scope, command.request_id, body, _json(result)) return result + async def _record_scada_presentation(self, task: dict, run: dict, payload: dict, now: datetime) -> None: + if not scada_presentation_coverage.applies(task): + return + action = payload["action"] + targets = [] + if action == "render_scada_analysis": + targets = [item["device_id"] for item in payload["items"]] + elif action in {"locate_features", "highlight_features"} and payload["layer"] == "scada": + targets = payload["feature_ids"] + elif action != "open_chart": + return + evidence = await self.repo.list_evidence(task["task_id"]) + observations = scada_presentation_coverage.ScadaObservations(task, evidence) + document = await self.repo.get_scada_presentation(task["task_id"], str(run["run_id"])) + if action == "open_chart": + if payload["chart"].get("scada"): + known = {item["evidence_id"] for item in evidence} + self._require_evidence_refs(payload["evidence_refs"], known) + chart = scada_presentation_coverage.verify_chart(task, payload["chart"], evidence, payload["evidence_refs"]) + targets = chart["device_ids"] + duplicates = [chart_id for chart_id, stored in (document or {}).get("charts", {}).items() + if chart_id != payload["chart_id"] and stored["role"] == chart["role"] == "comparison" + and any(set(stored.get("coverage", {}).get(device, {}).get("covered", [])) + & set(chart["coverage"][device]["covered"]) + for device in set(stored["device_ids"]) & set(targets))] + require(not duplicates, "SCADA_CHART_DUPLICATE_DEVICE", + f"设备同期图已存在，请更新原chart_id：{', '.join(duplicates)}", refs=duplicates) + document = scada_presentation_coverage.add_targets(document, targets, observations) + document["charts"][payload["chart_id"]] = chart + elif document and payload["chart_id"] in document["charts"]: + # Replacing a typed chart with a generic one removes its coverage, + # while cumulative targets prevent shrinking the required set. + del document["charts"][payload["chart_id"]] + else: + return + else: + document = scada_presentation_coverage.add_targets(document, targets, observations) + await self.repo.save_scada_presentation(task["task_id"], str(run["run_id"]), document, now) + + async def _verify_scada_report_presentation(self, task: dict, run: dict, report: dict, now: datetime) -> None: + if not scada_presentation_coverage.applies(task): + return + if report.get("stage") != "final" and not report.get("scada_device_ids"): + return + evidence = await self.repo.list_evidence(task["task_id"]) + observations = scada_presentation_coverage.ScadaObservations(task, evidence) + document = await self.repo.get_scada_presentation(task["task_id"], str(run["run_id"])) + targets = list(report.get("scada_device_ids") or []) + if targets: + document = scada_presentation_coverage.add_targets(document, targets, observations) + if report.get("stage") == "final": + scada_presentation_coverage.require_complete(task, document, observations) + if document: + await self.repo.save_scada_presentation(task["task_id"], str(run["run_id"]), document, now) + async def propose_action( self, task_id: UUID, command: ActionProposalCommand ) -> HumanApprovalAction: @@ -2049,6 +2109,7 @@ class OperationalTaskService: "Run requires a final report from the same generation" ) else: + await self._verify_scada_report_presentation(task, run, _json(report.report), now) actions = [ HumanApprovalAction.model_validate(item) for item in await self.repo.list_actions(str(task_id)) diff --git a/app/services/scada_presentation_coverage.py b/app/services/scada_presentation_coverage.py new file mode 100644 index 0000000..5bc9d28 --- /dev/null +++ b/app/services/scada_presentation_coverage.py @@ -0,0 +1,265 @@ +"""Verify monitoring charts against observations and require run-local device coverage.""" +from __future__ import annotations + +from copy import deepcopy +from datetime import datetime, timedelta, timezone +from math import isclose +import re + +from app.domain.domain_errors import require + + +DISPLAY_TIMEZONE = timezone(timedelta(hours=8)) +METRICS = {"pressure": "pressure", "pipe_flow": "flow", "flow": "flow"} + + +def applies(task: dict) -> bool: + if task.get("workflow") != "scada_anomaly" and not task.get("source", {}).get("replay_context"): + return False + # Recovery has its own Server-verified post-repair evidence and finalization gate. + return (task.get("handling_context") or {}).get("phase") not in {"verification_ready", "completed"} + + +def cutoff(task: dict) -> datetime: + replay = task.get("source", {}).get("replay_context") or {} + return datetime.fromisoformat(replay.get("knowledge_cutoff") or task["trigger"]["occurred_at"]) + + +def _unit(value: str) -> str: + return value.replace("³", "3").replace(" ", "").lower() + + +class ScadaObservations: + def __init__(self, task: dict, evidence: list[dict], refs: list[str] | None = None): + self.devices: dict[str, dict] = {} + self.values: dict[tuple[str, datetime], dict] = {} + self.windows: list[tuple[set[str], set[str], datetime, datetime]] = [] + self.frozen_limits: list[tuple[set[str], datetime]] = [] + target = cutoff(task) + replay = task.get("source", {}).get("replay_context") or {} + for item in evidence: + snapshot = item.get("data_snapshot") or {} + result = snapshot.get("data") or {} + if snapshot.get("source") != "query_scada_readings" or not isinstance(result, dict): + continue + data = result.get("data") + coverage = data.get("coverage", {}) if isinstance(data, dict) else {} + if replay and coverage and (coverage.get("replay_id") != replay["replay_id"] + or coverage.get("batch_sequence") != replay["batch_sequence"]): + 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 end > target: + continue + devices = data.get("devices", []) if isinstance(data, dict) else [] + for device in devices: + self.devices[str(device["device_id"])] = device + if refs is not None and item["evidence_id"] not in refs: + continue + rows = data.get("readings", []) if isinstance(data, dict) else data + if not isinstance(rows, list): + continue + ids = {str(row["device_id"]) for row in rows if "device_id" in row} + outside = set(coverage.get("outside_frozen_scope_device_ids", [])) + if coverage.get("query_complete") is True: + ids.update(coverage.get("queried_device_ids", [])) + ids.update(str(device["device_id"]) for device in devices) + page = result.get("page") or {} + complete = not (snapshot.get("truncated") or result.get("truncated") + or page.get("has_more") or page.get("offset", 0)) + if complete: + self.windows.append((ids, outside, start, end)) + if replay and coverage.get("query_complete") and coverage.get("frozen_start"): + self.frozen_limits.append((ids, datetime.fromisoformat(coverage["frozen_start"]))) + for row in rows: + 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 + + def absent_reason(self, device: str, at: datetime) -> str | None: + for ids, outside, start, end in self.windows: + if device in ids and start <= at <= end: + return "outside_frozen_scope" if device in outside else "no_readings" + for ids, frozen_start in self.frozen_limits: + if device in ids and at < frozen_start: + return "outside_frozen_window" + return None + + def window_is_empty(self, device: str, start: datetime, end: datetime) -> bool: + if any(key == device and start <= at <= end for key, at in self.values): + return False + intervals = [(left, right) for ids, _, left, right in self.windows if device in ids] + intervals.extend((start, frozen_start) for ids, frozen_start in self.frozen_limits if device in ids) + through = start + for left, right in sorted(intervals): + if left > through: + break + through = max(through, right) + if through >= end: + return True + return False + + def period(self, device: str, start: datetime, end: datetime) -> timedelta: + frequency = str(self.devices[device].get("transmission_frequency") or "") + match = re.fullmatch(r"(\d+):([0-5]\d):([0-5]\d)", frequency) + if match: + hours, minutes, seconds = map(int, match.groups()) + period = timedelta(hours=hours, minutes=minutes, seconds=seconds) + else: + minutes = re.fullmatch(r"(\d+)\s*(?:min|分钟)", frequency) + times = sorted(at for key, at in self.values if key == device and start <= at <= end) + period = (timedelta(minutes=int(minutes[1])) if minutes else + min((right - left for left, right in zip(times, times[1:])), default=timedelta(0))) + require(period > timedelta(0), "SCADA_CHART_EVIDENCE_REQUIRED", + f"设备 {device} 缺少可核实的采样周期，请读取完整设备属性", refs=[device]) + return period + + def slots(self, device: str, start: datetime, end: datetime) -> list[datetime]: + """Keep the business bounds exact, aligning slots to the observed sampling phase.""" + period = self.period(device, start, end) + observed = sorted(at for key, at in self.values if key == device and start <= at <= end) + historical = sorted(at for key, at in self.values if key == device and at < start) + # Referenced historical observations preserve phase even if current data is absent. + anchor = (observed[0] if observed else historical[-1] if historical else + start.astimezone(DISPLAY_TIMEZONE).replace(hour=0, minute=0, second=0, microsecond=0)) + first = start + (anchor - start) % period + return [first + period * index for index in range(int((end - first) // period) + 1)] + + +def add_targets(document: dict | None, devices: list[str], observations: ScadaObservations) -> dict: + result = deepcopy(document) if document else {"version": 1, "targets": [], "charts": {}} + targets = set(result["targets"]) + for device in devices: + require(device in observations.devices, "SCADA_CHART_EVIDENCE_REQUIRED", + f"设备 {device} 未出现在本轮截止时间内的完整设备证据中", refs=[device]) + if observations.devices[device].get("device_type") in METRICS: + targets.add(device) + result["targets"] = sorted(targets) + return result + + +def verify_chart(task: dict, chart: dict, evidence: list[dict], refs: list[str]) -> dict: + """Return verified coverage; mutate only the outgoing chart's Server-derived gaps.""" + context = chart["scada"] + end = cutoff(task) + start = end - timedelta(hours=3) + require(datetime.fromisoformat(context["window_start"]) == start + and datetime.fromisoformat(context["window_end"]) == end, + "SCADA_CHART_WINDOW_INVALID", "SCADA图表必须使用本轮截止时刻之前的完整三小时窗口") + observations = ScadaObservations(task, evidence, refs) + series = chart["series"] + ids = {item.get("device_id") for item in series} + require(None not in ids and all(device in observations.devices for device in ids), + "SCADA_CHART_EVIDENCE_REQUIRED", "每条SCADA序列须绑定已核实的 device_id", refs=sorted(ids - {None})) + comparison = context["role"] == "comparison" + require(chart["chart_type"] == ("time_comparison" if comparison else "line"), + "SCADA_CHART_INVALID", "SCADA总览须使用line，同期图须使用time_comparison") + bindings = [(item["device_id"], item.get("day_offset")) for item in series] + require(len(bindings) == len(set(bindings)), "SCADA_CHART_INVALID", "SCADA序列的设备与日期绑定不能重复") + require((len(ids) == 1 and {day for _, day in bindings} == set(range(8))) if comparison + else all(day == 0 for _, day in bindings), + "SCADA_CHART_INVALID", "总览每个设备绑定day_offset=0；同期图每个设备须有当天及此前七天共8组序列") + field = context.get("value_field", "monitored_value") + gaps = [] + coverage = {} + for item in series: + device, day = item["device_id"], item["day_offset"] + metadata = observations.devices[device] + require(METRICS.get(metadata.get("device_type")) == context["metric"] + and _unit(metadata.get("measurement_unit") or "") == _unit(context["unit"]), + "SCADA_CHART_INVALID", f"设备 {device} 的指标或单位与完整设备证据不一致", refs=[device]) + period = observations.period(device, start, end) + full_slots = observations.slots(device, start, end) + no_schedule = not full_slots + if no_schedule: + require(observations.window_is_empty(device, start - timedelta(days=day), end - timedelta(days=day)), + "SCADA_CHART_EVIDENCE_REQUIRED", f"设备 {device} 无计划采样槽，须完整读取对应三小时窗口并证明无读数", refs=[device]) + # This is an empty display entry, never an observation or a missed scheduled sample. + full_slots = [end] + labels = [at.astimezone(DISPLAY_TIMEZONE).strftime("%H:%M:%S" if period.total_seconds() % 60 else "%H:%M") + for at in full_slots] + first = labels.index(chart["x_data"][0]) if chart["x_data"][0] in labels else -1 + count = len(chart["x_data"]) + require(first >= 0 and chart["x_data"] == labels[first:first + count] + and len(item.get("sample_times") or []) == count, + "SCADA_CHART_INVALID", f"设备 {device} 须保留连续真实采样槽和sample_times；超限按时间分图", refs=[device]) + slots = full_slots[first:first + count] + coverage[device] = {"covered": [at.isoformat() for at in slots]} + require((item.get("type") or "line") == ("scatter" if day else "line"), + "SCADA_CHART_INVALID", "当前序列须为line，历史序列须为scatter") + missing = 0 + reason = "no_readings" + expected = {at - timedelta(days=day) for at in (full_slots[-3:] if day else full_slots)} + # Off-grid observations cannot be silently discarded by a regular-looking chart. + observed = {at for key, at in observations.values if key == device + and min(expected) <= at <= max(expected)} + require(observed <= expected, "SCADA_CHART_INVALID", + f"设备 {device} 含非规则采样时刻，须明确处理，不能静默丢点", refs=[device]) + for index, slot in enumerate(slots): + at = slot - timedelta(days=day) + value, sample_time = item["data"][index], item["sample_times"][index] + if at not in expected: + require(value is None and sample_time is None, "SCADA_CHART_INVALID", + "历史散点只保留每日末三个对应时刻") + continue + record = observations.values.get((device, at)) + actual = record.get(field) if record is not None else None + if actual is None: + absent = observations.absent_reason(device, at) + require(record is not None or absent is not None, "SCADA_CHART_EVIDENCE_REQUIRED", + f"设备 {device} 的 {at.isoformat()} 未被完整读取，不能将未查询当作缺测", refs=[device]) + require(value is None and sample_time is None, "SCADA_CHART_DATA_MISMATCH", + f"设备 {device} 缺测槽必须保留null，不能补零或推算", refs=[device]) + missing += 1 + if absent in {"outside_frozen_scope", "outside_frozen_window"}: + reason = absent + elif no_schedule: + reason = "no_scheduled_samples" + else: + require(value is not None and sample_time is not None + and datetime.fromisoformat(sample_time) == at + and (isclose(value, actual, rel_tol=1e-9, abs_tol=1e-9) + or any(isclose(value, round(actual, digits), rel_tol=0, abs_tol=1e-9) + for digits in range(1, 7))), + "SCADA_CHART_DATA_MISMATCH", + f"设备 {device} 的 {at.isoformat()} 图表值或采样时间与证据不符", refs=[device]) + if missing: + gaps.append({"device_id": device, "day_offset": day, "missing_count": missing, "reason": reason}) + context["gaps"] = gaps + return {"role": context["role"], "metric": context["metric"], "device_ids": sorted(ids), + "window_start": context["window_start"], "window_end": context["window_end"], "gaps": gaps, + "coverage": coverage} + + +def require_complete(task: dict, document: dict | None, observations: ScadaObservations) -> None: + targets = set((document or {}).get("targets", [])) + supported = {key for key, device in observations.devices.items() if device.get("device_type") in METRICS} + if not supported and not targets: + return + require(bool(targets), "SCADA_PRESENTATION_INCOMPLETE", + "请声明本轮SCADA目标设备并发布当前总览与七天同期图；旧格式图表不计入设备覆盖", status=409) + end = cutoff(task) + start = end - timedelta(hours=3) + covered = {"current": {}, "comparison": {}} + for chart in document["charts"].values(): + if (datetime.fromisoformat(chart["window_start"]) == start + and datetime.fromisoformat(chart["window_end"]) == end): + for device, slots in chart.get("coverage", {}).items(): + if device not in chart["device_ids"]: + continue + covered[chart["role"]].setdefault(device, set()).update( + datetime.fromisoformat(at) for at in slots["covered"]) + missing = {} + for role in covered: + missing[role] = sorted(device for device in targets + if not covered[role].get(device) or not set(observations.slots(device, start, end) or [end]) + <= covered[role][device]) + missing_current, missing_comparison = missing["current"], missing["comparison"] + require(not missing_current and not missing_comparison, "SCADA_PRESENTATION_INCOMPLETE", + f"本轮SCADA目标共{len(targets)}个；缺少当前总览设备：{', '.join(missing_current) or '无'}；" + f"缺少七天同期对比设备：{', '.join(missing_comparison) or '无'}。请补图后重试；仅作参考不能豁免展示。", + refs=sorted(set(missing_current + missing_comparison)), status=409, + recovery="retry_same_request") diff --git a/app/services/time_comparison_chart.py b/app/services/time_comparison_chart.py index b9074d3..c2379cf 100644 --- a/app/services/time_comparison_chart.py +++ b/app/services/time_comparison_chart.py @@ -13,7 +13,7 @@ DISPLAY_TIMEZONE = timezone(timedelta(hours=8)) DAY_SECONDS = 86400 -def validate_time_comparison_chart(chart: Mapping[str, Any]) -> None: +def validate_time_comparison_chart(chart: Mapping[str, Any], *, allow_missing_current: bool = False) -> None: """Validate new publications; persisted historical charts keep their read contract.""" chart_type = chart["chart_type"] series = chart["series"] @@ -70,8 +70,17 @@ def validate_time_comparison_chart(chart: Mapping[str, Any]) -> None: current_bases.update(bases) else: historical_bases.extend(bases) - check(len(current_bases) == 1, "当前 line 必须包含真实采样时间，且所有当前曲线使用同一窗口") - current = next(iter(current_bases)) + scada = chart.get("scada") + empty_scada = allow_missing_current and not current_bases and scada and scada["role"] == "comparison" + check(len(current_bases) == 1 or bool(empty_scada), + "当前 line 必须包含真实采样时间，且所有当前曲线使用同一窗口") + # SCADA publication separately verifies every null against persisted reads. + if empty_scada: + window_start = datetime.fromisoformat(scada["window_start"]).astimezone(DISPLAY_TIMEZONE) + start_seconds = window_start.hour * 3600 + window_start.minute * 60 + window_start.second + current = window_start.replace(microsecond=0) + timedelta(seconds=(seconds[0] - start_seconds) % DAY_SECONDS) + else: + current = next(iter(current_bases)) for historical in historical_bases: delta = (current - historical).total_seconds() check(delta > 0 and delta % DAY_SECONDS == 0, diff --git a/contracts/manifest.json b/contracts/manifest.json index e930bd3..7f9dc13 100644 --- a/contracts/manifest.json +++ b/contracts/manifest.json @@ -3,7 +3,7 @@ "contracts": { "server": { "file": "server-v1.openapi.json", - "sha256": "7ed3f6485909c3ef7fac8d2083d6931765e4d197e690c3b9914b905a8aa743b2" + "sha256": "d96df154b86b797d0db913d1e69cf335d92d34dd119f800c9346b90c6119c9bb" } } } diff --git a/contracts/server-v1.openapi.json b/contracts/server-v1.openapi.json index 5d5ea86..c30012b 100644 --- a/contracts/server-v1.openapi.json +++ b/contracts/server-v1.openapi.json @@ -4032,6 +4032,23 @@ "title": "Recommendations", "type": "array" }, + "scada_device_ids": { + "anyOf": [ + { + "items": { + "maxLength": 128, + "minLength": 1, + "type": "string" + }, + "maxItems": 1000, + "type": "array" + }, + { + "type": "null" + } + ], + "title": "Scada Device Ids" + }, "stage": { "anyOf": [ { @@ -4144,6 +4161,23 @@ "title": "Recommendations", "type": "array" }, + "scada_device_ids": { + "anyOf": [ + { + "items": { + "maxLength": 128, + "minLength": 1, + "type": "string" + }, + "maxItems": 1000, + "type": "array" + }, + { + "type": "null" + } + ], + "title": "Scada Device Ids" + }, "stage": { "enum": [ "interim", @@ -7383,6 +7417,110 @@ "title": "RunSimulationManuallyByDateRest", "type": "object" }, + "ScadaChartContext": { + "additionalProperties": false, + "properties": { + "gaps": { + "items": { + "$ref": "#/components/schemas/ScadaChartGap" + }, + "maxItems": 8, + "title": "Gaps", + "type": "array" + }, + "metric": { + "enum": [ + "pressure", + "flow" + ], + "title": "Metric", + "type": "string" + }, + "role": { + "enum": [ + "current", + "comparison" + ], + "title": "Role", + "type": "string" + }, + "unit": { + "maxLength": 128, + "minLength": 1, + "title": "Unit", + "type": "string" + }, + "value_field": { + "default": "monitored_value", + "enum": [ + "monitored_value", + "cleaned_value" + ], + "title": "Value Field", + "type": "string" + }, + "window_end": { + "format": "date-time", + "title": "Window End", + "type": "string" + }, + "window_start": { + "format": "date-time", + "title": "Window Start", + "type": "string" + } + }, + "required": [ + "role", + "metric", + "unit", + "window_start", + "window_end" + ], + "title": "ScadaChartContext", + "type": "object" + }, + "ScadaChartGap": { + "additionalProperties": false, + "properties": { + "day_offset": { + "maximum": 7.0, + "minimum": 0.0, + "title": "Day Offset", + "type": "integer" + }, + "device_id": { + "maxLength": 128, + "minLength": 1, + "title": "Device Id", + "type": "string" + }, + "missing_count": { + "maximum": 100.0, + "minimum": 1.0, + "title": "Missing Count", + "type": "integer" + }, + "reason": { + "enum": [ + "no_readings", + "outside_frozen_scope", + "outside_frozen_window", + "no_scheduled_samples" + ], + "title": "Reason", + "type": "string" + } + }, + "required": [ + "device_id", + "day_offset", + "missing_count", + "reason" + ], + "title": "ScadaChartGap", + "type": "object" + }, "ScadaDeviceResponse": { "description": "Project SCADA metadata keyed by the canonical device identifier.", "properties": { @@ -10450,6 +10588,16 @@ "title": "Chart Type", "type": "string" }, + "scada": { + "anyOf": [ + { + "$ref": "#/components/schemas/ScadaChartContext" + }, + { + "type": "null" + } + ] + }, "series": { "items": { "$ref": "#/components/schemas/TaskSafeChartSeries" @@ -10527,6 +10675,32 @@ "title": "Data", "type": "array" }, + "day_offset": { + "anyOf": [ + { + "maximum": 7.0, + "minimum": 0.0, + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Day Offset" + }, + "device_id": { + "anyOf": [ + { + "maxLength": 128, + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Device Id" + }, "name": { "maxLength": 128, "title": "Name", diff --git a/contracts/view-command.schema.json b/contracts/view-command.schema.json index 8064e81..e1f8dc9 100644 --- a/contracts/view-command.schema.json +++ b/contracts/view-command.schema.json @@ -207,6 +207,97 @@ "type": "string", "maxLength": 200 }, + "scada": { + "type": "object", + "properties": { + "role": { + "type": "string", + "enum": [ + "current", + "comparison" + ] + }, + "metric": { + "type": "string", + "enum": [ + "pressure", + "flow" + ] + }, + "unit": { + "type": "string", + "minLength": 1, + "maxLength": 128 + }, + "window_start": { + "type": "string", + "maxLength": 64, + "format": "date-time", + "pattern": "^(?:[1-9][0-9]{3}|0[1-9][0-9]{2}|00[1-9][0-9]|000[1-9])-[0-9]{2}-[0-9]{2}T(?:[01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9](?:\\.[0-9]+)?(?:Z|[+-](?:[01][0-9]|2[0-3]):[0-5][0-9])$" + }, + "window_end": { + "type": "string", + "maxLength": 64, + "format": "date-time", + "pattern": "^(?:[1-9][0-9]{3}|0[1-9][0-9]{2}|00[1-9][0-9]|000[1-9])-[0-9]{2}-[0-9]{2}T(?:[01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9](?:\\.[0-9]+)?(?:Z|[+-](?:[01][0-9]|2[0-3]):[0-5][0-9])$" + }, + "value_field": { + "type": "string", + "enum": [ + "monitored_value", + "cleaned_value" + ] + }, + "gaps": { + "type": "array", + "items": { + "type": "object", + "properties": { + "device_id": { + "type": "string", + "minLength": 1, + "maxLength": 128 + }, + "day_offset": { + "type": "integer", + "minimum": 0, + "maximum": 7 + }, + "missing_count": { + "type": "integer", + "minimum": 1, + "maximum": 100 + }, + "reason": { + "type": "string", + "enum": [ + "no_readings", + "outside_frozen_scope", + "outside_frozen_window", + "no_scheduled_samples" + ] + } + }, + "required": [ + "device_id", + "day_offset", + "missing_count", + "reason" + ], + "additionalProperties": false + }, + "maxItems": 8 + } + }, + "required": [ + "role", + "metric", + "unit", + "window_start", + "window_end" + ], + "additionalProperties": false + }, "chart_type": { "type": "string", "enum": [ @@ -241,6 +332,16 @@ "type": "string", "maxLength": 128 }, + "device_id": { + "type": "string", + "minLength": 1, + "maxLength": 128 + }, + "day_offset": { + "type": "integer", + "minimum": 0, + "maximum": 7 + }, "type": { "type": "string", "enum": [ diff --git a/docs/investigation-progress-and-presentation.md b/docs/investigation-progress-and-presentation.md index 6fe300d..2c8c267 100644 --- a/docs/investigation-progress-and-presentation.md +++ b/docs/investigation-progress-and-presentation.md @@ -149,6 +149,42 @@ time_comparison 的 x_data 使用精度一致、按窗口排列的 UTC+8 HH:mm 前端维护的脚本 scripts/export-view-command-contract.mjs 同步 Agent/Server 的结构协议及 tests/fixtures 下的统一 JSONL 和 chart-publication-cases.json；JSON Schema 不代替跨字段语义校验。Server 的独立测试用相同正反例校验 Pydantic 边界和 validate_time_comparison_chart；Agent 主机、原生工具及 Dev 的新图表输入也执行对应规则。 关键趋势与比较结论保留到报告中，最终报告的图表不替代过程发布。 + +### SCADA 设备展示覆盖 + +SCADA 异常初次调查的监测过程图使用可选的 `chart.scada` 结构： +`role=current|comparison`、`metric=pressure|flow`、`unit`、带时区的 `window_start/window_end`， +窗口固定为本轮截止时刻之前三小时；采样槽保持真实采样相位，截止非整刻不平移。超过100槽按连续时间分图，使用不同稳定chart_id，元数据仍声明完整三小时窗口；Server逐点核实后合并覆盖，缺少任一槽仍拒绝收尾。同期历史只取完整窗口末三个采样槽，同设备同期分图不得重叠。低频设备窗口无计划采样槽时，完整读取对应当天/历史三小时并核实无读数后，以截止时刻的单个null保留设备入口，标记no_scheduled_samples，不计作漏采。`value_field` 默认为 `monitored_value`，可显式选 `cleaned_value`。 +每条序列绑定 `device_id` 和 `day_offset`（当天 0，此前七天 1 至 7）。 +同期图每个设备包含当天 line 和七组 scatter，历史仅末三个对应采样槽有值。 +Server 校验设备属性、单位、完整采样槽、真实时间与引用快照的对应值，允许数值四舍五入保留 1 至 6 位小数， +保留负流量；缺测必须有完整查询或明确空值记录证明，未查询不能视为缺测。 +`chart.scada.gaps` 由 Server 生成并存入展示事件，前端保留设备入口并展示缺测点数。 +回放全量查询的 `coverage.frozen_start` 记录服务端已冻结的最早时刻；历史早于该边界时， +以本轮匹配 replay/batch 的完整回执证明 `outside_frozen_window`，无需虚构一次成功的历史读取。 + +地图的 SCADA 定位/标记、已验证图表和最终报告 `scada_device_ids` 累计本轮目标； +监测明细表使用该字段声明完整设备集合。按指标和单位分图，跨多张总览取设备并集， +同一 `chart_id` 更新只采用最新覆盖，累计目标不缩小。关闭图表及清空地图只影响浏览器， +不撤销产物覆盖。最终报告及成功结束均拒绝遗漏，返回 `409 SCADA_PRESENTATION_INCOMPLETE`， +`related_refs` 列出待补设备；Agent 补图后可重试原请求。普通进度和 interim 报告可继续发布。 +“仅作参考”不豁免展示；确有缺测时仍发布有空槽的图表并保留设备， +完整查询无法完成时保留 interim 和具体失败原因，不伪造缺测来通过 final。 + +覆盖事实保存在 `workbench.operational_task_runs.artifact_manifest.scada_presentation`， +JSON 包含累计 `targets`、按 `chart_id` 保存的最新验证 `charts` 及 `updated_at`。 +它与 Run 一起保留，独立于七天展示事件清理；不依赖浏览器在线或回执。 +不同 Run、项目和截止窗口不能抵数。恢复复核阶段沿用恢复证据门禁，不要求重画异常轮次图表。 +历史读取不要求新字段，已完成报告不回写；旧 Agent 的无绑定图表仍可读，但不能通过新轮次的覆盖门禁。 + +部署按[工作台数据库变更门禁](workbench-database-rollout.md)核实目标 BizDB、备份和回滚演练。 +先停止目标项目的任务写入并安排配套切换，在维护窗口应用 +[035 迁移](../resources/sql/035_scada_presentation_coverage.sql)，随后发布 Server、Agent 和前端。 +迁移只为 Run 新增通用 JSON 产物清单，默认空对象；不会自动回填旧图或自动迁移其他项目。 +未应用迁移的 SCADA 写入与收尾返回明确的 503，历史读取保持可用。 +Run 读取投影显式排除内部清单，REST 与 MCP 原有 Run 结构保持一致。 +若回退到本次读取兼容修复之前的 Server，须先备份清单再移除新增列；旧代码的整行读取不能直接带着新列回退。 +本次实现不代表迁移或上线已执行。 业务 Skill 说明本步骤应展示什么，调查指令统一说明发布方法、非空几何检查、去重和失败处理。 同一阶段批量展示，数据或方案没有变化时不重复请求；不为地图动作增加无关业务分析。 报告的 `map_targets` 仅为手动定位入口。关键步骤的数据缺失或展示失败需记录具体限制， diff --git a/resources/sql/035_scada_presentation_coverage.sql b/resources/sql/035_scada_presentation_coverage.sql new file mode 100644 index 0000000..ff33118 --- /dev/null +++ b/resources/sql/035_scada_presentation_coverage.sql @@ -0,0 +1,14 @@ +-- Run-scoped artifact facts survive domain-event retention without a new lifecycle table. +BEGIN; +SET LOCAL lock_timeout = '5s'; + +ALTER TABLE workbench.operational_task_runs + ADD COLUMN IF NOT EXISTS artifact_manifest jsonb NOT NULL DEFAULT '{}'::jsonb; +ALTER TABLE workbench.operational_task_runs + DROP CONSTRAINT IF EXISTS operational_task_runs_artifact_manifest_object; +ALTER TABLE workbench.operational_task_runs + ADD CONSTRAINT operational_task_runs_artifact_manifest_object + CHECK (jsonb_typeof(artifact_manifest) = 'object'); +COMMENT ON COLUMN workbench.operational_task_runs.artifact_manifest IS + 'Server-verified per-run artifact facts. scada_presentation holds cumulative targets and latest chart coverage, independent of browser receipts and event retention; not run status.'; +COMMIT; diff --git a/resources/work-order-skills/scada-anomaly/SKILL.md b/resources/work-order-skills/scada-anomaly/SKILL.md index ae5558d..0946dc4 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.14" + version: "1.0.15" --- # SCADA 数据异常调查 @@ -17,7 +17,7 @@ metadata: 通过 query_scada_readings 省略device_ids读取冻结异常窗内全部SCADA设备的完整属性和授权观测，覆盖压力、流量、水质等全部类型；正常对照也采用全量读取。Server自动读完所有设备页和观测页，完整返回devices、readings、coverage，结果仍受8MiB整份上限约束，超限明确报错后按时间分段，不能丢弃设备。query_complete表示查询已读完，缺测与无有效值另行判断。查询结束时间止于knowledge_cutoff，数据批次止于batch_sequence；冻结范围外设备单列，不回退实时数据。按需核查开始前15天基线，复用已有完整监测回执。结合monitored_value、cleaned_value和质量标记判断变化，可靠采集的低压与水质变化保留为业务异常。 -全量响应中的devices包含完整目录属性，readings包含实际读数，coverage列出全部查询设备、各设备读数数量、缺测和冻结范围外设备。不能只读取已标记、异常或代表设备后推断全网状态。单独读取list_scada_devices时，默认每页100个，持续翻页到page.has_more=false；属性缩略不等于已读取完整设备信息。全量读取后可显式指定device_ids做局部复核，局部读数仍按page.has_more接续，不能替代全量输入。最终报告引用覆盖本轮knowledge_cutoff和batch_sequence的全量证据，Server会拒绝仅有局部或旧批次证据的final。 +全量响应中的devices包含完整目录属性，readings包含实际读数，coverage列出全部查询设备、各设备读数数量、缺测和冻结范围外设备。不能用代表设备推断全网。单独读取list_scada_devices时，默认每页100个，持续翻页到page.has_more=false；属性缩略不等于已读取完整设备信息。全量读取后可显式指定device_ids做局部复核，局部读数仍按page.has_more接续，不能替代全量输入。最终报告引用覆盖本轮knowledge_cutoff和batch_sequence的全量证据，Server会拒绝仅有局部或旧批次证据的final。 分别说明压力异常模型得分、压力绝对下降及变化率、已配置出厂边界流量变化，再说明联合证据支持程度。使用实际权重、嫌疑门槛、动态阈值、参与测点与各自基线。综合分用于爆管嫌疑判断；得分越低越异常，目标告警与参考日分别解释。水质等其他异常按实际证据选择原因分支。 @@ -31,7 +31,9 @@ metadata: ## 对比展示与影响 -先查询并发布截至本轮knowledge_cutoff或异常时刻t的[t-3小时,t]当前趋势总览，覆盖本轮全部已标识SCADA设备，包括地图定位、等级标记及表格列出的目标设备。按指标和单位分图，同组各设备各占一条当前序列，系列名带device_id；每图最多8组，超过时分组发布，各组使用不同且稳定的chart_id。缺测填null，完全无读数的设备明确列为缺测并说明查询窗口，不能静默省略。压力和流量分别发布图表，包含正常与异常测点；某指标无设备或无有效读数时明确说明。每条当前序列保留三小时窗口内全部已取得采样点，不能用用于判断突变的[t-采样周期,t]相邻两点代替三小时趋势。15分钟数据正常应有13个时刻；缺测保留对应空槽并说明实际覆盖。收尾前逐项核对目标设备、查询覆盖、已发布图表和缺测清单，不用少量代表设备替代全量总览。图表“数据”视图沿用同一完整序列，报告按指标附测点、单位、窗口、有效点数和缺测说明表格。 +监测图必填chart.scada：role=current/comparison、metric=pressure/flow、unit、window_start/window_end（三小时）；series绑定device_id、day_offset=0..7。evidence_refs包含所有采样来源。报告scada_device_ids声明表格目标。Server在final及成功结束校验本轮总览和同期覆盖，SCADA_PRESENTATION_INCOMPLETE须补图；仅作参考不豁免。缺测须完整查询证明，保留null和设备图；scada.gaps由Server生成。恢复复核沿用恢复门禁。 + +先查询并发布以本轮knowledge_cutoff或异常时刻t为终点的[t-3小时,t]当前总览，覆盖地图定位、等级标记和表格列出的全部SCADA目标，含正常与异常测点。压力、流量按指标及单位分图，每设备一条序列，name带device_id；每图最多8组，超限按设备分图并用不同稳定chart_id。保留三小时内真实采样网格；截止非整刻不平移采样。超过100槽按连续时间分图，各用稳定chart_id，scada窗口仍填完整三小时；Server合并采样槽覆盖。低频无计划采样槽时，完整查询当天及历史窗口，以截止时刻单个null保留入口。缺测保留设备；“数据”视图沿用全部序列，报告附设备、单位、窗口、有效点数与缺测表。 定位输入具备后继续定位。压力、流量按设备另发七天同期复合图，不占用或替代全量当前总览。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的本地日期回退，不含当天。按实际周期保留缺口，按需翻页并报告参考点数；基线未冻结、读取失败或缺测时说明并保留当前趋势，不退化为前一天或相邻两点。 diff --git a/tests/support/operational_task_repository.py b/tests/support/operational_task_repository.py index a37bb60..2436b8e 100644 --- a/tests/support/operational_task_repository.py +++ b/tests/support/operational_task_repository.py @@ -57,6 +57,7 @@ class TaskRepository: self.plans = [] self.reports = [] self.progress = {} + self.scada_presentations = {} self.notifications = [] self.locks = [] self.access_requests = {} @@ -104,6 +105,12 @@ class TaskRepository: async def get_latest_progress(self, task_id, generation): return self.progress.get((task_id, generation)) + async def get_scada_presentation(self, task_id, run_id): + return deepcopy(self.scada_presentations.get((task_id, run_id))) + + async def save_scada_presentation(self, task_id, run_id, document, now): + self.scada_presentations[(task_id, run_id)] = deepcopy(document) + async def get_latest_graph_progress(self, task_id): snapshots = [value for (stored_id, _), value in self.progress.items() if stored_id == task_id diff --git a/tests/unit/test_complete_scada.py b/tests/unit/test_complete_scada.py index afb2d05..7c83729 100644 --- a/tests/unit/test_complete_scada.py +++ b/tests/unit/test_complete_scada.py @@ -82,6 +82,7 @@ def test_all_scada_replay_reads_only_frozen_devices_batches_and_window(monkeypat assert data["coverage"]["queried_device_ids"] == ["P1", "Q1"] assert data["coverage"]["missing_device_ids"] == ["Q1"] assert data["coverage"]["outside_frozen_scope_device_ids"] == ["NEW"] + assert data["coverage"]["frozen_start"] == START.isoformat() repository.detected_outputs.assert_awaited_once_with( context["replay_id"], ["P1", "Q1"], START, START + timedelta(hours=1), 9, limit=1001, offset=0, ) diff --git a/tests/unit/test_operational_task_storage_readiness.py b/tests/unit/test_operational_task_storage_readiness.py index 13fd2f2..e487029 100644 --- a/tests/unit/test_operational_task_storage_readiness.py +++ b/tests/unit/test_operational_task_storage_readiness.py @@ -98,3 +98,57 @@ async def test_presentation_history_storage_filters_project_task_and_committed_c assert "project_id=%s" in query and "aggregate_id=%s" in query assert "operational.presentation.requested" in query and "event_seq<=%s" in query assert "ORDER BY event_seq LIMIT %s" in query + + +@pytest.mark.anyio +async def test_missing_scada_manifest_migration_fails_closed(): + connection = SimpleNamespace(execute=AsyncMock(return_value=SimpleNamespace( + fetchone=AsyncMock(return_value={"available": False}), + ))) + repository = OperationalTaskRepository(connection, uuid4()) + with pytest.raises(DomainError) as failure: + await repository.get_scada_presentation(str(uuid4()), str(uuid4())) + assert failure.value.code == "DEPENDENCY_UNAVAILABLE" + assert failure.value.status == 503 + assert "035_scada_presentation_coverage.sql" in failure.value.message + + +@pytest.mark.anyio +async def test_scada_manifest_is_scoped_to_project_task_and_run_and_preserves_other_run_facts(): + from datetime import datetime, timezone + + project, task, run = map(str, (uuid4(), uuid4(), uuid4())) + document = {"version": 1, "targets": ["524529"], "charts": {}} + connection = SimpleNamespace(execute=AsyncMock(side_effect=[ + SimpleNamespace(fetchone=AsyncMock(return_value={"available": True})), + SimpleNamespace(fetchone=AsyncMock(return_value={"document": document})), + SimpleNamespace(rowcount=1), + ])) + repository = OperationalTaskRepository(connection, project) + assert await repository.get_scada_presentation(task, run) == document + query, params = connection.execute.call_args.args + assert "artifact_manifest->'scada_presentation'" in query + assert params == (project, task, run) + await repository.save_scada_presentation(task, run, document, datetime.now(timezone.utc)) + query, params = connection.execute.call_args.args + assert "jsonb_set(artifact_manifest,'{scada_presentation}'" in query + assert params[1:] == (project, task, run) + assert "revision=" not in query and "status=" not in query + + +@pytest.mark.anyio +async def test_internal_manifest_does_not_change_public_run_schemas_after_migration(): + from datetime import datetime, timezone + from app.domain.schemas.operational_task import OperationalTaskRun + from tests.support.operational_task_repository import _active_run, _context, _task_document + + context = _context() + task = _task_document(context, datetime.now(timezone.utc)) + run = _active_run(task, datetime.now(timezone.utc)) + connection = SimpleNamespace(execute=AsyncMock(return_value=SimpleNamespace(fetchall=AsyncMock(return_value=[{ + **run, "project_id": context.project_id, "artifact_manifest": {"scada_presentation": {"targets": ["524529"]}}, + }])))) + repository = OperationalTaskRepository(connection, context.project_id) + public = (await repository.list_task_runs(task["task_id"], 10, 0))[0] + assert "artifact_manifest" not in public + assert OperationalTaskRun.model_validate(public).run_id == OperationalTaskRun.model_validate(run).run_id diff --git a/tests/unit/test_scada_presentation_coverage.py b/tests/unit/test_scada_presentation_coverage.py new file mode 100644 index 0000000..2014324 --- /dev/null +++ b/tests/unit/test_scada_presentation_coverage.py @@ -0,0 +1,432 @@ +"""SCADA display coverage is independent of whether a device supports a conclusion.""" +from datetime import datetime, timedelta, timezone +from uuid import uuid4 + +import pytest + +from app.domain.domain_errors import DomainError +from app.domain.schemas.operational_data import TaskDataSnapshot +from app.domain.schemas.operational_task import AgentArtifactsCommand, AgentRunCommand, TaskPresentationEventCommand +from app.services.operational_tasks import OperationalTaskService +from tests.support.operational_task_repository import TaskRepository, _active_run, _context, _task_document + + +DEVICES = ["486033", "542017", "541061", "524529"] +END = datetime(2026, 7, 16, tzinfo=timezone.utc) +START = END - timedelta(hours=3) + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +async def investigation(): + context = _context() + task = _task_document(context, END, lease=END + timedelta(minutes=1)) + task["workflow"] = "scada_anomaly" + repository = TaskRepository(task, _active_run(task, END)) + service = OperationalTaskService(repository, context, set(), clock=lambda: END) + readings = [] + for day in range(8): + slots = range(13) if day == 0 else range(10, 13) + for slot in slots: + for device in DEVICES: + readings.append({"device_id": device, "time": (START + timedelta(minutes=15 * slot, days=-day)).isoformat(), + "monitored_value": float(slot + 1), "cleaned_value": float(slot + 1)}) + evidence, _ = await service.record_data_snapshot(task["task_id"], TaskDataSnapshot( + source="query_scada_readings", source_ref="mcp:observations", queried_at=END, + origin="system_data", truncated=False, data={ + "window": {"start": (START - timedelta(days=7)).isoformat(), "end": END.isoformat()}, + "data": {"devices": [{"device_id": device, "device_type": "pipe_flow", "measurement_unit": "m3/h", + "transmission_frequency": "0:15:00"} for device in DEVICES], + "readings": readings, "coverage": {"mode": "all_scada", "query_complete": True}}, + }, + )) + return task, repository, service, evidence.evidence_id + + +def coverage(comparison_devices): + slots = [(START + timedelta(minutes=15 * i)).isoformat() for i in range(13)] + common = {"coverage": {device: {"covered": slots} for device in DEVICES}, "metric": "flow", "window_start": START.isoformat(), "window_end": END.isoformat(), "gaps": []} + return {"version": 1, "targets": DEVICES.copy(), "charts": { + "overview": {**common, "role": "current", "device_ids": DEVICES.copy()}, + **{f"comparison-{device}": {**common, "role": "comparison", "device_ids": [device], + "coverage": {device: {"covered": slots}}} + for device in comparison_devices}, + }} + + +def final_report(evidence_id): + return {"stage": "final", "outcome": "concluded", "title": "调查结果", + "current_conclusion": "524529 仅作参考", "conclusion_boundary": "待现场核验", + "evidence_refs": [evidence_id], "data_gaps": [], "scada_device_ids": DEVICES} + + +@pytest.mark.anyio +async def test_final_rejects_four_target_devices_with_three_comparisons(): + task, repository, service, evidence_id = await investigation() + repository.scada_presentations = {(task["task_id"], repository.run["run_id"]): coverage(DEVICES[:3])} + with pytest.raises(DomainError) as failure: + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", report=final_report(evidence_id), + )) + assert failure.value.code == "SCADA_PRESENTATION_INCOMPLETE" + assert "524529" in failure.value.message + assert not repository.reports + + +@pytest.mark.anyio +async def test_final_accepts_all_four_devices(): + task, repository, service, evidence_id = await investigation() + repository.scada_presentations = {(task["task_id"], repository.run["run_id"]): coverage(DEVICES)} + result = await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", report=final_report(evidence_id), + )) + assert result.report_revision == 1 + + +def chart(role="current", devices=None): + devices = DEVICES if devices is None else devices + comparison = role == "comparison" + series = [] + for device in devices: + for day in (range(8) if comparison else [0]): + slots = range(13) if day == 0 else range(10, 13) + series.append({"name": f"{device} {day}", "device_id": device, "day_offset": day, + "type": "scatter" if day else "line", + "data": [float(index + 1) if index in slots else None for index in range(13)], + "sample_times": [(START + timedelta(minutes=15 * index, days=-day)).isoformat() + if index in slots else None for index in range(13)]}) + return {"title": "流量观测", "chart_type": "time_comparison" if comparison else "line", + "x_data": [(START + timedelta(minutes=15 * index, hours=8)).strftime("%H:%M") for index in range(13)], + "series": series, "scada": {"role": role, "metric": "flow", "unit": "m3/h", + "window_start": START.isoformat(), "window_end": END.isoformat()}} + + +async def publish(task, repository, service, evidence_id, chart_id, data): + command = TaskPresentationEventCommand( + request_id=uuid4(), run_id=repository.run["run_id"], source_session_id="session-1", + command={"action": "open_chart", "chart_id": chart_id, "chart": data, "evidence_refs": [evidence_id]}, + ) + return await service.publish_presentation_event(task["task_id"], command) + + +@pytest.mark.anyio +async def test_publications_require_complete_targets_and_retry_without_a_browser(): + task, repository, service, evidence_id = await investigation() + await publish(task, repository, service, evidence_id, "overview", chart()) + for device in DEVICES[:3]: + await publish(task, repository, service, evidence_id, device, chart("comparison", [device])) + interim = {**final_report(evidence_id), "stage": "interim"} + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", report=interim, + )) + final = AgentArtifactsCommand(request_id=uuid4(), source_session_id="session-1", report=final_report(evidence_id)) + with pytest.raises(DomainError) as failure: + await service.publish_artifacts(task["task_id"], final) + assert failure.value.code == "SCADA_PRESENTATION_INCOMPLETE" + assert failure.value.refs == ["524529"] + await publish(task, repository, service, evidence_id, "524529", chart("comparison", ["524529"])) + await service.publish_presentation_event(task["task_id"], TaskPresentationEventCommand( + request_id=uuid4(), run_id=repository.run["run_id"], source_session_id="session-1", + command={"action": "close_chart", "chart_id": "524529"}, + )) + result = await service.publish_artifacts(task["task_id"], final) + assert result.task.latest_report.report.stage == "final" + # Neither retained event history nor a browser acknowledgement is required. + repository.events.clear() + complete = await service.complete_run(task["task_id"], repository.run["run_id"], AgentRunCommand( + request_id=uuid4(), expected_revision=repository.run["revision"], source_session_id="session-1", + )) + assert complete.latest_run.status == "succeeded" + + +@pytest.mark.anyio +async def test_successful_completion_rechecks_coverage_after_final_report(): + task, repository, service, evidence_id = await investigation() + repository.scada_presentations[(task["task_id"], repository.run["run_id"])] = coverage(DEVICES) + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", report=final_report(evidence_id), + )) + # Replacing a chart cannot erase its fourth target from the required set. + await publish(task, repository, service, evidence_id, "overview", chart(devices=DEVICES[:3])) + with pytest.raises(DomainError) as failure: + await service.complete_run(task["task_id"], repository.run["run_id"], AgentRunCommand( + request_id=uuid4(), expected_revision=repository.run["revision"], source_session_id="session-1", + )) + assert failure.value.code == "SCADA_PRESENTATION_INCOMPLETE" + assert "524529" in failure.value.message + assert repository.run["status"] == "running" + + +@pytest.mark.anyio +@pytest.mark.parametrize("missing_current", [False, True]) +async def test_missing_observations_keep_the_object_and_server_derived_gaps(missing_current): + task, repository, service, evidence_id = await investigation() + data = repository.evidence[evidence_id]["data_snapshot"]["data"]["data"] + data["readings"] = [row for row in data["readings"] if row["device_id"] != "524529" + or not missing_current and datetime.fromisoformat(row["time"]) >= START] + value = chart("comparison", ["524529"]) + for series in value["series"]: + if missing_current or series["day_offset"]: + series["data"] = [None] * 13 + series["sample_times"] = [None] * 13 + await publish(task, repository, service, evidence_id, "524529", value) + saved = repository.events[-1]["metadata"]["command"]["chart"] + assert len(saved["scada"]["gaps"]) == (8 if missing_current else 7) + assert sum(gap["missing_count"] for gap in saved["scada"]["gaps"]) == (34 if missing_current else 21) + document = await repository.get_scada_presentation(task["task_id"], repository.run["run_id"]) + assert document["targets"] == ["524529"] + assert document["charts"]["524529"]["device_ids"] == ["524529"] + + +@pytest.mark.anyio +@pytest.mark.parametrize("defect,code", [ + ("null_existing_value", "SCADA_CHART_DATA_MISMATCH"), + ("invented_value", "SCADA_CHART_DATA_MISMATCH"), + ("wrong_window", "SCADA_CHART_WINDOW_INVALID"), + ("missing_day", "SCADA_CHART_INVALID"), + ("wrong_unit", "SCADA_CHART_INVALID"), + ("unqueried_history", "SCADA_CHART_EVIDENCE_REQUIRED"), +]) +async def test_scada_chart_cannot_fake_coverage(defect, code): + task, repository, service, evidence_id = await investigation() + value = chart("comparison", ["524529"]) + if defect == "null_existing_value": + value["series"][1]["data"][10] = None + value["series"][1]["sample_times"][10] = None + elif defect == "invented_value": + value["series"][0]["data"][0] = 999. + elif defect == "wrong_window": + value["scada"]["window_end"] = (END - timedelta(minutes=15)).isoformat() + elif defect == "missing_day": + value["series"].pop() + elif defect == "wrong_unit": + value["scada"]["unit"] = "m" + elif defect == "unqueried_history": + repository.evidence[evidence_id]["data_snapshot"]["data"]["window"]["start"] = START.isoformat() + with pytest.raises(DomainError) as failure: + await publish(task, repository, service, evidence_id, "524529", value) + assert failure.value.code == code + assert not repository.scada_presentations + + +@pytest.mark.anyio +async def test_previous_run_charts_never_satisfy_a_new_run(): + task, repository, service, evidence_id = await investigation() + original = repository.run["run_id"] + repository.scada_presentations[(task["task_id"], original)] = coverage(DEVICES) + task["execution_generation"] += 1 + repository.run = _active_run(task, END) + repository.runs[repository.run["run_id"]] = repository.run + with pytest.raises(DomainError) as failure: + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", report=final_report(evidence_id), + )) + assert failure.value.code == "SCADA_PRESENTATION_INCOMPLETE" + assert original != repository.run["run_id"] + assert (task["task_id"], original) in repository.scada_presentations + + +@pytest.mark.anyio +async def test_one_device_cannot_inflate_comparison_object_count_with_a_new_chart_id(): + task, repository, service, evidence_id = await investigation() + await publish(task, repository, service, evidence_id, "524529", chart("comparison", ["524529"])) + with pytest.raises(DomainError) as failure: + await publish(task, repository, service, evidence_id, "524529-duplicate", chart("comparison", ["524529"])) + assert failure.value.code == "SCADA_CHART_DUPLICATE_DEVICE" + assert failure.value.refs == ["524529"] + await publish(task, repository, service, evidence_id, "524529", chart("comparison", ["524529"])) + assert len(repository.scada_presentations[(task["task_id"], repository.run["run_id"])]["charts"]) == 1 + + +@pytest.mark.anyio +async def test_server_frozen_boundary_proves_unavailable_history_without_fabricating_a_query(): + task, repository, service, evidence_id = await investigation() + replay_id = str(uuid4()) + task["source"]["replay_context"] = {"replay_id": replay_id, "knowledge_cutoff": END.isoformat(), "batch_sequence": 3} + snapshot = repository.evidence[evidence_id]["data_snapshot"]["data"] + snapshot["window"]["start"] = START.isoformat() + snapshot["data"]["coverage"].update(replay_id=replay_id, batch_sequence=3, frozen_start=START.isoformat()) + value = chart("comparison", ["524529"]) + for series in value["series"][1:]: + series["data"] = [None] * 13 + series["sample_times"] = [None] * 13 + await publish(task, repository, service, evidence_id, "524529", value) + gaps = repository.events[-1]["metadata"]["command"]["chart"]["scada"]["gaps"] + assert len(gaps) == 7 + assert {gap["reason"] for gap in gaps} == {"outside_frozen_window"} + + +@pytest.mark.anyio +async def test_interim_monitoring_table_targets_cannot_be_removed_from_a_later_final_report(): + task, repository, service, evidence_id = await investigation() + incomplete = coverage(DEVICES[:3]) + incomplete["targets"] = DEVICES[:3] + incomplete["charts"]["overview"]["device_ids"] = DEVICES[:3] + repository.scada_presentations[(task["task_id"], repository.run["run_id"])] = incomplete + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", report={**final_report(evidence_id), "stage": "interim"}, + )) + with pytest.raises(DomainError) as failure: + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", report={**final_report(evidence_id), "scada_device_ids": DEVICES[:3]}, + )) + assert failure.value.code == "SCADA_PRESENTATION_INCOMPLETE" + assert failure.value.refs == ["524529"] + + +@pytest.mark.anyio +async def test_narrative_diagram_references_do_not_declare_business_display_targets(): + task, repository, service, evidence_id = await investigation() + selected = DEVICES[:3] + document = coverage(selected) + document["targets"] = selected + document["charts"]["overview"]["device_ids"] = selected + repository.scada_presentations[(task["task_id"], repository.run["run_id"])] = document + report = {**final_report(evidence_id), "scada_device_ids": selected, "diagram": { + "title": "待选参考测点", "mermaid": "flowchart LR\nreference[参考测点]", "node_refs": [{ + "node_id": "reference", "label": "参考测点", "map_targets": [{"source_id": "scada", "feature_id": "524529"}], + }], + }} + result = await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", report=report, + )) + assert result.report_revision == 1 + + +@pytest.mark.anyio +async def test_unaligned_cutoff_keeps_actual_quarter_hour_samples(): + task, repository, service, evidence_id = await investigation() + task["trigger"]["occurred_at"] = (END + timedelta(minutes=8)).isoformat() + value = chart("comparison", ["524529"]) + value["scada"].update(window_start=(START + timedelta(minutes=8)).isoformat(), + window_end=(END + timedelta(minutes=8)).isoformat()) + value["x_data"] = value["x_data"][1:] + for item in value["series"]: + item["data"] = item["data"][1:] + item["sample_times"] = item["sample_times"][1:] + await publish(task, repository, service, evidence_id, "unaligned", value) + + +@pytest.mark.anyio +@pytest.mark.parametrize("frequency", ["15min", "15分钟"]) +async def test_wholly_missing_devices_accept_supported_cadence_formats(frequency): + task, repository, service, evidence_id = await investigation() + data = repository.evidence[evidence_id]["data_snapshot"]["data"]["data"] + data["readings"] = [] + for device in data["devices"]: + device["transmission_frequency"] = frequency + value = chart("comparison", ["524529"]) + for item in value["series"]: + item["data"] = [None] * 13 + item["sample_times"] = [None] * 13 + await publish(task, repository, service, evidence_id, "missing", value) + + +@pytest.mark.anyio +async def test_one_minute_chunks_require_union_of_verified_slots(): + task, repository, service, evidence_id = await investigation() + data = repository.evidence[evidence_id]["data_snapshot"]["data"]["data"] + data["devices"] = [{**data["devices"][0], "transmission_frequency": "1min"}] + device = DEVICES[0] + data["readings"] = [{"device_id": device, "time": (START + timedelta(minutes=i, days=-day)).isoformat(), + "monitored_value": float(i + 1)} + for day in range(8) for i in (range(181) if day == 0 else range(178, 181))] + + def chunk(role, first, stop, missing_current=False): + value = chart(role, [device]) + value["x_data"] = [(START + timedelta(minutes=i, hours=8)).strftime("%H:%M") for i in range(first, stop)] + for item in value["series"]: + day = item["day_offset"] + present = lambda i: (not day or i >= 178) and not (missing_current and day == 0) + item["data"] = [float(i + 1) if present(i) else None for i in range(first, stop)] + item["sample_times"] = [(START + timedelta(minutes=i, days=-day)).isoformat() if present(i) else None + for i in range(first, stop)] + return value + + report = {**final_report(evidence_id), "scada_device_ids": [device]} + for role in ("current", "comparison"): + await publish(task, repository, service, evidence_id, role + "-first", chunk(role, 0, 100)) + with pytest.raises(DomainError) as failure: + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", report=report)) + assert failure.value.code == "SCADA_PRESENTATION_INCOMPLETE" + with pytest.raises(DomainError) as duplicate: + await publish(task, repository, service, evidence_id, "overlap", chunk("comparison", 99, 181)) + assert duplicate.value.code == "SCADA_CHART_DUPLICATE_DEVICE" + for role in ("current", "comparison"): + await publish(task, repository, service, evidence_id, role + "-last", chunk(role, 100, 181)) + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", report=report)) + # Missing current data still permits the last chunk's historical dates across midnight. + data["readings"] = [row for row in data["readings"] if datetime.fromisoformat(row["time"]) < START] + await publish(task, repository, service, evidence_id, "comparison-last", chunk("comparison", 100, 181, True)) + + +@pytest.mark.parametrize("start,end,labels,historical", [ + ("2026-07-16T05:08:00+08:00", "2026-07-16T08:08:00+08:00", ["07:45", "08:00"], + ["2026-07-15T07:45:00+08:00", "2026-07-15T08:00:00+08:00"]), + ("2026-07-15T22:08:00+08:00", "2026-07-16T01:08:00+08:00", ["00:45", "01:00"], + ["2026-07-15T00:45:00+08:00", "2026-07-15T01:00:00+08:00"]), +]) +def test_missing_current_chunk_uses_its_first_slot_date(start, end, labels, historical): + from app.services.time_comparison_chart import validate_time_comparison_chart + + validate_time_comparison_chart({"chart_type": "time_comparison", "x_data": labels, + "scada": {"role": "comparison", "window_start": start, "window_end": end}, + "series": [{"type": "line", "data": [None, None], "sample_times": [None, None]}, + {"type": "scatter", "data": [1., 2.], "sample_times": historical}]}, + allow_missing_current=True) + + +@pytest.mark.anyio +async def test_missing_current_preserves_observed_historical_sampling_phase(): + task, repository, service, evidence_id = await investigation() + data = repository.evidence[evidence_id]["data_snapshot"]["data"]["data"] + data["readings"] = [{**row, "time": (datetime.fromisoformat(row["time"]) + timedelta(minutes=5)).isoformat()} + for row in data["readings"] if datetime.fromisoformat(row["time"]) < START] + value = chart("comparison", ["524529"]) + value["x_data"] = [(START + timedelta(minutes=15 * i + 5, hours=8)).strftime("%H:%M") for i in range(12)] + for item in value["series"]: + day = item["day_offset"] + item["data"] = [float(i + 1) if day and i >= 9 else None for i in range(12)] + item["sample_times"] = [(START + timedelta(minutes=15 * i + 5, days=-day)).isoformat() + if day and i >= 9 else None for i in range(12)] + # The final three corresponding slots are now 23:20,23:35,23:50 UTC. + for day in range(1, 8): + data["readings"].append({"device_id": "524529", "time": (START + timedelta(minutes=140, days=-day)).isoformat(), + "monitored_value": 10.}) + await publish(task, repository, service, evidence_id, "historical-phase", value) + + +@pytest.mark.anyio +@pytest.mark.parametrize("point_only", [False, True]) +async def test_six_hour_devices_with_no_scheduled_slots_require_full_absence_proof(point_only): + task, repository, service, evidence_id = await investigation() + end = END + timedelta(hours=2) + start = end - timedelta(hours=3) + task["trigger"]["occurred_at"] = end.isoformat() + snapshot = repository.evidence[evidence_id]["data_snapshot"]["data"] + snapshot["window"] = {"start": (end if point_only else start - timedelta(days=7)).isoformat(), "end": end.isoformat()} + snapshot["data"]["readings"] = [] + for device in snapshot["data"]["devices"]: + device["transmission_frequency"] = "6:00:00" + for role in ("current", "comparison"): + value = chart(role, ["524529"]) + value["scada"].update(window_start=start.isoformat(), window_end=end.isoformat()) + value["x_data"] = ["10:00"] + for item in value["series"]: + item["data"], item["sample_times"] = [None], [None] + if point_only: + with pytest.raises(DomainError) as failure: + await publish(task, repository, service, evidence_id, role, value) + assert failure.value.code == "SCADA_CHART_EVIDENCE_REQUIRED" + else: + await publish(task, repository, service, evidence_id, role, value) + gaps = repository.events[-1]["metadata"]["command"]["chart"]["scada"]["gaps"] + assert all(gap["reason"] == "no_scheduled_samples" for gap in gaps) + if not point_only: + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", report={**final_report(evidence_id), "scada_device_ids": ["524529"]}))