fix(tasks): isolate peripheral feedback writes by execution scope

An old development synchronizer could overwrite a production recovery timestamp in a shared project database even after production deployment. Check task scope before feedback or source-withdrawal writes and before idempotency replay.

Cover cross-scope rejection, same-scope success and replay, and preserve actual field feedback time when later order metadata changes.

.env.example

@@ -6,6 +6,8 @@
 # 基础运行
 PROJECT_NAME="TJWater Server"
 ENVIRONMENT="production"
+# 任务领取范围；留空时沿用 ENVIRONMENT。历史任务需先执行 034 迁移。
+OPERATIONAL_TASK_EXECUTION_SCOPE=""
 API_V1_STR="/api/v1"
 NETWORK_NAME="tjwater"
 

app/api/v1/endpoints/operational_tasks.py

@@ -9,6 +9,7 @@ from fastapi import APIRouter, Depends, Header, HTTPException, Query, status
 from psycopg import AsyncConnection
 from app.auth.metadata_dependencies import get_metadata_repository
 
+from app.core.config import settings
 from app.auth.permissions import (
     OPERATIONAL_TASK_EXECUTE,
     OPERATIONAL_TASK_EXTERNAL_SYNC,
@@ -122,6 +123,7 @@ def _service(ctx: ProjectContext, conn: AsyncConnection) -> OperationalTaskServi
         OperationalTaskRepository(conn, ctx.project_id),
         ctx,
         _task_permissions(ctx),
+        execution_scope=settings.operational_task_execution_scope,
     )
 
 

app/api/v1/endpoints/workbench_events.py

@@ -36,6 +36,7 @@ def _service(ctx: ProjectContext, conn: AsyncConnection) -> OperationalTaskServi
         OperationalTaskRepository(conn, ctx.project_id),
         ctx,
         permissions_for_context(ctx),
+        execution_scope=settings.operational_task_execution_scope,
     )
 
 

app/core/config.py

@@ -8,6 +8,12 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
 class Settings(BaseSettings):
     PROJECT_NAME: str = "TJWater Server"
     ENVIRONMENT: str = "production"
+    OPERATIONAL_TASK_EXECUTION_SCOPE: str = ""
+
+    @property
+    def operational_task_execution_scope(self) -> str:
+        return self.OPERATIONAL_TASK_EXECUTION_SCOPE.strip() or self.ENVIRONMENT
+
     API_V1_STR: str = "/api/v1"
     NETWORK_NAME: str = "default_network"
     # 敏感配置加密密钥 (Fernet)

app/domain/schemas/operational_task.py

@@ -574,6 +574,7 @@ class OperationalTaskRun(StrictModel):
 
 
 class OperationalTaskSummary(StrictModel):
+    execution_scope: NonEmptyStr
     project_id: UUID
     task_id: UUID
     task_type: TaskType
@@ -721,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)]

app/infra/db/postgresql/operational_tasks.py

@@ -264,6 +264,7 @@ class OperationalTaskRepository:
     async def list_runs(
         self, *, status: str | None, limit: int, cursor_created_at=None, cursor_run_id=None,
         task_type: str | None = None, task_created_from: datetime | None = None,
+        execution_scope: str = "production",
     ) -> list[dict]:
         result = await self.connection.execute(
             """SELECT run_id,task_id,generation,execution_request_id,status,revision,
@@ -282,18 +283,20 @@ class OperationalTaskRepository:
                        AND t.task_id=operational_task_runs.task_id
                        AND (%s::text IS NULL OR t.task_type=%s::text)
                        AND (%s::timestamptz IS NULL OR t.created_at>=%s::timestamptz)
+                       AND t.document->>'execution_scope'=%s
                  )
                ORDER BY created_at,run_id LIMIT %s""",
             (
                 self.project_id, status, status, cursor_created_at,
                 cursor_created_at, cursor_run_id,
-                task_type, task_type, task_created_from, task_created_from, limit,
+                task_type, task_type, task_created_from, task_created_from, execution_scope, limit,
             ),
         )
         return [dict(row) for row in await result.fetchall()]
 
     async def claim_expired_runs(
         self, limit: int, *, task_type: str | None = None, task_created_from: datetime | None = None,
+        execution_scope: str = "production",
     ) -> list[dict]:
         cursor = await self.connection.execute(
             """SELECT run_id,task_id,generation,execution_request_id,status,revision,
@@ -310,9 +313,10 @@ class OperationalTaskRepository:
                        AND t.task_id=operational_task_runs.task_id
                        AND (%s::text IS NULL OR t.task_type=%s::text)
                        AND (%s::timestamptz IS NULL OR t.created_at>=%s::timestamptz)
+                       AND t.document->>'execution_scope'=%s
                  )
                ORDER BY lease_expires_at,run_id FOR UPDATE SKIP LOCKED LIMIT %s""",
-            (self.project_id, task_type, task_type, task_created_from, task_created_from, limit),
+            (self.project_id, task_type, task_type, task_created_from, task_created_from, execution_scope, limit),
         )
         return [dict(row) for row in await cursor.fetchall()]
 
