fix(skills): keep comparison guidance within inline response budget

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,16 +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#>>'{source,replay_context}' IS NULL
+                 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/application.py

@@ -429,6 +429,9 @@ class McpQueryApplication:
                 output = display_query_result(result.model_dump(mode="json"))
                 content = []
                 from app.mcp.extended_service import ARTIFACT_TOOLS
+                # Complete SCADA inputs must reach the model intact, not as a first-page preview.
+                # The shared 8 MiB result ceiling above still applies before persistence and output.
+                complete_scada = name == "query_scada_readings" and args.device_ids is None
                 if (not include_full and name == "list_scada_devices"
                         and len(json.dumps(output, ensure_ascii=False).encode()) > 12000):
                     preview = compact_scada_catalog(output, budget=12000)
@@ -436,6 +439,7 @@ class McpQueryApplication:
                         output = preview
                 if (
                     not include_full
+                    and not complete_scada
                     and name not in ARTIFACT_TOOLS
                     and len(json.dumps(output, ensure_ascii=False).encode()) > 12000
                 ):

app/mcp/catalog.py

@@ -35,6 +35,12 @@ class QueryTool:
                 "。默认每页100个，最多500个；page.has_more=true时继续递增offset读取后续页面。"
                 "大目录内联保留本页全部设备的编号、类型和单位，模型映射及坐标用get_scada_device核验。"
             )
