From b45c5c6f46e97ae9d23cdb048d1af00f6ca7f5d1 Mon Sep 17 00:00:00 2001
From: Huarch <huarch97@outlook.com>
Date: Sat, 14 Nov 2026 23:00:15 +0800
Subject: [PATCH] feat(investigations): refine report authoring and replay
 flushing

---
 app/domain/schemas/operational_task.py        | 108 +++++++-
 app/domain/schemas/report_authoring.py        |  73 +++++
 app/mcp/analysis_models.py                    |  15 +-
 app/mcp/analysis_service.py                   |   9 +
 app/mcp/catalog.py                            |   2 +-
 app/mcp/mutation_service.py                   |   3 +-
 app/mcp/query_service.py                      |  18 +-
 app/mcp/scenario_service.py                   |  28 +-
 app/services/operational_tasks.py             |   9 +
 app/services/scada_replay.py                  |  34 +--
 app/services/simulation_scenarios.py          |   2 +
 contracts/manifest.json                       |   2 +-
 contracts/server-v1.openapi.json              | 255 +++++++++++++++++-
 ...investigation-progress-and-presentation.md |  31 ++-
 docs/investigation-report-contract.md         |  70 ++++-
 docs/mcp-analysis-tools.md                    |   2 +-
 docs/water-pollution-response-workflow.md     |  38 ++-
 .../work-order-skills/scada-anomaly/SKILL.md  |  10 +-
 .../water-pollution-event/SKILL.md            |  68 +++--
 tests/unit/test_flushing_valve_plan.py        |  78 ++++++
 .../test_investigation_report_authoring.py    |  96 +++++++
 tests/unit/test_mcp_work_order_skills.py      |  12 +
 tests/unit/test_operational_tasks.py          |  74 +++++
 tests/unit/test_pollution_event_workflow.py   |   2 +-
 tests/unit/test_presentation_authoring.py     |  72 +++++
 tests/unit/test_replay_flushing.py            |  91 +++++++
 tests/unit/test_scada_replay_service.py       |  17 +-
 27 files changed, 1125 insertions(+), 94 deletions(-)
 create mode 100644 app/domain/schemas/report_authoring.py
 create mode 100644 tests/unit/test_flushing_valve_plan.py
 create mode 100644 tests/unit/test_investigation_report_authoring.py
 create mode 100644 tests/unit/test_presentation_authoring.py
 create mode 100644 tests/unit/test_replay_flushing.py

diff --git a/app/domain/schemas/operational_task.py b/app/domain/schemas/operational_task.py
index 29550e8..5edcc3a 100644
--- a/app/domain/schemas/operational_task.py
+++ b/app/domain/schemas/operational_task.py
@@ -5,13 +5,16 @@ import re
 from typing import Annotated, Literal
 from uuid import UUID
 
-from pydantic import Field, JsonValue, field_validator, model_serializer, model_validator
+from pydantic import Field, JsonValue, StringConstraints, field_validator, model_serializer, model_validator
 from app.domain.schemas.answer_ui import (
     CHART_SAMPLE_TIME_PATTERN, ReportUiBlock, validate_chart_sample_time, validate_report_ui_blocks,
 )
 
 from app.domain.schemas.task_question import TaskQuestionRequest
 from app.domain.schemas.shared import NonEmptyStr, PositiveInt, StrictBool, StrictModel
+from app.domain.schemas.report_authoring import (
+    validate_presentation_command_text, validate_report_text, validate_report_ui_text,
+)
 from app.domain.schemas.operational_data import TaskDataSnapshot, TaskDataSource
 from app.domain.schemas.workbench_runtime import (
     WorkbenchDateTime,
@@ -349,12 +352,44 @@ class InvestigationReport(StrictModel):
         return self
 
 
+class InvestigationDataGapInput(InvestigationDataGap):
+    """New gap descriptions use business prose; references remain opaque."""
+
+    missing_data: NonEmptyStr = Field(max_length=4000, description="缺失的业务资料，不写执行字段")
+    scope: NonEmptyStr = Field(max_length=4000, description="缺口涉及的对象、地点与当地时间范围")
+    impact: NonEmptyStr = Field(max_length=4000, description="资料缺项对当前判断的实际影响")
+    required_data: NonEmptyStr = Field(max_length=4000, description="需要补充的具体业务资料")
+
+    @field_validator("missing_data", "scope", "impact", "required_data")
+    @classmethod
+    def validate_reader_text(cls, value: str) -> str:
+        return validate_report_text(value)
+
+
 class InvestigationReportInput(InvestigationReport):
     """Write model: every new report explicitly declares its business result."""
 
+    title: NonEmptyStr = Field(max_length=80, description="简短中文业务标题，写明对象和当前结果，不含执行轮次、流程版本或原始时间戳")
+    current_conclusion: NonEmptyStr = Field(max_length=300, description="先写当前业务判断，再写关键依据与下一步，建议 150 字以内；完整数值和对象比较使用 ui_blocks.Table")
+    conclusion_boundary: NonEmptyStr = Field(max_length=240, description="只写影响判断的证据范围、不确定性和适用限制，不复述结论、执行步骤或参数")
+    recommendations: list[Annotated[NonEmptyStr, Field(max_length=160)]] = Field(default_factory=list)
+    pending_items: list[Annotated[NonEmptyStr, Field(max_length=160)]] = Field(default_factory=list)
+    stop_reason: Annotated[NonEmptyStr, Field(max_length=160)] | None = None
     stage: Literal["interim", "final"]
     outcome: Literal["concluded", "insufficient_data"]
-    data_gaps: list[InvestigationDataGap]
+    data_gaps: list[InvestigationDataGapInput]
+
+    @field_validator("title", "current_conclusion", "conclusion_boundary", "stop_reason")
+    @classmethod
+    def validate_reader_text(cls, value: str | None) -> str | None:
+        return validate_report_text(value) if value is not None else value
+
+    @model_validator(mode="after")
+    def validate_authoring(self):
+        for value in [*self.recommendations, *self.pending_items]:
+            validate_report_text(value)
+        validate_report_ui_text(self.ui_blocks)
+        return self
 
 
 class InvestigationProgress(StrictModel):
@@ -365,8 +400,14 @@ class InvestigationProgress(StrictModel):
 
 
 class InvestigationProgressInput(InvestigationProgress):
+    summary: NonEmptyStr = Field(max_length=300, description="用简短中文说明本轮新增发现、当前依赖和下一步；不复制报告、执行字段、版本号或证据编号")
     graph_update: InvestigationGraphUpdate | None = None
 
+    @field_validator("summary")
+    @classmethod
+    def validate_reader_summary(cls, value: str) -> str:
+        return validate_report_text(value)
+
     @model_validator(mode="after")
     def validate_graph_input(self):
         if self.graph_update is not None and self.diagram is not None:
@@ -908,11 +949,72 @@ TaskPresentationCommand = Annotated[
 ]
 
 
+class TaskPresentationNarration(StrictModel):
+    stage: Literal[
+        "anomaly_detection", "evidence_verification", "asset_localization",
+        "event_impact", "source_control", "isolation_plan", "flushing_plan",
+        "field_feedback", "clearance_verification", "recovery_verification",
+    ]
+    summary: Annotated[str, Field(strict=True, min_length=1, max_length=160)]
+    evidence_refs: Annotated[list[TaskFeatureId], Field(min_length=1, max_length=100)]
+    next_step: Annotated[str, Field(strict=True, min_length=1, max_length=120)] | None = None
+
+    @model_validator(mode="after")
+    def validate_narration(self):
+        if not self.summary.strip() or (self.next_step is not None and not self.next_step.strip()):
+            raise ValueError("presentation narration must contain readable text")
+        if len(set(self.evidence_refs)) != len(self.evidence_refs):
+            raise ValueError("presentation narration evidence IDs must be unique")
+        return self
+
+
+TaskPresentationPurpose = Annotated[
+    str, StringConstraints(strict=True, strip_whitespace=True, min_length=1, max_length=160),
+]
+
+
+class TaskPresentationNarrationInput(TaskPresentationNarration):
+    """New live findings use business prose; saved narration remains readable."""
+
+    summary: Annotated[str, Field(strict=True, min_length=1, max_length=160,
+        description="本步已取得的关键发现，建议 80 字以内；区分观测、模型候选和现场反馈，不复述动作或核查目的")]
+    next_step: Annotated[str, Field(strict=True, min_length=1, max_length=120,
+        description="实际下一步或等待条件，建议 60 字以内；无后续时省略，不把展示当作现场执行")] | None = None
+
+    @field_validator("summary", "next_step")
+    @classmethod
+    def validate_reader_text(cls, value: str | None) -> str | None:
+        return validate_report_text(value) if value is not None else value
+
+
 class TaskPresentationEventCommand(StrictModel):
     request_id: UUID
     run_id: UUID
     source_session_id: NonEmptyStr
     command: TaskPresentationCommand
+    narration: TaskPresentationNarrationInput | None = None
+    purpose: TaskPresentationPurpose | None = Field(default=None,
+        description="一句中文说明本步为何查看这些对象，建议 60 字以内；不堆砌编号、执行参数或复述当前发现")
+
+    @field_validator("purpose")
+    @classmethod
+    def validate_reader_purpose(cls, value: str | None) -> str | None:
+        return validate_report_text(value) if value is not None else value
+
+    @model_validator(mode="after")
+    def validate_visible_labels(self):
+        validate_presentation_command_text(self.command.model_dump())
+        return self
+
+    @model_serializer(mode="wrap")
+    def serialize_legacy_command(self, handler):
+        value = handler(self)
+        # Keep old idempotency digests stable when no narration was supplied.
+        if self.narration is None:
+            value.pop("narration", None)
+        if self.purpose is None:
+            value.pop("purpose", None)
+        return value
 
 
 class TaskPresentationEventResult(StrictModel):
@@ -943,6 +1045,8 @@ class TaskPresentationHistoryItem(StrictModel):
     # Stored records are validated against the current command contract by the
     # reader; an obsolete record must not prevent reading the rest of the page.
     command: dict[str, JsonValue]
+    narration: TaskPresentationNarration | None = None
+    purpose: TaskPresentationPurpose | None = None
 
 
 class TaskPresentationHistoryPage(StrictModel):
diff --git a/app/domain/schemas/report_authoring.py b/app/domain/schemas/report_authoring.py
new file mode 100644
index 0000000..a8b1107
--- /dev/null
+++ b/app/domain/schemas/report_authoring.py
@@ -0,0 +1,73 @@
+"""Reader-facing text checks for new publications, never historical reads."""
+from __future__ import annotations
+
+import re
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from app.domain.schemas.answer_ui import ReportUiBlock
+
+
+_execution_text = re.compile(
+    r"\b(?:execution_generation|handling_context|verification_ready|waiting_work_order|"
+    r"waiting_scada|returned_at|verified_through|required_samples|interval_seconds|"
+    r"knowledge_cutoff|batch_sequence|query_complete|evidence_id|action_id|run_id|"
+    r"task_id|source_session_id|approval_status|execution_status|verification_evidence_id|"
+    r"normal_day_count|baseline_method|selection_method|get_task_context|query_scada_readings|"
+    r"run_burst_location|publish_task_report|publish_task_progress)\b|"
+    r"\b(?:status|phase|interval)\s*[:=]|\b(?:mcp|recovery|sha256):|"
+    r"\bskill(?:\.(?:name|version))?\s*[:=]|\bskill\s+[\w-]+\s+v?\d|"
+    r"\b(?:final|interim)/concluded\b|"
+    r"[0-9a-f]{8}-(?:[0-9a-f]{4}-){3}[0-9a-f]{12}",
+    re.IGNORECASE | re.ASCII,
+)
+_source_time = re.compile(
+    r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}|UTC\s*[+-]\s*\d|Asia/Shanghai|上海时间|"
+    r"\d{2}:\d{2}(?::\d{2})?\s*\+\d{2}:\d{2}",
+    re.IGNORECASE,
+)
+_unrendered_markup = re.compile(r"`|\*\*|</?[A-Za-z][^>]*>|^\s{0,3}#{1,6}\s", re.MULTILINE)
+
+
+def validate_report_text(value: str) -> str:
+    if _execution_text.search(value):
+        raise ValueError("请将执行字段和状态码改写为中文业务说明，完整编号保留在结构化引用中")
+    if _source_time.search(value):
+        raise ValueError("请将来源时间转换为当地日期和时刻，正文省略时区名称、偏移和 ISO 时间戳")
+    if _unrendered_markup.search(value):
+        raise ValueError("可见文字使用纯文本；重点与明细使用对应结果组件，不写 Markdown 或 HTML")
+    return value
+
+
+def validate_report_ui_text(blocks: list[ReportUiBlock]) -> None:
+    for item in blocks:
+        validate_report_text(item.block.fallback_text)
+        for element in item.block.spec.elements.values():
+            props = element.props
+            for key in ("title", "summary", "text", "description", "x_axis_name", "y_axis_name"):
+                if isinstance(props.get(key), str):
+                    validate_report_text(props[key])
+            for value in props.get("columns", []):
+                validate_report_text(value)
+            for row in props.get("rows", []):
+                for value in row:
+                    validate_report_text(value)
+            for value in props.get("x_data", []):
+                validate_report_text(value)
+            for series in props.get("series", []):
+                validate_report_text(series["name"])
+
+
+def validate_presentation_command_text(command: dict[str, Any]) -> None:
+    """Check visible labels without touching geometry, references or timestamps."""
+    if command["action"] == "set_overlay":
+        validate_report_text(command["title"])
+    elif command["action"] == "open_chart":
+        chart = command["chart"]
+        for key in ("title", "x_axis_name", "y_axis_name"):
+            if isinstance(chart.get(key), str):
+                validate_report_text(chart[key])
+        for value in chart["x_data"]:
+            validate_report_text(value)
+        for series in chart["series"]:
+            validate_report_text(series["name"])
diff --git a/app/mcp/analysis_models.py b/app/mcp/analysis_models.py
index 79a865c..6a54447 100644
--- a/app/mcp/analysis_models.py
+++ b/app/mcp/analysis_models.py
@@ -171,9 +171,20 @@ class ValveClosureInput(ScenarioInput):
 
 
 class FlushingInput(ScenarioInput):