@@ -711,15 +715,16 @@ class OperationalTaskRepository:
         row = await cursor.fetchone()
         return dict(row) if row else None
 
-    async def recovery_candidates(self, limit: int, created_from=None) -> list[dict]:
+    async def recovery_candidates(self, limit: int, created_from=None, execution_scope: str = "production") -> list[dict]:
         cursor = await self.connection.execute(
             """SELECT document FROM workbench.operational_tasks
                WHERE project_id=%s AND task_type='event_investigation'
                  AND status NOT IN ('cancelled','failed')
                  AND document#>>'{handling_context,phase}'='waiting_scada'
+                 AND document->>'execution_scope'=%s
                  AND (%s::timestamptz IS NULL OR created_at>=%s)
                ORDER BY COALESCE((document#>>'{handling_context,last_checked_at}')::timestamptz,created_at),task_id LIMIT %s""",
-            (self.project_id, created_from, created_from, limit),
+            (self.project_id, execution_scope, created_from, created_from, limit),
         )
         return [row["document"] for row in await cursor.fetchall()]
 

app/mcp/query_service.py

@@ -12,6 +12,7 @@ from datetime import datetime, timezone
 from uuid import uuid4
 from fastapi import HTTPException
 from fastapi.encoders import jsonable_encoder
+from app.core.config import settings
 from app.auth.permissions import permissions_for_context
 from app.infra.db.postgresql import read_queries as reads
 from app.infra.db.postgresql.spatial_queries import nearby_elements
@@ -52,6 +53,7 @@ def task_service(scope: QueryScope) -> OperationalTaskService:
         OperationalTaskRepository(scope.connection, scope.project.project_id),
         scope.project,
         set(permissions_for_context(scope.project)),
+        execution_scope=settings.operational_task_execution_scope,
     )
 
 

app/services/operational_recovery.py

@@ -131,9 +131,11 @@ async def reconcile_recoveries(service: OperationalTaskService, timescale: Async
     from app.infra.db.timescaledb.repositories.scada import ScadaRepository
 
     async with service.repo.transaction():
-        candidates = await service.repo.recovery_candidates(limit, created_from)
+        candidates = await service.repo.recovery_candidates(limit, created_from, service.execution_scope)
     resumed = 0
     for task in candidates:
+        if task.get("execution_scope") != service.execution_scope:
+            continue
         context = task["handling_context"]
         returned_at = datetime.fromisoformat(context["returned_at"])
         incident_at = datetime.fromisoformat(task["trigger"]["occurred_at"])

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)
@@ -88,11 +89,13 @@ def _task_json(model: OperationalTaskSummary) -> dict:
 
 
 class OperationalTaskService:
-    def __init__(self, repository, context, permissions: Iterable[str], *, clock=None):
+    def __init__(self, repository, context, permissions: Iterable[str], *, clock=None,
+                 execution_scope: str = "production"):
         self.repo = repository
         self.context = context
         self.permissions = frozenset(permissions)
         self.clock = clock or (lambda: datetime.now(timezone.utc))
+        self.execution_scope = execution_scope
 
     def _now(self) -> datetime:
         now = self.clock()
