fix(charts): validate same-time comparison publications

Prompt-only projection rules allowed historical samples to be appended to the current axis. Enforce typed comparison semantics at presentation and report publication, preserving historical read contracts and returning actionable validation errors.

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

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,35 @@ 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())

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