+        elif self.name == "query_scada_readings":
+            suffix = (
+                "。首次调查省略device_ids，Server读取完整设备属性并自动翻页读取窗口内全部授权观测，"
+                "返回devices、readings和coverage，包括缺测及冻结范围外设备；全量结果完整内联。"
+                "显式device_ids仅用于后续局部复核，不能替代全量核查。"
+            )
         else:
             suffix = (
                 "。先限定对象和时间范围，再读取明细；集合默认20条，按需逐页扩大，"
@@ -64,7 +70,7 @@ _specs = [
     ("optimize_sensor_placement", "使用灵敏度或K-means算法推荐压力监测点；返回候选节点，正式部署另行保存并核验观测能力", SensorPlacementInput, "optimization.run", "analysis", False, True),
     ("run_leakage_identification", "基于SCADA压力与模型进行DMA漏损分配识别，需有依据的总漏损流量和时间对齐；返回待现场核验的区域估计。内部执行遗传算法与水力计算", LeakageIdentificationInput, "burst.run", "analysis", False, True),
     ("run_burst_detection", "爆管异常侦测：使用至少5个完整压力传感器此前15天数据；返回压力异常信号，候选管段由定位工具分析", BurstDetectionInput, "burst.run", "analysis", False, True),
-    ("run_burst_location", "爆管候选定位：对比异常与正常压力，可辅助流量；依据现有异常形成候选，水力算法使用有依据的burst_leakage作为输入。任务上下文location_demo.available=true时可省略漏量，Server直接返回候选管段与相似度，随后继续地图展示、隔离边界与影响分析。压力节点可省略，自动读取项目配置；monitoring正常窗省略时默认前一天同时间段，须说明假设；回放必须显式提供冻结正常窗。simulation须有已有分析运行UUID及同窗模型基线，结果标注模型来源", BurstLocationInput, "burst.run", "analysis", False, True),
+    ("run_burst_location", "爆管候选定位：对比异常与正常压力，可辅助流量；依据现有异常形成候选，水力算法使用有依据的burst_leakage作为输入。任务上下文location_capability.available=true时可省略漏量，Server直接返回候选管段与相似度，随后继续地图展示、隔离边界与影响分析。压力节点可省略，自动读取项目配置；monitoring正常窗省略时默认前一天同时间段，须说明假设；回放必须显式提供冻结正常窗。simulation须有已有分析运行UUID及同窗模型基线，结果标注模型来源", BurstLocationInput, "burst.run", "analysis", False, True),
     ("analyze_valve_isolation", "按事故候选要素分析拓扑隔离边界、建议关闭阀门和受影响节点；候选分析无需等待现场确诊。输出为拓扑隔离与供水失源范围；水力影响另用关阀模拟评估，现场操作前核实阀门映射及可用性", ValveIsolationInput, "simulation.run", "analysis", False, True),
     ("generate_node_coverage", "为当前项目已核实的模型节点集合生成空间覆盖示意，适用于候选分布、监测覆盖比较、区域成员、隔离或模拟结果节点等，支持各类项目分析。支持最多8组、合计10000个唯一节点，从真实GIS查询坐标；监测设备、投诉或地址须先关联模型节点。可按连接分组，记录凹包/缓冲参数、缺失节点和display_only；结果为节点空间覆盖示意，保留分组及几何覆盖范围。返回result.map_payload及coverage，任务中用set_overlay引用本次evidence_id、省略data展示；地图通过展示工具发布", NodeCoverageInput, "webgis.view", "analysis", False, True),
     (

app/mcp/models.py

@@ -106,8 +106,20 @@ class WindowInput(PageInput):
 
 
 class ScadaInput(WindowInput):
-    device_ids: list[Id] = Field(min_length=1, max_length=20)
-    field: Literal["monitored_value", "cleaned_value"] = "monitored_value"
+    device_ids: list[Id] | None = Field(
+        default=None, min_length=1, max_length=20,
+        description="Omit to read the complete SCADA catalogue and all authorized readings in the window; explicit IDs select a paginated detail query.",
+    )
+    field: Literal["monitored_value", "cleaned_value"] = Field(
+        default="monitored_value",
+        description="Field for explicit-device detail queries; complete all-device reads return both raw and cleaned values.",
+    )
+
+    @model_validator(mode="after")
+    def complete_query_starts_at_zero(self):
+        if self.device_ids is None and self.offset:
+            raise ValueError("All-device SCADA queries read every page internally; offset must be zero")
+        return self
 
 
 class BaselineSeriesInput(WindowInput):

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,
     )
 
 
@@ -129,6 +131,11 @@ async def authorize_query(
     if not granted and binding.task_id and source in binding.task_scope_cap:
         async with scope.connection.transaction():
             granted = await task_service(scope).can_read_source(binding.task_id, source)
+            if granted and tool.name == "query_scada_readings" and args.device_ids is None:
+                granted = (
+                    "scada_devices" in binding.task_scope_cap
+                    and await task_service(scope).can_read_source(binding.task_id, "scada_devices")
+                )
     if not granted:
         raise HTTPException(403, "Data source permission is required")
     if tool.name == "run_burst_location" and args.burst_leakage is None:
@@ -175,13 +182,15 @@ async def execute_query(
     source_ref = "mcp:" + str(uuid4())
     evidence_id = None
     replay_context = getattr(binding, "replay_context", None)
-    if name == "query_scada_readings" and replay_context:
+    if name == "query_scada_readings" and args.device_ids is None:
+        from app.mcp.scada_readings import read_complete_scada
+        data, units, gaps, origin = await read_complete_scada(scope, binding, args)
+    elif name == "query_scada_readings" and replay_context:
         from app.services.scada_replay_queries import replay_snapshot_rows
         rows = await replay_snapshot_rows(conn, scope.project.project_id, replay_context,
                                           args.device_ids, args.start, args.end, args.limit + 1, args.offset)
         data, page, truncated = page_result(rows, args)
         origin = "synthetic"
-        gaps.append("历史数据回放证据，仅覆盖该轮调查已提交的批次，不表示现场实时观测。")
     elif name == "run_burst_location" and replay_context:
         from app.mcp.analysis_service import calculate_replay_burst_location
         from app.services.scada_replay_queries import replay_burst_location_inputs
@@ -190,7 +199,6 @@ async def execute_query(
             conn, scope.project.project_id, replay_context, args,
         )
         result = await location_demo_result(conn, scope.project.project_code, args)
-        used_demo = result is not None
         await conn.commit()
         if result is None:
             async with scope.simulation_routing():
@@ -204,11 +212,7 @@ async def execute_query(
             "result": result,
         }
         origin = "synthetic"
-        gaps = [
-            "定位只使用任务知识截止前已固化的回放清洗输出；候选排序依赖漏量假设、模型和测点覆盖，仍需现场核实。"
-        ]
-        if used_demo:
-            gaps = [result["notice"]] if result.get("notice") else []
+        gaps = [result["notice"]] if result.get("notice") else []
     elif name == "run_burst_location" and (
         demo_result := await location_demo_result(conn, scope.project.project_code, args)
     ) is not None:
@@ -488,7 +492,7 @@ async def execute_query(
                     and "run_burst_location" in binding.tool_cap):
                 from app.services.burst_location_demo import location_demo_context
 
-                data["location_demo"] = location_demo_context(
+                data["location_capability"] = location_demo_context(
                     getattr(scope.project, "project_code", None), replay_context,
                 )
         else:

app/mcp/scada_readings.py

@@ -0,0 +1,121 @@
+"""Complete project SCADA reads with explicit frozen-scope and missing-data accounting."""
+
+import json
+from collections import Counter
+
+from fastapi import HTTPException
+from fastapi.encoders import jsonable_encoder
+
+from app.mcp.models import BoundContext, ScadaInput
+from app.mcp.scope import QueryScope
+from app.infra.db.postgresql import read_queries as reads
+from app.infra.db.postgresql.scada_replays import ScadaReplayRepository
+from app.services.scada_replay_queries import check_replay_window, check_snapshot_start
+
+
+MAX_COMPLETE_SCADA_BYTES = 8 * 1024 * 1024
+
+
+async def read_complete_scada(
+    scope: QueryScope, binding: BoundContext, args: ScadaInput,
+) -> tuple[dict, dict, list[str], str]:
+    """Exhaust source pages; never treat a first page or selected device subset as a full scan."""
+    devices = []
+    offset = 0
+    size = 0
+
+    def account(items):
+        nonlocal size
+        size += len(json.dumps(jsonable_encoder(items), ensure_ascii=False).encode())
+        if size > MAX_COMPLETE_SCADA_BYTES:
+            raise HTTPException(413, "Complete SCADA result exceeds 8 MiB; split the time window and retain every device")
+
+    while True:
+        page = await reads.scada_devices(scope.connection, limit=501, offset=offset)
+        batch = page[:500]
+        account(batch)
+        devices.extend(batch)
+        if len(page) <= 500:
+            break
+        offset += len(batch)
+
+    project_device_count = len(devices)
+    context = getattr(binding, "replay_context", None)
+    catalog = {str(device["device_id"]): device for device in devices}
+    outside_scope = []
+    if context:
+        check_replay_window(context, args.start, args.end)
+        repository = ScadaReplayRepository(scope.connection, scope.project.project_id)
+        run = await repository.get(context["replay_id"])
+        if not run:
+            raise HTTPException(404, "SCADA replay was not found in the current project")
+        check_snapshot_start(run, args.start)
+        frozen = {str(device["device_id"]): device for device in run.get("devices", [])}
+        selected = list(dict.fromkeys(run["config"]["device_ids"]))
+        outside_scope = sorted(set(catalog) - set(selected))
+        # Retain devices removed from the current catalogue but present in the frozen source.
+        for device in selected:
+            if device not in catalog:
+                catalog[device] = frozen.get(device, {"device_id": device})
+        devices = list(catalog.values())
+    else:
+        selected = list(catalog)
+
+    rows = []
+    offset = 0
+    # The public limit controls source page size, not completeness of this operation.
+    page_size = min(max(args.limit, 1000), 10000)
+
+    async def read_pages(connection=None):
+        nonlocal offset
+        while selected:
+            if context:
+                page = await repository.detected_outputs(
+                    context["replay_id"], selected, args.start, args.end, context["batch_sequence"],
+                    limit=page_size + 1, offset=offset,
+                )
+            else:
+                page = await reads.series_rows(
+                    connection, source="scada", element_type="junction", ids=selected,
+                    start=args.start, end=args.end, field=None,
+                    limit=page_size + 1, offset=offset,
+                )
+            batch = page[:page_size]
+            if context:
+                # Match the existing frozen-reading contract; internal source hashes are not observations.
+                fields = ("time", "device_id", "monitored_value", "cleaned_value", "quality", "reason")
+                batch = [{key: row[key] for key in fields if key in row} for row in batch]
+            account(batch)
+            rows.extend(batch)
+            if len(page) <= page_size:
+                break
+            offset += len(batch)
+
+    if context:
+        await read_pages()
+    elif selected:
+        async with scope.timescale() as connection:
+            await read_pages(connection)
+
+    counts = Counter(str(row["device_id"]) for row in rows)
+    missing = sorted(set(selected) - counts.keys())
+    coverage = {
+        "mode": "all_scada", "query_complete": True,
+        "project_device_count": project_device_count, "queried_device_count": len(selected),
+        "devices_with_readings": len(counts), "reading_count": len(rows),
+        "queried_device_ids": selected, "missing_device_ids": missing,
+        "outside_frozen_scope_device_ids": outside_scope,
+        "readings_per_device": {device: counts[device] for device in selected},
+        "replay_id": context["replay_id"] if context else None,
+        "batch_sequence": context["batch_sequence"] if context else None,
+    }
+    data = {"devices": devices, "readings": rows, "coverage": coverage}
+    account([coverage])
+    gaps = []
+    if missing:
+        gaps.append(f"{len(missing)} queried SCADA devices have no readings in this window; see coverage.missing_device_ids")
+    if outside_scope:
+        gaps.append(f"{len(outside_scope)} project devices are outside this frozen replay; see coverage.outside_frozen_scope_device_ids")
+    units = {device: catalog[device]["measurement_unit"] for device in selected
+             if catalog[device].get("measurement_unit")}
+    return data, units, gaps, "synthetic" if context else "system_data"

app/services/operational_recovery.py

@@ -20,8 +20,7 @@ RECOVERY_WORKFLOWS = {"burst_report", "scada_anomaly"}
 
 
 def track_work_order(task: dict, action: dict) -> None:
-    if (task.get("workflow") not in RECOVERY_WORKFLOWS or action["kind"] != "create_work_order"
-            or task.get("source", {}).get("replay_context")):
+    if task.get("workflow") not in RECOVERY_WORKFLOWS or action["kind"] != "create_work_order":
         return
     receipt = action.get("receipt") or {}
     if action.get("execution_status") != "succeeded" or receipt.get("adapter") != "peripheral_sample":
@@ -132,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"])
@@ -160,6 +161,9 @@ async def reconcile_recoveries(service: OperationalTaskService, timescale: Async
                     devices.extend(row["device_id"] for row in data if isinstance(row, dict) and isinstance(row.get("device_id"), str))
             devices = list(dict.fromkeys(devices))
         async with timescale.transaction():
+            # Field completion authorizes a separate recovery observation even
+            # when the incident was found in a frozen replay. Keep that replay's
+            # query window and batch boundary unchanged.
             rows = await ScadaRepository.get_recovery_scada_readings(
                 timescale, devices, incident_at, returned_at, through, baseline_limit=100_001,
             ) if through > returned_at and devices else []

app/services/operational_scada_coverage.py

@@ -0,0 +1,40 @@
+"""Require persisted complete SCADA input before final investigation conclusions."""
+
+from datetime import datetime
+
+from app.domain.domain_errors import require
+
+
+def require_complete_scada_evidence(task: dict, evidence: list[dict], refs: list[str]) -> None:
+    replay = task.get("source", {}).get("replay_context")
+    if task.get("workflow") != "scada_anomaly" and not replay:
+        return
+    target = datetime.fromisoformat(
+        replay["knowledge_cutoff"] if replay else task["trigger"]["occurred_at"]
+    )
+    for item in evidence:
+        if item["evidence_id"] not in refs:
+            continue
+        snapshot = item.get("data_snapshot") or {}
+        if snapshot.get("source") != "query_scada_readings" or snapshot.get("truncated"):
+            continue
+        result = snapshot.get("data") or {}
+        data = result.get("data") if isinstance(result, dict) else None
+        coverage = data.get("coverage") if isinstance(data, dict) else None
+        if not coverage or coverage.get("mode") != "all_scada" or coverage.get("query_complete") is not True:
+            continue
+        window = result.get("window") or {}
+        if not window.get("start") or not window.get("end"):
+            continue
+        start, end = (datetime.fromisoformat(window[key]) for key in ("start", "end"))
+        if not start <= target <= end:
+            continue
+        if replay and (coverage.get("replay_id") != replay["replay_id"]
+                       or coverage.get("batch_sequence") != replay["batch_sequence"]
+                       or end != target):
+            continue
+        return
+    require(False, "SCADA_COVERAGE_REQUIRED",
+            "最终SCADA报告必须引用本轮全量读取证据。调用query_scada_readings并省略device_ids，"
+            "在任务时间窗内读取全部授权设备、完整属性与观测，核对coverage和缺测清单后再发布。",
+            status=409)

app/services/operational_tasks.py

@@ -73,6 +73,8 @@ from app.domain.schemas.operational_data import TaskDataSnapshot, TaskDataSource
 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)
@@ -87,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()
@@ -194,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,
@@ -488,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,
@@ -509,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"])
@@ -758,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
@@ -828,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"
             ):
@@ -929,6 +944,10 @@ class OperationalTaskService:
                 current_plan = snapshot
             if command.report:
                 self._require_evidence_refs(command.report.evidence_refs, known_evidence)
+                if command.report.stage == "final":
+                    require_complete_scada_evidence(
+                        task, await self.repo.list_evidence(str(task_id)), command.report.evidence_refs,
+                    )
                 for gap in command.report.data_gaps:
                     require(set(gap.evidence_refs) <= set(command.report.evidence_refs),
                             "REPORT_GAP_EVIDENCE_INVALID",
@@ -1082,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":
@@ -1132,7 +1152,7 @@ class OperationalTaskService:
             await self.repo.lock(f"task:{task_id}")
             task = await self._task(task_id)
             self._require_actionable_task(task)
-            self._require_real_action(task)
+            self._require_real_action(task, command.kind)
             self._validate_action_payload(command)
             if task["task_type"] == "event_investigation":
                 report = await self.repo.get_latest_report(str(task_id))
@@ -1203,8 +1223,8 @@ class OperationalTaskService:
             await self.repo.lock(f"action:{action_id}")
             task = await self._task(task_id)
             self._require_actionable_task(task)
-            self._require_real_action(task)
             action = await self._action(task_id, action_id)
+            self._require_real_action(task, action.kind)
             require(action.kind == "create_work_order", "INVALID_ACTION_KIND",
                     "Only work-order drafts can be edited", status=422)
             require(action.version == command.expected_version, "VERSION_CONFLICT",
@@ -1259,8 +1279,8 @@ class OperationalTaskService:
             await self.repo.lock(f"action:{action_id}")
             task = await self._task(task_id)
             self._require_actionable_task(task)
-            self._require_real_action(task)
             action = await self._action(task_id, action_id)
+            self._require_real_action(task, action.kind)
             require(action.version == command.expected_version, "VERSION_CONFLICT",
                     "Action version changed", status=409)
             require(action.approval_status == "pending", "INVALID_TRANSITION",
@@ -1306,8 +1326,8 @@ class OperationalTaskService:
             await self.repo.lock(f"action:{action_id}")
             task = await self._task(task_id)
             self._require_actionable_task(task)
-            self._require_real_action(task)
             action = await self._action(task_id, action_id)
+            self._require_real_action(task, action.kind)
             require(action.version == command.expected_version, "VERSION_CONFLICT",
                     "Action version changed", status=409)
             require(action.approval_status == "approved", "INVALID_TRANSITION",
@@ -2151,9 +2171,10 @@ class OperationalTaskService:
         return task
 
     @staticmethod
-    def _require_real_action(task: dict) -> None:
-        require(not (task.get("source") or {}).get("replay_context"), "REPLAY_ACTION_DISABLED",
-                "SCADA 回放调查仅生成分析与报告，不执行真实工单或设备动作", status=409)
+    def _require_real_action(task: dict, kind: str) -> None:
+        require(kind == "create_work_order" or not (task.get("source") or {}).get("replay_context"),
+                "REPLAY_ACTION_DISABLED",
+                "回放任务不支持设备命令；可生成工单草稿，由有权限的用户确认发布", status=409)
 
     @staticmethod
     def _require_actionable_task(task: dict) -> None:
@@ -2510,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"]
@@ -532,7 +536,7 @@ class ScadaReplayService:
             f"爆管侦测得分 {score_text}；"
             f"压力下降测点 {len(pressure.get('hit_device_ids', []))} 个；"
             f"出厂边界流量增加测点 {len(flow.get('hit_device_ids', []))} 个。"
-            "任一指标异常即启动调查，加权分仅用于评估爆管嫌疑，不能据此提前确诊。"
+            "任一指标异常即启动调查，加权分用于评估爆管嫌疑；达到门槛后继续候选定位、隔离与影响分析。"
         )
         if monitoring.get("anomaly_signals"):
             signals = monitoring["anomaly_signals"]

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/burst-response-workflow.md

@@ -55,6 +55,10 @@ SCADA作为事件影响的辅助证据，按目标测点和事件／对照窗口
 
 `POST /operational-tasks/recovery/reconcile` 由现有 operational worker 每轮调用，按项目及任务创建边界处理等待复核的任务；SCADA 读取失败不阻断其他调查派发。复核直接读取项目时序库中完工反馈时间之后最近时刻的 SCADA，不依赖回放状态或推进回放时间。
 
+这条恢复链路同样跟踪带 `replay_context` 的调查工单。原异常快照的 `knowledge_cutoff` 和 `batch_sequence` 保留，普通回放查询继续受其限制；工单完工反馈单独触发 Server 的 `ScadaRepository.get_recovery_scada_readings()`，以实际完工业务时间为 T，读取严格晚于 T 的最近一帧。例如 2026-07-16 13:00 完工，15 分钟采样时复核 13:15，而非原异常截止的 08:00。Agent 从 `handling_context.verification_evidence_id` 读取保存的恢复证据，并与原异常证据共同引用，不需放开回放查询或重建复测工单。
+
+已有调查在收到新的有效完工反馈时会补登记成功派发的关联工单。旧回报误用了电脑当前时间时，需在来源工单更正现场反馈业务时间并提交新版本；系统不猜测完工时刻，也不改写原报告或旧证据。
+
 ## 验证
 
 检查Skill名称、版本、摘要与正文digest一致、内联体积、角色权限、工单工具隔离和SCADA定位回归。Agent类型检查与会话工具策略测试共同覆盖运行时入口。发布后分别验收工单事件处理报告和SCADA候选定位，原有报告按既有修订保留。

docs/mcp-query-implementation.md

@@ -14,12 +14,16 @@ Agent 根据401 challenge 或路径/根 well-known 发现受保护资源元数
 
 调查数据返回前，在短事务内追加现有任务证据并保存调用回执。同一主体、会话、任务、run 和原生 call ID 的重试复用证据；不同参数或失效范围明确报错。七类原有访问申请 scope 保持原契约，MCP 的新增来源只扩充快照来源枚举。没有新增报告发布、审批、控制、Shell 或递归委派权限。
 
-普通查询默认20条；目录类最多500条，时序最多10000条、20个标识、24小时；offset 最大10000。Agent 首次优先单对象、必要的短时间窗和小页，仅在证据不足时按需扩大，不自动遍历全网。SQL 有 limit/offset 和 statement_timeout。历史对比在同一条SQL中按序列、时间对数据点分页，返回 page.has_more；非零offset也标明覆盖不完整。历史曲线展示调用初始每页500点，`RESULT_TOO_LARGE` 时原位减半重试，最多读取1000条、发出16次请求，共用120秒超时（单次请求同时受该剩余总预算约束），图表最多100个有效点；这不改变模型首次查询默认值或时序接口上限。
+普通查询默认20条；目录类最多500条，显式指定设备的时序查询最多10000条、20个标识、24小时；offset 最大10000。SCADA调查首次省略 `query_scada_readings.device_ids`，由Server读取项目完整设备目录及窗口内全部授权观测，内部自动翻页，`offset` 必须为0。其余查询按问题选定对象和窗口。SQL 有 limit/offset 和 statement_timeout。历史对比在同一条SQL中按序列、时间对数据点分页，返回 page.has_more；非零offset也标明覆盖不完整。历史曲线展示调用初始每页500点，`RESULT_TOO_LARGE` 时原位减半重试，最多读取1000条、发出16次请求，共用120秒超时（单次请求同时受该剩余总预算约束），图表最多100个有效点；这不改变模型首次查询默认值或时序接口上限。
 
 单份结果超过8MiB报错；超过12KB的调查工具消息默认返回证据资源引用，设备目录的精简路径见下段。完整快照仍保存，Server 的 `resources/read` 可以读取它。Agent 本地 MCP 只提供工具，将上游资源引用转为 `evidence_read` 中的 `mcp_tjwater_get_task_evidence` 参数，不向 OpenCode 泄露不可在本地读取的链接。`get_task_evidence` 在数据库中按 data_path、limit、offset 选页，嵌套集合和超过256字符的字符串返回路径描述；字符串按 Unicode 字符分页。页和完整资源均在数据库返回前检查8MiB上限，保留原证据编号和合成来源；普通查询要求缩小范围。其他结果仍保留执行后的整体8MiB检查，不能把它理解为所有查询均有数据库端字节限制。原始/清洗实测、模型基线、分析结果、比较和 synthetic 均明确标记。非有限数值转为 null 并记录数据缺口。
 
 `list_scada_devices` 默认每页100个、最多500个，其他集合仍沿用各自的分页默认值。设备目录超过12,000字节时，先内联本页全部设备的 `device_id`、`device_type` 和 `measurement_unit`，省略其他属性并明确标注；模型映射、坐标等通过 `get_scada_device` 核验，任务证据和幂等回执保留完整属性。此时 `truncated` 表示属性省略，设备是否还有后续页以 `page.has_more` 为准，按本页实际设备数递增 `offset`。精简目录仍超预算时沿用证据引用或缩页路径。
 
+全量查询同时要求设备目录与读数权限；通过任务授权读取时，捕获范围及当前批准状态都必须同时包含 `scada_devices` 和 `scada_readings`，显式设备读数查询仍沿用读数授权。
+
+全量 `query_scada_readings` 返回 `data.devices` 的完整属性、`data.readings` 的原始和清洗值，以及 `data.coverage` 的查询设备、读数数量和缺测清单；该模式保留完整内联数据，不经过12KB缩略，仍执行8MiB整份结果上限。回放仅读取指定批次之前的冻结观测，项目中未纳入该回放的设备单列 `outside_frozen_scope_device_ids`。`query_complete=true` 表示授权范围内的查询已读完，不表示设备均有有效数据。SCADA及历史回放任务发布final时，必须引用窗口覆盖本轮时刻、且回放批次匹配的全量证据；单设备、局部集合、未引用或旧批次证据不能通过此校验。既有报告不回写，后续重跑适用新约束。
+
 `get_task_context` 默认 section=summary，只查固定元数据、160字符任务标题及200字符结论预览，并返回 available_sections；preview_only 和 truncated 明确说明不是完整报告。按 section=task/report/evidence/actions/progress/access_requests/plan/run 读取详情，report_revision 保留选择旧报告的语义。data_path 选择嵌套值，limit 默认20、最多500，offset 最大10000；证据、动作和授权申请集合以稳定 ID 为键，按 ID 字典序分页。字符串按字符分页。详情在数据库返回前检查8KB预算，超限缩页或选更深路径。缺少可选进度或授权申请表时返回 metadata.available=false，不冒充空事实。绑定权限范围只读取七类既有 scope，不再加载整个任务上下文。不增加工具数量、数据库表或迁移，也不改变原 REST 完整上下文契约。
 
 外围 MCP 直接读取所选项目 BizDB 的工单事实和既有时序。外围工单统一返回 `system_data`；测试数据和 DEMO-* 观测仍按各自数据来源处理。DEMO-* 观测在数据库端拒绝超过8MiB的完整样本，预算内仍需读全样本、核验摘要后筛选；此限制不能通过缩小查询窗口解决。监测点列表沿用设备目录，不把未登记的测试点冒充实测设备。

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 不会因代码更新恢复。

docs/scada-peripheral-replay-design.md

@@ -18,6 +18,8 @@
 
 Agent 查询本轮已检测的固定批次快照，受任务的 replay_id、knowledge_cutoff、batch_sequence 和设备集合约束。新回放首次监测以只读一致快照冻结开始前 15 天历史，负序号静默基线按最多 2,000 行分块暂存为 `clean_prepared`；全部读取完成后，在一个 BizDB 短事务中把批次转为 `detected` 并原子发布基线与初始化标记，历史查询和定位对照才能读取。
 
+关联现场工单完工后的恢复复核采用独立证据链：Server 以现场反馈业务时间 T 读取项目时序库中严格晚于 T 的最近一帧及事件前对照，保存恢复监测证据后接续原调查。此路径不依赖回放推进，也不修改原异常快照的知识上限或批次；Agent 读取保存的恢复证据，不通过普通回放查询越界取数。详见[恢复复核流程](burst-response-workflow.md)。
+
 读取各页期间不持有 BizDB 事务，内存只保留当前页和设备短历史。固定准备预算为 1,000,000 行、512 MiB 序列化批次，覆盖 500 设备十五天每 15 分钟一条所需的 720,000 行并留有余量。超限不保存超限页，状态明确返回 `BASELINE_BUDGET_EXCEEDED` 及中文原因；未发布批次不进入观测、监测或待处理批次。失败重试只清理当前项目、本轮回放的未发布基线，从新的一致源快照重新准备，已发布基线保持不变。
 
 旧回放没有冻结基线时明确拒绝回放前查询，不回落到实时 TSDB。爆管定位支持受回放截止约束的冻结输入适配，其他未适配计算工具继续拒绝。[SCADA 调查 Skill](../resources/work-order-skills/scada-anomaly/SKILL.md) 要求据实说明证据限制。运行异常自动建立调查，数据质量异常仅建立告警；恢复时推进调查代次。真实模型推理与报告效果尚未联调。
@@ -573,7 +575,14 @@ next_cursor 绑定项目、设备、查询区间并固定本轮分页时间上
 
 ### 自动播放与简短通知（2026-11-07）
 
-调查任务仍按 `replay_context.knowledge_cutoff` 与 `batch_sequence` 截断知识，后续播放不自动扩大当前调查窗口。任务摘要只保留指标与异常数量，逐测点明细通过冻结 SCADA 证据查询。`scada-anomaly` 要求高嫌疑诊断继续候选定位；当前 Case3 的任务摘要和 skill 响应返回 `location_demo` 能力提示。提示可用时 `run_burst_location` 允许省略 `burst_leakage`，仍先核验冻结异常窗、正常窗及压力设备，再调用固定候选方法；水力算法仍要求漏量依据。候选输出不再含 mock 提示，MCP回放证据保留 `origin=synthetic`，不向 Agent 暴露事件真值。
+调查任务仍按 `replay_context.knowledge_cutoff` 与 `batch_sequence` 截断知识，后续播放不自动扩大当前调查窗口。任务摘要只保留指标与异常数量，逐测点明细通过冻结 SCADA 证据查询。`scada-anomaly` 要求高嫌疑诊断继续候选定位；当前 Case3 的任务摘要和 skill 响应返回 `location_capability` 能力提示。提示可用时 `run_burst_location` 允许省略 `burst_leakage`，仍先核验冻结异常窗、正常窗及压力设备，再调用固定候选方法；水力算法仍要求漏量依据。候选输出不再含 mock 提示，MCP回放证据保留 `origin=synthetic`，不向 Agent 暴露事件真值。
+
+Agent可见能力字段统一为 `location_capability`，不再输出旧 `location_demo` 字段。
+公共指令与 SCADA Skill 明确候选及相似度返回后继续地图展示、隔离、节点覆盖、影响分析与报告，
+来源字段本身不构成停止条件。有效 SCADA 查询不再将来源说明列为 `data_gaps`；实际缺测、
+窗口与设备校验仍按原规则处理。现场工单与设备动作维持服务端权限限制，动作拒绝文案明确
+允许继续已授权分析。旧工具回执和历史消息不改写；既有 OpenCode 进程的 system 配置副本
+需刷新后才使用新的公共指令。
 
 回放调查开放项目资产属性、空间查询、拓扑隔离和节点覆盖，沿用原有业务权限。它们依据当前模型，不代表历史现场状态。实时 TSDB、在线模拟等未接入冻结输入的路径仍不可用；候选定位不以出厂流量命中或现场确诊为前提。
 

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

@@ -2,22 +2,22 @@
 name: scada-anomaly
 description: 从 SCADA 指标异常开展数据核验、爆管诊断、候选定位与影响分析。
 metadata:
-  version: "1.0.8"
+  version: "1.0.13"
 ---
 
-# SCADA 回放异常调查
+# SCADA 数据异常调查
 
-读取任务与 replay_context，核对设备、单位、异常时刻、knowledge_cutoff 和 batch_sequence。来源标注为已存 SCADA 的演示回放；同一持续事件沿用已有调查。
+读取任务与 replay_context，核对设备、单位、异常时刻、knowledge_cutoff 和 batch_sequence。使用本轮已提交的 SCADA 观测开展调查，同一持续事件沿用已有调查。证据来源字段用于记录数据取得方式，不构成停止分析的条件。
 
-报告与图表的阅读时区统一为Asia/Shanghai（UTC+8）。工具返回的+08:00时间按原值使用；Z时间换算后同步更新日期、历史系列名和横轴标签，例如2026-07-16T00:00:00Z对应7月16日08:00。sample_times记录相同真实时间点并使用+08:00偏移，查询与冻结边界按带时区的时间点比较。
+报告与图表的阅读时区统一为Asia/Shanghai（UTC+8）。工具返回的+08:00时间按原值使用；Z时间换算后同步更新日期、历史系列名和横轴标签，例如2026-07-16T00:00:00Z对应7月16日08:00。sample_times记录相同真实时间点并使用+08:00偏移，查询与冻结边界按带时区的时间点比较。图表标题、横轴和系列名只写日期、时刻与单位，不写“上海时间”“Asia/Shanghai”或“UTC+8”；横轴命名为“时间”。
 
 ## 数据核验与诊断
 
 任一已配置指标异常即进入调查，包含压力、流量、水质规则和压力模型信号。先核验数据质量与测点，再判别原因；模型或组合指标有缺口时，继续核查已发现的单项异常。
 
-通过 query_scada_readings 读取冻结异常窗和正常对照，按需核查开始前15天基线；查询结束时间止于 knowledge_cutoff，数据批次止于 batch_sequence。复用已有监测回执。结合 monitored_value、cleaned_value 和质量标记判断变化，可靠采集的低压与水质变化保留为业务异常。
+通过 query_scada_readings 省略device_ids读取冻结异常窗内全部SCADA设备的完整属性和授权观测，覆盖压力、流量、水质等全部类型；正常对照也采用全量读取。Server自动读完所有设备页和观测页，完整返回devices、readings、coverage，结果仍受8MiB整份上限约束，超限明确报错后按时间分段，不能丢弃设备。query_complete表示查询已读完，缺测与无有效值另行判断。查询结束时间止于knowledge_cutoff，数据批次止于batch_sequence；冻结范围外设备单列，不回退实时数据。按需核查开始前15天基线，复用已有完整监测回执。结合monitored_value、cleaned_value和质量标记判断变化，可靠采集的低压与水质变化保留为业务异常。
 
-设备目录用 list_scada_devices，默认每页100个；page.has_more=true时递增offset继续读取，直到覆盖本轮目标设备。大目录内联可能仅包含本页全部设备的编号、类型和单位，模型映射与坐标用get_scada_device核验。读数查询的limit按设备数乘时间点数安排，检查page.has_more；缩略响应通过evidence_read指定的工具和data_path分小页读取已保存证据，不能把首屏、首个设备或属性缩略当作完整覆盖。
+全量响应中的devices包含完整目录属性，readings包含实际读数，coverage列出全部查询设备、各设备读数数量、缺测和冻结范围外设备。不能只读取已标记、异常或代表设备后推断全网状态。单独读取list_scada_devices时，默认每页100个，持续翻页到page.has_more=false；属性缩略不等于已读取完整设备信息。全量读取后可显式指定device_ids做局部复核，局部读数仍按page.has_more接续，不能替代全量输入。最终报告引用覆盖本轮knowledge_cutoff和batch_sequence的全量证据，Server会拒绝仅有局部或旧批次证据的final。
 
 分别说明压力异常模型得分、压力绝对下降及变化率、已配置出厂边界流量变化，再说明联合证据支持程度。使用实际权重、嫌疑门槛、动态阈值、参与测点与各自基线。综合分用于爆管嫌疑判断；得分越低越异常，目标告警与参考日分别解释。水质等其他异常按实际证据选择原因分支。
 
@@ -25,20 +25,28 @@ metadata:
 
 综合分达到嫌疑门槛，或压力与模型证据支持高嫌疑时，继续候选定位。流量变化和现场确认作为证据维度，定位输入按当前工具要求准备。
 
-检查摘要或 skill 响应顶层的 location_demo。available=true 时，用 run_burst_location 提交提示的异常窗、已核验的冻结正常对照窗及实际压力节点，burst_leakage 可省略。读取返回的候选管段与相似度，优先按 similarity_percent 展示百分比；只有 similarity 时，按0–1小数换算。取得候选后继续地图展示、隔离边界与影响分析。其他定位路径使用有依据的漏量估计或明确情景假设。缺少定位输入时记录待补项，继续独立的资产、空间与拓扑核查。
+检查摘要或 skill 响应顶层的 location_capability。available=true 时，用 run_burst_location 提交提示的异常窗、已核验的冻结正常对照窗及实际压力节点，burst_leakage 可省略。读取返回的候选管段与相似度，优先按 similarity_percent 展示百分比；只有 similarity 时，按0–1小数换算。取得候选后继续地图展示、隔离边界、节点覆盖与影响分析，完成报告后再收尾。其他定位路径使用有依据的漏量估计或明确情景假设。缺少定位输入时记录待补项，继续独立的资产、空间与拓扑核查。
 
 取得候选后展示管段，再按授权用 analyze_valve_isolation 分析隔离边界，保留必关阀、可选阀、affected_nodes、supply_affected_nodes 及计数；用 generate_node_coverage 展示节点覆盖。候选与拓扑建议标注为待现场核验方案。最终报告前核对定位调用及返回结果，调用失败记录具体错误与输入缺口。
 
 ## 对比展示与影响
 
-先发布覆盖本轮全部已标识SCADA设备的当前趋势总览，包括地图定位、等级标记及表格列出的目标设备。按指标和单位分图，同组各设备各占一条当前序列，系列名带device_id；每图最多8组，超过时分组发布，各组使用不同且稳定的chart_id。缺测填null，完全无读数的设备明确列为缺测并说明查询窗口，不能静默省略。收尾前逐项核对目标设备、查询覆盖、已发布图表和缺测清单，不用少量代表设备替代全量总览。
+先查询并发布截至本轮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保留真实带时区时刻，坐标轴注明单位。各采样周期保留实际缺口，按需翻页并报告实际参考点数。
+定位输入具备后继续定位。压力、流量按设备另发七天同期复合图，不占用或替代全量当前总览。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的本地日期回退，不含当天。按实际周期保留缺口，按需翻页并报告参考点数；基线未冻结、读取失败或缺测时说明并保留当前趋势，不退化为前一天或相邻两点。
 
 影响说明列出异常前参考、异常期间和已有恢复观测中的压力、流量或水质变化、持续时间、涉及测点及已核实资产。可比对象、单位、时段与运行条件下给出变化量，分别标明实际观测、模型预测和拓扑隔离。停水范围与漏量引用相应拓扑、水力、流量平衡或现场依据；恢复观测、用户关联或现场证据缺失时列为具体待核实项。
 
 ## 范围与报告
 
-本轮时序分析保持任务冻结窗口，后续播放形成的观测由相应轮次处理。资产属性、空间查询和拓扑隔离使用项目当前模型，注明模型时态；水力工具按已开放的冻结输入能力调用。演示定位继续执行权限、设备与冻结观测校验；基线或覆盖缺口按实际返回记录。
+本轮时序分析保持任务冻结窗口，后续播放形成的观测由相应轮次处理。资产属性、空间查询和拓扑隔离使用项目当前模型，注明模型时态；水力工具按已开放的冻结输入能力调用。定位调用执行权限、设备与冻结观测校验；基线或覆盖缺口按实际返回记录。
 
-报告给出观测事实、规则信号、原因候选、定位与隔离结果、影响说明和建议。摘要以150字以内为目标，测点明细用表格、完整依据留在证据。演示分析只发布回放成果，本轮分析完成与现场事件闭环分别说明。
+报告给出观测事实、规则信号、原因候选、定位与隔离结果、影响说明和建议。摘要以150字以内为目标，测点明细用表格、完整依据留在证据。候选、相似度和拓扑结果具备时，完成相应分析、展示与报告后再收尾；现场核验、审批和维修列为后续待办。现场工单与设备操作遵守服务端动作权限，本轮分析完成与现场事件闭环分别说明。
+
+需要现场核验时，先发布本轮outcome=concluded的调查报告，再调用propose_task_action提出kind=create_work_order、work_order_type=general的关联预约工单草稿，引用候选与影响证据，写明核验对象、内容和回执要求。带replay_context的任务同样可提出草稿，描述中注明历史异常时刻及回放来源；草稿须由有权限的用户编辑、确认后发布，后续派发与反馈进入预约任务。草稿待确认时说明等待人工处理，不将调查报告完成写成现场处置完成。回放任务不支持设备命令。
+
+## 完工后的恢复复核
+
+收到关联工单的completed反馈后，先读取get_task_context摘要中的handling_context，核对实际完工业务时间returned_at，不以工单创建、预约或电脑当前时间代替。Server通过现有恢复监测读取项目时序库中严格晚于returned_at的最近一帧及事件前15天同相位对照，独立保存恢复证据；例如13:00完工、15分钟采样时核对13:15。这条恢复链路适用于回放来源工单，不要求推进回放，原异常knowledge_cutoff与batch_sequence保持不变。
+
+phase=waiting_work_order时等待现场完工；phase=waiting_scada时查看last_check，分别说明尚无后续采样、基线或测点缺项、实际仍异常，等待监测复核，不将原回放查询越界解释为恢复数据不存在。phase=verification_ready时，用get_task_evidence读取verification_evidence_id的完整恢复证据，比较实际样本、正常范围与现场回报，最终报告同时引用原异常全量证据与当前恢复证据。恢复数据已由Server保存，不用query_scada_readings越过原冻结上限重复查询，也不因该上限再建复测工单；确有独立现场工作时才提议新动作。完工时间修订后使用新返回的证据编号，旧证据仅作历史。正常读数不替代阀位、漏点照片等现场事实，报告分别说明监测恢复、现场待核验项与事件闭环。

tests/unit/test_burst_location_demo.py

@@ -131,6 +131,8 @@ def test_location_capability_and_tool_contract_do_not_inject_mock_instructions()
     assert "mock" not in str(context) and "模拟" not in context["notice"]
     tool = TOOLS["run_burst_location"].protocol_tool()
     assert "相似度" in tool.description
+    assert "location_capability.available=true" in tool.description
+    assert "location_demo" not in tool.description
     assert "demo模拟候选" not in tool.description
     assert "明确标记的演示候选" not in str(tool.input_schema)
 

tests/unit/test_complete_scada.py

@@ -0,0 +1,179 @@
+import asyncio
+from contextlib import asynccontextmanager
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+from uuid import uuid4
+
+import pytest
+from fastapi import HTTPException
+
+from app.mcp import query_service, scada_readings
+from app.mcp.catalog import TOOLS
+from app.mcp.models import ScadaInput
+from app.domain.domain_errors import DomainError
+from app.services.operational_scada_coverage import require_complete_scada_evidence
+
+
+START = datetime(2026, 7, 16, tzinfo=timezone.utc)
+
+
+def _args(**extra):
+    return ScadaInput(context_handle="a" * 43, start=START, end=START + timedelta(hours=1), **extra)
+
+
+def test_all_scada_reads_every_catalogue_and_observation_page_with_full_attributes(monkeypatch):
+    devices = [{"device_id": f"D{i}", "device_type": ["pressure", "pipe_flow", "quality"][i % 3],
+                "measurement_unit": ["m", "m3/h", "mg/L"][i % 3], "node_id": f"J{i}",
+                "longitude": 121.5, "latitude": 31.0, "reliability": 1} for i in range(503)]
+    rows = [{"device_id": f"D{i % 502}", "time": START.isoformat(),
+             "monitored_value": float(i), "cleaned_value": None} for i in range(1205)]
+
+    async def catalog(connection, *, limit, offset):
+        return devices[offset:offset + limit]
+
+    async def series(connection, **args):
+        assert args["ids"] == [device["device_id"] for device in devices]
+        assert args["field"] is None
+        return rows[args["offset"]:args["offset"] + args["limit"]]
+
+    catalog_reader, series_reader = AsyncMock(side_effect=catalog), AsyncMock(side_effect=series)
+    monkeypatch.setattr(scada_readings.reads, "scada_devices", catalog_reader)
+    monkeypatch.setattr(scada_readings.reads, "series_rows", series_reader)
+    monkeypatch.setattr(query_service, "authorize_query", AsyncMock())
+
+    @asynccontextmanager
+    async def timescale(): yield object()
+
+    scope = SimpleNamespace(connection=object(), timescale=timescale)
+    result = asyncio.run(query_service.execute_query(scope, SimpleNamespace(replay_context=None),
+                                                    TOOLS["query_scada_readings"], _args()))
+    assert result.data["devices"] == devices and result.data["readings"] == rows
+    assert [call.kwargs["offset"] for call in catalog_reader.await_args_list] == [0, 500]
+    assert [call.kwargs["offset"] for call in series_reader.await_args_list] == [0, 1000]
+    coverage = result.data["coverage"]
+    assert coverage["query_complete"] is True and coverage["queried_device_count"] == 503
+    assert coverage["devices_with_readings"] == 502 and coverage["missing_device_ids"] == ["D502"]
+    assert coverage["reading_count"] == 1205 and len(result.units) == 503
+    assert result.origin == "system_data" and result.page is None and not result.truncated
+
+
+def test_all_scada_replay_reads_only_frozen_devices_batches_and_window(monkeypatch):
+    devices = [{"device_id": key, "device_type": kind, "measurement_unit": unit, "node_id": key}
+               for key, kind, unit in [("P1", "pressure", "m"), ("Q1", "quality", "mg/L"), ("NEW", "pressure", "m")]]
+    rows = [{"device_id": "P1", "time": START.isoformat(), "monitored_value": 20.,
+             "cleaned_value": 20., "quality": "valid", "reason": "冻结观测", "source_hash": "internal-hash"}]
+    context = {"replay_id": str(uuid4()), "knowledge_cutoff": (START + timedelta(hours=1)).isoformat(),
+               "batch_sequence": 9}
+    repository = SimpleNamespace(
+        get=AsyncMock(return_value={"start_time": START.isoformat(), "config": {"device_ids": ["P1", "Q1"]},
+                                   "devices": devices[:2]}),
+        detected_outputs=AsyncMock(return_value=rows),
+    )
+    monkeypatch.setattr(scada_readings.reads, "scada_devices", AsyncMock(return_value=devices))
+    monkeypatch.setattr(scada_readings, "ScadaReplayRepository", lambda *args: repository)
+    timescale = AsyncMock(side_effect=AssertionError("Frozen reads must never use live TSDB"))
+    scope = SimpleNamespace(connection=object(), project=SimpleNamespace(project_id=uuid4()), timescale=timescale)
+    data, units, gaps, origin = asyncio.run(scada_readings.read_complete_scada(
+        scope, SimpleNamespace(replay_context=context), _args(),
+    ))
+    assert data["devices"] == devices
+    assert data["readings"] == [{key: value for key, value in rows[0].items() if key != "source_hash"}]
+    assert data["coverage"]["queried_device_ids"] == ["P1", "Q1"]
+    assert data["coverage"]["missing_device_ids"] == ["Q1"]
+    assert data["coverage"]["outside_frozen_scope_device_ids"] == ["NEW"]
+    repository.detected_outputs.assert_awaited_once_with(
+        context["replay_id"], ["P1", "Q1"], START, START + timedelta(hours=1), 9, limit=1001, offset=0,
+    )
+    assert origin == "synthetic" and len(gaps) == 2 and units == {"P1": "m", "Q1": "mg/L"}
+    timescale.assert_not_called()
+
+
+def test_all_scada_empty_catalogue_does_not_query_unfiltered_timeseries(monkeypatch):
+    monkeypatch.setattr(scada_readings.reads, "scada_devices", AsyncMock(return_value=[]))
+    timescale = AsyncMock(side_effect=AssertionError("An empty catalogue must not select every TSDB row"))
+    data, units, _, _ = asyncio.run(scada_readings.read_complete_scada(
+        SimpleNamespace(connection=object(), timescale=timescale), SimpleNamespace(replay_context=None), _args(),
+    ))
+    assert data["coverage"]["queried_device_count"] == 0 and data["readings"] == [] and units == {}
+    timescale.assert_not_called()
+
+
+def test_all_scada_rejects_offset_and_empty_explicit_selection():
+    for values in ({"offset": 1}, {"device_ids": []}):
+        with pytest.raises(ValueError):
+            _args(**values)
+    assert _args(device_ids=["P1"], offset=100).offset == 100
+
+
+@pytest.mark.parametrize("captured,approved,explicit,direct,allowed", [
+    ({"scada_readings", "scada_devices"}, {"scada_readings"}, False, False, False),
+    ({"scada_readings"}, {"scada_readings", "scada_devices"}, False, False, False),
+    ({"scada_readings", "scada_devices"}, {"scada_devices"}, False, False, False),
+    ({"scada_readings", "scada_devices"}, {"scada_readings", "scada_devices"}, False, False, True),
+    ({"scada_readings"}, {"scada_readings"}, True, False, True),
+    (set(), set(), False, True, True),
+])
+def test_complete_scada_requires_both_captured_and_current_source_grants(
+    monkeypatch, captured, approved, explicit, direct, allowed,
+):
+    @asynccontextmanager
+    async def transaction():
+        yield
+
+    can_read = AsyncMock(side_effect=lambda task_id, source: source in approved)
+    monkeypatch.setattr(query_service, "permissions_for_context", lambda _: {"scada.view"} if direct else set())
+    monkeypatch.setattr(query_service, "task_service", lambda _: SimpleNamespace(can_read_source=can_read))
+    binding = SimpleNamespace(
+        replay_context=None, tool_cap=["query_scada_readings"],
+        permission_cap=["scada.view"] if direct else [], task_id=uuid4(), task_scope_cap=captured,
+    )
+    scope = SimpleNamespace(project=object(), connection=SimpleNamespace(transaction=transaction))
+    args = _args(**({"device_ids": ["P1"]} if explicit else {}))
+    call = query_service.authorize_query(scope, binding, TOOLS["query_scada_readings"], args)
+    if allowed:
+        asyncio.run(call)
+    else:
+        with pytest.raises(HTTPException) as error:
+            asyncio.run(call)
+        assert error.value.status_code == 403
+    if direct:
+        can_read.assert_not_awaited()
+    elif allowed:
+        expected = {"scada_readings"} if explicit else {"scada_readings", "scada_devices"}
+        assert {call.args[1] for call in can_read.await_args_list} == expected
+
+
+def test_all_scada_reports_oversize_instead_of_successfully_returning_a_partial_scan(monkeypatch):
+    monkeypatch.setattr(scada_readings.reads, "scada_devices", AsyncMock(return_value=[
+        {"device_id": "P1", "description": "x" * 200},
+    ]))
+    monkeypatch.setattr(scada_readings, "MAX_COMPLETE_SCADA_BYTES", 100)
+    with pytest.raises(HTTPException) as error:
+        asyncio.run(scada_readings.read_complete_scada(
+            SimpleNamespace(connection=object()), SimpleNamespace(replay_context=None), _args(),
+        ))
+    assert error.value.status_code == 413
+
+
+@pytest.mark.parametrize("defect", [None, "old_batch", "other_replay", "old_window", "unreferenced", "truncated"])
+def test_final_scada_evidence_must_match_the_current_frozen_scope(defect):
+    replay_id = str(uuid4())
+    replay = {"replay_id": replay_id, "knowledge_cutoff": START.isoformat(), "batch_sequence": 9}
+    task = {"workflow": "burst_report", "source": {"replay_context": replay}}
+    coverage = {"mode": "all_scada", "query_complete": True, "replay_id": replay_id, "batch_sequence": 9}
+    window = {"start": (START - timedelta(hours=1)).isoformat(), "end": START.isoformat()}
+    if defect == "old_batch": coverage["batch_sequence"] = 8
+    if defect == "other_replay": coverage["replay_id"] = str(uuid4())
+    if defect == "old_window": window["end"] = (START - timedelta(minutes=15)).isoformat()
+    evidence = [{"evidence_id": "mcp:all", "data_snapshot": {
+        "source": "query_scada_readings", "truncated": defect == "truncated",
+        "data": {"data": {"coverage": coverage}, "window": window},
+    }}]
+    refs = [] if defect == "unreferenced" else ["mcp:all"]
+    if defect is None:
+        require_complete_scada_evidence(task, evidence, refs)
+    else:
+        with pytest.raises(DomainError) as error:
+            require_complete_scada_evidence(task, evidence, refs)
+        assert error.value.code == "SCADA_COVERAGE_REQUIRED"

tests/unit/test_mcp_query.py

@@ -774,6 +774,34 @@ def test_scada_catalog_default_100_paginates_without_changing_other_queries(runt
     assert TOOLS["list_elements"].model(context_handle=value.context_handle, element_type="pipe").limit == 20
 
 
+@pytest.mark.parametrize("complete", [False, True])
+def test_complete_scada_input_is_not_replaced_by_12kb_preview_but_details_stay_bounded(runtime, complete):
+    app, client, _, _ = runtime
+    value = binding()
+    data = {"devices": [{"device_id": f"D{index}", "node_id": f"J{index}", "description": "x" * 100}
+                        for index in range(168)],
+            "readings": [{"device_id": f"D{index}", "monitored_value": 20., "cleaned_value": 19.}
+                         for index in range(168)],
+            "coverage": {"mode": "all_scada", "query_complete": True, "queried_device_count": 168}}
+    assert len(json.dumps(data).encode()) > 12000
+    app.executor = AsyncMock(return_value=QueryResult(
+        source="query_scada_readings", source_ref="mcp:all-scada", queried_at=datetime.now(timezone.utc),
+        semantics="observed", data=data,
+    ))
+    arguments = {"context_handle": value.context_handle, "start": "2026-07-16T05:00:00+08:00",
+                 "end": "2026-07-16T08:00:00+08:00"}
+    if not complete:
+        arguments["device_ids"] = ["D1"]
+    response = request(client, "tools/call", {"name": "query_scada_readings", "arguments": arguments}).json()["result"]
+    if complete:
+        assert not response.get("isError"), response
+        assert response["structuredContent"]["data"] == data
+        assert response["structuredContent"]["truncated"] is False
+    else:
+        assert response["isError"] is True
+        assert response["structuredContent"]["error"]["code"] == "RESULT_TOO_LARGE"
+
+
 def test_business_skill_is_delivered_over_mcp_without_recording_event_evidence(runtime, monkeypatch):
     app, client, _, _ = runtime
     value = binding(task=True)

tests/unit/test_mcp_work_order_skills.py

@@ -49,6 +49,16 @@ def test_work_order_handling_and_legacy_scada_replay_receive_separate_skills():
     assert "run_burst_location" in replay["instructions"]
 
 
+def test_scada_instructions_advance_candidate_analysis_independently_of_source_labels():
+    skill = read_work_order_skill("scada_anomaly", include_content=True)
+    text = skill["instructions"]
+    assert "location_capability" in text
+    assert "证据来源字段用于记录数据取得方式，不构成停止分析的条件" in text
+    assert "隔离边界、节点覆盖与影响分析，完成报告后再收尾" in text
+    assert "现场工单与设备操作遵守服务端动作权限" in text
+    assert not any(word in text for word in ("演示", "mock", "location_demo", "只发布回放成果"))
+
+
 def test_task_skill_uses_persisted_workflow_and_preserves_task_scope(monkeypatch):
     task_id, project_id = uuid4(), uuid4()
     reader = AsyncMock(return_value={"task": {"workflow": "burst_report"}})
@@ -103,7 +113,8 @@ def test_replay_context_advertises_demo_location_without_revealing_truth(monkeyp
     for section in ("summary", "skill"):
         result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["get_task_context"],
             TaskContextInput(context_handle=binding.context_handle, section=section)))
-        capability = result.data["location_demo"]
+        capability = result.data["location_capability"]
+        assert "location_demo" not in result.data
         assert capability["available"] is True
         assert capability["tool"] == "run_burst_location"
         assert capability["burst_leakage_required"] is False
@@ -113,4 +124,4 @@ def test_replay_context_advertises_demo_location_without_revealing_truth(monkeyp
     binding.replay_context["knowledge_cutoff"] = "2026-07-15T23:45:00Z"
     result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["get_task_context"],
         TaskContextInput(context_handle=binding.context_handle)))
-    assert result.data["location_demo"]["available"] is False
+    assert result.data["location_capability"]["available"] is False

tests/unit/test_operational_recovery.py

@@ -97,8 +97,13 @@ def test_off_grid_feedback_uses_first_later_sample_even_without_monitoring_inter
     assert result["verified_through"] == (RETURNED + timedelta(minutes=15)).isoformat()
 
 
-def test_dispatch_waits_for_all_orders_and_accepts_corrected_field_completion_time():
+@pytest.mark.parametrize("replay", [False, True])
+def test_dispatch_waits_for_all_orders_and_accepts_corrected_field_completion_time(replay):
     task = {"workflow": "burst_report"}
+    if replay:
+        task["source"] = {"replay_context": {
+            "replay_id": str(uuid4()), "knowledge_cutoff": INCIDENT.isoformat(), "batch_sequence": 3,
+        }}
     for order_id in ("one", "two"):
         track_work_order(task, {"kind": "create_work_order", "action_id": str(uuid4()), "execution_status": "succeeded",
                                 "receipt": {"adapter": "peripheral_sample", "receipt_id": order_id}})
@@ -120,6 +125,9 @@ def test_dispatch_waits_for_all_orders_and_accepts_corrected_field_completion_ti
     assert task["handling_context"]["returned_at"] == corrected.isoformat()
     assert task["handling_context"]["phase"] == "waiting_scada"
     assert "verification_evidence_id" not in task["handling_context"] and "last_check" not in task["handling_context"]
+    if replay:
+        assert task["source"]["replay_context"]["knowledge_cutoff"] == INCIDENT.isoformat()
+        assert task["source"]["replay_context"]["batch_sequence"] == 3
 
 
 def test_field_feedback_time_requires_a_real_note_and_explicit_timezone():

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"],
@@ -577,7 +601,7 @@ async def test_presentation_cursor_and_event_are_read_only_and_idempotent():
 
 
 @pytest.mark.anyio
-async def test_replay_investigation_can_publish_evidenced_composite_chart_but_cannot_propose_real_action():
+async def test_replay_investigation_can_publish_chart_and_propose_work_order():
     now = datetime(2026, 10, 10, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
@@ -593,44 +617,117 @@ async def test_replay_investigation_can_publish_evidenced_composite_chart_but_ca
     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"]},
                            ]}},
     ))
     assert result.recorded
     assert repository.events[-1]["metadata"]["command"]["chart"]["series"][1]["data"] == [None, 28.]
-    with pytest.raises(DomainError) as caught:
-        await service.propose_action(task["task_id"], ActionProposalCommand(
-            request_id=uuid4(), kind="create_work_order", reason="演示不得创建真实工单",
-            payload={"title": "现场核查", "description": "核查", "work_order_type": "general"},
-        ))
-    assert caught.value.code == "REPLAY_ACTION_DISABLED"
+    _seed_concluded_report(repository, now)
+    action = await service.propose_action(task["task_id"], ActionProposalCommand(
+        request_id=uuid4(), kind="create_work_order", reason="候选管段待现场核验",
+        evidence_refs=[evidence.evidence_id],
+        payload={"title": "现场核查", "description": "核查", "work_order_type": "general"},
+    ))
+    assert action.approval_status == "pending"
+    assert action.execution_status == "not_submitted"
+    assert repository.action["action_id"] == str(action.action_id)
 
 
 @pytest.mark.anyio
-@pytest.mark.parametrize("operation", ["draft", "approve", "submit"])
-async def test_replay_cannot_edit_approve_or_submit_preexisting_real_actions(operation):
+@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):
     now = datetime(2026, 10, 10, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
     task["source"]["replay_context"] = {
         "replay_id": str(uuid4()), "knowledge_cutoff": now.isoformat(), "batch_sequence": 1,
     }
+    action = HumanApprovalAction(
+        action_id=uuid4(), task_id=task["task_id"], version=1, kind="issue_command",
+        reason="关阀", payload={"command_type": "close_valve", "target": {"kind": "valve", "id": "V1"},
+            "parameters": {}, "impact_summary": "隔离候选管段", "preconditions": ["现场核验"],
+            "expires_at": now + timedelta(minutes=5)},
+        evidence_refs=[], required_approval_permission="command.approve",
+        required_execution_permission="command.execute", approval_status="pending",
+        execution_status="not_submitted", proposed_by="principal:test", proposed_at=now,
+    )
     repository = RunTaskRepository(task, _active_run(task, now))
+    repository.action = action.model_dump(mode="json")
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     dispatcher = SimpleNamespace(dispatch=AsyncMock())
     args = {"request_id": uuid4(), "expected_version": 1}
     with pytest.raises(DomainError) as caught:
-        if operation == "draft":
-            await service.update_work_order_draft(task["task_id"], uuid4(), WorkOrderDraftUpdateCommand(
-                **args, payload={"title": "现场核查", "description": "核查", "work_order_type": "general"}))
+        if operation == "propose":
+            await service.propose_action(task["task_id"], ActionProposalCommand(
+                request_id=uuid4(), kind="issue_command", reason=action.reason, payload=action.payload))
         elif operation == "approve":
-            await service.decide_action(task["task_id"], uuid4(), ActionDecisionCommand(**args, decision="approved"))
+            await service.decide_action(task["task_id"], action.action_id, ActionDecisionCommand(**args, decision="approved"))
         else:
-            await service.submit_action(task["task_id"], uuid4(), ActionExecuteCommand(**args), dispatcher, None)
+            await service.submit_action(task["task_id"], action.action_id, ActionExecuteCommand(**args), dispatcher, None)
     assert caught.value.code == "REPLAY_ACTION_DISABLED"
     dispatcher.dispatch.assert_not_awaited()
 
@@ -965,6 +1062,36 @@ def _current_report():
     }
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize("complete", [False, True])
+async def test_scada_final_report_requires_complete_current_input_not_selected_device_evidence(complete):
+    now = datetime(2026, 10, 10, tzinfo=timezone.utc)
+    context = _context()
+    task = _task_document(context, now, lease=now + timedelta(minutes=1))
+    task["workflow"] = "scada_anomaly"
+    repository = RunTaskRepository(task, _active_run(task, now))
+    service = OperationalTaskService(repository, context, set(), clock=lambda: now)
+    data = {"devices": [{"device_id": "P1"}], "readings": [], "coverage": {
+        "mode": "all_scada", "query_complete": True,
+    }} if complete else [{"device_id": "P1", "monitored_value": 20.}]
+    evidence, _ = await service.record_data_snapshot(task["task_id"], TaskDataSnapshot(
+        source="query_scada_readings", source_ref="mcp:scada-coverage", queried_at=now,
+        origin="system_data", truncated=False,
+        data={"data": data, "window": {"start": (now - timedelta(hours=1)).isoformat(), "end": now.isoformat()}},
+    ))
+    command = AgentArtifactsCommand(
+        request_id=uuid4(), source_session_id="session-1",
+        report={**_current_report(), "evidence_refs": [evidence.evidence_id]},
+    )
+    if complete:
+        result = await service.publish_artifacts(task["task_id"], command)
+        assert result.task.latest_report.report.stage == "final"
+    else:
+        with pytest.raises(DomainError) as error:
+            await service.publish_artifacts(task["task_id"], command)
+        assert error.value.code == "SCADA_COVERAGE_REQUIRED" and not repository.reports
+
+
 def _insufficient_report():
     return {
         **_current_report(), "outcome": "insufficient_data",
@@ -1085,21 +1212,39 @@ def _seed_concluded_report(repository, now):
 
 
 @pytest.mark.anyio
-async def test_burst_waits_for_completion_then_scada_evidence_then_agent_report(monkeypatch):
+@pytest.mark.parametrize("replay", [False, True])
+async def test_burst_waits_for_completion_then_scada_evidence_then_agent_report(monkeypatch, replay):
     from copy import deepcopy
     from app.domain.schemas.project_settings import ProjectSettings
     from app.services.operational_recovery import reconcile_recoveries, track_work_order
 
-    now = datetime(2026, 7, 16, 4, tzinfo=timezone.utc)
-    incident = now - timedelta(hours=4)
-    returned = incident + timedelta(hours=1)
+    now = datetime(2026, 7, 16, 6, tzinfo=timezone.utc)
+    incident = now - timedelta(hours=6)
+    returned = datetime.fromisoformat("2026-07-16T13:00:00+08:00")
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
     task.update(workflow="burst_report", trigger={"kind": "event", "occurred_at": incident.isoformat(), "detail": "爆管"})
+    if replay:
+        task["workflow"] = "scada_anomaly"
+        task["source"]["replay_context"] = {
+            "replay_id": str(uuid4()), "knowledge_cutoff": incident.isoformat(), "batch_sequence": 3,
+        }
     repository = RunTaskRepository(task, _active_run(task, now))
     repository.connection = object()
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     monkeypatch.setattr(service, "_resolve_investigation_alerts", AsyncMock())
+    incident_evidence_id = None
+    if replay:
+        frozen = task["source"]["replay_context"]
+        evidence, _ = await service.record_data_snapshot(task["task_id"], TaskDataSnapshot(
+            source="query_scada_readings", source_ref="mcp:frozen-incident", queried_at=now,
+            origin="synthetic", truncated=False,
+            data={"data": {"devices": [{"device_id": "P1"}], "readings": [], "coverage": {
+                "mode": "all_scada", "query_complete": True,
+                "replay_id": frozen["replay_id"], "batch_sequence": 3,
+            }}, "window": {"start": (incident - timedelta(hours=3)).isoformat(), "end": incident.isoformat()}},
+        ))
+        incident_evidence_id = evidence.evidence_id
     _seed_concluded_report(repository, now)
     action = await service.propose_action(task["task_id"], ActionProposalCommand(
         request_id=uuid4(), kind="create_work_order", reason="隔离抢修与恢复供水",
@@ -1135,6 +1280,11 @@ async def test_burst_waits_for_completion_then_scada_evidence_then_agent_report(
     reader.return_value = rows
     assert await reconcile_recoveries(service, timescale, settings) == 1
     assert await reconcile_recoveries(service, timescale, settings) == 0
+    reader.assert_awaited_with(timescale, ["P1"], incident, returned, now, baseline_limit=100_001)
+    if replay:
+        assert repository.task["source"]["replay_context"]["replay_id"] == task["source"]["replay_context"]["replay_id"]
+        assert datetime.fromisoformat(repository.task["source"]["replay_context"]["knowledge_cutoff"]) == incident
+        assert repository.task["source"]["replay_context"]["batch_sequence"] == 3
     recovery = repository.task["handling_context"]
     evidence_id = recovery["verification_evidence_id"]
     assert repository.evidence[evidence_id]["data_snapshot"]["data"]["status"] == "passed"
@@ -1165,7 +1315,14 @@ async def test_burst_waits_for_completion_then_scada_evidence_then_agent_report(
         await service.complete_run(task["task_id"], repository.run["run_id"], AgentRunCommand(
             request_id=uuid4(), expected_revision=repository.run["revision"], source_session_id="session-1"))
     assert rejected.value.code == "RECOVERY_EVIDENCE_REQUIRED"
-    repository.reports[-1]["report"]["evidence_refs"] = [evidence_id]
+    if replay:
+        await service.publish_artifacts(task["task_id"], AgentArtifactsCommand(
+            request_id=uuid4(), source_session_id="session-1",
+            report={**_current_report(), "evidence_refs": [incident_evidence_id, evidence_id]},
+        ))
+        assert repository.reports[-1]["report"]["evidence_refs"] == [incident_evidence_id, evidence_id]
+    else:
+        repository.reports[-1]["report"]["evidence_refs"] = [evidence_id]
     result = await service.complete_run(task["task_id"], repository.run["run_id"], AgentRunCommand(
         request_id=uuid4(), expected_revision=repository.run["revision"], source_session_id="session-1"))
     assert result.status == "completed" and result.handling_context["phase"] == "completed"
@@ -1684,10 +1841,15 @@ async def test_completed_gate_requires_access_request_decision():
 
 
 @pytest.mark.anyio
-async def test_work_order_draft_edit_creates_new_review_version():
+@pytest.mark.parametrize("replay", [False, True])
+async def test_work_order_draft_edit_creates_new_review_version(replay):
     now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
+    if replay:
+        task["source"]["replay_context"] = {
+            "replay_id": str(uuid4()), "knowledge_cutoff": now.isoformat(), "batch_sequence": 1,
+        }
     repository = RunTaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, {OPERATIONAL_ACTION_WORK_ORDER_APPROVE}, clock=lambda: now)
     _seed_concluded_report(repository, now)
@@ -1710,10 +1872,15 @@ async def test_work_order_draft_edit_creates_new_review_version():
 
 
 @pytest.mark.anyio
-async def test_one_click_work_order_confirmation_replays_without_duplicate_dispatch():
+@pytest.mark.parametrize("is_replay", [False, True])
+async def test_one_click_work_order_confirmation_replays_without_duplicate_dispatch(is_replay):
     now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc)
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
+    if is_replay:
+        task["source"]["replay_context"] = {
+            "replay_id": str(uuid4()), "knowledge_cutoff": now.isoformat(), "batch_sequence": 1,
+        }
     repository = RunTaskRepository(task, _active_run(task, now))
     service = OperationalTaskService(repository, context, {
         OPERATIONAL_ACTION_WORK_ORDER_APPROVE, OPERATIONAL_ACTION_WORK_ORDER_EXECUTE,
@@ -1732,6 +1899,11 @@ async def test_one_click_work_order_confirmation_replays_without_duplicate_dispa
             return ActionDispatchOutcome("unsupported", "测试适配器未配置")
 
     dispatcher = Dispatcher()
+    with pytest.raises(DomainError) as caught:
+        await service.submit_action(task["task_id"], action.action_id,
+            ActionExecuteCommand(request_id=uuid4(), expected_version=action.version), dispatcher, None)
+    assert caught.value.code == "INVALID_TRANSITION"
+    assert dispatcher.calls == 0
     command = WorkOrderConfirmDispatchCommand(request_id=uuid4(), expected_version=action.version)
     first = await service.confirm_work_order_dispatch(task["task_id"], action.action_id,
                                                        command, dispatcher, None)
@@ -2283,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_queries.py

@@ -10,7 +10,7 @@ from fastapi import HTTPException
 
 from app.mcp.analysis_models import BurstLocationInput
 from app.mcp.catalog import TOOLS
-from app.mcp.models import BoundContext
+from app.mcp.models import BoundContext, ScadaInput
 from app.mcp import query_service
 from app.services.scada_replay_queries import replay_burst_location_inputs, replay_snapshot_rows
 
@@ -18,6 +18,30 @@ from app.services.scada_replay_queries import replay_burst_location_inputs, repl
 START = datetime(2026, 10, 8, tzinfo=timezone.utc)
 
 
+def test_replay_origin_is_preserved_without_a_provenance_only_data_gap(monkeypatch):
+    from app.services import scada_replay_queries
+
+    monkeypatch.setattr(query_service, "authorize_query", AsyncMock())
+    reading = {"time": START.isoformat(), "device_id": "P1", "cleaned_value": 20.0,
+               "monitored_value": 20.0, "quality": "valid"}
+    read = AsyncMock(return_value=[reading])
+    monkeypatch.setattr(scada_replay_queries, "replay_snapshot_rows", read)
+    project_id = uuid4()
+    context = {"replay_id": str(uuid4()), "knowledge_cutoff": (START + timedelta(minutes=15)).isoformat(),
+               "batch_sequence": 9}
+    connection = object()
+    scope = SimpleNamespace(connection=connection, project=SimpleNamespace(project_id=project_id))
+    binding = SimpleNamespace(replay_context=context)
+    args = ScadaInput(context_handle="a" * 43, device_ids=["P1"], start=START,
+                      end=START + timedelta(minutes=15))
+    result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["query_scada_readings"], args))
+    assert result.origin == "synthetic"
+    assert result.data[0]["device_id"] == "P1"
+    assert result.data_gaps == []
+    read.assert_awaited_once_with(connection, project_id, context, ["P1"], args.start, args.end,
+                                  args.limit + 1, args.offset)
+
+
 class Cursor:
     def __init__(self, rows): self.rows = rows
     async def fetchone(self): return self.rows[0] if self.rows else None

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