-    drainage_node_id: Id
+    drainage_node_id: Id = Field(description="已核实的冲洗排放节点；清洁水来源、流向及污染水处置须在方案中说明")
     flushing_flow_m3h: float = Field(gt=0, allow_inf_nan=False, description="明确的冲洗流量，m³/h；提供流量来源与估算依据")
-    closed_valves: list[Id] = Field(default_factory=list, max_length=20)
+    closed_valves: list[Id] = Field(default_factory=list, max_length=20,
+        description="本次冲洗情景中保持关闭的完整阀门清单，用于隔离用户或约束流向")
+    opened_valves: list[Id] = Field(default_factory=list, max_length=20,
+        description="本次冲洗情景中明确打开的阀门，如清洁水进水和导流阀；不表示现场已开阀或恢复用户供水。未列阀门沿用当前模型状态，不继承上次独立模拟")
+
+    @model_validator(mode="after")
+    def distinct_valve_states(self):
+        if any(len(values) != len(set(values)) for values in (self.closed_valves, self.opened_valves)):
+            raise ValueError("flushing valve lists must contain distinct IDs")
+        if set(self.closed_valves) & set(self.opened_valves):
+            raise ValueError("a flushing valve cannot be both opened and closed")
+        return self
 
 
 NonnegativeValue = Annotated[float, Field(ge=0, allow_inf_nan=False)]
diff --git a/app/mcp/analysis_service.py b/app/mcp/analysis_service.py
index c8d6ce3..aa3e354 100644
--- a/app/mcp/analysis_service.py
+++ b/app/mcp/analysis_service.py
@@ -132,3 +132,12 @@ async def calculate_replay_contaminant_simulation(args: BaseModel, *, network: s
         args.model_dump(exclude={"context_handle", "timeout_seconds"}, exclude_none=True),
         network=network, username=username, model_only=True,
     )
+
+
+async def calculate_replay_flushing(args: BaseModel, *, network: str, username: str) -> dict[str, Any]:
+    """Evaluate a model-only flushing plan without live SCADA or field execution."""
+    return await bounded_calculation(
+        calculate_scenario, "analyze_flushing",
+        args.model_dump(exclude={"context_handle", "timeout_seconds"}, exclude_none=True),
+        network=network, username=username, model_only=True,
+    )
diff --git a/app/mcp/catalog.py b/app/mcp/catalog.py
index 752d88a..171efc9 100644
--- a/app/mcp/catalog.py
+++ b/app/mcp/catalog.py
@@ -66,7 +66,7 @@ _specs = [
     ("run_simulation", "按水力步长在在线业务模型上执行指定时间段的实时模拟；更新该时段的模型条件及实时基线，返回指定节点和管段结果；duration单位为秒，输出为在线模型结果，正式分析成果通过方案工具保存", ScenarioInput, "simulation.run", "analysis", False, True),
     ("run_contaminant_simulation", "在隔离模型中模拟假设污染源的浓度传播；include_impact_coverage=true时汇总全网节点整个模拟时段超过impact_threshold_mg_l的累计范围、首个超阈值报告时刻和峰值，并保存地图证据。阈值为情景展示条件，非饮用水安全限值；回放按当前模型独立估计，不读取实时SCADA。定位后先提现场核查/采样处置工单，工单审核派发与本计算并行，结果补入同一调查", ContaminantSimulationInput, "simulation.run", "analysis", False, True),
     ("analyze_valve_closure", "在隔离模型中模拟关阀的水力影响，返回指定节点/管段结果；现场阀门操作按业务审批执行", ValveClosureInput, "simulation.run", "analysis", False, True),
-    ("analyze_flushing", "在隔离模型中评估指定排放节点、冲洗流量和关阀组合的水力结果；现场效果由执行回执和复测核验", FlushingInput, "simulation.run", "analysis", False, True),
+    ("analyze_flushing", "在独立模型中评估指定排放节点、流量及明确开关阀组合的冲洗水力结果；未列阀门沿用当前模型，不继承上次模拟；不判定污染清除或恢复供水", FlushingInput, "simulation.run", "analysis", False, True),
     ("analyze_pressure_regulation", "在隔离模型中评估给定调泵序列和水箱初始水位的压力影响；输出为给定方案评估，现场调度按审批执行", PressureRegulationInput, "simulation.run", "analysis", False, True),
     ("optimize_sensor_placement", "使用灵敏度或K-means算法推荐压力监测点；返回候选节点，正式部署另行保存并核验观测能力", SensorPlacementInput, "optimization.run", "analysis", False, True),
     ("run_leakage_identification", "基于SCADA压力与模型进行DMA漏损分配识别，需有依据的总漏损流量和时间对齐；返回待现场核验的区域估计。内部执行遗传算法与水力计算", LeakageIdentificationInput, "burst.run", "analysis", False, True),
diff --git a/app/mcp/mutation_service.py b/app/mcp/mutation_service.py
index afe4496..56dd662 100644
--- a/app/mcp/mutation_service.py
+++ b/app/mcp/mutation_service.py
@@ -189,11 +189,12 @@ def saved_scenario(network: str, username: str, args: BaseModel) -> Any:
             modify_valve_opening={valve: 0.0 for valve in parameters["valves"]})
     if args.scenario == "flushing":
         from app.native.wndb.model.options import get_option
+        from app.mcp.scenario_service import flushing_valve_controls
         if get_option(network).get("UNITS") not in {"LPS", "CMH"}:
             raise ValueError("Flushing supports LPS or CMH models only")
         return scenarios.flushing_analysis(**common, username=username,
             drainage_node_ID=parameters["drainage_node_id"], flushing_flow=parameters["flushing_flow_m3h"],
-            modify_valve_opening={valve: 0.0 for valve in parameters["closed_valves"]})
+            valve_control=flushing_valve_controls(parameters))
     mappings = query_corresponding_element_id_and_query_id(network)
     controls = parameters["pump_control"]
     if set(controls) - (set(mappings.fixed_pumps) | set(mappings.variable_pumps)):
diff --git a/app/mcp/query_service.py b/app/mcp/query_service.py
index 223a0d9..567ebd7 100644
--- a/app/mcp/query_service.py
+++ b/app/mcp/query_service.py
@@ -79,7 +79,7 @@ async def authorize_query(
                    "get_element_properties", "list_elements", "query_nearby_elements", "get_network_options",
                    "get_node_connections", "get_element_gis_properties", "query_network_extent",
                    "list_regions", "get_region", "get_region_nodes",
-                   "analyze_valve_isolation", "generate_node_coverage", "run_burst_simulation", "run_contaminant_simulation",
+                   "analyze_valve_isolation", "generate_node_coverage", "run_burst_simulation", "run_contaminant_simulation", "analyze_flushing",
                    "get_calculation_job", "get_calculation_result", "list_calculation_jobs"}
         if tool.name not in allowed:
             raise HTTPException(403, "回放调查仅开放冻结 SCADA、资产、拓扑分析及指定模型情景；时序观测请使用 query_scada_readings")