@@ -195,6 +198,7 @@ class OperationalTaskService:
             now = self._now()
             task_id = uuid4()
             task = OperationalTaskSummary(
+                execution_scope=self.execution_scope,
                 project_id=self.context.project_id,
                 task_id=task_id,
                 task_type=command.task_type,
@@ -489,6 +493,7 @@ class OperationalTaskService:
         task_type: str | None = None, task_created_from: datetime | None = None,
     ) -> tuple[list[OperationalTaskRun], str | None]:
         rows = await self.repo.list_runs(
+            execution_scope=self.execution_scope,
             status=status,
             limit=limit + 1,
             cursor_created_at=cursor_created_at,
@@ -510,6 +515,9 @@ class OperationalTaskService:
         scope = f"run:{run_id}:claim"
         async with self.repo.transaction():
             await self.repo.require_execution_storage()
+            task = await self._task(task_id)
+            require(task.get("execution_scope") == self.execution_scope,
+                    "EXECUTION_SCOPE_MISMATCH", "当前环境不能领取其他执行范围的任务", status=403)
             replay = await self._request_replay(scope, command.request_id, body)
             if replay:
                 return OperationalTaskRun.model_validate(replay["run"])
@@ -759,6 +767,7 @@ class OperationalTaskService:
         async with self.repo.transaction():
             for run in await self.repo.claim_expired_runs(
                 limit, task_type=task_type, task_created_from=task_created_from,
+                execution_scope=self.execution_scope,
             ):
                 # Lease recovery already holds a row lock. Try rather than wait
                 # for the shared run lock to avoid reversing the lock order of
@@ -829,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"
             ):
@@ -1087,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":
@@ -1498,11 +1513,13 @@ class OperationalTaskService:
         body = _json(command)
         scope = f"task:{task_id}:external-feedback"
         async with self.repo.transaction():
+            await self.repo.lock(f"task:{task_id}")
+            task = await self._task(task_id)
+            require(task.get("execution_scope") == self.execution_scope,
+                    "EXECUTION_SCOPE_MISMATCH", "当前环境不能同步其他执行范围的任务", status=403)
             replay = await self._request_replay(scope, command.request_id, body)
             if replay:
                 return ExternalFeedbackResult(task=await self.detail(task_id), **replay)
-            await self.repo.lock(f"task:{task_id}")
-            task = await self._task(task_id)
             if task["task_type"] == "event_investigation":
                 self._require_actionable_task(task)
                 own_source = task["source"].get("external_id")
@@ -1659,11 +1676,13 @@ class OperationalTaskService:
         body = _json(command)
         scope = f"task:{task_id}:withdraw-external-source"
         async with self.repo.transaction():
+            await self.repo.lock(f"task:{task_id}")
+            task = await self._task(task_id)
+            require(task.get("execution_scope") == self.execution_scope,
+                    "EXECUTION_SCOPE_MISMATCH", "当前环境不能同步其他执行范围的任务", status=403)
             replay = await self._request_replay(scope, command.request_id, body)
             if replay:
                 return await self.detail(task_id)
-            await self.repo.lock(f"task:{task_id}")
-            task = await self._task(task_id)
             source = task["source"]
             require(
                 task["task_type"] == "external_work_order"
@@ -2516,6 +2535,7 @@ class OperationalTaskService:
         due_at: datetime, now: datetime,
     ) -> OperationalTaskSummary:
         return OperationalTaskSummary(
+            execution_scope=self.execution_scope,
             project_id=self.context.project_id, task_id=task_id,
             task_type="scheduled_flow", workflow=schedule.schedule_key,
             lifecycle_status="open", status="queued", priority=schedule.priority, revision=1,

app/services/scada_replay.py

@@ -6,6 +6,7 @@ import json
 from uuid import NAMESPACE_URL, uuid4, uuid5
 
 from fastapi import HTTPException
+from app.core.config import settings
 from app.auth.permissions import permissions_for_context
 
 from app.domain.schemas.project_settings import ScadaMonitoringSettings
@@ -439,7 +440,8 @@ class ScadaReplayService:
         alerts = TaskAlertService(TaskAlertRepository(self.repo.connection, self.context.project_id),
                                   self.context, internal_producer_id=producer)
         tasks = OperationalTaskService(OperationalTaskRepository(self.repo.connection, self.context.project_id),
-                                       self.context, permissions_for_context(self.context))
+                                       self.context, permissions_for_context(self.context),
+                                       execution_scope=settings.operational_task_execution_scope)
         for event in changes:
             # Period diagnosis owns measurement notifications during manual steps.
             # Retain rule events as evidence; quality alerts remain independent.
