diff --git a/app/domain/schemas/operational_task.py b/app/domain/schemas/operational_task.py index 60ad8ae..c1e41fc 100644 --- a/app/domain/schemas/operational_task.py +++ b/app/domain/schemas/operational_task.py @@ -722,7 +722,7 @@ class TaskSafeChartSeries(StrictModel): class TaskSafeChart(StrictModel): title: Annotated[str, Field(max_length=200)] - chart_type: Literal["line", "bar"] + 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 x_data: Annotated[list[Annotated[str, Field(max_length=128)]], Field(min_length=1, max_length=100)] diff --git a/app/services/operational_tasks.py b/app/services/operational_tasks.py index ea77c55..776cb07 100644 --- a/app/services/operational_tasks.py +++ b/app/services/operational_tasks.py @@ -74,6 +74,7 @@ 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 +from app.services.time_comparison_chart import validate_time_comparison_chart LEASE_DURATION = timedelta(seconds=60) @@ -837,6 +838,11 @@ class OperationalTaskService: "Only a running run accepts Agent artifacts", status=409) now = self._now() self._require_active_run_lease(active_run, now) + if command.report: + for item in command.report.ui_blocks: + for element in item.block.spec.elements.values(): + if element.type == "Chart": + validate_time_comparison_chart(element.props) if command.report and ( command.report.stage == "final" or command.report.outcome == "insufficient_data" ): @@ -1095,6 +1101,7 @@ class OperationalTaskService: self._require_active_run_lease(run, now) payload = _json(command.command) if command.command.action == "open_chart": + validate_time_comparison_chart(payload["chart"]) 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": diff --git a/app/services/time_comparison_chart.py b/app/services/time_comparison_chart.py new file mode 100644 index 0000000..b9074d3 --- /dev/null +++ b/app/services/time_comparison_chart.py @@ -0,0 +1,78 @@ +"""Publication-only validation for same-time-of-day line/scatter comparisons.""" +from __future__ import annotations + +import re +from collections.abc import Mapping +from datetime import datetime, timedelta, timezone +from typing import Any + +from app.domain.domain_errors import require + + +DISPLAY_TIMEZONE = timezone(timedelta(hours=8)) +DAY_SECONDS = 86400 + + +def validate_time_comparison_chart(chart: Mapping[str, Any]) -> None: + """Validate new publications; persisted historical charts keep their read contract.""" + chart_type = chart["chart_type"] + series = chart["series"] + kinds = {item.get("type") or chart_type for item in series} + if chart_type != "time_comparison": + require(not {"line", "scatter"} <= kinds, "TIME_COMPARISON_INVALID", + "曲线与历史散点复合图须使用 chart_type=time_comparison", status=422) + return + + def check(condition: bool, message: str) -> None: + require(condition, "TIME_COMPARISON_INVALID", message, status=422) + + kinds = {item.get("type") or "line" for item in series} + check(kinds == {"line", "scatter"}, + "time_comparison 的 series 必须包含当前 line 和历史 scatter，不能混入 bar") + labels = chart["x_data"] + check(all(re.fullmatch(r"(?:[01][0-9]|2[0-3]):[0-5][0-9](?::[0-5][0-9])?", label) + for label in labels), + "time_comparison.x_data 只接受 HH:mm 或 HH:mm:ss 时刻；历史日期保留在 sample_times，不能追加到横轴") + check(len({len(label) for label in labels}) == 1, "x_data 必须使用一致的时刻精度") + seconds = [sum(int(part) * factor for part, factor in zip(label.split(":"), (3600, 60, 1))) + for label in labels] + offsets = [0] + for index in range(1, len(seconds)): + step = (seconds[index] - seconds[index - 1]) % DAY_SECONDS + check(step > 0, f"x_data[{index}] 重复了上一时刻；历史点应复用当前槽位") + offsets.append(offsets[-1] + step) + check(offsets[-1] < DAY_SECONDS, "x_data 必须按时间递增且小于一天，不能追加多天历史槽位") + + current_bases: set[datetime] = set() + historical_bases: list[datetime] = [] + for series_index, item in enumerate(series): + times = item.get("sample_times") + path = f"series[{series_index}]" + check(times is not None and len(times) == len(labels), f"{path}.sample_times 必须提供并与 x_data 等长") + bases: set[datetime] = set() + for index, (value, at) in enumerate(zip(item["data"], times, strict=True)): + check((value is None) == (at is None), + f"{path}.sample_times[{index}]：每个有效值必须保留真实带时区时间，缺测值和时间均填 null") + if at is None: + continue + try: + local = datetime.fromisoformat(at).astimezone(DISPLAY_TIMEZONE) + base = local - timedelta(seconds=offsets[index]) + except OverflowError: + check(False, f"{path}.sample_times[{index}] 超出同期对比可表达的日期范围") + label = labels[index] + expected = local.strftime("%H:%M:%S" if len(label) == 8 else "%H:%M") + check(expected == label and local.microsecond == 0 and (len(label) == 8 or local.second == 0), + f"{path}.sample_times[{index}] 换算为 UTC+8 后须与 x_data[{index}]={label} 对齐") + bases.add(base) + check(len(bases) <= 1, f"{path}.sample_times 必须按同一窗口对齐，不能将不同日期拼入同一组") + if (item.get("type") or "line") == "line": + current_bases.update(bases) + else: + historical_bases.extend(bases) + check(len(current_bases) == 1, "当前 line 必须包含真实采样时间，且所有当前曲线使用同一窗口") + current = next(iter(current_bases)) + for historical in historical_bases: + delta = (current - historical).total_seconds() + check(delta > 0 and delta % DAY_SECONDS == 0, + "历史 scatter 必须来自当前窗口之前的对应日期，不能使用当天或未来采样") diff --git a/contracts/answer-ui-catalog.json b/contracts/answer-ui-catalog.json index eba32b4..dc1b4f4 100644 --- a/contracts/answer-ui-catalog.json +++ b/contracts/answer-ui-catalog.json @@ -310,7 +310,8 @@ "type": "string", "enum": [ "line", - "bar" + "bar", + "time_comparison" ] }, "x_axis_name": { diff --git a/contracts/manifest.json b/contracts/manifest.json index 0caeb25..50226fc 100644 --- a/contracts/manifest.json +++ b/contracts/manifest.json @@ -3,7 +3,7 @@ "contracts": { "server": { "file": "server-v1.openapi.json", - "sha256": "50bb8322494c830ed209b0acc4f02fcc9e49d1046ac4ea5a2c3e6d7ec665aa04" + "sha256": "426b8d9170f253aa7132e9376031a7da8c04ac6f6e693948890d6f343420af1b" } } } diff --git a/contracts/server-v1.openapi.json b/contracts/server-v1.openapi.json index 5001ecd..43bbbb4 100644 --- a/contracts/server-v1.openapi.json +++ b/contracts/server-v1.openapi.json @@ -10223,7 +10223,8 @@ "chart_type": { "enum": [ "line", - "bar" + "bar", + "time_comparison" ], "title": "Chart Type", "type": "string" diff --git a/contracts/view-command.schema.json b/contracts/view-command.schema.json index 1b2c101..8064e81 100644 --- a/contracts/view-command.schema.json +++ b/contracts/view-command.schema.json @@ -211,7 +211,8 @@ "type": "string", "enum": [ "line", - "bar" + "bar", + "time_comparison" ] }, "x_axis_name": { diff --git a/resources/work-order-skills/scada-anomaly/SKILL.md b/resources/work-order-skills/scada-anomaly/SKILL.md index 1b3ccf4..4547774 100644 --- a/resources/work-order-skills/scada-anomaly/SKILL.md +++ b/resources/work-order-skills/scada-anomaly/SKILL.md @@ -33,7 +33,7 @@ metadata: 先查询并发布截至本轮knowledge_cutoff或异常时刻t的[t-3小时,t]当前趋势总览，覆盖本轮全部已标识SCADA设备，包括地图定位、等级标记及表格列出的目标设备。按指标和单位分图，同组各设备各占一条当前序列，系列名带device_id；每图最多8组，超过时分组发布，各组使用不同且稳定的chart_id。缺测填null，完全无读数的设备明确列为缺测并说明查询窗口，不能静默省略。压力和流量分别发布图表，包含正常与异常测点；某指标无设备或无有效读数时明确说明。每条当前序列保留三小时窗口内全部已取得采样点，不能用用于判断突变的[t-采样周期,t]相邻两点代替三小时趋势。15分钟数据正常应有13个时刻；缺测保留对应空槽并说明实际覆盖。收尾前逐项核对目标设备、查询覆盖、已发布图表和缺测清单，不用少量代表设备替代全量总览。图表“数据”视图沿用同一完整序列，报告按指标附测点、单位、窗口、有效点数和缺测说明表格。 -定位输入具备后继续推进定位。压力、流量均需发布按设备分图的七天历史复合对照，作为全量当前趋势总览之外的必要展示；不能因历史散点占用序列额度而遗漏其他设备的当前趋势。复合图采用 [t-3小时,t] 当前折线和前7天每天 t-30分钟、t-15分钟、t 的21个参考位置。15分钟数据用13槽，当前一个line、历史七个scatter；历史散点只填末3槽，其他为null。sample_times保留真实带时区时刻，坐标轴注明单位。前7天按t转换后的当地日期逐日回退，不包含当天；按实际采样周期保留缺口，按需翻页并报告实际参考点数。基线未冻结、读取失败或缺测时明确说明并保留已有三小时趋势，不能静默退化为前一天或相邻两个时刻。 +定位输入具备后继续推进定位。压力、流量均需发布按设备分图的七天历史复合对照，作为全量当前趋势总览之外的必要展示；不能因历史散点占用序列额度而遗漏其他设备的当前趋势。复合图采用 [t-3小时,t] 当前折线和前7天每天 t-30分钟、t-15分钟、t 的21个参考位置。chart_type=time_comparison表示同期复合对比，区别于普通line和bar。x_data只放UTC+8的HH:mm或HH:mm:ss共享时刻槽，不追加历史日期或历史槽位；15分钟数据用13槽，series中当前一个type=line、历史七个type=scatter，历史散点只填末3槽，其他为null。每组sample_times与data等长，每个有效值保留真实带时区日期和时刻，缺测值及其时间均填null；历史点按本地时刻放入对应槽，当前曲线使用同一窗口，每个历史点集使用此前某一天的对应窗口，跨午夜按窗口时间顺序排列。坐标轴注明单位。前7天按t转换后的当地日期逐日回退，不包含当天；按实际采样周期保留缺口，按需翻页并报告实际参考点数。基线未冻结、读取失败或缺测时明确说明并保留已有三小时趋势，不能静默退化为前一天或相邻两个时刻。 影响说明列出异常前参考、异常期间和已有恢复观测中的压力、流量或水质变化、持续时间、涉及测点及已核实资产。可比对象、单位、时段与运行条件下给出变化量，分别标明实际观测、模型预测和拓扑隔离。停水范围与漏量引用相应拓扑、水力、流量平衡或现场依据；恢复观测、用户关联或现场证据缺失时列为具体待核实项。 diff --git a/tests/unit/test_operational_tasks.py b/tests/unit/test_operational_tasks.py index ef3be47..b352d17 100644 --- a/tests/unit/test_operational_tasks.py +++ b/tests/unit/test_operational_tasks.py @@ -617,9 +617,10 @@ async def test_replay_investigation_can_publish_chart_and_propose_work_order(): result = 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": "open_chart", "chart_id": "pressure", "evidence_refs": [evidence.evidence_id], - "chart": {"title": "压力", "chart_type": "line", "x_data": ["19:45", "20:00"], + "chart": {"title": "压力", "chart_type": "time_comparison", "x_data": ["19:45", "20:00"], "y_axis_name": "压力 m", "series": [ - {"name": "当前", "type": "line", "data": [27., 26.]}, + {"name": "当前", "type": "line", "data": [27., 26.], + "sample_times": ["2026-10-10T19:45:00+08:00", "2026-10-10T20:00:00+08:00"]}, {"name": "前一天", "type": "scatter", "data": [None, 28.], "sample_times": [None, "2026-10-09T20:00:00+08:00"]}, ]}}, @@ -637,6 +638,65 @@ async def test_replay_investigation_can_publish_chart_and_propose_work_order(): assert repository.action["action_id"] == str(action.action_id) +@pytest.mark.anyio +@pytest.mark.parametrize("destination", ["presentation", "report"]) +@pytest.mark.parametrize("defect", [None, "legacy_type", "dates_on_axis", "missing_times", "wrong_slot", "future_day"]) +async def test_comparison_chart_publication_is_validated_at_both_write_boundaries(destination, defect): + now = datetime(2026, 10, 10, tzinfo=timezone.utc) + context = _context() + task = _task_document(context, now, lease=now + timedelta(minutes=1)) + repository = RunTaskRepository(task, _active_run(task, now)) + service = OperationalTaskService(repository, context, set(), clock=lambda: now) + evidence, _ = await service.record_data_snapshot(task["task_id"], TaskDataSnapshot( + source="query_scada_readings", source_ref="mcp:comparison", queried_at=now, + origin="synthetic", data={"items": []}, truncated=False, + )) + chart = {"title": "同期对照", "chart_type": "time_comparison", "x_data": ["08:00"], "series": [ + {"name": "当前", "data": [25.], "sample_times": ["2026-10-10T08:00:00+08:00"]}, + {"name": "历史", "type": "scatter", "data": [26.], "sample_times": ["2026-10-09T08:00:00+08:00"]}, + ]} + if defect == "legacy_type": + chart["chart_type"] = "line" + elif defect == "dates_on_axis": + chart["x_data"] = ["10-10 08:00"] + elif defect == "missing_times": + del chart["series"][1]["sample_times"] + elif defect == "wrong_slot": + chart["series"][1]["sample_times"] = ["2026-10-09T07:45:00+08:00"] + elif defect == "future_day": + chart["series"][1]["sample_times"] = ["2026-10-11T08:00:00+08:00"] + events_before = len(repository.events) + + async def publish(): + if destination == "presentation": + 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": "open_chart", "chart_id": "comparison", "evidence_refs": [evidence.evidence_id], + "chart": chart}, + )) + else: + await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( + request_id=uuid4(), source_session_id="session-1", report={ + **_current_report(), "evidence_refs": [evidence.evidence_id], "ui_blocks": [{ + "evidence_refs": [evidence.evidence_id], "block": { + "version": "answer-ui@1", "block_id": "comparison", "fallback_text": "同期对照", + "spec": {"root": "chart", "elements": {"chart": { + "type": "Chart", "children": [], "props": {**chart, "source": "合成数据"}, + }}}, + }, + }], + }, + )) + if defect is None: + await publish() + assert len(repository.events) > events_before + else: + with pytest.raises(DomainError) as error: + await publish() + assert error.value.code == "TIME_COMPARISON_INVALID" + assert len(repository.events) == events_before + + @pytest.mark.anyio @pytest.mark.parametrize("operation", ["propose", "approve", "submit"]) async def test_replay_cannot_propose_approve_or_submit_device_commands(operation): diff --git a/tests/unit/test_time_comparison_chart.py b/tests/unit/test_time_comparison_chart.py new file mode 100644 index 0000000..462a264 --- /dev/null +++ b/tests/unit/test_time_comparison_chart.py @@ -0,0 +1,105 @@ +from copy import deepcopy +from datetime import datetime, timedelta, timezone + +import pytest + +from app.domain.domain_errors import DomainError +from app.domain.schemas.operational_task import TaskSafeChart +from app.services.time_comparison_chart import validate_time_comparison_chart + + +def comparison_chart(start="2026-07-16T05:00:00+08:00", interval=15): + times = [datetime.fromisoformat(start) + timedelta(minutes=index * interval) for index in range(13)] + return {"title": "七天同期对比", "chart_type": "time_comparison", + "x_data": [at.strftime("%H:%M") for at in times], "series": [ + {"name": "当前", "type": "line", "data": [25.] * 13, + "sample_times": [at.isoformat() for at in times]}, + *[{"name": f"前{day}天", "type": "scatter", "data": [None] * 10 + [26.] * 3, + "sample_times": [None] * 10 + [(at - timedelta(days=day)).isoformat() for at in times[10:]]} + for day in range(1, 8)], + ]} + + +@pytest.mark.parametrize("start,interval", [ + ("2026-07-16T05:00:00+08:00", 15), + ("2026-07-16T23:00:00+08:00", 15), + ("2026-07-16T05:00:00+08:00", 5), +]) +def test_same_time_comparison_preserves_seven_sparse_days_and_sampling_intervals(start, interval): + chart = comparison_chart(start, interval) + # A missing current observation is still a real empty slot. + chart["series"][0]["data"][11] = None + chart["series"][0]["sample_times"][11] = None + # UTC and local-offset spellings refer to the same real historical instant. + chart["series"][1]["sample_times"][12] = ( + datetime.fromisoformat(chart["series"][1]["sample_times"][12]) + .astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + ) + before = deepcopy(chart) + validate_time_comparison_chart(TaskSafeChart.model_validate(chart).model_dump(mode="json")) + assert chart == before + + +@pytest.mark.parametrize("defect", [ + "appended_history", "duplicate_slots", "wrong_slot", "missing_times", "missing_point_time", + "mixed_dates", "same_day", "future_day", "bar_series", "only_line", "no_current_observations", +]) +def test_special_comparison_rejects_misleading_projections(defect): + chart = comparison_chart() + history = chart["series"][1] + if defect == "appended_history": + chart["x_data"] = [f"07-16 {slot}" for slot in chart["x_data"]] + ["07-15 07:30", "07-15 07:45", "07-15 08:00"] + for series in chart["series"]: + series["data"] += [None] * 3 + series["sample_times"] += [None] * 3 + elif defect == "duplicate_slots": + chart["x_data"][11] = chart["x_data"][10] + elif defect == "wrong_slot": + history["sample_times"][10] = "2026-07-15T07:45:00+08:00" + elif defect == "missing_times": + del history["sample_times"] + elif defect == "missing_point_time": + history["sample_times"][10] = None + elif defect == "mixed_dates": + history["sample_times"][10] = "2026-07-14T07:30:00+08:00" + elif defect in {"same_day", "future_day"}: + day = 16 if defect == "same_day" else 17 + history["sample_times"] = [at.replace("07-15", f"07-{day}") if at else None for at in history["sample_times"]] + elif defect == "bar_series": + history["type"] = "bar" + elif defect == "only_line": + chart["series"] = chart["series"][:1] + elif defect == "no_current_observations": + chart["series"][0]["data"] = chart["series"][0]["sample_times"] = [None] * 13 + with pytest.raises(DomainError) as error: + validate_time_comparison_chart(TaskSafeChart.model_validate(chart).model_dump(mode="json")) + assert error.value.code == "TIME_COMPARISON_INVALID" + + +def test_ordinary_charts_skip_comparison_validation_and_old_mixed_charts_remain_readable(): + chart = {"title": "普通趋势", "chart_type": "line", "x_data": ["07-16 08:00"], + "series": [{"name": "当前", "data": [25.]}]} + validate_time_comparison_chart(chart) + chart["chart_type"] = "bar" + validate_time_comparison_chart(chart) + chart["chart_type"] = "line" + chart["series"].append({"name": "旧版散点", "type": "scatter", "data": [26.]}) + assert TaskSafeChart.model_validate(chart).model_dump(mode="json") == chart + with pytest.raises(DomainError, match="time_comparison"): + validate_time_comparison_chart(chart) + + +@pytest.mark.parametrize("labels,times", [ + (["07:45"], ["9999-12-31T23:45:00Z"]), + (["08:00"], ["0001-01-01T00:00:00+23:59"]), + (["23:45", "00:15"], [None, "0001-01-01T00:15:00+08:00"]), +]) +def test_calendar_range_overflow_returns_a_domain_rejection(labels, times): + chart = {"title": "日期边界", "chart_type": "time_comparison", "x_data": labels, "series": [ + {"name": "当前", "type": "line", "data": [1. if at else None for at in times], "sample_times": times}, + {"name": "历史缺测", "type": "scatter", "data": [None] * len(labels), "sample_times": [None] * len(labels)}, + ]} + with pytest.raises(DomainError, match="日期范围") as error: + validate_time_comparison_chart(TaskSafeChart.model_validate(chart).model_dump(mode="json")) + assert error.value.code == "TIME_COMPARISON_INVALID" + assert error.value.status == 422