@@ -91,6 +91,8 @@ async def authorize_query(
             if not args.include_impact_coverage:
                 raise HTTPException(422, "回放污染情景必须启用影响覆盖，按当前模型估计传播范围")
             check_replay_window(replay_context, args.start_time, args.start_time)
+        if tool.name == "analyze_flushing":
+            check_replay_window(replay_context, args.start_time, args.start_time)
         if tool.name == "query_scada_readings":
             check_replay_window(replay_context, args.start, args.end)
         if tool.name == "run_contaminant_source_trace":
@@ -233,6 +235,17 @@ async def execute_query(
         data = {"parameters": args.model_dump(mode="json", exclude={"context_handle"}), "result": result}
         origin = "synthetic"
         gaps = ["回放污染传播为当前模型的独立假设情景，不读取实时SCADA，不代表冻结观测或历史模型重建。未来模拟样本仅为预测；浓度阈值仅用于情景展示，不是饮用水安全限值。"]
+    elif name == "analyze_flushing" and replay_context:
+        from app.mcp.analysis_service import calculate_replay_flushing
+
+        await conn.commit()
+        async with scope.simulation_routing():
+            result = await calculate_replay_flushing(
+                args, network=scope.project.project_code, username=scope.user.username,
+            )
+        data = {"parameters": args.model_dump(mode="json", exclude={"context_handle"}), "result": result}
+        origin = "synthetic"
+        gaps = ["回放冲洗为当前模型的独立水力情景，不读取实时SCADA，不继承上次模拟；未来模拟样本仅为预测，不验证污染清除、水质合格或现场操作。"]
     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
@@ -540,7 +553,8 @@ async def execute_query(
                     "algorithm_status": "provisional",
                     "required_inputs": ["candidate_node_ids", "start", "end"],
                     "next_steps": ["present_node", "interim_report", "propose_task_action", "run_contaminant_simulation",
-                                   "analyze_valve_isolation", "generate_node_coverage", "report"],
+                                   "source_control", "analyze_valve_isolation", "generate_node_coverage",
+                                   "analyze_flushing", "field_feedback", "clearance_verification", "report"],
                     "work_order_wait_blocks_analysis": False,
                     "simulation_parameters": {"include_impact_coverage": True},
                 }
diff --git a/app/mcp/scenario_service.py b/app/mcp/scenario_service.py
index 0ab69d7..87d9bcf 100644
--- a/app/mcp/scenario_service.py
+++ b/app/mcp/scenario_service.py
@@ -10,6 +10,12 @@ SCENARIO_TOOLS = frozenset({
 })
 
 
+def flushing_valve_controls(parameters: dict[str, Any]) -> dict[str, dict[str, str]]:
+    """Apply explicit states without changing regulating-valve settings."""
+    return {**{valve: {"status": "CLOSED"} for valve in parameters.get("closed_valves", [])},
+            **{valve: {"status": "OPEN"} for valve in parameters.get("opened_valves", [])}}
+
+
 def calculate_scenario(name: str, parameters: dict[str, Any], *, network: str, username: str,
                        model_only: bool = False) -> dict[str, Any]:
     from app.services import simulation_scenarios as scenarios
@@ -60,9 +66,16 @@ def calculate_scenario(name: str, parameters: dict[str, Any], *, network: str, u
         from app.native.wndb.model.options import get_option
         if get_option(network).get("UNITS") not in {"LPS", "CMH"}:
             raise ValueError("Flushing currently supports LPS or CMH models only")
+        boundary = {}
+        if model_only:
+            from app.infra.db.postgresql.scada import ScadaElementMappings
+            boundary["scada_mappings"] = ScadaElementMappings(
+                reservoirs={}, tanks={}, fixed_pumps={}, variable_pumps={},
+                pressure={}, demand={}, quality={},
+            )
         result = scenarios.flushing_analysis(**common, username=username,
             drainage_node_ID=parameters["drainage_node_id"], flushing_flow=parameters["flushing_flow_m3h"],
-            modify_valve_opening={valve: 0.0 for valve in parameters["closed_valves"]})
+            valve_control=flushing_valve_controls(parameters), **boundary)
     elif name == "analyze_pressure_regulation":
         from app.services.simulation import query_corresponding_element_id_and_query_id
         mappings = query_corresponding_element_id_and_query_id(network)
@@ -102,4 +115,17 @@ def calculate_scenario(name: str, parameters: dict[str, Any], *, network: str, u
                                              threshold_mg_l=impact_threshold, model_only=model_only)
         impact["coverage"] = {**output["coverage"], **impact.get("coverage", {})}
         output.update(impact)
+    if name == "analyze_flushing":
+        output["flushing_plan"] = {
+            "opened_valves": parameters.get("opened_valves", []),
+            "closed_valves": parameters.get("closed_valves", []),
+            "drainage_node_id": parameters["drainage_node_id"],
+            "flushing_flow_m3h": parameters["flushing_flow_m3h"],
+            "estimated_discharge_volume_m3": parameters["flushing_flow_m3h"] * duration / 3600,
+            "state_basis": "independent_current_model_scenario",
+            "boundary_source": "model_only_no_live_scada" if model_only else "model_with_start_time_scada",
+            "result_scope": "hydraulic_only",
+            "field_execution_verified": False,
+            "water_quality_clearance_verified": False,
+        }
     return output
diff --git a/app/services/operational_tasks.py b/app/services/operational_tasks.py
index 2526665..c533972 100644
--- a/app/services/operational_tasks.py
+++ b/app/services/operational_tasks.py
@@ -1067,6 +1067,8 @@ class OperationalTaskService:
                 run_id=row["metadata"]["run_id"],
                 execution_generation=row["metadata"]["execution_generation"],
                 command=row["metadata"]["command"],
+                narration=row["metadata"].get("narration"),
+                purpose=row["metadata"].get("purpose"),
             ) for row in rows[:limit]]
             return TaskPresentationHistoryPage(
                 task_id=task_id, event_cursor=boundary, items=items,
@@ -1097,6 +1099,9 @@ class OperationalTaskService:
             now = self._now()
             self._require_active_run_lease(run, now)
             payload = _json(command.command)
+            if command.narration is not None:
+                known = {item["evidence_id"] for item in await self.repo.list_evidence(str(task_id))}
+                self._require_evidence_refs(command.narration.evidence_refs, known)
             await self._record_scada_presentation(task, run, payload, now)
             if command.command.action == "open_chart":
                 validate_time_comparison_chart(payload["chart"], allow_missing_current=bool(
@@ -1122,6 +1127,10 @@ class OperationalTaskService:
                 payload["data"] = resolved
             metadata = {"task_id": str(task_id), "run_id": str(run["run_id"]),
                         "execution_generation": run["generation"], "command": payload}
+            if command.narration is not None:
+                metadata["narration"] = _json(command.narration)
+            if command.purpose is not None:
+                metadata["purpose"] = command.purpose
             try:
                 size = len(json.dumps(metadata, ensure_ascii=False, allow_nan=False).encode("utf-8"))
             except (ValueError, TypeError) as exc:
diff --git a/app/services/scada_replay.py b/app/services/scada_replay.py
index 6a26aa7..882fdc7 100644
--- a/app/services/scada_replay.py
+++ b/app/services/scada_replay.py
@@ -4,6 +4,7 @@ from __future__ import annotations
 from datetime import datetime, timedelta, timezone
 import json
 from uuid import NAMESPACE_URL, uuid4, uuid5
+from zoneinfo import ZoneInfo
 
 from fastapi import HTTPException
 from app.core.config import settings
@@ -531,35 +532,24 @@ class ScadaReplayService:
         external = incident["external_id"]
         pressure = monitoring["pressure"]
         flow = monitoring["outlet_flow"]
-        detector = monitoring["burst_detection"]
-        score = detector.get("target_score")
-        score_text = f"{score:.4f}" if score is not None else "数据不足"
-        summary = (
-            f"SCADA 数据诊断发现指标异常，进入异常调查与爆管诊断。"
-            f"爆管侦测得分 {score_text}；"
-            f"压力下降测点 {len(pressure.get('hit_device_ids', []))} 个；"
-            f"出厂边界流量增加测点 {len(flow.get('hit_device_ids', []))} 个。"
-            "任一指标异常即启动调查，加权分用于评估爆管嫌疑；达到门槛后继续候选定位、隔离与影响分析。"
-        )
+        findings = []
+        if pressure.get("hit_device_ids"):
+            findings.append(f"{len(pressure['hit_device_ids'])} 个压力测点下降")
+        if flow.get("hit_device_ids"):
+            findings.append(f"{len(flow['hit_device_ids'])} 个出厂边界流量测点增加")
+        summary = f"SCADA 监测发现{'、'.join(findings) or '指标异常'}，已启动调查。"
         if monitoring.get("anomaly_signals"):
             signals = monitoring["anomaly_signals"]
             devices = {device for item in signals for device in item.get("device_ids", [])}
-            summary += f" 单项异常 {len(signals)} 项，涉及 {len(devices)} 个测点；明细通过冻结 SCADA 查询核验。"
-        if monitoring.get("weighted"):
-            weighted = monitoring["weighted"]
-            weights = weighted["weights"]
-            summary += (
-                f" 综合评分 {weighted['score']:.2f}，爆管嫌疑门槛 {weighted['threshold']:.2f}"
-                f"（侦测 {weights['burst_detection']:.0%}、压力 {weights['pressure']:.0%}、"
-                f"流量 {weights['outlet_flow']:.0%}）。"
-            )
-            if detector.get("score_threshold") is not None:
-                summary += f" 侦测动态阈值 {detector['score_threshold']:.4f}。"
+            summary += f"本次异常涉及 {len(devices)} 个测点。"
+        summary += ("现有监测结果提示爆管嫌疑，需结合管网与现场核验。"
+                    if monitoring["status"] == "suspected_burst" else "异常原因及影响范围待核查。")
+        event_time = parse_time(monitoring["period_end"]).astimezone(ZoneInfo("Asia/Shanghai"))
         command = OperationalTaskCreate(
             request_id=uuid5(NAMESPACE_URL, external),
             task_type="event_investigation",
             workflow="scada_anomaly",
-            title=f"SCADA 异常调查 · {monitoring['period_end']}",
+            title=f"SCADA 异常调查 · {event_time:%Y-%m-%d %H:%M}",
             summary=summary,
             source=TaskSource(
                 kind="external_system",
diff --git a/app/services/simulation_scenarios.py b/app/services/simulation_scenarios.py
index aa7da94..dcf1ba5 100644
--- a/app/services/simulation_scenarios.py
+++ b/app/services/simulation_scenarios.py
@@ -316,6 +316,7 @@ def flushing_analysis(
     valve_control: dict[str, dict] = None,
     _temporary_project: str | None = None,
     persist_results: bool = True,
+    scada_mappings: ScadaElementMappings | None = None,
 ) -> dict | str | None:
     """
     管道冲洗模拟
@@ -445,6 +446,7 @@ def flushing_analysis(
         modify_total_duration=modify_total_duration,
         modify_valve_opening=modify_valve_opening,
         valve_control=valve_control,
+        scada_mappings=scada_mappings,
         scheme_type="flushing_analysis",
         scheme_name=scheme_name,
         result_db_name=name,
diff --git a/contracts/manifest.json b/contracts/manifest.json
index b8fbe90..8fde00a 100644
--- a/contracts/manifest.json
+++ b/contracts/manifest.json
@@ -3,7 +3,7 @@
   "contracts": {
     "server": {
       "file": "server-v1.openapi.json",
-      "sha256": "aeff2c644006bf7eddd6cfe1c8aa0bad7f38c4632a456b5d3be1bb8201efc0db"
+      "sha256": "c31e9b4c71216951d1525acf1bcb8aeba4158b6cdc6b0b9dc01be02f9b6e6af3"
     }
   }
 }
diff --git a/contracts/server-v1.openapi.json b/contracts/server-v1.openapi.json
index 5e08730..2302a22 100644
--- a/contracts/server-v1.openapi.json
+++ b/contracts/server-v1.openapi.json
@@ -3537,6 +3537,81 @@
         "title": "InvestigationDataGap",
         "type": "object"
       },
+      "InvestigationDataGapInput": {
+        "additionalProperties": false,
+        "description": "New gap descriptions use business prose; references remain opaque.",
+        "properties": {
+          "blocking": {
+            "title": "Blocking",
+            "type": "boolean"
+          },
+          "evidence_refs": {
+            "items": {
+              "minLength": 1,
+              "type": "string"
+            },
+            "minItems": 1,
+            "title": "Evidence Refs",
+            "type": "array"
+          },
+          "gap_id": {
+            "minLength": 1,
+            "title": "Gap Id",
+            "type": "string"
+          },
+          "impact": {
+            "description": "资料缺项对当前判断的实际影响",
+            "maxLength": 4000,
+            "minLength": 1,
+            "title": "Impact",
+            "type": "string"
+          },
+          "missing_data": {
+            "description": "缺失的业务资料，不写执行字段",
+            "maxLength": 4000,
+            "minLength": 1,
+            "title": "Missing Data",
+            "type": "string"
+          },
+          "node_id": {
+            "anyOf": [
+              {
+                "minLength": 1,
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Node Id"
+          },
+          "required_data": {
+            "description": "需要补充的具体业务资料",
+            "maxLength": 4000,
+            "minLength": 1,
+            "title": "Required Data",
+            "type": "string"
+          },
+          "scope": {
+            "description": "缺口涉及的对象、地点与当地时间范围",
+            "maxLength": 4000,
+            "minLength": 1,
+            "title": "Scope",
+            "type": "string"
+          }
+        },
+        "required": [
+          "gap_id",
+          "missing_data",
+          "scope",
+          "impact",
+          "required_data",
+          "evidence_refs",
+          "blocking"
+        ],
+        "title": "InvestigationDataGapInput",
+        "type": "object"
+      },
       "InvestigationDiagram": {
         "additionalProperties": false,
         "properties": {
@@ -3907,7 +3982,8 @@
             ]
           },
           "summary": {
-            "maxLength": 20000,
+            "description": "用简短中文说明本轮新增发现、当前依赖和下一步；不复制报告、执行字段、版本号或证据编号",
+            "maxLength": 300,
             "minLength": 1,
             "title": "Summary",
             "type": "string"
@@ -4110,18 +4186,22 @@
         "description": "Write model: every new report explicitly declares its business result.",
         "properties": {
           "conclusion_boundary": {
+            "description": "只写影响判断的证据范围、不确定性和适用限制，不复述结论、执行步骤或参数",
+            "maxLength": 240,
             "minLength": 1,
             "title": "Conclusion Boundary",
             "type": "string"
           },
           "current_conclusion": {
+            "description": "先写当前业务判断，再写关键依据与下一步，建议 150 字以内；完整数值和对象比较使用 ui_blocks.Table",
+            "maxLength": 300,
             "minLength": 1,
             "title": "Current Conclusion",
             "type": "string"
           },
           "data_gaps": {
             "items": {
-              "$ref": "#/components/schemas/InvestigationDataGap"
+              "$ref": "#/components/schemas/InvestigationDataGapInput"
             },
             "title": "Data Gaps",
             "type": "array"
@@ -4154,6 +4234,7 @@
           },
           "pending_items": {
             "items": {
+              "maxLength": 160,
               "minLength": 1,
               "type": "string"
             },
@@ -4162,6 +4243,7 @@
           },
           "recommendations": {
             "items": {
+              "maxLength": 160,
               "minLength": 1,
               "type": "string"
             },
@@ -4196,6 +4278,7 @@
           "stop_reason": {
             "anyOf": [
               {
+                "maxLength": 160,
                 "minLength": 1,
                 "type": "string"
               },
@@ -4206,6 +4289,8 @@
             "title": "Stop Reason"
           },
           "title": {
+            "description": "简短中文业务标题，写明对象和当前结果，不含执行轮次、流程版本或原始时间戳",
+            "maxLength": 80,
             "minLength": 1,
             "title": "Title",
             "type": "string"
@@ -10159,6 +10244,30 @@
             ],
             "title": "Command"
           },
+          "narration": {
+            "anyOf": [
+              {
+                "$ref": "#/components/schemas/TaskPresentationNarrationInput"
+              },
+              {
+                "type": "null"
+              }
+            ]
+          },
+          "purpose": {
+            "anyOf": [
+              {
+                "maxLength": 160,
+                "minLength": 1,
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "一句中文说明本步为何查看这些对象，建议 60 字以内；不堆砌编号、执行参数或复述当前发现",
+            "title": "Purpose"
+          },
           "request_id": {
             "format": "uuid",
             "title": "Request Id",
@@ -10237,11 +10346,34 @@
             "title": "Execution Generation",
             "type": "integer"
           },
+          "narration": {
+            "anyOf": [
+              {
+                "$ref": "#/components/schemas/TaskPresentationNarration"
+              },
+              {
+                "type": "null"
+              }
+            ]
+          },
           "occurred_at": {
             "format": "date-time",
             "title": "Occurred At",
             "type": "string"
           },
+          "purpose": {
+            "anyOf": [
+              {
+                "maxLength": 160,
+                "minLength": 1,
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Purpose"
+          },
           "run_id": {
             "format": "uuid",
             "title": "Run Id",
@@ -10299,6 +10431,125 @@
         "title": "TaskPresentationHistoryPage",
         "type": "object"
       },
+      "TaskPresentationNarration": {
+        "additionalProperties": false,
+        "properties": {
+          "evidence_refs": {
+            "items": {
+              "maxLength": 128,
+              "minLength": 1,
+              "type": "string"
+            },
+            "maxItems": 100,
+            "minItems": 1,
+            "title": "Evidence Refs",
+            "type": "array"
+          },
+          "next_step": {
+            "anyOf": [
+              {
+                "maxLength": 120,
+                "minLength": 1,
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Next Step"
+          },
+          "stage": {
+            "enum": [
+              "anomaly_detection",
+              "evidence_verification",
+              "asset_localization",
+              "event_impact",
+              "source_control",
+              "isolation_plan",
+              "flushing_plan",
+              "field_feedback",
+              "clearance_verification",
+              "recovery_verification"
+            ],
+            "title": "Stage",
+            "type": "string"
+          },
+          "summary": {
+            "maxLength": 160,
+            "minLength": 1,
+            "title": "Summary",
+            "type": "string"
+          }
+        },
+        "required": [
+          "stage",
+          "summary",
+          "evidence_refs"
+        ],
+        "title": "TaskPresentationNarration",
+        "type": "object"
+      },
+      "TaskPresentationNarrationInput": {
+        "additionalProperties": false,
+        "description": "New live findings use business prose; saved narration remains readable.",
+        "properties": {
+          "evidence_refs": {
+            "items": {
+              "maxLength": 128,
+              "minLength": 1,
+              "type": "string"
+            },
+            "maxItems": 100,
+            "minItems": 1,
+            "title": "Evidence Refs",
+            "type": "array"
+          },
+          "next_step": {
+            "anyOf": [
+              {
+                "description": "实际下一步或等待条件，建议 60 字以内；无后续时省略，不把展示当作现场执行",
+                "maxLength": 120,
+                "minLength": 1,
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Next Step"
+          },
+          "stage": {
+            "enum": [
+              "anomaly_detection",
+              "evidence_verification",
+              "asset_localization",
+              "event_impact",
+              "source_control",
+              "isolation_plan",
+              "flushing_plan",
+              "field_feedback",
+              "clearance_verification",
+              "recovery_verification"
+            ],
+            "title": "Stage",
+            "type": "string"
+          },
+          "summary": {
+            "description": "本步已取得的关键发现，建议 80 字以内；区分观测、模型候选和现场反馈，不复述动作或核查目的",
+            "maxLength": 160,
+            "minLength": 1,
+            "title": "Summary",
+            "type": "string"
+          }
+        },
+        "required": [
+          "stage",
+          "summary",
+          "evidence_refs"
+        ],
+        "title": "TaskPresentationNarrationInput",
+        "type": "object"
+      },
       "TaskQuestion": {
         "additionalProperties": false,
         "properties": {
diff --git a/docs/investigation-progress-and-presentation.md b/docs/investigation-progress-and-presentation.md
index 0b268de..87ca368 100644
--- a/docs/investigation-progress-and-presentation.md
+++ b/docs/investigation-progress-and-presentation.md
@@ -32,7 +32,9 @@ Server 将阅读进展、业务报告和浏览器展示请求分别保存。运
 }
 ```
 
-示例中的编号由实际任务和资产替换。正文最长 20,000 字符，`graph_update` 可省略。
+示例中的编号由实际任务和资产替换。新发布的 `summary` 最多 300 字符，建议 120 字以内，
+只说明本轮新增发现、当前依赖和下一步；使用中文业务语言与当地时间，不复制报告或执行参数。
+详细结果和引用留在图节点、证据与报告明细中，历史正文仍按原值读取。`graph_update` 可省略。
 图示沿用 [调查报告契约](investigation-report-contract.md) 的 Mermaid 安全限制。
 节点标记可省略或为 null，均显示为参考步骤；新增量更新只接受以下五种：
 
@@ -107,6 +109,33 @@ Server 校验当前 run、绑定会话、有效租约和幂等身份，记录 `o
 响应为 `recorded/event_seq/run_id/execution_generation`，表示展示请求已保存，
 不表示浏览器已显示或现场动作已执行。
 
+### 展示表达与阅读
+
+实时展示与正式报告共用中文业务表达和时间要求，字段沿用现有结构：
+
+| 字段 | 表达职责 | 长度 |
+| --- | --- | --- |
+| purpose | 为什么核查这些对象；不复述动作名称、当前发现或完整报告 | 建议 60 字以内，最多 160 字符 |
+| narration.stage | 本步所属业务阶段，前端显示中文阶段名称 | 使用既有枚举 |
+| narration.summary | 本步已有证据支持的关键发现，区分观测、候选、模型与现场反馈 | 建议 80 字以内，最多 160 字符 |
+| narration.next_step | 实际下一步或等待条件；无后续时省略 | 建议 60 字以内，最多 120 字符 |
+| narration.evidence_refs | 该发现引用的实际证据，不作为正文展示 | 保持现有引用校验 |
+
+已有明确分析结果时提供 `narration`；纯视角调整或清除操作不编造发现。
+目的、发现和后续分别组织，不重复粘贴同一段文字。同一核查目的的批量图表可以复用有依据的
+相同讲解，保留合并阅读能力；不同发现分别保存，不为合并而抹去结论差异。
+
+讲解、目的和图表、覆盖层的可见标签均采用纯文本，省略执行字段、版本号、技术状态码、
+完整 UUID、证据哈希、时区名称、偏移和 ISO 时间戳。必要设备编号、单位与历史日期保留；
+标题交代对象、指标或预计范围，算法参数留在证据明细。结构化采样时刻、设备标识、数值、
+几何和证据引用保持原值，不因文字整理改变分析范围。
+
+Server 的新请求模型、Agent 原生工具及宿主入口校验这些文字规则；历史读取模型保持兼容。
+前端在当前动作说明中突出已保存的发现，再呈现目的和后续，缺少讲解的历史步骤只显示用途，
+不推断结论。讲解进入已有滚动阅读区，历史与播放控件留在标题区，长文字不挤出控件。
+历史选择固定在所选步骤；自动停留时间考虑实际可见的目的、发现与后续文字，
+用户手动选择的停留时间保持优先。上述文字和播放行为不控制后台调查、审批或现场执行。
+
 | command.action | 参数 |
 | --- | --- |
 | locate_features | layer、feature_ids |
diff --git a/docs/investigation-report-contract.md b/docs/investigation-report-contract.md
index 4cdc968..4cecb96 100644
--- a/docs/investigation-report-contract.md
+++ b/docs/investigation-report-contract.md
@@ -9,17 +9,19 @@
 
 | 字段 | 类型 | 新写入规则 |
 | --- | --- | --- |
-| title | 非空字符串 | 报告标题 |
-| current_conclusion | 非空字符串 | 当前判断文字；无结论时明确写无法判断，不编造原因 |
-| conclusion_boundary | 非空字符串 | 判断范围及限制 |
+| title | 非空字符串，最多 80 字符 | 简短中文业务标题，建议 30 字以内，不包含执行轮次和流程版本 |
+| current_conclusion | 非空字符串，最多 300 字符 | 首句写当前判断，接关键依据和实际下一步，建议 150 字以内；完整数值放入表格 |
+| conclusion_boundary | 非空字符串，最多 240 字符 | 影响判断的证据范围、不确定性和适用限制；不复述执行过程、参数或完整资产清单 |
 | evidence_refs | 字符串数组 | 引用该任务中已持久化、不可变的证据编号 |
-| recommendations | 字符串数组 | 保持原类型，处置建议不等于已执行动作 |
-| pending_items | 字符串数组 | 保持原类型，不从文字推断状态 |
+| recommendations | 字符串数组，每项最多 160 字符 | 具体对象与后续动作；处置建议不等于已执行动作 |
+| pending_items | 字符串数组，每项最多 160 字符 | 实际尚未完成或等待外部条件的事项，不从文字推断状态 |
 | stage | interim / final | 必填；阶段性报告不能用于结束运行 |
 | outcome | concluded / insufficient_data | 必填；业务调查结果，与运行成功/失败独立 |
-| stop_reason | 非空字符串或 null | 证据不足时必填；已形成结论时必须为 null |
+| stop_reason | 最多 160 字符的非空字符串或 null | 证据不足时必填；已形成结论时必须为 null |
 | data_gaps | 缺口对象数组 | 必填；空数组表示已记录且没有缺口 |
 | diagram | 调查脉络对象或 null | 可选；给人阅读的 Mermaid 说明，不控制执行或任务状态 |
+| ui_blocks | 结果组件数组，最多 8 块、合计 400 KB | 以现有 Result、Section、Table 等组件表达重点与明细，块内引用必须属于报告证据 |
+| scada_device_ids | 设备编号数组，最多 1000 项 | 监测明细与展示目标的完整集合，不能因精简摘要删除目标 |
 
 缺口对象除 node_id 外的字段必填：
 
@@ -27,10 +29,10 @@
 | --- | --- | --- |
 | gap_id | 非空字符串 | 报告内唯一的缺口编号 |
 | node_id | 非空字符串或 null | 可选的展示关联，不要求有执行计划 |
-| missing_data | 非空字符串 | 缺失数据内容 |
-| scope | 非空字符串 | 查询对象、地点及时间范围 |
-| impact | 非空字符串 | 因此无法回答的调查问题 |
-| required_data | 非空字符串 | 新一轮调查需要补齐的具体数据 |
+| missing_data | 非空字符串，最多 4000 字符 | 缺失的业务资料 |
+| scope | 非空字符串，最多 4000 字符 | 查询对象、地点及当地时间范围 |
+| impact | 非空字符串，最多 4000 字符 | 因此无法回答的调查问题 |
+| required_data | 非空字符串，最多 4000 字符 | 新一轮调查需要补齐的具体数据 |
 | evidence_refs | 至少一个证据编号 | 实际查询或可信输入记录；必须也出现在报告 evidence_refs 中 |
 | blocking | boolean | 是否阻止形成结论 |
 
@@ -42,6 +44,48 @@
 不包含 Markdown 围栏、初始化指令和 click 命令。前端使用 strict 安全级别渲染 SVG。
 完整图语法的解析失败只影响图示，正文、依据和审核仍可阅读。
 
+## 报告表达与生成要求
+
+报告面向供水运维人员，按当前判断、关键依据、判断限制、后续安排组织内容。
+字段仍按现有模型发布，不将多段报告重新堆入 `current_conclusion` 或 `conclusion_boundary`。
+
+- 结论首句直接回答当前发生了什么、形成什么判断。随后只保留最关键的变化和实际下一步，
+  正文最多举两个关键对象。评分权重、算法参数、全部测点、阀门和影响对象放入依据明细。
+- 多个对象或多组数值比较使用有证据引用的 `ui_blocks.Table`。表格按实际资料选择对象、
+  指标、时刻、异常值、当前值、基线与结果等列，单位放在列名或单元格；缺值明确写未取得。
+  全部目标按需分表，不以精简正文为由缩小调查、证据或 SCADA 展示覆盖。
+- 使用 `Result` 突出有依据的判断、关键变化或实际待办，使用带业务标题的 `Section` 组织明细。
+  模型候选或尚待现场核验用 `unconfirmed`，部分完成用 `partial`；`success` 必须有相应通过依据。
+  现有文字字段按纯文本展示，不写 Markdown 加粗、标题、表格或 HTML，也不引入自定义强调字段。
+- 正文、进展、缺口说明和组件可见文字采用中文业务语言。执行字段、技术状态码、流程版本、
+  完整 UUID 和证据哈希留在结构化引用或执行记录中。必要资产编号、SCADA、DN600、单位、
+  模型相似度和有业务用途的工单简称可以保留，不能机械删除所有英文或数字。
+- 阅读时间先换算为 `Asia/Shanghai`，再以当地日期和时刻显示，例如 `2026-07-16 13:15`。
+  正文、标题、进展、表格和图表均省略时区名称、偏移及 ISO 原始时间戳。已有 `+08:00`
+  不重复换算，跨日同步更新日期；查询时间、`sample_times` 和证据时间戳仍保留明确偏移。
+  事件发生、工单完工、恢复采样和报告生成时间各自标明用途，不能互相代替。
+- 区分观测、模型估计、候选和现场反馈。相似度不是发生概率，提交工单不证明现场完工，
+  单帧复核通过不证明长期稳定；不同来源或阶段的设备数分别说明范围，不能混用分母。
+  单条流量规则未触发不表示所有流量未变化；不同影响集合不能直接相加。
+- 判断边界只保留影响理解的限制。实际资料缺项放入 `data_gaps`；未来常规采样和日常监测
+  写入建议，不将其包装成当前关键缺口。非阻塞资料缺项不得新增事件闭环条件。
+
+初次调查说明异常、候选及预计影响；等待处置说明已完成事项、实际等待条件和下一步；
+恢复报告先说明现场反馈与完工时间、恢复复核时刻与结果，再给出必要限制和后续监测安排。
+恢复明细来自本次恢复证据，原异常分析作为历史依据保留，不在摘要中重复全部定位与隔离参数。
+
+`progress.summary` 最多 300 字符，建议 120 字以内，只写本轮新增发现、当前依赖和下一步，
+不复制正式报告、不汇报执行字段和流程版本。整理报告时核对脉络图：已解除的等待条件应按
+原节点更新业务说明和参考标记，保留失败尝试与原路线，不重复添加相同连线。
+这不要求逐步更新图示，也不改变任何运行、审批、等待或闭环判据。
+
+Server 的新写入模型及 Agent 的原生工具、宿主发布入口校验长度、明显执行字段与技术编号、
+原始时间表达和无法渲染的格式标记；组件的可见文字同样校验，结构化证据编号与采样时间除外。
+校验错误返回具体原因，由 Agent 依据原证据改写后重试，不截断正文、不删除引用或补造缺口。
+语义准确性、信息取舍和表格完整性仍需遵循提示词与既有证据、覆盖校验，不能由正则代替。
+
+上述长度和文字规则仅约束新发布，不应用于历史读取或原证据快照；不回写已有报告。
+
 每条 node_ref 包含 `node_id`、`label`、`evidence_refs`、`action_refs`；后两项默认空数组。
 node_id 为最多 80 字符的 ASCII 标识符，必须以字母或下划线开头，在本图引用内唯一，
 并与 Mermaid 源码中实际节点使用完全相同的 ID；禁止为同一节点另造引用 ID；
@@ -133,7 +177,7 @@ failure_code 为 ARTIFACT_INCOMPLETE；调用方须读取返回的业务状态
     "gap_id": "terminal-sample-gap",
     "node_id": null,
     "missing_data": "投诉时段末梢水样检测结果",
-    "scope": "投诉小区，2026-09-24 08:00 至 09:00（北京时间）",
+    "scope": "投诉小区，2026-09-24 08:00 至 09:00",
     "impact": "无法判断异常来自供水管网还是用户端。",
     "required_data": "该地点和时段的采样检测记录",
     "evidence_refs": ["sample-query-empty"],
@@ -150,6 +194,10 @@ failure_code 为 ARTIFACT_INCOMPLETE；调用方须读取返回的业务状态
 - 已有未决动作阻止矛盾的缺数退出；重试幂等，不重复生成报告。
 - 新轮次保留旧报告历史标识，不能显示成当前轮次已结束。
 - 桌面及 320/375px 展示结果、缺口、证据，审批区不出现数据授权入口。
+- 新发布拒绝超长摘要、正文执行字段、完整技术编号、原始时间表达和未支持的格式标记；
+  相同历史报告仍可读取，结构化证据引用与带偏移的采样时间保持不变。
+- 生成结果结论在前、明细入表，明确单位、时间和各自覆盖范围；
+  候选、现场反馈、单帧复核与长期监测保持各自含义，进展和图示不与当前事实冲突。
 
 ## 发布、阅读与协作
 
diff --git a/docs/mcp-analysis-tools.md b/docs/mcp-analysis-tools.md
index b5f7f1b..1516bf7 100644
--- a/docs/mcp-analysis-tools.md
+++ b/docs/mcp-analysis-tools.md
@@ -12,7 +12,7 @@ Agent 原生工具统一使用 `mcp_tjwater_` 前缀。Server目录共67项：42
 | `run_leakage_identification` | SCADA压力约束下的DMA漏损分配估计；需总漏损量依据 | `burst.run` + `scada.view` |
 | `run_contaminant_simulation` | 假设源正向传播；可选按全网全时段浓度汇总累计范围、首个超阈值报告时刻与峰值，详见[污染流程](water-pollution-response-workflow.md)；与工单并行，不承担反演或安全判定 | `simulation.run` + `scada.view` |
 | `analyze_valve_closure` | 给定关阀组合的水力影响 | `simulation.run` + `scada.view` |
-| `analyze_flushing` | 指定排放节点和流量的冲洗水力影响；不证明消毒合格 | `simulation.run` + `scada.view` |
+| `analyze_flushing` | 指定排放点、流量与互斥的 `opened_valves` / `closed_valves`，评估独立冲洗水力情景并保存阀态和预计排放量；不证明污染清除或水质合格 | `simulation.run` + `scada.view` |
 | `analyze_pressure_regulation` | 给定调泵序列、水箱初始水位的压力影响；不自动优化调度 | `simulation.run` + `scada.view` |
 | `optimize_sensor_placement` | 灵敏度或K-means压力监测点建议；不保存正式布点方案 | `optimization.run`，项目成员角色 |
 
diff --git a/docs/water-pollution-response-workflow.md b/docs/water-pollution-response-workflow.md
index c5691e2..3b6b7dd 100644
--- a/docs/water-pollution-response-workflow.md
+++ b/docs/water-pollution-response-workflow.md
@@ -6,20 +6,26 @@
 flowchart TD
     detection[有效水质观测命中业务规则] --> task[独立水污染事件调查]
     task --> evidence[全量观测及七天同期核验]
-    evidence --> trace[快速候选溯源]
+    evidence --> protection[并行提出用户保护与应急供水措施]
+    evidence --> trace[候选溯源与采样核验]
     trace --> node[展示已核验节点]
     node --> report[候选源阶段结论]
     report --> action[立即提出现场核查采样工单供审批派发]
     report --> simulation[污染传播模拟与累计浓度影响范围]
-    report --> isolation[节点隔离与阀门边界]
-    simulation --> impact[同一调查补充证据地图与报告]
-    isolation --> impact
-    action --> feedback[采样处置复测反馈]
-    feedback --> review[接续原调查复核]
-    impact --> review
-    review --> final[最终报告及事件收尾]
+    simulation --> source[污染输入控制建议]
+    source --> isolation[保护用户与隔离边界]
+    isolation --> treatment[按污染物选择清除方法]
+    treatment --> flushing[清洁水入口 导流阀门 排放处置与分步冲洗]
+    action --> feedback[现场操作与采样反馈]
+    flushing --> feedback
+    feedback --> review[核对实际水质复测与恢复依据]
+    review --> decision{复测依据是否支持恢复}
+    decision -->|否或资料不足| treatment
+    decision -->|是| final[分区恢复建议与最终报告]
 ```
 
+图中连线表示业务与展示层次，先期保护、工单审核和独立计算可以并行。阀门操作均为方案，现场动作继续经过原审批与派发；阶段记录不能替代执行回执。首次展示保留处置前传播结果或明确的数据缺口，再展示隔离及清除方案。
+
 ## 侦测入口
 
 现有 `ScadaReplayService` 使用冻结的水质设备类型 `quality` / `water_quality` 判别水质规则事件。仪表无效值、缺测产生的 `data_quality` 告警单独处理。水质业务规则命中直接创建污染调查，即使同时运行压力／流量周期诊断，也不被合并成爆管调查。持续规则事件沿用同一任务，规则恢复后接续该任务的新一轮。压力／流量与水质同时异常时各自保留调查。
@@ -46,6 +52,22 @@ flowchart TD
 
 ## 展示、反馈与收尾
 
+`water-pollution-event` 1.2.0 将观测核验、候选定位、处置前传播、源头控制、用户隔离、清除冲洗、现场反馈、水质复测和恢复复核分开。每次正式展示保存 `purpose`；有证据的阶段通过 `narration.stage` 记录，新增 `source_control`、`flushing_plan`、`clearance_verification`。历史保存格式继续使用原事件 metadata JSON，不增加数据库列。前端显示阶段和本步目的，逐步回看保留独立画面；同阶段兼容图表仍可合并。
+
+源头控制与保护下游用户分别决策。用已核实的连接和流向判断水力下游，不能按地理方位判断；源头未核实不得声称污染输入已停止。清除方法按污染物及现场条件选择，不统一推荐煮沸、加药或排水。通知保留，工单按执行职责和阶段复用或必要拆分，不能同时要求同一阀门开和关。
+
+### 冲洗模型与阀门方案
+
+`analyze_flushing` 增加可选 `opened_valves`，与 `closed_valves` 互斥且各自去重。明确开阀用于清洁水进入或导流，关阀用于保护用户和约束路径；不改变阀门原调节设定。普通计算和保存冲洗方案使用相同显式阀态。未列阀门沿用当前模型状态，每次计算为独立情景，不能继承上次计算的阀位或浓度场。
+
+每一步操作表保存：目的、前置条件、清洁水源及依据、排放节点与污染水处置去向、完整开关阀清单、计划顺序、流量、持续时间、预计排放量、监测要求、实际执行反馈。清洁进水阀打开不表示用户可以恢复供水。路径只采用已核实资产几何，不用直线伪造流向。
+
+计算返回 `flushing_plan`，包含开关阀、排放节点、流量和 `estimated_discharge_volume_m3`（给定流量乘持续时间）。它仅是水力情景，`water_quality_clearance_verified=false`、`field_execution_verified=false`；估计水量不是实测排放量。当前不模拟停止污染输入后的连续清除衰减，也不验证进水洁净、排放合规或实验室合格。缺少模型或现场依据时报告具体限制，不伪造清除效果。带回放上下文的调查仅新增独立模型冲洗情景：起算不晚于知识截止，始终传入空 SCADA 映射，禁止实时观测回填；原角色、任务、权限及作业所有权校验保留。历史播放读取已保存方案，不重新计算或现场操作。
+
+### 复测与恢复依据
+
+复核记录采样地点、时间、指标、数值、单位、适用判定依据、受影响范围覆盖及结论，区分实际结果和模型预测。工单完成、压力恢复或余氯单项正常不能证明目标污染物已清除。关键复测资料不足或未通过时，Agent 保存 `interim` 并提出具体补测或调整方案；不统一新增固定轮数、时长或阈值。已有足够复测与恢复依据时，展示可恢复区域并发布引用有效现场反馈及复测证据的最终报告。Server 继续校验反馈引用，不自动判定实验室合格，也不授权设备操作或现场复供。
+
 水质图表使用 `chart.scada.metric=quality`，沿用真实单位、三小时当前趋势、前七天同期散点及证据校验；不同指标分图。同单位不同水质指标通过图表纵轴名称区分展示分组。Server、Agent、Frontend 的图表协议同步支持 quality，正式报告仍需引用本轮全量 SCADA 证据。
 
 关联现场任务使用 general 工单。成功创建后保存在 `handling_context.work_orders`，进入 `waiting_work_order`；全部有效工单 completed 后进入 `field_review`，接续 Agent，而非执行爆管压力恢复计算。最终 concluded 报告必须引用全部有效工单最新的 `feedback_evidence_id`，然后由 Server 完成任务并解除关联告警。完工反馈或新版本复测资料会重新进入复核；证据不齐时 Agent 使用现有提问／新动作流程等待事实。该机制校验反馈引用与调查收尾，不自动判断实验室水质合格。
diff --git a/resources/work-order-skills/scada-anomaly/SKILL.md b/resources/work-order-skills/scada-anomaly/SKILL.md
index 02491f3..0b5f188 100644
--- a/resources/work-order-skills/scada-anomaly/SKILL.md
+++ b/resources/work-order-skills/scada-anomaly/SKILL.md
@@ -2,7 +2,7 @@
 name: scada-anomaly
 description: 从 SCADA 指标异常开展数据核验、爆管诊断、候选定位与影响分析。
 metadata:
-  version: "1.0.18"
+  version: "1.0.19"
 ---
 
 # SCADA 数据异常调查
@@ -11,7 +11,7 @@ metadata:
 
 先按handling_context接续：待审批/工单/监测且无独立工作时，保存本轮interim/concluded及action_id后退出，不轮询。verification_ready复用原异常证据，只补恢复复核；初次分析按下文。
 
-报告与图表的阅读时区统一为Asia/Shanghai（UTC+8）。工具返回的+08:00时间按原值使用；Z时间换算后同步更新日期、历史系列名和横轴标签，例如2026-07-16T00:00:00Z对应7月16日08:00。sample_times记录相同真实时间点并使用+08:00偏移，查询与冻结边界按带时区的时间点比较。图表标题、横轴和系列名只写日期、时刻与单位，不写“上海时间”“Asia/Shanghai”或“UTC+8”；横轴命名为“时间”。
+阅读时间按 Asia/Shanghai 换算，已有 +08:00 不重复换算，跨日更新日期。正文、标题、进展、表格和图表只写当地日期、时刻与单位，省略时区名称、偏移及原始时间戳；横轴用“时间”。查询、冻结边界和 sample_times 保留明确偏移。
 
 ## 数据核验与诊断
 
@@ -21,7 +21,7 @@ metadata:
 
 全量devices含完整目录，readings含观测，coverage含各设备读数数量、缺测及冻结范围外清单。不得由代表测点推断全网。单查list_scada_devices默认每页100，读至page.has_more=false，属性缩略不代表完整信息。全量后可按device_ids局部复核并持续翻页，不能替代全量输入。final引用本轮knowledge_cutoff及batch_sequence的全量证据；Server拒绝局部或旧批次证据。
 
-分别说明压力异常模型得分、压力绝对下降及变化率、已配置出厂边界流量变化，再说明联合证据支持程度。使用实际权重、嫌疑门槛、动态阈值、参与测点与各自基线。综合分用于爆管嫌疑判断；得分越低越异常，目标告警与参考日分别解释。水质等其他异常按实际证据选择原因分支。
+核对模型得分、压力变化与出厂边界流量的联合证据；权重、门槛、阈值、测点与基线留在依据。得分越低越异常，相似度非发生概率。区分边界与其他管流测点，规则零命中不代表无流量变化；其他异常按证据选择原因。
 
 ## 候选定位与隔离
 
@@ -49,7 +49,7 @@ run_burst_simulation设include_impact_coverage=true，用定位管道、异常
 
 本轮时序分析保持任务冻结窗口，后续播放形成的观测由相应轮次处理。资产属性、空间查询和拓扑隔离使用项目当前模型，注明模型时态；水力工具按已开放的冻结输入能力调用。定位调用执行权限、设备与冻结观测校验；基线或覆盖缺口按实际返回记录。
 
-报告含观测、规则、原因、定位隔离、影响和建议；摘要≤150字，测点用表格、完整依据留证据。完成分析、展示与报告后收尾；现场核验、审批维修按回执列待办。现场工单与设备操作遵守服务端动作权限，恢复复核通过后闭环。
+报告先写判断与关键依据，摘要建议 150 字、最多 300 字；多测点数值入 Table，保留单位、方向与缺测。区分各阶段测点数，不相加不同影响集合。现场事项按回执列待办。现场工单与设备操作遵守服务端动作权限，恢复复核通过后闭环。
 
 现场工单用propose_task_action（kind=create_work_order、work_order_type=general），注明对象、条件、回执及回放来源；有权限者确认派发，通知保留，报告完成不等于现场完成；回放禁设备命令。
 
@@ -59,4 +59,6 @@ run_burst_simulation设include_impact_coverage=true，用定位管道、异常
 
 waiting_work_order等待完工；waiting_scada按last_check说明尚无后续采样、数据缺项或仍异常，等待监测复核。verification_ready时用get_task_evidence读取verification_evidence_id完整快照；工单completed且恢复证据status=passed时，发布final/concluded恢复闭环报告，同时引用原异常全量证据与当前恢复证据。Server完成运行后关闭任务并解除关联告警，phase=completed表示已闭环。阀位、漏点照片、现场复测数值等未提供时仅记为资料缺项，不新增现场核验、人工确认或更多采样的闭环条件。
 
+恢复报告先写现场反馈与完工时间、采样时刻和结果，表格列全部复核测点及基线；边界说明单帧与模型限制，日常监测列建议。不重列旧定位参数或执行字段；更新原图中已解除的等待标记，保留路线与失败记录。
+
 恢复观测由Server独立保存，原冻结上限不阻断此证据的使用；query_scada_readings仍遵守冻结窗口。完工时间修订后使用新证据编号，旧证据仅作历史。确有独立工作时再提议新动作。
diff --git a/resources/work-order-skills/water-pollution-event/SKILL.md b/resources/work-order-skills/water-pollution-event/SKILL.md
index 6ad77eb..2b00745 100644
--- a/resources/work-order-skills/water-pollution-event/SKILL.md
+++ b/resources/work-order-skills/water-pollution-event/SKILL.md
@@ -1,62 +1,72 @@
 ---
 name: water-pollution-event
-description: 从水质监测侦测信号调查水污染事件，完成数据核验、候选溯源、隔离与影响分析、处置报告和反馈复核。
+description: 水污染侦测后的观测核验、候选溯源、源头控制、用户隔离、清除冲洗与水质复测。
 metadata:
-  version: "1.1.0"
+  version: "1.2.0"
 ---
 
 # 水污染侦测事件调查
 
-适用于 `workflow=water_pollution`，入口是水质监测规则侦测异常，与供水水质投诉 `water_quality_complaint` 分开。无需投诉或人工上报才能调查，也不需要压力下降、流量增加或爆管综合评分达到门槛。侦测异常、候选污染源和污染确诊分别说明。
+适用于 workflow=water_pollution，与 water_quality_complaint 投诉分开，不依赖爆管评分、漏量或压力流量异常。侦测信号、候选污染源、污染确诊和恢复供水分别说明。
 
-先读取任务、规则信号、异常设备、发生时刻、已有证据与报告。任务携带 replay_context 时，查询严格遵守 knowledge_cutoff 和 batch_sequence；普通侦测任务使用实际异常时刻。时间统一按 Asia/Shanghai 阅读，查询和 sample_times 保留明确时区，图表标题只写日期、时刻与单位。后续轮次优先处理新事实，复用已有资产、溯源、隔离和展示成果。
+读取任务、信号、时间、handling_context 和已有成果。回放遵守 knowledge_cutoff、batch_sequence；阅读时间 Asia/Shanghai，查询与 sample_times 带时区。后续轮次复用成果，优先新增事实。
 
-## 1. 核验侦测与观测
+## 1. 核验观测与先期保护
 
-调用 query_scada_readings，省略 device_ids，全量读取异常时刻之前三小时的授权 SCADA；核对 devices、readings、coverage、分页完成与缺测。核查水质指标名称、单位、设备状态、清洗质量、模型节点映射以及实际规则上下限和连续命中条件。超出仪表量程、无效值和缺测属于数据质量问题，不能直接当作污染。有效水质读数超出配置业务范围支持异常调查，不自行认定为法定水质超标。
+query_scada_readings 省略 device_ids，读取异常前三小时全量 SCADA，核查分页完成、覆盖、质量、单位、映射及配置阈值。无效值或缺测不等于污染，业务异常不等于法定超标。逐设备对照七天同期，保留日期和 null；压力正常不能排除污染。
 
-对照同一设备前七天同期及实际采样记录，保留真实时间和 null 缺口。压力、流量仅作关联依据，正常压力不能排除污染，爆管得分不用于替代水污染判断。没有适用的限值或采样结果时记录具体缺口，继续候选分析。
+核验、采样与用户保护并行。按风险提出用水限制、通知、重点用户保障及应急供水；缺关联、适配器或批准依据则交人工核定，不称已执行。保护措施不等溯源或模拟，未知污染物不统一建议煮沸、加药或冲洗。
 
-## 2. 候选溯源与节点展示
+## 2. 候选溯源与现场工单
 
-使用水质设备映射取得异常关联节点，以 get_element_properties、get_node_connections 核对。映射缺失时查询设备坐标附近模型节点，不猜编号。候选必须来自当前项目已查询资产。
+用设备映射、get_element_properties、get_node_connections 核实节点，缺映射则查坐标附近资产。检查 source_trace_capability，run_contaminant_source_trace 提交已核实 candidate_node_ids 和核查窗，读取候选、缺失节点及 selection_basis。当前为临时确定性选点，非正式反演，不编造概率或确诊。synthetic 不阻止分析，无节点仅停依赖分支。
 
-检查 source_trace_capability，调用 run_contaminant_source_trace：candidate_node_ids 提交已核验的候选节点，start/end 为实际核查窗。读取 located_node、top_candidates、validated_node_ids、missing_node_ids 和 selection_basis。当前接口以稳定规则快速返回确定节点，正式反演算法后续接入，不虚构相似度、确诊概率或污染范围。
+取得候选后立即 publish_task_presentation 展示节点，保存 interim/concluded，注明“候选污染源待现场核验、传播范围计算中”；propose_task_action 提出 kind=create_work_order、work_order_type=general 的现场核查／采样处置草稿，引用定位证据和地点。沿用已有 action_id，不等模拟完成才发起工单。审核派发与模拟、隔离分析并行，结果补入同一 task_id，不覆盖已审核工单或绕过版本校验。
 
-临时算法和 synthetic 来源只说明结果取得方式，不构成停止调查的条件。取得 located_node 后立即通过 publish_task_presentation 展示候选节点，继续隔离、覆盖、影响和报告。无候选时修正资产映射；确实无可用节点时仅暂停依赖源节点的分支，完成已有监测事实与处置建议。
+## 3. 处置前污染影响
 
-定位到源节点后，立即保存 interim/concluded 阶段报告，注明“候选污染源待现场核验、传播范围计算中”，调用 propose_task_action 提出 kind=create_work_order、work_order_type=general 的现场核查／采样处置草稿，引用节点、地点和定位证据。沿用已有 action_id，不因模拟返回再建同类工单。工单审核、派发与传播模拟、隔离分析并行，不等模拟完成才发起工单；审批权限与动作版本校验沿用 Server 流程。
+run_contaminant_simulation 设 source=located_node、include_impact_coverage=true，采用事件起算时刻、明确预测时长和需画曲线的 output_node_ids。concentration 为 mg/L，须有来源或说明情景假设，不能把浊度、余氯下降或感官描述直接换算成污染源浓度。缺依据时列具体参数缺口，继续工单和独立分析。
 
-## 3. 处置前的污染传播模拟
+impact_threshold_mg_l 默认0、严格大于，仅供展示，非饮用水安全限值。全网整个时段汇总累计节点、首个超阈值采样时刻、峰值与覆盖；沿用初始水质和反应条件，非源独立贡献，范围外不等于安全。
 
-继续调用 run_contaminant_simulation，source=located_node、include_impact_coverage=true；start_time 使用事件业务时刻，duration 使用明确的情景预测时长，output_node_ids 选择源节点和需展示曲线的监测／重要节点。源浓度 concentration 单位 mg/L，使用有来源的浓度或有明确依据的情景假设，保留依据与不确定性，不能把浊度、余氯下降或感官描述直接换算成污染源浓度。缺少可用依据时记录具体参数缺口，继续工单和隔离分支。
+queued/running 时继续独立工作，get_calculation_job 跟踪、get_calculation_result 读终态。未完计算不能 final，由宿主等待接续；待审批不阻止分析，模拟失败不撤回工单。
 
-impact_threshold_mg_l 仅作模型影响范围的展示条件；默认0表示正浓度，非饮用水安全限值。工具在筛选展示节点前汇总全网节点整个模拟时段，返回累计超阈值节点、首个超阈值报告时刻、峰值和地图覆盖。沿用模型初始水质及反应条件，不能将总模型浓度当作源独立贡献，不能据此宣布范围外安全。
+按 presentation_preview.evidence_id 发布“水污染传播预计影响范围（处置前情景）”、参数与计数表及浓度曲线；模型曲线不填观测专用 chart.scada。保存正式事件供历史复用；空范围、缺测、单位及几何限制单列。先保留污染传播步骤，再展示隔离范围，不以隔离覆盖替代传播范围。
 
-queued/running 时继续独立工作，用 get_calculation_job 跟踪并以 get_calculation_result 读取终态，不以回执代替结果。待审批或 waiting_work_order 不阻止继续可执行分析；模拟结果补入同一 task_id 的证据、进度与报告，不覆盖已审核工单或绕过版本校验。模拟失败只记录该分支问题，不撤回已有现场核查工单。存在未完成计算时不能结束调查运行或发布 final，按宿主计算等待机制接续。
+回放允许传播及冲洗情景，起算不晚于知识截止，使用当前模型且不读取实时 SCADA；未来样本仅预测，非历史重建。
 
-读取结果后，用 presentation_preview 的 evidence_id 发布“水污染传播预计影响范围（处置前情景）”地图；将源浓度、阈值、预测窗、影响节点数、首达采样时刻和峰值写入报告表格，按真实时刻展示模型浓度曲线，模型曲线不填观测专用 chart.scada。地图和图表进入正式调查展示事件，历史播放读取已保存证据，不重新运行模拟。空范围、缺测、不支持的单位或几何超限按实际结果说明。保留污染传播步骤，再展示隔离范围，不以隔离覆盖替代传播范围。
+## 4. 源头控制与用户隔离
 
-带 replay_context 时同样可运行此情景，起算不得晚于知识截止；使用当前模型且不读取实时 SCADA。知识截止后的模拟时刻是预测，不得当作已观测数据或历史模型重建。
+分别提出污染输入控制、污染区边界与用户保护。源头措施依据污染物、进入路径和现场事实；候选节点不是可关设备，不假定源已停止。用流向及连接核对水力下游，不能按地理方位推断。
 
-## 4. 隔离边界与影响
+analyze_valve_isolation 设 accident_element=[located_node]、accident_element_type="node"，保留 isolatable、must_close_valves、optional_valves、affected_nodes、supply_affected_nodes 及计数。失效阀门用 disabled_valves 核查替代方案，不能隔离时保留原因。
 
-调用 analyze_valve_isolation，accident_element=[located_node]，明确 accident_element_type="node"，避免节点与管段重号。保留 isolatable、must_close_valves、optional_valves、affected_nodes、supply_affected_nodes 及各自计数。已知失效阀门通过 disabled_valves 分析替代边界；不能隔离时记录原因并继续其他处置建议。
+展示阀门，generate_node_coverage 分组生成直接隔离及新增失源覆盖，再 publish_task_presentation；空组不算。拓扑覆盖非污染边界，无资料不编户数。有权限和输入时 analyze_valve_closure 核验水力影响，不证明污染清除。
 
-展示候选节点、必关与可选阀门。用 generate_node_coverage 分组生成直接隔离、新增供水失源覆盖，再通过 publish_task_presentation 发布。空节点组不提交覆盖计算。拓扑隔离、供水失源、污染传播、水力变化和用户影响分别说明，节点覆盖不能当作污染传播边界，无用户资料不虚构户数。
+## 5. 污染物清除与分步冲洗
 
-输入和权限具备时使用 analyze_valve_closure 评估关阀水力影响，污染传播复用第3节结果。有排放节点、流量和前置条件时用 analyze_flushing 评估冲洗，缺少可选参数不阻止形成隔离、采样和保障建议。回放未开放的其他水力工具按实际权限记录限制，继续已授权分析。
+先说明污染物性质、处理依据和未知项，再提出排空、冲洗、消毒或其他适用处理；不得默认任何污染都能冲洗清除。排放去向及污染水收集处理须有依据，不默认排入雨水、河道或普通排水口。
 
-## 5. 过程图表与报告
+核对清洁水源、排放点、流量及许可条件后用 analyze_flushing。每阶段保存目的、前置条件、清洁进水依据、排放去向、完整开关阀清单、顺序、流量、时长、预计水量、监测和待办。隔离和冲洗不能共用固定关阀清单，进水阀打开不等于用户复供；缺参数仅停依赖分支。
 
-水质曲线填写 chart.scada：metric="quality"、真实 unit、role=current/comparison、完整三小时 window_start/window_end；series 绑定 device_id 和 day_offset。不同水质指标或单位分图，不能把浊度、余氯或其他指标混成同一浓度。当前总览覆盖报告与地图列出的全部目标；各设备七天同期图采用 time_comparison，当前三小时为 line，前七天每天末三个实际采样时刻为 scatter，sample_times 与 data 等长且保留真实日期。缺测保留 null，超过槽位上限按时间分图。所有数据引用本轮证据；展示不要求等待浏览器。
+opened_valves 明确开阀、closed_valves 指定关闭阀门，两者互斥。每次是独立当前模型情景，未列阀门沿用模型状态，不继承上次模拟；重列本阶段必要阀态，不能称为连续污染清除。flushing_plan 保存方案及估计排放量，仅证明水力结果，不验证清洁进水、排放合规或现场执行。尚未耦合停止输入后的水质衰减，压力、流量或模型零浓度不能替代复测。
 
-报告包含“事件概况、侦测与数据核验、候选溯源、隔离方案、影响说明、处置建议、反馈复核与待办”。影响说明列出实际异常指标、单位、时间、测点、直接隔离与失源节点数、水力或传播预测和用户保障。明确临时选点、观测事实、模型假设及待现场确认事项。最终报告引用本轮全量 SCADA 证据及已取得的溯源、隔离和覆盖证据。查询失败或权限拒绝记录为执行异常，不能写成无数据或未见异常。
+展示清洁水入口、保护边界、导流阀及排放点；有真实GIS才画路径。拟开、拟关、现场实际阀位分列，不改写实时资产。回放未授权工具记录限制，复用已有证据。
 
-## 6. 处置与反馈接续
+## 6. 图表、报告与展示节奏
 
-第2节已提出的 general 工单继续沿用，按新增证据组织污染进入控制、隔离核验、通知保障、清洗冲洗及复测；已有工单确有不足时再提出必要的新动作。标题写“地点＋动作”，写明执行条件、真实节点、时间要求与回执内容；审批、派发和现场执行遵守 Server 权限。没有用户通知适配器时交由人工实施。回放草稿注明异常业务时间及来源，不直接执行设备命令。
+水质 chart.scada 填 metric=quality、unit、role、完整三小时 window_start/window_end，总览覆盖全部目标。逐设备 time_comparison 用当前 line、七天每天末三真实槽 scatter；series 带 device_id、day_offset、等长 sample_times，缺测 null。不同指标或单位分图，超限分段，不标成流量。
 
-待审批、waiting_work_order 且无独立工作时，发布本轮 interim/concluded 并引用真实 action_id 后结束本轮，释放租约，不循环查询或重复建单。field_review 时核对各关联工单反馈及实际采样、处置、复测的时间、位置、数值和判定依据，再查询反馈业务时间之后已有水质观测；不要重跑原始调查。缺少关键事实时先完成独立工作，再用 ask_task_user 明确待补内容并结束本轮等待。
+展示依次为观测、候选、传播、源头控制、隔离、冲洗、反馈、复测、恢复；计算可并行，首次隔离展示须在传播结果或缺口说明之后，通知和工单不等待展示。发布填写具体 purpose；有证据时带 narration（stage、summary、evidence_refs）。沿用原阶段，源头控制用 source_control、冲洗用 flushing_plan、水质复测用 clearance_verification。阶段不等于批准或已执行，不为凑阶段造证据或空操作。
 
-本轮调查成果完成与现场事件结案分别表述。工单 completed 或压力恢复不证明水质合格，不套用爆管压力恢复的自动结案条件。反馈复核具备依据时发布 final/concluded，引用最新现场反馈证据；仍异常时调整建议或新建必要动作，未来采样、复测和恢复不能写成已完成。
+报告依次写事件概况、观测、候选与传播、源头控制、用户隔离、清除冲洗、反馈复测及恢复条件；各项列结论、对象、预计或实际、证据和待办。操作表保留阶段、阀态、进排水、条件、反馈及复测依据。引用本轮全量 SCADA 和已取得的分析证据；权限拒绝不写成无数据。各阶段独立保存，回放不合并吞步骤，不派单或操作设备。
+
+## 7. 工单反馈与复测恢复
+
+复用 general 工单及 action_id，仅因职责或阶段确需拆分才建新单，保留前置条件，不能同时要求同阀开关。回传实际阀位、进排水、执行时间和异常；复测含采样地点、时间、指标、结果、单位、判定依据及结论。通知保留，操作遵守原权限。
+
+待审批或 waiting_work_order 且无独立工作时，发布本轮 interim/concluded 并引用 action_id 后退出释放租约，不轮询或重复建单。field_review 复用原调查，只核对最新反馈与相关水质观测；关键事实缺失时先完成独立工作，再用 ask_task_user 提问并等待，必要时提出复测或调整处置工单。
+
+工单 completed、压力恢复、余氯单项正常或模型预测均不能证明污染物已清除。恢复依据须说明目标污染物或已核实风险、适用检测项目及判定标准、实际采样结果与时间地点、受影响区域覆盖和有依据的恢复结论；无适用依据或结果未通过时保持 interim，列待复测区域与下一步，不能宣布恢复供水或发布恢复闭环 final。不能为所有事件统一增设固定轮数、等待时长或未配置阈值。
+
+有复测与恢复依据后展示可恢复区域，final/concluded 引用所有有效工单最新 feedback_evidence_id 及复测证据。不套用爆管压力结案，仍异常调整方案，未来采样和恢复不写成已完成。Server不自动认定实验室合格或授权现场复供。
diff --git a/tests/unit/test_flushing_valve_plan.py b/tests/unit/test_flushing_valve_plan.py
new file mode 100644
index 0000000..7dd6760
--- /dev/null
+++ b/tests/unit/test_flushing_valve_plan.py
@@ -0,0 +1,78 @@
+from datetime import datetime, timezone
+from uuid import uuid4
+
+import pytest
+from pydantic import ValidationError
+
+from app.mcp.analysis_models import FlushingInput
+from app.mcp.extended_models import SavedScenarioInput
+from app.mcp.mutation_service import saved_scenario
+from app.mcp.scenario_service import calculate_scenario, flushing_valve_controls
+
+
+def parameters(**changes):
+    return {"context_handle": "a" * 43, "start_time": datetime(2026, 7, 16, tzinfo=timezone.utc),
+            "duration": 1800, "output_node_ids": ["J1"], "drainage_node_id": "J2",
+            "flushing_flow_m3h": 12, **changes}
+
+
+@pytest.mark.parametrize("changes", [
+    {"closed_valves": ["V1"], "opened_valves": ["V1"]},
+    {"closed_valves": ["V1", "V1"]},
+    {"opened_valves": ["V1", "V1"]},
+])
+def test_conflicting_or_duplicate_valves_are_rejected_for_preview_and_saved_scenarios(changes):
+    values = parameters(**changes)
+    with pytest.raises(ValidationError):
+        FlushingInput.model_validate(values)
+    with pytest.raises(ValidationError):
+        SavedScenarioInput.model_validate({"context_handle": values.pop("context_handle"),
+            "request_id": uuid4(), "name": "冲洗", "scenario": "flushing", "parameters": values})
+
+
+def test_preview_and_saved_flushing_apply_identical_open_and_closed_states(monkeypatch):
+    from app.services import simulation_scenarios
+
+    calls = []
+    monkeypatch.setattr(simulation_scenarios, "flushing_analysis", lambda **kw: calls.append(kw) or {
+        "node_results": [{"node": "J1", "result": []}], "link_results": []})
+    monkeypatch.setattr("app.native.wndb.model.options.get_option", lambda _: {"UNITS": "LPS"})
+    args = FlushingInput.model_validate(parameters(opened_valves=["V-in"], closed_valves=["V-user"]))
+    output = calculate_scenario("analyze_flushing", args.model_dump(exclude={"context_handle", "timeout_seconds"}),
+                                network="owned", username="actor")
+    values = SavedScenarioInput.model_validate({"context_handle": args.context_handle,
+        "request_id": uuid4(), "name": "分区冲洗", "scenario": "flushing",
+        "parameters": args.model_dump(exclude={"context_handle", "timeout_seconds"})})
+    saved_scenario("owned", "actor", values)
+    for call in calls:
+        assert call["valve_control"] == {"V-in": {"status": "OPEN"}, "V-user": {"status": "CLOSED"}}
+        assert "modify_valve_opening" not in call
+        assert call["drainage_node_ID"] == "J2" and call["flushing_flow"] == 12
+    assert calls[0]["persist_results"] is False and calls[1]["persist_results"] is True
+    assert output["flushing_plan"] == {
+        "opened_valves": ["V-in"], "closed_valves": ["V-user"], "drainage_node_id": "J2",
+        "flushing_flow_m3h": 12, "estimated_discharge_volume_m3": 6,
+        "state_basis": "independent_current_model_scenario", "result_scope": "hydraulic_only",
+        "boundary_source": "model_with_start_time_scada",
+        "field_execution_verified": False, "water_quality_clearance_verified": False,
+    }
+
+
+def test_legacy_closed_only_inputs_remain_valid():
+    args = FlushingInput.model_validate(parameters(closed_valves=["V1"]))
+    assert args.opened_valves == []
+    assert flushing_valve_controls(args.model_dump()) == {"V1": {"status": "CLOSED"}}
+
+
+def test_explicit_opening_reopens_a_closed_valve_without_overwriting_its_setting(monkeypatch):
+    from app.services import simulation
+
+    model = {"V-in": {"link": "V-in", "status": "CLOSED", "setting": 15},
+             "V-user": {"link": "V-user", "status": "ACTIVE", "setting": 23}}
+    monkeypatch.setattr(simulation, "get_status", lambda project, valve: dict(model[valve]))
+    monkeypatch.setattr(simulation, "set_status", lambda project, changes: model.update({
+        row["link"]: row for row in changes.operations}))
+    simulation._apply_valve_control("temporary", flushing_valve_controls({
+        "opened_valves": ["V-in"], "closed_valves": ["V-user"]}))
+    assert model["V-in"] == {"link": "V-in", "status": "OPEN", "setting": 15}
+    assert model["V-user"] == {"link": "V-user", "status": "CLOSED", "setting": 23}
diff --git a/tests/unit/test_investigation_report_authoring.py b/tests/unit/test_investigation_report_authoring.py
new file mode 100644
index 0000000..6adb990
--- /dev/null
+++ b/tests/unit/test_investigation_report_authoring.py
@@ -0,0 +1,96 @@
+"""New publications are readable; historical reports keep their original evidence."""
+from copy import deepcopy
+
+import pytest
+
+from app.domain.schemas.operational_task import (
+    InvestigationProgress,
+    InvestigationProgressInput,
+    InvestigationReport,
+    InvestigationReportInput,
+)
+
+
+def report():
+    return {
+        "title": "压力异常恢复复核报告",
+        "current_conclusion": "恢复复核通过。7 月 16 日 13:15，13 台复核设备均处于历史正常范围。",
+        "conclusion_boundary": "依据完工后的一个采样时刻判断恢复，长期稳定性继续由日常监测跟踪。",
+        "stage": "final", "outcome": "concluded", "data_gaps": [],
+        "evidence_refs": ["recovery:sha256:stored-evidence"],
+        "recommendations": ["继续监测压力与流量。"], "pending_items": [],
+    }
+
+
+@pytest.mark.parametrize("text", [
+    "恢复复核 status=passed，13 台设备 normal。",
+    "第 3 轮 execution_generation=3，handling_context.phase=verification_ready。",
+    "完工后 required_samples=1，interval_seconds=900，verified_through=13:15。",
+    "已读取 recovery:sha256:01234567，复核通过。",
+    "Skill scada-anomaly v1.0.16 已按 Server 分发执行。",
+    "query_scada_readings 已执行，query_complete=true。",
+    "关联工单 60000000-0000-4000-8000-000000000002 已完工。",
+    "恢复采样时间为 2026-07-16T05:15:00Z。",
+    "恢复采样时间为 2026-07-16 13:15（UTC+8）。",
+    "恢复采样时间为 2026-07-16 13:15+08:00。",
+    "**恢复复核通过**，其余设备正常。",
+    "<strong>恢复复核通过</strong>。",
+])
+def test_new_report_and_progress_reject_execution_details_and_raw_display_markup(text):
+    legacy = {**report(), "current_conclusion": text}
+    assert InvestigationReport.model_validate(legacy).current_conclusion == text
+    assert InvestigationProgress.model_validate({"summary": text}).summary == text
+    with pytest.raises(ValueError):
+        InvestigationReportInput.model_validate(legacy)
+    with pytest.raises(ValueError):
+        InvestigationProgressInput.model_validate({"summary": text})
+
+
+@pytest.mark.parametrize(("field", "limit"), [
+    ("title", 80), ("current_conclusion", 300), ("conclusion_boundary", 240),
+])
+def test_new_report_bounds_summary_fields_without_truncating_history(field, limit):
+    valid = {**report(), field: "测" * limit}
+    assert getattr(InvestigationReportInput.model_validate(valid), field) == valid[field]
+    too_long = {**valid, field: "测" * (limit + 1)}
+    assert getattr(InvestigationReport.model_validate(too_long), field) == too_long[field]
+    with pytest.raises(ValueError):
+        InvestigationReportInput.model_validate(too_long)
+
+
+def test_progress_summary_is_short_and_keeps_technical_references_structured():
+    progress = {"summary": "测" * 300, "graph_update": {"nodes": [{
+        "node_id": "recovery", "label": "恢复监测复核通过", "status": "recorded",
+        "evidence_refs": ["recovery:sha256:stored-evidence"],
+    }]}}
+    assert InvestigationProgressInput.model_validate(progress).summary == progress["summary"]
+    with pytest.raises(ValueError):
+        InvestigationProgressInput.model_validate({**progress, "summary": "测" * 301})
+
+
+def test_asset_ids_measurements_and_evidence_linked_tables_remain_publishable():
+    value = report()
+    value["current_conclusion"] = "05:00-08:00 的候选管段 515870（DN600）相似度 97.08%，尚待现场核验。压力测点 170486 回升至 26.12 m。"
+    value["ui_blocks"] = [{"evidence_refs": value["evidence_refs"], "block": {
+        "version": "answer-ui@1", "block_id": "recovery-table", "phase": "final",
+        "fallback_text": "压力测点 170486 恢复至 26.12 m。",
+        "spec": {"root": "table", "elements": {"table": {
+            "type": "Table", "children": [], "props": {
+                "title": "恢复监测明细", "columns": ["测点", "修复后", "正常范围"],
+                "rows": [["170486", "26.12 m", "26.02 至 26.30 m"]],
+            },
+        }}},
+    }}]
+    assert InvestigationReportInput.model_validate(value).ui_blocks
+    leaked = deepcopy(value)
+    leaked["ui_blocks"][0]["block"]["spec"]["elements"]["table"]["props"]["rows"][0][2] = "status=passed"
+    assert InvestigationReport.model_validate(leaked).ui_blocks
+    with pytest.raises(ValueError):
+        InvestigationReportInput.model_validate(leaked)
+
+
+@pytest.mark.parametrize("field", ["recommendations", "pending_items"])
+def test_follow_up_is_business_prose(field):
+    value = {**report(), field: ["等待 handling_context.phase=waiting_scada。"]}
+    with pytest.raises(ValueError):
+        InvestigationReportInput.model_validate(value)
diff --git a/tests/unit/test_mcp_work_order_skills.py b/tests/unit/test_mcp_work_order_skills.py
index 2ffae21..171efef 100644
--- a/tests/unit/test_mcp_work_order_skills.py
+++ b/tests/unit/test_mcp_work_order_skills.py
@@ -70,6 +70,17 @@ def test_pollution_instructions_offer_work_order_before_simulation_and_reuse_the
     assert "不能把浊度、余氯下降或感官描述直接换算成污染源浓度" in text
 
 
+def test_pollution_workflow_distinguishes_protection_flushing_and_clearance():
+    skill = read_work_order_skill("water_pollution", include_content=True)
+    text = skill["instructions"]
+    assert skill["version"] == "1.2.0"
+    assert text.index("## 4. 源头控制与用户隔离") < text.index("## 5. 污染物清除与分步冲洗")
+    assert "opened_valves" in text and "closed_valves" in text
+    assert "不继承上次模拟" in text and "排放去向" in text
+    assert all(stage in text for stage in ("source_control", "flushing_plan", "clearance_verification"))
+    assert "不能证明污染物已清除" in text and "结果未通过时保持 interim" in text
+
+
 @pytest.mark.parametrize("workflow", ["burst_report", "water_quality_complaint", "water_pollution"])
 def test_task_skill_uses_persisted_workflow_and_preserves_task_scope(monkeypatch, workflow):
     task_id, project_id = uuid4(), uuid4()
@@ -102,6 +113,7 @@ def test_task_skill_uses_persisted_workflow_and_preserves_task_scope(monkeypatch
             assert capability["simulation_parameters"] == {"include_impact_coverage": True}
             assert capability["next_steps"].index("propose_task_action") < capability["next_steps"].index("run_contaminant_simulation")
     assert body.data["skill"]["instructions"] and not body.truncated
+    assert len(json.dumps(body.model_dump(mode="json"), ensure_ascii=False).encode()) < 10000
     assert body.evidence_id is None and body.semantics == "task"
     reader.assert_awaited_with(scope.connection, project_id=project_id, task_id=task_id, report_revision=None)
     binding.permission_cap = []
diff --git a/tests/unit/test_operational_tasks.py b/tests/unit/test_operational_tasks.py
index 95959bc..a34ce2f 100644
--- a/tests/unit/test_operational_tasks.py
+++ b/tests/unit/test_operational_tasks.py
@@ -2363,3 +2363,77 @@ async def test_external_sync_only_mutates_tasks_in_its_execution_scope(operation
         await getattr(other, method)(task["task_id"], command)
     assert error.value.code == "EXECUTION_SCOPE_MISMATCH" and error.value.status == 403
     assert repository.task == before
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("stage", ["asset_localization", "source_control", "flushing_plan", "clearance_verification"])
+async def test_presentation_narration_is_evidence_backed_saved_and_replayed_without_mutating_task(stage):
+    now = datetime(2026, 11, 14, tzinfo=timezone.utc)
+    context = _context()
+    task = _task_document(context, now, lease=now + timedelta(minutes=1))
+    repository = TaskRepository(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='observations', queried_at=now,
+        origin='synthetic', data={'items': []}, truncated=False,
+    ))
+    narration = {'stage': stage, 'summary': '本阶段保存的候选方案及核验依据。',
+                 'evidence_refs': [evidence.evidence_id], 'next_step': '提出现场工单草稿，并行开展影响分析。'}
+    command = TaskPresentationEventCommand(
+        request_id=uuid4(), run_id=repository.run['run_id'], source_session_id='session-1',
+        command={'action': 'locate_features', 'layer': 'pipes', 'feature_ids': ['P1']}, narration=narration,
+        purpose='核对候选管段与异常测点的空间关系，供后续影响分析。',
+    )
+    result = await service.publish_presentation_event(task['task_id'], command)
+    assert repository.events[-1]['metadata']['narration'] == narration
+    assert repository.events[-1]['metadata']['purpose'] == command.purpose
+    assert await service.publish_presentation_event(task['task_id'], command) == result
+    page = await service.presentation_history(task['task_id'])
+    assert page.items[-1].narration.model_dump(exclude_none=True) == narration
+    assert page.items[-1].purpose == command.purpose
+    assert repository.task['revision'] == task['revision']
+    with pytest.raises(DomainError) as conflict:
+        await service.publish_presentation_event(task['task_id'], command.model_copy(update={
+            'narration': command.narration.model_copy(update={'summary': '不同判断'}),
+        }))
+    assert conflict.value.code == 'IDEMPOTENCY_CONFLICT'
+    foreign = TaskPresentationEventCommand(
+        request_id=uuid4(), run_id=repository.run['run_id'], source_session_id='session-1',
+        command={'action': 'clear_highlights'}, narration={**narration, 'evidence_refs': ['other-task']},
+    )
+    before = len(repository.events)
+    with pytest.raises(DomainError) as rejected:
+        await service.publish_presentation_event(task['task_id'], foreign)
+    assert rejected.value.code == 'EVIDENCE_NOT_FOUND'
+    assert len(repository.events) == before
+
+
+@pytest.mark.parametrize('changes', [
+    {'stage': 'invented_stage'}, {'summary': 'x' * 161}, {'summary': '   '},
+    {'evidence_refs': []}, {'evidence_refs': ['e1', 'e1']}, {'next_step': 'x' * 121},
+])
+def test_presentation_narration_rejects_unbounded_or_unsubstantiated_fields(changes):
+    with pytest.raises(ValidationError):
+        TaskPresentationEventCommand(
+            request_id=uuid4(), run_id=uuid4(), source_session_id='session-1',
+            command={'action': 'clear_highlights'},
+            narration={'stage': 'evidence_verification', 'summary': '测点异常待核验',
+                       'evidence_refs': ['e1'], **changes},
+        )
+
+
+def test_legacy_presentation_idempotency_body_does_not_gain_an_empty_narration_field():
+    command = TaskPresentationEventCommand(
+        request_id=uuid4(), run_id=uuid4(), source_session_id='session-1',
+        command={'action': 'clear_highlights'},
+    )
+    assert 'narration' not in command.model_dump(mode='json')
+    assert 'purpose' not in command.model_dump(mode='json')
+
+@pytest.mark.parametrize('purpose', ['', '   ', 'x' * 161, 123])
+def test_presentation_purpose_rejects_invalid_intent(purpose):
+    with pytest.raises(ValidationError):
+        TaskPresentationEventCommand(
+            request_id=uuid4(), run_id=uuid4(), source_session_id='session-1',
+            command={'action': 'zoom_to_map', 'center': [121, 31], 'zoom': 15}, purpose=purpose,
+        )
diff --git a/tests/unit/test_pollution_event_workflow.py b/tests/unit/test_pollution_event_workflow.py
index 8420f1f..8d1041b 100644
--- a/tests/unit/test_pollution_event_workflow.py
+++ b/tests/unit/test_pollution_event_workflow.py
@@ -70,7 +70,7 @@ async def test_pollution_work_order_can_be_approved_and_dispatched_while_simulat
     await service.publish_artifacts(task["task_id"], AgentArtifactsCommand(
         request_id=uuid4(), source_session_id="session-1", report={**report,
             "current_conclusion": "传播模拟完成，现场核查继续", "evidence_refs": [evidence.evidence_id],
-            "pending_items": [f"等待现场工单反馈，action_id={action.action_id}"]}))
+            "pending_items": ["等待现场核查和采样处置工单反馈"]}))
     assert len(repository.reports) == 2 and repository.action["version"] == dispatched.version
     assert repository.action["approval_status"] == "approved" and repository.run["status"] == "running"
 
diff --git a/tests/unit/test_presentation_authoring.py b/tests/unit/test_presentation_authoring.py
new file mode 100644
index 0000000..f855732
--- /dev/null
+++ b/tests/unit/test_presentation_authoring.py
@@ -0,0 +1,72 @@
+"""Presentation prose is validated on writes without rejecting historical scenes."""
+from copy import deepcopy
+from datetime import datetime, timezone
+from uuid import uuid4
+
+import pytest
+
+from app.domain.schemas.operational_task import TaskPresentationEventCommand, TaskPresentationHistoryItem
+
+
+def presentation():
+    return {
+        "request_id": uuid4(), "run_id": uuid4(), "source_session_id": "fixture-session",
+        "purpose": "核对压力变化与异常时刻的对应关系。",
+        "narration": {"stage": "evidence_verification", "summary": "压力已回升，恢复情况仍需结合现场反馈复核。",
+                      "next_step": "核对完工后的恢复监测。", "evidence_refs": ["recovery:sha256:fixture"]},
+        "command": {"action": "open_chart", "chart_id": "pressure", "evidence_refs": ["recovery:sha256:fixture"],
+                    "chart": {"title": "170486 压力恢复趋势", "chart_type": "line", "x_axis_name": "时间",
+                              "y_axis_name": "压力（m）", "x_data": ["13:00", "13:15"],
+                              "series": [{"name": "170486 当前压力", "data": [20.09, 26.12],
+                                          "sample_times": ["2026-07-16T05:00:00Z", "2026-07-16T05:15:00Z"]}]}},
+    }
+
+
+@pytest.mark.parametrize("text", [
+    "status=passed，required_samples=1。", "Skill scada-anomaly v1.0.16 已执行。",
+    "已读取 recovery:sha256:01234567。", "恢复时刻 2026-07-16T05:15:00Z。",
+    "恢复时刻 13:15（UTC+8）。", "**恢复复核通过**。",
+])
+def test_presentation_rejects_execution_text_in_purpose_finding_and_follow_up(text):
+    value = presentation()
+    for field in ("purpose", "summary", "next_step"):
+        changed = deepcopy(value)
+        if field == "purpose":
+            changed[field] = text
+        else:
+            changed["narration"][field] = text
+        history = TaskPresentationHistoryItem(
+            event_seq=1, run_id=changed["run_id"], execution_generation=1,
+            occurred_at=datetime(2026, 7, 16, tzinfo=timezone.utc),
+            command=changed["command"], purpose=changed["purpose"], narration=changed["narration"],
+        )
+        assert history.purpose == changed["purpose"]
+        with pytest.raises(ValueError):
+            TaskPresentationEventCommand.model_validate(changed)
+
+
+@pytest.mark.parametrize("field", ["title", "x_axis_name", "y_axis_name", "x_data", "series"])
+def test_new_chart_visible_labels_follow_the_same_prose_contract(field):
+    value = presentation()
+    chart = value["command"]["chart"]
+    if field == "x_data":
+        chart[field][0] = "2026-07-16T05:00:00Z"
+    elif field == "series":
+        chart[field][0]["name"] = "execution_generation=3"
+    else:
+        chart[field] = "压力复核 status=passed"
+    with pytest.raises(ValueError):
+        TaskPresentationEventCommand.model_validate(value)
+
+
+def test_display_validation_preserves_real_timestamps_identifiers_and_values():
+    value = presentation()
+    parsed = TaskPresentationEventCommand.model_validate(value).model_dump(mode="json")
+    assert parsed["command"] == value["command"]
+    assert parsed["narration"]["evidence_refs"] == value["narration"]["evidence_refs"]
+    value["command"] = {"action": "set_overlay", "overlay_id": "estimated-impact",
+                        "title": "候选管段预计影响范围", "mode": "geometry", "evidence_id": "mcp:fixture"}
+    assert TaskPresentationEventCommand.model_validate(value).command.title == "候选管段预计影响范围"
+    value["command"]["title"] = "候选影响范围（UTC+8）"
+    with pytest.raises(ValueError):
+        TaskPresentationEventCommand.model_validate(value)
diff --git a/tests/unit/test_replay_flushing.py b/tests/unit/test_replay_flushing.py
new file mode 100644
index 0000000..10f722d
--- /dev/null
+++ b/tests/unit/test_replay_flushing.py
@@ -0,0 +1,91 @@
+import asyncio
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+from uuid import uuid4
+
+import pytest
+from fastapi import HTTPException
+
+from app.mcp import analysis_service, query_service
+from app.mcp.analysis_models import FlushingInput
+from app.mcp.calculation_jobs import should_queue
+from app.mcp.catalog import TOOLS
+
+
+def parameters(**changes):
+    return FlushingInput.model_validate({"context_handle": "a" * 43,
+        "start_time": "2026-07-16T08:00:00+08:00", "duration": 1800,
+        "output_node_ids": ["J1"], "drainage_node_id": "J2", "flushing_flow_m3h": 12,
+        "opened_valves": ["V-in"], "closed_valves": ["V-user"], **changes})
+
+
+def context(monkeypatch):
+    monkeypatch.setattr(query_service, "permissions_for_context", lambda _: {"simulation.run", "scada.view"})
+
+    @asynccontextmanager
+    async def routing():
+        yield
+
+    scope = SimpleNamespace(connection=SimpleNamespace(commit=AsyncMock()),
+        project=SimpleNamespace(project_code="owned", project_role="member"), user=SimpleNamespace(username="actor"),
+        simulation_routing=routing)
+    binding = SimpleNamespace(tool_cap=["analyze_flushing"], permission_cap=["simulation.run", "scada.view"],
+        purpose="task_execution", task_id=uuid4(), task_scope_cap=[],
+        replay_context={"knowledge_cutoff": "2026-07-16T08:00:00+08:00"})
+    return scope, binding
+
+
+def test_replay_flushing_uses_the_model_only_adapter_and_retains_prediction_boundary(monkeypatch):
+    scope, binding = context(monkeypatch)
+    worker = AsyncMock(return_value={"flushing_plan": {"water_quality_clearance_verified": False}})
+    monkeypatch.setattr(analysis_service, "calculate_replay_flushing", worker)
+    result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["analyze_flushing"], parameters()))
+    assert result.origin == "synthetic" and "预测" in result.data_gaps[0]
+    assert result.data["result"]["flushing_plan"]["water_quality_clearance_verified"] is False
+    assert result.data["parameters"]["opened_valves"] == ["V-in"]
+    scope.connection.commit.assert_awaited_once()
+    worker.assert_awaited_once()
+    assert worker.await_args.kwargs == {"network": "owned", "username": "actor"}
+    assert should_queue(TOOLS["analyze_flushing"], parameters())
+
+
+@pytest.mark.parametrize("problem", ["future_start", "reader", "permission", "tool_cap"])
+def test_replay_flushing_preserves_time_permissions_and_reader_boundaries(monkeypatch, problem):
+    scope, binding = context(monkeypatch)
+    args = parameters()
+    if problem == "future_start": args = parameters(start_time="2026-07-16T08:15:00+08:00")
+    if problem == "reader": binding.purpose = "task_reader"
+    if problem == "permission": binding.permission_cap.remove("scada.view")
+    if problem == "tool_cap": binding.tool_cap.clear()
+    with pytest.raises(HTTPException) as caught:
+        asyncio.run(query_service.authorize_query(scope, binding, TOOLS["analyze_flushing"], args))
+    assert caught.value.status_code == (422 if problem == "future_start" else 403)
+
+
+def test_model_only_flushing_and_native_runner_both_receive_empty_scada_mappings(monkeypatch):
+    from app.services import simulation_scenarios as scenarios
+
+    captures = []
+    native = scenarios.flushing_analysis.__wrapped__
+    monkeypatch.setattr("app.native.wndb.model.options.get_option", lambda _: {"UNITS": "LPS"})
+
+    def isolated(**kwargs):
+        captures.append(kwargs)
+        return native(**kwargs, _temporary_project="temporary")
+
+    monkeypatch.setattr(scenarios, "flushing_analysis", isolated)
+    monkeypatch.setattr(scenarios, "is_junction", lambda *_: True)
+    monkeypatch.setattr(scenarios, "get_option", lambda _: {"UNITS": "LPS"})
+    monkeypatch.setattr(scenarios, "get_time", lambda _: {"HYDRAULIC TIMESTEP": "00:15:00"})
+    monkeypatch.setattr(scenarios, "get_demand", lambda *_: {"demands": []})
+    for name in ("add_pattern", "set_demand", "set_option"):
+        monkeypatch.setattr(scenarios, name, lambda *_: None)
+    monkeypatch.setattr(scenarios.simulation, "run_simulation", lambda **kw: captures.append(kw) or {
+        "node_results": [], "link_results": []})
+    output = asyncio.run(analysis_service.calculate_replay_flushing(parameters(), network="owned", username="actor"))
+    assert len(captures) == 2
+    assert captures[0]["scada_mappings"] is captures[1]["scada_mappings"]
+    assert all(not values for values in vars(captures[0]["scada_mappings"]).values())
+    assert captures[1]["name"] == "temporary" and captures[1]["persist_results"] is False
+    assert output["flushing_plan"]["boundary_source"] == "model_only_no_live_scada"
diff --git a/tests/unit/test_scada_replay_service.py b/tests/unit/test_scada_replay_service.py
index 169a2bc..1d82082 100644
--- a/tests/unit/test_scada_replay_service.py
+++ b/tests/unit/test_scada_replay_service.py
@@ -603,15 +603,20 @@ def test_suspected_monitoring_creates_burst_report_with_frozen_replay_context(mo
 
     command = create_task.await_args.args[0]
     assert command.workflow == "scada_anomaly"
+    assert command.title == "SCADA 异常调查 · 2026-10-08 08:15"
+    assert "2 个压力测点下降" in command.summary
+    assert "1 个出厂边界流量测点增加" in command.summary
+    assert "爆管嫌疑" in command.summary
     assert command.source.replay_context.batch_sequence == 8
     assert command.source.replay_context.knowledge_cutoff == START + timedelta(minutes=15)
     assert monitoring["task_id"] == str(create_task.return_value.task_id)
     assert document["monitoring_incident"]["alert"]["condition_id"] == f"operational-task:{monitoring['task_id']}"
     assert document["monitoring_incident"]["alert"]["description"] == "已启动调查，详细诊断见调查报告。"
     if weighted:
-        assert "-0.0843" in command.summary
-        assert "综合评分 0.90" in command.summary
-        assert "流量 10%" in command.summary
+        assert "-0.0843" not in command.summary
+        assert "综合评分" not in command.summary
+        assert "10%" not in command.summary
+        assert monitoring["weighted"]["score"] == 0.9
 
 
 def test_monitoring_summary_counts_signals_without_copying_reading_details(monkeypatch):
@@ -916,8 +921,10 @@ def test_any_single_measurement_anomaly_enters_diagnosis_below_burst_threshold(m
     command = create_task.await_args.args[0]
     assert command.workflow == "scada_anomaly"
     assert command.source.event_kind == "scada_anomaly"
-    assert "任一指标异常即启动调查" in command.summary
-    assert "爆管嫌疑门槛 0.80" in command.summary
+    assert "已启动调查" in command.summary
+    assert "异常原因及影响范围待核查" in command.summary
+    assert "爆管嫌疑门槛" not in command.summary
+    assert monitoring["weighted"]["score"] == 0.1
     assert monitoring["task_id"] == str(create_task.return_value.task_id)
     assert monitoring["incident"]["action"] == "raised"