@@ -503,6 +505,7 @@ class ScadaReplayService:
                 tasks = OperationalTaskService(
                     OperationalTaskRepository(self.repo.connection, self.context.project_id),
                     self.context, permissions_for_context(self.context),
+                    execution_scope=settings.operational_task_execution_scope,
                 )
                 await tasks.advance_replay_investigation(UUID(incident["task_id"]), {
                     "replay_id": document["replay_id"], "knowledge_cutoff": monitoring["knowledge_cutoff"],
@@ -520,6 +523,7 @@ class ScadaReplayService:
             OperationalTaskRepository(self.repo.connection, self.context.project_id),
             self.context,
             permissions_for_context(self.context),
+            execution_scope=settings.operational_task_execution_scope,
         )
         external = incident["external_id"]
         pressure = monitoring["pressure"]

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 必须来自当前窗口之前的对应日期，不能使用当天或未来采样")

contracts/answer-ui-catalog.json

@@ -310,7 +310,8 @@
           "type": "string",
           "enum": [
             "line",
-            "bar"
+            "bar",
+            "time_comparison"
           ]
         },
         "x_axis_name": {

contracts/manifest.json

@@ -3,7 +3,7 @@
   "contracts": {
     "server": {
       "file": "server-v1.openapi.json",
-      "sha256": "939f38733e943f21cc2e3d9f344fa8122ca7a7b35cfebf96ff3f00f469026ceb"
+      "sha256": "426b8d9170f253aa7132e9376031a7da8c04ac6f6e693948890d6f343420af1b"
     }
   }
 }

contracts/server-v1.openapi.json

@@ -4683,6 +4683,11 @@
             "title": "Execution Request Id",
             "type": "string"
           },
+          "execution_scope": {
+            "minLength": 1,
+            "title": "Execution Scope",
+            "type": "string"
+          },
           "handling_context": {
             "additionalProperties": {
               "$ref": "#/components/schemas/JsonValue"
@@ -4846,6 +4851,7 @@
           }
         },
         "required": [
+          "execution_scope",
           "project_id",
           "task_id",
           "task_type",
@@ -5660,6 +5666,11 @@
             "title": "Execution Request Id",
             "type": "string"
           },
+          "execution_scope": {
+            "minLength": 1,
+            "title": "Execution Scope",
+            "type": "string"
+          },
           "handling_context": {
             "additionalProperties": {
               "$ref": "#/components/schemas/JsonValue"
@@ -5773,6 +5784,7 @@
           }
         },
         "required": [
+          "execution_scope",
           "project_id",
           "task_id",
           "task_type",
@@ -10211,7 +10223,8 @@
           "chart_type": {
             "enum": [
               "line",
-              "bar"
+              "bar",
+              "time_comparison"
             ],
             "title": "Chart Type",
             "type": "string"

contracts/view-command.schema.json

@@ -211,7 +211,8 @@
               "type": "string",
               "enum": [
                 "line",
-                "bar"
+                "bar",
+                "time_comparison"
               ]
             },
             "x_axis_name": {

docs/operational-task-workers.md

@@ -190,3 +190,29 @@ OPERATIONAL_TASKS_STORAGE_TEST=1 conda run -n server python -m pytest tests/inte
 ```
 
 第二条需要当前配置的 PostgreSQL 可连接及测试 DDL 权限。测试在强制回滚事务中创建表结构和隔离项目数据，结束后不保留这些修改；不代替正式迁移。外部工单调用仍使用隔离 HTTP transport，不会操作实际工单系统。该测试覆盖持久化报告、反馈重新排队、旧会话拒绝、人工确认与丢响应回执核对，而非真实模型推理质量。
+
+### 共用数据库时的执行范围
+
+任务创建时，Server 将 `OPERATIONAL_TASK_EXECUTION_SCOPE`（留空则使用
+`ENVIRONMENT`）写入 `workbench.operational_tasks.document.execution_scope`。
+本地配置 `development`，线上配置 `production`；浏览器不提交此字段，
+重试保留原范围。`execution_scope` 为必填字段，运行时代码不为缺失值兜底。
+现有任务通过 `resources/sql/034_operational_task_execution_scope.sql` 一次性补为
+`production`；已有明确范围的任务保持原值。此回填是历史任务的迁移策略，
+不代表通过旧记录推断出了创建环境。
+
+迁移需要与新版本切换协调执行：暂停任务写入，执行每个业务库的 SQL，
+启动新版本并恢复任务写入。旧版本的严格模型不接受新增 JSON 字段，
+新版本也拒绝未迁移数据；不能在旧版本仍服务任务请求时单独回填。
+SQL 同时增加必填范围约束，不设置数据库默认值。回滚版本前需同步处理该约束
+与 JSON 字段，不得直接切回旧镜像。
+
+Worker 应连接所属环境的 `OPERATIONAL_WORKER_SERVER_URL` 和
+`OPERATIONAL_WORKER_AGENT_URL`。Server 在运行列表、领取、租约回收与
+SCADA recovery 候选查询上限制执行范围；直接领取其他范围返回
+`403 EXECUTION_SCOPE_MISMATCH`。外部工单反馈与来源撤回也必须匹配任务的执行范围，
+并在幂等重放前校验；开发环境同步器不能更新生产任务或重置其恢复监测起点。
+这属于配置隔离，不能替代服务凭据授权：
+不要让本地 Worker 通过线上 Server 领取任务。普通任务与历史记录仍共用存储。
+
+修改范围配置需要重启相应 Server。暂停的本地 Worker 不会因代码更新恢复。

resources/sql/034_operational_task_execution_scope.sql

@@ -0,0 +1,26 @@
+-- Run against each project business database during the coordinated Server cutover.
+-- Historical tasks are explicitly assigned to production; retain existing scopes.
+BEGIN;
+SET LOCAL lock_timeout = '3s';
+SET LOCAL statement_timeout = '30s';
+
+UPDATE workbench.operational_tasks
+SET document = jsonb_set(document, '{execution_scope}', '"production"'::jsonb)
+WHERE NOT (document ? 'execution_scope');
+
+DO $$
+BEGIN
+    IF NOT EXISTS (
+        SELECT 1 FROM pg_constraint
+        WHERE conrelid = 'workbench.operational_tasks'::regclass
+          AND conname = 'operational_tasks_execution_scope_required'
+    ) THEN
+        ALTER TABLE workbench.operational_tasks
+            ADD CONSTRAINT operational_tasks_execution_scope_required CHECK (
+                (jsonb_typeof(document->'execution_scope') = 'string'
+                 AND length(btrim(document->>'execution_scope')) > 0) IS TRUE
+            );
+    END IF;
+END $$;
+
+COMMIT;

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转换后的当地日期逐日回退，不包含当天；按实际采样周期保留缺口，按需翻页并报告实际参考点数。基线未冻结、读取失败或缺测时明确说明并保留已有三小时趋势，不能静默退化为前一天或相邻两个时刻。
+定位输入具备后继续定位。压力、流量按设备另发七天同期复合图，不占用或替代全量当前总览。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的本地日期回退，不含当天。按实际周期保留缺口，按需翻页并报告参考点数；基线未冻结、读取失败或缺测时说明并保留当前趋势，不退化为前一天或相邻两点。
 
 影响说明列出异常前参考、异常期间和已有恢复观测中的压力、流量或水质变化、持续时间、涉及测点及已核实资产。可比对象、单位、时段与运行条件下给出变化量，分别标明实际观测、模型预测和拓扑隔离。停水范围与漏量引用相应拓扑、水力、流量平衡或现场依据；恢复观测、用户关联或现场证据缺失时列为具体待核实项。
 

tests/unit/test_operational_tasks.py

@@ -235,6 +235,7 @@ async def test_targeted_notification_cannot_be_acknowledged_without_permission()
 def _task_document(context, now, *, lease=None):
     task_id = uuid4()
     return OperationalTaskSummary(
+        execution_scope="production",
         project_id=context.project_id,
         task_id=task_id,
         task_type="event_investigation",
@@ -436,16 +437,39 @@ def _active_run(task, now):
 
 
 @pytest.mark.anyio
-async def test_run_claim_persists_agent_instance_ownership():
+@pytest.mark.parametrize("task_scope,worker_scope,allowed", [
+    (None, "production", False),
+    (None, "development", False),
+    ("production", "production", True),
+    ("production", "development", False),
+    ("development", "production", False),
+    ("development", "development", True),
+])
+async def test_run_claim_persists_agent_instance_ownership(task_scope, worker_scope, allowed):
     now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=None)
     task.update(status="queued", agent_session_id=None, agent_lease_expires_at=None)
+    if task_scope is None:
+        task.pop("execution_scope", None)
+    else:
+        task["execution_scope"] = task_scope
     run = _active_run(task, now)
     run.update(status="queued", revision=1, agent_session_id=None,
                lease_expires_at=None, heartbeat_at=None, started_at=None)
     repository = RunTaskRepository(task, run)
-    service = OperationalTaskService(repository, context, set(), clock=lambda: now)
+    service = OperationalTaskService(repository, context, set(), clock=lambda: now,
+                                     execution_scope=worker_scope)
+
+    if not allowed:
+        with pytest.raises(DomainError) as failure:
+            await service.claim_run(task["task_id"], run["run_id"], AgentRunCommand(
+                request_id=run["execution_request_id"], expected_revision=1,
+                source_session_id="session-1", agent_instance_id="agent-a"))
+        assert failure.value.code == "EXECUTION_SCOPE_MISMATCH"
+        assert repository.run["status"] == "queued"
+        assert repository.task["status"] == "queued"
+        return
 
     claimed = await service.claim_run(
         task["task_id"], run["run_id"],
@@ -593,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"]},
                            ]}},
@@ -613,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):
@@ -2371,3 +2455,91 @@ async def test_recovery_resolves_alerts_with_original_facts_and_stable_event_id(
         assert command.event_id == uuid5(NAMESPACE_URL, f"recovery:{task_id}:recovery:evidence-1:incident-1")
         assert command.source.model_dump(mode="json") == alert["source"]
         assert command.title == alert["title"] and command.description == alert["description"]
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("scope", ["production", "development"])
+async def test_task_creation_persists_server_execution_scope(monkeypatch, scope):
+    from app.domain.schemas.operational_task import OperationalTaskCreate
+
+    now = datetime(2026, 11, 8, tzinfo=timezone.utc)
+    context = _context()
+    document = _task_document(context, now)
+    repository = TaskRepository(document)
+    repository.insert_task = AsyncMock()
+    repository.insert_run = AsyncMock()
+    service = OperationalTaskService(repository, context, set(), clock=lambda: now,
+                                     execution_scope=scope)
+    monkeypatch.setattr(service, "_task_event", AsyncMock())
+    monkeypatch.setattr(service, "_run_event", AsyncMock())
+    monkeypatch.setattr(service, "detail", AsyncMock())
+    command = OperationalTaskCreate(request_id=uuid4(), **{
+        key: document[key] for key in ("task_type", "workflow", "title", "summary", "source", "trigger")
+    })
+    await service.create(command)
+    assert repository.insert_task.await_args.args[0]["execution_scope"] == scope
+
+
+def test_task_scope_is_required_in_persisted_contract():
+    task = _task_document(_context(), datetime(2026, 11, 8, tzinfo=timezone.utc))
+    task.pop("execution_scope")
+    with pytest.raises(ValidationError) as error:
+        OperationalTaskSummary.model_validate(task)
+    assert any(item["loc"] == ("execution_scope",) and item["type"] == "missing"
+               for item in error.value.errors())
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("operation", ["feedback", "withdrawal"])
+@pytest.mark.parametrize("task_scope,server_scope,allowed", [
+    (None, "production", False),
+    (None, "development", False),
+    ("production", "production", True),
+    ("production", "development", False),
+    ("development", "production", False),
+    ("development", "development", True),
+])
+async def test_external_sync_only_mutates_tasks_in_its_execution_scope(operation, task_scope, server_scope, allowed):
+    from copy import deepcopy
+
+    now = datetime(2026, 11, 8, tzinfo=timezone.utc)
+    context = _context()
+    task = _task_document(context, now, lease=None)
+    source_id = f"peripheral_sample:{uuid4()}"
+    task.update(task_type="external_work_order", source={
+        "kind": "external_system", "external_id": source_id,
+        "external_revision": 1, "event_kind": "general",
+    })
+    if task_scope is None:
+        task.pop("execution_scope")
+    else:
+        task["execution_scope"] = task_scope
+    repository = TaskRepository(task)
+    service = OperationalTaskService(repository, context, set(), clock=lambda: now, execution_scope=server_scope)
+    if operation == "feedback":
+        command = ExternalFeedbackCommand(request_id=uuid4(), source_event_id=source_id,
+            source_revision=2, status="completed", occurred_at=now, detail="已关阀修复管道")
+        method = "external_feedback"
+    else:
+        command = ExternalSourceWithdrawalCommand(request_id=uuid4(), source_event_id=source_id)
+        method = "withdraw_external_source"
+    before = deepcopy(repository.task)
+    if not allowed:
+        with pytest.raises(DomainError) as error:
+            await getattr(service, method)(task["task_id"], command)
+        assert error.value.code == "EXECUTION_SCOPE_MISMATCH" and error.value.status == 403
+        assert repository.task == before
+        assert repository.feedback == repository.evidence == repository.requests == {}
+        return
+
+    await getattr(service, method)(task["task_id"], command)
+    assert repository.task["status"] == ("completed" if operation == "feedback" else "cancelled")
+    # Check scope before idempotency replay as well as before the first write.
+    await getattr(service, method)(task["task_id"], command)
+    before = deepcopy(repository.task)
+    other = OperationalTaskService(repository, context, set(), clock=lambda: now,
+        execution_scope="development" if server_scope == "production" else "production")
+    with pytest.raises(DomainError) as error:
+        await getattr(other, method)(task["task_id"], command)
+    assert error.value.code == "EXECUTION_SCOPE_MISMATCH" and error.value.status == 403
+    assert repository.task == before

tests/unit/test_peripheral_task_sync.py

@@ -226,7 +226,7 @@ async def test_sample_sync_rejects_cross_project_data_before_writing():
 
 @pytest.mark.anyio
 @pytest.mark.parametrize("targeted", [False, True])
-@pytest.mark.parametrize("order_revision", [1, 2])
+@pytest.mark.parametrize("order_revision", [1, 4])
 @pytest.mark.parametrize(
     ("investigation_status", "expected_feedback_count", "routes_to_investigation"),
     [
@@ -254,9 +254,13 @@ async def test_created_work_order_receipt_preserves_projection_and_routes_active
                 "updated_at": "2026-09-23T09:00:00Z",
                 "title": "采样工单", "priority": "normal",
                 "work_order_type": "general", "created_at": "2026-09-23T08:00:00Z",
-                "scheduled_at": None,
+                "scheduled_at": "2026-09-23T08:30:00Z",
                 "activity": ([{"sequence": 2, "occurred_at": "2026-09-23T08:45:00Z", "actor": "现场人员",
-                               "kind": "feedback", "title": "现场反馈", "detail": "复测余氯0.2mg/L"}]
+                               "kind": "feedback", "title": "现场反馈", "detail": "复测余氯0.2mg/L"},
+                              {"sequence": 3, "occurred_at": "2026-09-23T08:50:00Z", "actor": "调度员",
+                               "kind": "updated", "title": "工单信息更新", "detail": "更新了状态。"},
+                              {"sequence": 4, "occurred_at": "2026-09-23T09:00:00Z", "actor": "调度员",
+                               "kind": "updated", "title": "工单信息更新", "detail": "更新了优先级。"}]
                              if order_revision > 1 else []),
             }
             if targeted:

tests/unit/test_scada_replay_service.py

@@ -579,7 +579,7 @@ def test_suspected_monitoring_creates_burst_report_with_frozen_replay_context(mo
     monkeypatch.setattr("app.infra.db.postgresql.operational_tasks.OperationalTaskRepository", lambda *_: object())
     monkeypatch.setattr(
         "app.services.operational_tasks.OperationalTaskService",
-        lambda *_: SimpleNamespace(create=create_task),
+        lambda *_, **_kwargs: SimpleNamespace(create=create_task),
     )
     monitoring = {
         "status": "suspected_burst",
@@ -618,7 +618,7 @@ def test_monitoring_summary_counts_signals_without_copying_reading_details(monke
     _, service, *_ = setup(monkeypatch)
     create_task = AsyncMock(return_value=SimpleNamespace(task_id=uuid4()))
     monkeypatch.setattr("app.infra.db.postgresql.operational_tasks.OperationalTaskRepository", lambda *_: object())
-    monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService", lambda *_: SimpleNamespace(create=create_task))
+    monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService", lambda *_, **_kwargs: SimpleNamespace(create=create_task))
     monitoring = incident_monitoring(0)
     monitoring["anomaly_signals"] = [
         {"device_ids": [f"P{i}"], "detail": f"P{i}=28.14402136077437 m，超出冻结正常同时段范围 [28.3253, 28.8685] m"}
@@ -641,7 +641,7 @@ def test_continuing_burst_reuses_task_across_persisted_periods(monkeypatch):
     monkeypatch.setattr("app.services.scada_replay.permissions_for_context", lambda _: {"task.write"})
     monkeypatch.setattr("app.infra.db.postgresql.operational_tasks.OperationalTaskRepository", lambda *_: object())
     monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService",
-                        lambda *_: SimpleNamespace(create=create_task, advance_replay_investigation=advance_task))
+                        lambda *_, **_kwargs: SimpleNamespace(create=create_task, advance_replay_investigation=advance_task))
     document = {"replay_id": str(uuid4())}
 
     async def run():
@@ -683,7 +683,7 @@ def test_incident_recovers_once_then_allows_new_accident(monkeypatch):
     monkeypatch.setattr("app.services.scada_replay.permissions_for_context", lambda _: {"task.write"})
     monkeypatch.setattr("app.infra.db.postgresql.operational_tasks.OperationalTaskRepository", lambda *_: object())
     monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService",
-                        lambda *_: SimpleNamespace(create=create_task))
+                        lambda *_, **_kwargs: SimpleNamespace(create=create_task))
     document = {"replay_id": str(uuid4())}
 
     async def run():
@@ -784,7 +784,7 @@ def test_combined_monitoring_keeps_rule_evidence_and_quality_notifications(monke
     monkeypatch.setattr("app.infra.db.postgresql.task_alerts.TaskAlertRepository", lambda *_: object())
     monkeypatch.setattr("app.services.task_alerts.TaskAlertService", lambda *_args, **_kwargs: SimpleNamespace(push=push))
     monkeypatch.setattr("app.infra.db.postgresql.operational_tasks.OperationalTaskRepository", lambda *_: object())
-    monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService", lambda *_: SimpleNamespace(create=AsyncMock()))
+    monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService", lambda *_, **_kwargs: SimpleNamespace(create=AsyncMock()))
     def event(rule, category=None):
         return {"event_id": str(uuid4()), "rule_id": rule, "change": "raised", "occurred_at": START.isoformat(),
                 "detail": "检测信号", "category": category, "task_id": None}
@@ -812,7 +812,7 @@ def test_failed_incident_publication_rolls_back_state_and_retries_same_task_requ
     create_task = AsyncMock(return_value=SimpleNamespace(task_id=uuid4()))
     monkeypatch.setattr("app.services.scada_replay.permissions_for_context", lambda _: {"task.write"})
     monkeypatch.setattr("app.infra.db.postgresql.operational_tasks.OperationalTaskRepository", lambda *_: object())
-    monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService", lambda *_: SimpleNamespace(create=create_task))
+    monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService", lambda *_, **_kwargs: SimpleNamespace(create=create_task))
     def evaluate(_doc, _rows, *, period_start, period_end, batch_sequence):
         return {**incident_monitoring(0), "period_start": period_start.isoformat(), "period_end": period_end.isoformat(),
                 "knowledge_cutoff": period_end.isoformat(), "batch_sequence": batch_sequence}
@@ -843,7 +843,7 @@ def test_any_single_measurement_anomaly_enters_diagnosis_below_burst_threshold(m
     _, service, *_ = setup(monkeypatch)
     create_task = AsyncMock(return_value=SimpleNamespace(task_id=uuid4()))
     monkeypatch.setattr("app.infra.db.postgresql.operational_tasks.OperationalTaskRepository", lambda *_: object())
-    monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService", lambda *_: SimpleNamespace(create=create_task))
+    monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService", lambda *_, **_kwargs: SimpleNamespace(create=create_task))
     monitoring = incident_monitoring(0, status="attention")
     for key in ("pressure", "outlet_flow", "burst_detection"):
         monitoring[key]["hit"] = key == indicator

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