feat(investigations): add event impact and work-order feedback flows

Add isolated burst and pollution impact evidence, parallel work-order workflows, persisted notifications and graph updates.

Checkpoint current code. Known follow-ups remain for large pollution result previews and graceful geometry failure.

tests/unit/test_burst_impact.py

@@ -0,0 +1,197 @@
+from datetime import datetime, timedelta, timezone
+from unittest.mock import Mock
+
+import pytest
+
+from app.services import burst_impact as service
+from app.services.node_coverage import build_node_coverage
+
+START = datetime(2026, 7, 16, 5, tzinfo=timezone.utc)
+
+
+def simulation(drops, unit="m"):
+    return {"units": {"pressure": unit}, "times": {"report_start": 0},
+            "node_results": [{"node": f"J{i}", "result": [{"pressure": 30 - drop}]}
+                             for i, drop in enumerate(drops)]}
+
+
+@pytest.fixture
+def inputs(monkeypatch):
+    monkeypatch.setattr(service, "get_option", lambda _: {"PRESSURE": "METERS"})
+    monkeypatch.setattr(service, "get_option_v3", lambda _: {})
+    read = Mock(side_effect=lambda ids, *a, **kw: {
+        node: [{"time": START.isoformat(), "value": 30}] for node in ids})
+    monkeypatch.setattr(service.InternalQueries, "query_realtime_simulation_by_ids_timerange", read)
+    def coverage(network, groups):
+        ids = sorted({node for group in groups for node in group["node_ids"]})
+        data = build_node_coverage(groups, [
+            {"id": node, "longitude": 121.5 + i * .001, "latitude": 30.9}
+            for i, node in enumerate(ids)], [])
+        payload = data.pop("map_payload")
+        return {"coverage": data, "map_payload": payload}
+    monkeypatch.setattr(service, "generate_node_coverage", coverage)
+    return read
+
+
+def test_thresholds_are_exclusive_with_no_rounding_and_no_closure(inputs):
+    result = service.estimate_burst_impact("own", START, simulation([.999, 1, 2.999, 3, 4.999, 5, 8, -1]))
+    impact = result["impact"]
+    assert [group["node_ids"] for group in impact["grades"]] == [["J1", "J2"], ["J3", "J4"], ["J5", "J6"]]
+    assert impact["below_threshold_node_count"] == 2
+    assert impact["compared_node_count"] == 8 and impact["affected_node_count"] == 6
+    assert impact["phase"] == "after_burst_before_intervention"
+    assert impact["baseline_frozen"] is False
+    inputs.assert_called_once_with([f"J{i}" for i in range(8)], START, START, "node", "pressure", db_name="own")
+    assert len(result["map_payload"]["features"]) == 3
+    assert all(feature["properties"]["model_estimate"] for feature in result["map_payload"]["features"])
+
+
+@pytest.mark.parametrize("unit,factor", [("kPa", 1 / 9.80665), ("PSI", 6894.757293168 / 9806.65),
+                                         ("bar", 100000 / 9806.65), ("m", 1)])
+def test_normalizes_baseline_and_simulation_independently(monkeypatch, inputs, unit, factor):
+    inputs.return_value = {"J0": [{"time": START.isoformat(), "value": 30 / factor}]}
+    inputs.side_effect = None
+    monkeypatch.setattr(service, "get_option", lambda _: {"PRESSURE": unit})
+    result = service.estimate_burst_impact("own", START, simulation([4]))
+    assert result["impact"]["comparisons"][0]["pressure_drop_m"] == pytest.approx(4)
+    assert result["impact"]["grades"][1]["node_ids"] == ["J0"]
+
+
+def test_v3_units_take_precedence_over_legacy_settings(monkeypatch, inputs):
+    monkeypatch.setattr(service, "get_option_v3", lambda _: {"PRESSURE_UNITS": "KPA"})
+    inputs.side_effect = None
+    inputs.return_value = {"J0": [{"time": START.isoformat(), "value": 30 * 9.80665}]}
+    impact = service.estimate_burst_impact("own", START, simulation([4]))["impact"]
+    assert impact["comparisons"][0]["pressure_drop_m"] == pytest.approx(4)
+    assert impact["baseline_units_source"] == "current_v3_model_settings"
+
+
+def test_missing_nonfinite_duplicate_and_wrong_time_are_unavailable_not_zero(inputs):
+    inputs.side_effect = None
+    inputs.return_value = {
+        "J0": [{"time": (START + timedelta(seconds=1)).isoformat(), "value": 30}],
+        "J1": [{"time": START.isoformat(), "value": float("nan")}],
+        "J2": [{"time": START.isoformat(), "value": 30}] * 2,
+        "J3": [{"time": START.isoformat(), "value": 30}],
+        "J4": [{"time": START.isoformat(), "value": 30}],
+    }
+    native = simulation([1] * 6)
+    native["node_results"][3]["result"] = [{"pressure": float("inf")}]
+    result = service.estimate_burst_impact("own", START, native)
+    assert result["impact"]["status"] == "partial"
+    assert result["impact"]["unavailable_node_ids"] == ["J0", "J1", "J2", "J3", "J5"]
+    assert result["impact"]["compared_node_count"] == 1
+
+
+@pytest.mark.parametrize("times", [{}, {"report_start": 900}, {"sample_times": ["2026-07-16T05:00:01Z"]}])
+def test_unknown_or_shifted_first_sample_never_queries_baseline(inputs, times):
+    native = simulation([2])
+    native["times"] = times
+    result = service.estimate_burst_impact("own", START, native)
+    assert result["impact"]["reason"] == "simulation_has_no_verified_start_time_sample"
+    assert "map_payload" not in result
+    inputs.assert_not_called()
+
+
+def test_unknown_units_do_not_draw_an_impact_area(inputs):
+    result = service.estimate_burst_impact("own", START, simulation([2], "unknown"))
+    assert result["impact"]["reason"] == "unsupported_or_missing_pressure_units"
+    inputs.assert_not_called()
+
+
+def test_full_native_network_compared_before_selected_series_filter(monkeypatch, inputs):
+    from app.mcp import scenario_service
+    from app.services import simulation_scenarios
+    native = simulation([4] * 25)
+    run = Mock(return_value=native)
+    monkeypatch.setattr(simulation_scenarios, "burst_analysis", run)
+    result = scenario_service.calculate_scenario("run_burst_simulation", {
+        "output_node_ids": ["J0"], "output_link_ids": [], "start_time": START, "duration": 900,
+        "burst_pipe_ids": ["P1"], "orifice_areas_cm2": [2], "include_impact_coverage": True,
+    }, network="own", username="user", model_only=True)
+    assert len(result["node_results"]) == 1
+    assert result["impact"]["affected_node_count"] == 25
+    assert result["coverage"]["requested_nodes"] == ["J0"]
+    assert result["persisted_baseline"] is False
+    assert run.call_args.kwargs["persist_results"] is False
+    assert run.call_args.kwargs["burst_ID"] == ["P1"]
+    assert not run.call_args.kwargs["scada_mappings"].reservoirs
+    assert result["impact"]["boundary_source"] == "model_only_no_live_scada"
+
+
+def test_large_affected_sets_report_full_counts_without_truncating(inputs):
+    result = service.estimate_burst_impact("own", START, simulation([4] * 10001))
+    assert result["impact"]["affected_node_count"] == 10001
+    assert result["impact"]["coverage_status"] == "unavailable"
+    assert "map_payload" not in result
+
+
+def test_replay_passes_empty_boundary_mappings_to_both_native_runs(monkeypatch):
+    from app.services import simulation_scenarios as scenarios
+    from app.infra.db.postgresql.scada import ScadaElementMappings
+    mappings = ScadaElementMappings({}, {}, {}, {}, {}, {}, {})
+    runs = Mock(return_value={"node_results": []})
+    monkeypatch.setattr(scenarios.simulation, "run_simulation", runs)
+    monkeypatch.setattr(scenarios, "get_pipe", lambda *a: {"node1": "A", "node2": "B", "diameter": 100})
+    monkeypatch.setattr(scenarios, "is_junction", lambda *a: True)
+    monkeypatch.setattr(scenarios, "get_emitter", lambda *a: None)
+    monkeypatch.setattr(scenarios, "set_emitter", lambda *a: None)
+    monkeypatch.setattr(scenarios, "get_option", lambda *a: {"UNITS": "LPS"})
+    monkeypatch.setattr(scenarios, "set_option", lambda *a: None)
+    scenarios.burst_analysis.__wrapped__(
+        name="own", modify_pattern_start_time=START.isoformat(), burst_ID=["P1"], burst_size=[2],
+        persist_results=False, _temporary_project="tjw_tmp_own", scada_mappings=mappings,
+    )
+    assert len(runs.call_args_list) == 2
+    assert all(call.kwargs["scada_mappings"] is mappings for call in runs.call_args_list)
+    assert runs.call_args.kwargs["persist_results"] is False
+
+
+@pytest.mark.anyio
+async def test_order_notification_precedes_durable_impact_and_keeps_run_available(inputs):
+    from copy import deepcopy
+    from uuid import uuid4
+    from fastapi.encoders import jsonable_encoder
+    from app.domain.schemas.operational_task import ActionProposalCommand, TaskPresentationEventCommand
+    from app.domain.schemas.operational_data import TaskDataSnapshot
+    from app.mcp.models import QueryResult
+    from app.mcp.presentation_preview import saved_presentation_preview
+    from app.services.operational_tasks import OperationalTaskService
+    from tests.support.operational_task_repository import TaskRepository, _context, _task_document, _active_run
+
+    context = _context()
+    task = _task_document(context, START, lease=START + timedelta(minutes=1))
+    task["workflow"] = "burst_report"
+    repository = TaskRepository(task, _active_run(task, START))
+    repository.reports.append({"revision": 1, "execution_generation": 1,
+                               "report": {"stage": "interim", "outcome": "concluded"}})
+    tasks = OperationalTaskService(repository, context, set(), clock=lambda: START)
+    order = await tasks.propose_action(task["task_id"], ActionProposalCommand(
+        request_id=uuid4(), kind="create_work_order", reason="P1候选管段待现场核验",
+        payload={"title": "P1现场核查", "description": "核查定位管道并反馈", "work_order_type": "general"},
+    ))
+    assert repository.task["status"] == "running"
+    assert repository.notifications[-1]["kind"] == "action_approval_pending"
+    assert order.approval_status == "pending" and order.execution_status == "not_submitted"
+    result = QueryResult(
+        source="run_burst_simulation", source_ref="mcp:burst-impact", queried_at=START,
+        semantics="analysis", origin="system_data", truncated=False,
+        data=jsonable_encoder({"result": service.estimate_burst_impact("own", START, simulation([2, 4, 6]))}),
+    )
+    evidence, _ = await tasks.record_data_snapshot(task["task_id"], TaskDataSnapshot(
+        source=result.source, source_ref=result.source_ref, queried_at=START,
+        origin="system_data", truncated=False, data=jsonable_encoder(result),
+    ))
+    result.evidence_id = evidence.evidence_id
+    hint = saved_presentation_preview(result)
+    await tasks.publish_presentation_event(task["task_id"], TaskPresentationEventCommand(
+        request_id=uuid4(), run_id=repository.run["run_id"], source_session_id="session-1",
+        command=hint["command"],
+    ))
+    recorded = deepcopy(repository.events[-1]["metadata"]["command"])
+    assert recorded["data"] == jsonable_encoder(result.data["result"]["map_payload"])
+    assert len(recorded["data"]["features"]) == 3
+    # Saved presentation remains complete even if the live query result changes.
+    result.data["result"]["map_payload"]["features"].clear()
+    assert len(repository.events[-1]["metadata"]["command"]["data"]["features"]) == 3
+    assert repository.run["status"] == "running"

tests/unit/test_contaminant_impact.py

@@ -0,0 +1,135 @@
+from copy import deepcopy
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from app.services import contaminant_impact as impact
+from app.services.node_coverage import build_node_coverage
+
+
+START = datetime(2026, 7, 16, 0, tzinfo=timezone.utc)
+
+
+def simulation(unit="MGL"):
+    return {"units": {"quality": unit}, "times": {"report_start": 0, "report_step": 900, "num_periods": 3},
+        "node_results": [{"node": node, "result": [{"quality": value} for value in values]}
+            for node, values in {"source": [1, 1, 1], "late": [0, 0.2, 0.5], "clean": [0, 0, 0]}.items()]}
+
+
+@pytest.fixture
+def geometry(monkeypatch):
+    calls = []
+
+    def generate(network, groups):
+        calls.append((network, deepcopy(groups)))
+        coverage = build_node_coverage(groups, [{"id": "source", "longitude": 121.4, "latitude": 31.1}], [])
+        payload = coverage.pop("map_payload")
+        return {"coverage": coverage, "map_payload": payload}
+
+    monkeypatch.setattr(impact, "generate_node_coverage", generate)
+    return calls
+
+
+def test_impact_uses_whole_prediction_window_and_real_concentration_units(geometry):
+    result = impact.estimate_contaminant_impact("project", START, 1800, simulation(), threshold_mg_l=0.2, model_only=True)
+    data = result["impact"]
+    assert data["affected_node_ids"] == ["source", "late"]
+    late = next(row for row in data["nodes"] if row["node_id"] == "late")
+    assert late["first_exceedance_sample_at"] == (START + timedelta(minutes=30)).isoformat()
+    assert late["peak_mg_l"] == 0.5 and late["peak_at"] == late["first_exceedance_sample_at"]
+    assert data["node_count"] == 3 and data["status"] == "partial"  # Late node has no GIS position.
+    assert data["boundary_source"] == "model_only_no_live_scada"
+    assert data["threshold_purpose"] == "scenario_display_not_drinking_water_standard"
+    assert geometry[0][1][0]["node_ids"] == ["source", "late"]
+    assert result["coverage"]["groups"]["contaminant_exposure"]["missing_node_ids"] == ["late"]
+    assert result["map_payload"]["features"][0]["properties"]["impact_phase"] == "after_contamination_before_intervention"
+    from app.domain.schemas.operational_data import TaskDataSnapshot
+    from app.services.operational_presentation import resolve_overlay
+    from fastapi.encoders import jsonable_encoder
+    snapshot = TaskDataSnapshot(source="run_contaminant_simulation", source_ref="calculation:propagation",
+        queried_at=START, origin="synthetic", truncated=False, data=jsonable_encoder({"result": result}))
+    saved = resolve_overlay(snapshot, "geometry")
+    assert saved["features"][0]["properties"]["model_estimate"] is True
+
+
+@pytest.mark.parametrize("unit", ["UGL", "ug/L", "µg/L"])
+def test_micrograms_are_converted_before_threshold_comparison(geometry, unit):
+    result = impact.estimate_contaminant_impact("project", START, 1800, simulation(unit), threshold_mg_l=0.0002)
+    assert result["impact"]["nodes"][1]["peak_mg_l"] == 0.0005
+    assert result["impact"]["affected_node_count"] == 2
+
+
+@pytest.mark.parametrize("problem", ["unit", "times", "duplicate", "too_many_samples", "no_values"])
+def test_invalid_results_do_not_turn_into_an_empty_safe_range(geometry, problem):
+    data = simulation()
+    if problem == "unit": data["units"]["quality"] = "HOURS"
+    if problem == "times": data["times"] = {"report_step": 0}
+    if problem == "duplicate": data["node_results"].append(data["node_results"][0])
+    if problem == "too_many_samples": data["node_results"][0]["result"].append({"quality": 2})
+    if problem == "no_values":
+        for row in data["node_results"]: row["result"] = [{"quality": float("nan")}] * 3
+    result = impact.estimate_contaminant_impact("project", START, 1800, data)
+    assert result["impact"]["status"] == "unavailable"
+    assert "reason" in result["impact"] and "map_payload" not in result
+    assert not geometry
+
+
+def test_missing_samples_remain_explicit_and_do_not_replace_first_observation(geometry):
+    data = simulation()
+    data["node_results"][1]["result"][0] = {"quality": None}
+    result = impact.estimate_contaminant_impact("project", START, 1800, data)
+    assert result["impact"]["incomplete_node_ids"] == ["late"]
+    assert result["impact"]["nodes"][1]["complete"] is False
+
+
+def test_geometry_limit_retains_the_numeric_prediction(monkeypatch):
+    monkeypatch.setattr(impact, "MAX_COVERAGE_NODES", 1)
+    result = impact.estimate_contaminant_impact("project", START, 1800, simulation())
+    assert result["impact"]["affected_node_count"] == 2
+    assert result["impact"]["coverage_status"] == "unavailable"
+    assert "map_payload" not in result
+
+
+def test_adapter_estimates_all_native_nodes_before_selecting_display_series(monkeypatch):
+    from app.mcp.analysis_models import ContaminantSimulationInput
+    from app.mcp.scenario_service import calculate_scenario
+    from app.services import simulation_scenarios
+
+    native = simulation()
+    calls = []
+    monkeypatch.setattr(simulation_scenarios, "contaminant_simulation", lambda **kw: calls.append(kw) or native)
+    seen = []
+    monkeypatch.setattr(impact, "estimate_contaminant_impact",
+        lambda network, start, duration, result, **kw: seen.append((network, start, duration, result, kw))
+        or {"impact": {"affected_node_count": 2}})
+    args = ContaminantSimulationInput(context_handle="a" * 43, start_time=START, duration=1800,
+        output_node_ids=["source"], source="source", concentration=1,
+        include_impact_coverage=True, impact_threshold_mg_l=0.2)
+    result = calculate_scenario("run_contaminant_simulation",
+        args.model_dump(exclude={"context_handle", "timeout_seconds"}), network="project", username="actor", model_only=True)
+    assert len(result["node_results"]) == 1 and len(seen[0][3]["node_results"]) == 3
+    assert seen[0][4] == {"threshold_mg_l": 0.2, "model_only": True}
+    assert result["impact"]["affected_node_count"] == 2
+    assert calls[0]["scada_mappings"].reservoirs == {} and calls[0]["scada_mappings"].quality == {}
+    assert calls[0]["persist_results"] is False and calls[0]["scheme_name"] is None
+
+
+def test_native_contaminant_service_passes_empty_mappings_to_the_engine(monkeypatch):
+    from app.infra.db.postgresql.scada import ScadaElementMappings
+    from app.services import simulation_scenarios as native
+
+    for name in ("set_time", "set_option", "add_source", "set_demand"):
+        monkeypatch.setattr(native, name, lambda *args: None)
+    monkeypatch.setattr(native, "get_time", lambda *_: {"HYDRAULIC TIMESTEP": "0:15:00"})
+    monkeypatch.setattr(native, "get_pattern", lambda *_: {"id": "pattern"})
+    monkeypatch.setattr(native, "get_source", lambda *_: {})
+    monkeypatch.setattr(native, "get_demand", lambda *_: {"demands": []})
+    monkeypatch.setattr(native, "get_option", lambda *_: {})
+    calls = []
+    monkeypatch.setattr(native.simulation, "run_simulation", lambda **kw: calls.append(kw) or simulation())
+    mappings = ScadaElementMappings(reservoirs={}, tanks={}, fixed_pumps={}, variable_pumps={}, pressure={}, demand={}, quality={})
+    native.contaminant_simulation.__wrapped__(name="project", modify_pattern_start_time=START.isoformat(),
+        modify_total_duration=1800, source="source", concentration=1, source_pattern="pattern",
+        username="actor", _temporary_project="temporary", persist_results=False, scada_mappings=mappings)
+    assert calls[0]["scada_mappings"] is mappings
+    assert calls[0]["name"] == "temporary" and calls[0]["persist_results"] is False

tests/unit/test_contaminant_simulation_mcp.py

@@ -0,0 +1,68 @@
+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 pydantic import ValidationError
+
+from app.mcp import analysis_service, query_service
+from app.mcp.analysis_models import ContaminantSimulationInput
+from app.mcp.calculation_jobs import should_queue
+from app.mcp.catalog import TOOLS
+
+
+def args(**extra):
+    return ContaminantSimulationInput(context_handle="a" * 43, **({
+        "start_time": "2026-07-16T08:00:00+08:00", "duration": 1800,
+        "output_node_ids": ["Q1"], "source": "Q1", "concentration": 1,
+        "include_impact_coverage": True, "impact_threshold_mg_l": 0.01,
+    } | extra))
+
+
+@pytest.mark.parametrize("extra", [{"impact_threshold_mg_l": -1}, {"impact_threshold_mg_l": float("nan")},
+                                    {"include_impact_coverage": "true"}])
+def test_impact_inputs_reject_invalid_thresholds_and_flags(extra):
+    with pytest.raises(ValidationError): args(**extra)
+
+
+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="project", project_role="member"), user=SimpleNamespace(username="actor"),
+        simulation_routing=routing)
+    binding = SimpleNamespace(tool_cap=["run_contaminant_simulation"], 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_prediction_can_extend_past_cutoff_without_reading_future_observations(monkeypatch):
+    scope, binding = context(monkeypatch)
+    worker = AsyncMock(return_value={"impact": {"affected_node_count": 2}})
+    monkeypatch.setattr(analysis_service, "calculate_replay_contaminant_simulation", worker)
+    result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["run_contaminant_simulation"], args()))
+    assert result.origin == "synthetic" and result.data["result"]["impact"]["affected_node_count"] == 2
+    assert "预测" in result.data_gaps[0]
+    scope.connection.commit.assert_awaited_once()
+    worker.assert_awaited_once()
+    assert worker.await_args.kwargs == {"network": "project", "username": "actor"}
+    assert should_queue(TOOLS["run_contaminant_simulation"], args())
+
+
+@pytest.mark.parametrize("problem", ["future_start", "no_impact", "reader", "permission"])
+def test_replay_admission_keeps_time_permission_and_reader_boundaries(monkeypatch, problem):
+    scope, binding = context(monkeypatch)
+    parameters = args()
+    if problem == "future_start": parameters = args(start_time="2026-07-16T08:15:00+08:00")
+    if problem == "no_impact": parameters = args(include_impact_coverage=False)
+    if problem == "reader": binding.purpose = "task_reader"
+    if problem == "permission": binding.permission_cap.remove("scada.view")
+    with pytest.raises(HTTPException) as caught:
+        asyncio.run(query_service.authorize_query(scope, binding, TOOLS["run_contaminant_simulation"], parameters))
+    assert caught.value.status_code == (422 if problem in {"future_start", "no_impact"} else 403)

tests/unit/test_contaminant_source_trace.py

@@ -0,0 +1,73 @@
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+from uuid import uuid4
+
+import pytest
+from fastapi import HTTPException
+from pydantic import ValidationError
+
+from app.mcp import query_service
+from app.mcp.analysis_models import ContaminantSourceTraceInput
+from app.mcp.calculation_jobs import should_queue
+from app.mcp.catalog import TOOLS
+from app.services import contaminant_source_trace as trace
+
+
+def arguments(**extra):
+    return ContaminantSourceTraceInput(context_handle="a" * 43,
+        **({"candidate_node_ids": ["J2", "J1", "missing"],
+            "start": "2026-07-16T08:00:00+08:00", "end": "2026-07-16T08:15:00+08:00"} | extra))
+
+
+@pytest.mark.parametrize("nodes", [[{"id": "J2"}, {"id": "J1"}], [{"id": "J1"}, {"id": "J2"}]])
+def test_trace_is_immediate_project_scoped_stable_and_explicit_about_algorithm(monkeypatch, nodes):
+    read = AsyncMock(return_value=nodes)
+    monkeypatch.setattr(trace, "read_rows", read)
+    connection = object()
+    args = arguments()
+    result = asyncio.run(trace.trace_contaminant_source(connection, args))
+    assert result["located_node"] == "J1"
+    assert result["missing_node_ids"] == ["missing"]
+    assert result["top_candidates"] == [{"node_id": "J1"}]
+    assert result["algorithm_status"] == "provisional"
+    assert result["selection_basis"] == "lexicographically_first_existing_candidate_node"
+    assert "相似度" not in result and "confidence" not in result
+    assert read.await_args.args[0] is connection
+    assert "network.nodes" in read.await_args.args[1] and "ANY(%s)" in read.await_args.args[1]
+    assert read.await_args.args[2] == (args.candidate_node_ids,)
+    assert not should_queue(TOOLS["run_contaminant_source_trace"], args)
+
+
+@pytest.mark.parametrize("change", [{"candidate_node_ids": []}, {"candidate_node_ids": ["J1", "J1"]},
+    {"candidate_node_ids": [str(index) for index in range(21)]},
+    {"start": "2026-07-16T08:00:00"}, {"end": "2026-07-16T08:00:00+08:00"},
+    {"end": "2026-07-18T08:00:00+08:00"}])
+def test_trace_rejects_unbounded_or_ambiguous_inputs(change):
+    with pytest.raises(ValidationError):
+        arguments(**change)
+
+
+@pytest.mark.parametrize("nodes", [[], [{"id": "J1"}]])
+def test_dispatch_preserves_provenance_and_reports_only_actual_node_gaps(monkeypatch, nodes):
+    monkeypatch.setattr(trace, "read_rows", AsyncMock(return_value=nodes))
+    monkeypatch.setattr(query_service, "permissions_for_context", lambda _: {"simulation.run"})
+    scope = SimpleNamespace(connection=object(), project=SimpleNamespace(project_role="member"))
+    binding = SimpleNamespace(tool_cap=["run_contaminant_source_trace"], purpose="task_execution",
+        permission_cap=["simulation.run"], task_id=uuid4(), task_scope_cap=[], replay_context=None)
+    result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["run_contaminant_source_trace"], arguments()))
+    assert result.origin == "synthetic" and result.semantics == "analysis"
+    assert (result.data["result"]["located_node"] is None) == (not nodes)
+    assert bool(result.data_gaps) == (not nodes)
+    assert "context_handle" not in result.data["parameters"]
+
+
+def test_replay_trace_cannot_escape_frozen_window(monkeypatch):
+    monkeypatch.setattr(query_service, "permissions_for_context", lambda _: {"simulation.run"})
+    scope = SimpleNamespace(project=SimpleNamespace(project_role="member"))
+    binding = SimpleNamespace(tool_cap=["run_contaminant_source_trace"], purpose="task_execution",
+        permission_cap=["simulation.run"], task_id=uuid4(), task_scope_cap=[],
+        replay_context={"knowledge_cutoff": "2026-07-16T08:00:00+08:00"})
+    with pytest.raises(HTTPException) as caught:
+        asyncio.run(query_service.authorize_query(scope, binding, TOOLS["run_contaminant_source_trace"], arguments()))
+    assert caught.value.status_code == 422

tests/unit/test_investigation_graph.py

@@ -80,6 +80,78 @@ async def test_failed_attempt_cannot_be_relabelled_as_a_success():
         ])
 
 
+@pytest.mark.anyio
+async def test_existing_waiting_node_accepts_a_status_only_update_without_extending_the_graph():
+    task, repo, service = setup_service()
+    first = await publish(service, repo, task["task_id"], nodes=[
+        {"node_id": "read", "label": "读取监测", "status": "recorded"},
+        {"node_id": "review", "label": "核对补充资料", "status": "needs_input", "detail": "等待现场记录"},
+    ], edges=[{"source": "read", "target": "review"}])
+    original = first.task.latest_progress.progress.diagram
+    await service.publish_artifacts(task["task_id"], AgentArtifactsCommand.model_validate({
+        "request_id": str(uuid4()), "source_session_id": repo.run["agent_session_id"],
+        "run_id": repo.run["run_id"], "report": {
+            "title": "阶段报告", "current_conclusion": "仍需核对现场资料", "conclusion_boundary": "现有资料范围",
+            "stage": "interim", "outcome": "concluded", "data_gaps": [],
+        },
+    }))
+    result = await publish(service, repo, task["task_id"], nodes=[
+        {"node_id": "review", "status": "recorded", "detail": "已核对收到的现场记录"},
+    ])
+    updated = result.task.latest_progress.progress.diagram
+    assert [node.node_id for node in updated.node_refs] == ["read", "review"]
+    assert updated.node_refs[1].label == "核对补充资料"
+    assert updated.node_refs[1].status == "recorded"
+    assert updated.node_refs[1].detail == "已核对收到的现场记录"
+    assert updated.edges == original.edges
+    assert updated.mermaid == original.mermaid
+    assert original.node_refs[1].status == "needs_input"
+    archived = (await service.report(task["task_id"], 1)).report.diagram
+    assert archived.node_refs[1].status == "needs_input"
+    assert archived.node_refs[1].detail == "等待现场记录"
+
+
+@pytest.mark.parametrize("label_patch", [{}, {"label": None}])
+def test_partial_updates_keep_labels_and_accumulated_references_including_legacy_nodes(label_patch):
+    evidence_id = "saved-reading"
+    action_id = uuid4()
+    original = InvestigationDiagram.model_validate({
+        "title": "旧调查图", "mermaid": "flowchart LR\nold{核对资料} --> next[继续调查]",
+        "node_refs": [{"node_id": "old", "label": "核对资料", "status": "waiting_human",
+                       "detail": "等待资料", "evidence_refs": [evidence_id], "action_refs": [action_id],
+                       "map_targets": [{"source_id": "scada", "feature_id": "P1"}]}],
+    })
+    updated = update_investigation_graph(original, InvestigationGraphUpdate.model_validate({
+        "nodes": [{"node_id": "old", "status": "recorded", **label_patch}],
+    }))
+    assert updated.mermaid == updated.source_mermaid == original.mermaid
+    assert updated.node_refs[0] == original.node_refs[0].model_copy(update={"status": "recorded"})
+    assert original.node_refs[0].status == "needs_input"
+
+
+def test_status_only_updates_cannot_create_an_unlabelled_node_or_rewrite_a_failed_attempt():
+    with pytest.raises(DomainError) as error:
+        update_investigation_graph(None, InvestigationGraphUpdate.model_validate({
+            "nodes": [{"node_id": "missing", "status": "recorded"}],
+        }))
+    assert error.value.code == "GRAPH_NODE_LABEL_REQUIRED"
+    failed = update_investigation_graph(None, InvestigationGraphUpdate.model_validate({
+        "nodes": [{"node_id": "attempt", "label": "失败尝试", "status": "failed", "detail": "接口超时"}],
+    }))
+    enriched = update_investigation_graph(failed, InvestigationGraphUpdate.model_validate({
+        "nodes": [{"node_id": "attempt", "evidence_refs": ["timeout-record"]}],
+    }))
+    assert enriched.node_refs[0].label == "失败尝试"
+    assert enriched.node_refs[0].detail == "接口超时"
+    assert enriched.node_refs[0].status == "failed"
+    assert enriched.node_refs[0].evidence_refs == ["timeout-record"]
+    with pytest.raises(DomainError) as error:
+        update_investigation_graph(enriched, InvestigationGraphUpdate.model_validate({
+            "nodes": [{"node_id": "attempt", "status": "recorded"}],
+        }))
+    assert error.value.code == "GRAPH_FAILED_ATTEMPT_IMMUTABLE"
+
+
 @pytest.mark.anyio
 async def test_new_execution_round_continues_the_same_task_graph():
     task, repo, service = setup_service()

tests/unit/test_mcp_analysis.py

@@ -26,6 +26,7 @@ HANDLE = "x" * 43
 START = "2026-10-08T12:00:00+08:00"
 END = "2026-10-08T13:00:00+08:00"
 INPUTS = {
+    "run_contaminant_source_trace": {"candidate_node_ids": ["J1"], "start": START, "end": END},
     "run_simulation": {"start_time": START, "duration": 3600, "output_node_ids": ["J1"], "output_link_ids": ["L1"]},
     "run_contaminant_simulation": {"start_time": START, "output_node_ids": ["J1"], "source": "J2", "concentration": 1.0},
     "analyze_valve_closure": {"start_time": START, "output_node_ids": ["J1"], "valves": ["V1"]},

tests/unit/test_mcp_calculation_jobs.py

@@ -204,3 +204,44 @@ def test_legacy_execution_timeout_does_not_block_owner_authorization(monkeypatch
     assert args.timeout_seconds <= 240
     assert row['parameters']['timeout_seconds'] == 3600
     authorize.assert_awaited_once()
+
+
+def test_large_burst_result_keeps_saved_map_command_and_full_evidence(monkeypatch):
+    from copy import deepcopy
+    from unittest.mock import AsyncMock
+    from app.mcp import calculation_jobs as jobs
+    from app.mcp.calculation_models import CalculationResultInput
+    from app.mcp.models import QueryResult
+    original = binding()
+    original.task_id = uuid4()
+    full = {"parameters": {"burst_pipe_ids": ["P1"]}, "result": {
+        "impact": {"phase": "after_burst_before_intervention", "status": "ready",
+                   "compared_node_count": 1000, "comparisons": [{"node_id": f"J{i}", "pressure_drop_m": 4} for i in range(1000)],
+                   "grades": [{"group_id": "drop_3_5", "node_ids": [f"J{i}" for i in range(1000)]}],
+                   "unavailable_node_ids": []},
+        "coverage": {"status": "ready", "groups": {"drop_3_5": {"node_ids": [f"J{i}" for i in range(1000)],
+                                                                  "mapped_node_count": 1000, "missing_node_ids": []}}},
+        "map_payload": {"type": "FeatureCollection", "features": [{"large": "recorded-geometry"}]},
+    }}
+    result = QueryResult(source='run_burst_simulation', source_ref='mcp:burst', queried_at=datetime.now(timezone.utc),
+                         semantics='analysis', data=full, evidence_id='saved-burst')
+    row = {'job_id': uuid4(), 'binding': durable_binding(original), 'status': 'succeeded', 'result': result.model_dump(mode='json')}
+    persisted = deepcopy(row)
+    monkeypatch.setattr(jobs, 'CalculationJobRepository', lambda *a: SimpleNamespace(get=AsyncMock(return_value=row)))
+    monkeypatch.setattr(jobs, 'authorize_original', AsyncMock())
+    monkeypatch.setattr(jobs, 'attach_calculation_evidence', AsyncMock(side_effect=lambda scope, row, result: result))
+    scope = SimpleNamespace(connection=None, project=SimpleNamespace(project_id=original.project_id, user_id=uuid4()))
+    args = CalculationResultInput(context_handle='x' * 32, job_id=row['job_id'])
+    response = asyncio.run(jobs.read_calculation_job(scope, original, SimpleNamespace(name='get_calculation_result'), args))
+    assert response.truncated and len(response.model_dump_json().encode()) < 11000
+    assert response.data['result']['impact']['compared_node_count'] == 1000
+    assert response.data['result']['impact']['grades'][0]['node_count'] == 1000
+    assert response.data['presentation']['command']['evidence_id'] == 'saved-burst'
+    assert response.data['presentation']['command']['action'] == 'set_overlay'
+    assert response.data['resource_uri'].endswith('/evidence/saved-burst')
+    assert 'comparisons' not in response.data['result']['impact']
+    assert row == persisted
+    # Explicit path pagination still reads every original node without summary replacement.
+    page = asyncio.run(jobs.read_calculation_job(scope, original, SimpleNamespace(name='get_calculation_result'),
+        args.model_copy(update={'data_path': ['result', 'impact', 'comparisons'], 'offset': 990, 'limit': 10})))
+    assert page.data[0]['node_id'] == 'J990' and page.page.total == 1000

tests/unit/test_mcp_extended.py

@@ -292,7 +292,7 @@ def test_artifacts_are_integrity_checked_bounded_binary_envelopes():
 
 def test_no_health_prediction_tool_and_every_write_requests_native_permission():
     assert not any("health" in name for name in TOOLS)
-    assert len(TOOLS) == 72
+    assert len(TOOLS) == 73
     assert sum(t.mutating for t in TOOLS.values()) == 13
     for tool in TOOLS.values():
         if tool.mutating:

tests/unit/test_mcp_presentation_preview.py

@@ -34,3 +34,32 @@ def test_missing_geometry_does_not_offer_a_broken_overlay_command():
         data={"result": {"coverage": {"status": "unavailable"}}},
     )
     assert saved_presentation_preview(result) is None
+
+
+def test_pollution_impact_keeps_a_saved_overlay_reference_for_history():
+    result = QueryResult(source="run_contaminant_simulation", source_ref="calculation:pollution",
+        evidence_id="pollution:impact", queried_at=datetime.now(timezone.utc), semantics="analysis", origin="synthetic",
+        data={"result": {"impact": {"phase": "after_contamination_before_intervention"},
+            "map_payload": {"type": "FeatureCollection", "features": [{"geometry": "saved"}]},
+            "coverage": {"status": "ready", "groups": {"contaminant_exposure": {"node_ids": ["Q1"], "mapped_node_count": 1}}}}})
+    preview = saved_presentation_preview(result)
+    assert preview["command"]["evidence_id"] == "pollution:impact"
+    assert preview["command"]["title"] == "水污染传播预计影响范围（处置前情景）"
+    assert "data" not in preview["command"] and preview["groups"]["contaminant_exposure"]["node_count"] == 1
+    result.data["result"]["impact"]["phase"] = "unknown"
+    assert saved_presentation_preview(result) is None
+
+
+def test_burst_impact_hint_uses_saved_evidence_and_distinct_pre_intervention_title():
+    result = QueryResult(
+        source="run_burst_simulation", source_ref="mcp:burst", evidence_id="mcp:burst",
+        queried_at=datetime.now(timezone.utc), semantics="analysis", origin="system_data", truncated=False,
+        data={"result": {"impact": {"phase": "after_burst_before_intervention"},
+                         "map_payload": {"type": "FeatureCollection", "features": [{"large": "geometry"}]}}},
+    )
+    hint = saved_presentation_preview(result)
+    assert hint["command"]["evidence_id"] == "mcp:burst"
+    assert "爆管后、处置前" in hint["command"]["title"]
+    assert "data" not in hint["command"]
+    result.data["result"].pop("impact")
+    assert saved_presentation_preview(result) is None

tests/unit/test_mcp_query.py

@@ -373,6 +373,30 @@ def test_replay_binding_denies_future_and_allows_device_metadata(monkeypatch):
     asyncio.run(query_service.authorize_query(SimpleNamespace(project=object()), bound, TOOLS["list_scada_devices"], SimpleNamespace()))
 
 
+def test_replay_burst_scenario_requires_coverage_permission_and_known_start(monkeypatch):
+    bound = binding(task=True)
+    bound.replay_context = {"knowledge_cutoff": "2026-10-08T01:00:00Z"}
+    permissions = PERMISSIONS | {"simulation.run", "scada.view", "simulation.view"}
+    bound.permission_cap = sorted(permissions)
+    monkeypatch.setattr(query_service, "permissions_for_context", lambda _: permissions)
+    scope = SimpleNamespace(project=SimpleNamespace(project_role="member"))
+    args = TOOLS["run_burst_simulation"].model.model_validate({
+        "context_handle": bound.context_handle, "start_time": "2026-10-08T01:00:00Z",
+        "output_node_ids": ["J1"], "burst_pipe_ids": ["P1"], "orifice_areas_cm2": [2],
+        "include_impact_coverage": True,
+    })
+    asyncio.run(query_service.authorize_query(scope, bound, TOOLS["run_burst_simulation"], args))
+    for changed, status in (({"include_impact_coverage": False}, 422),
+                            ({"start_time": datetime(2026, 10, 8, 2, tzinfo=timezone.utc)}, 422)):
+        with pytest.raises(HTTPException) as error:
+            asyncio.run(query_service.authorize_query(scope, bound, TOOLS["run_burst_simulation"], args.model_copy(update=changed)))
+        assert error.value.status_code == status
+    bound.permission_cap = sorted(permissions - {"simulation.view"})
+    with pytest.raises(HTTPException) as error:
+        asyncio.run(query_service.authorize_query(scope, bound, TOOLS["run_burst_simulation"], args))
+    assert error.value.status_code == 403
+
+
 def test_model_response_displays_utc8_without_changing_query_instants_or_source(runtime):
     app, client, _, _ = runtime
     at = datetime(2026, 12, 31, 16, tzinfo=timezone.utc)

tests/unit/test_mcp_work_order_skills.py

@@ -19,6 +19,7 @@ from app.mcp.work_order_skills import read_work_order_skill
     ("water_quality_complaint", "water-quality-complaint"),
     ("general", "appointment-follow-up"),
     ("scada_anomaly", "scada-anomaly"),
+    ("water_pollution", "water-pollution-event"),
 ])
 def test_skill_descriptor_and_body_have_same_version_and_digest(workflow, name):
     descriptor = read_work_order_skill(workflow)
@@ -59,9 +60,20 @@ def test_scada_instructions_advance_candidate_analysis_independently_of_source_l
     assert not any(word in text for word in ("演示", "mock", "location_demo", "只发布回放成果"))
 
 
-def test_task_skill_uses_persisted_workflow_and_preserves_task_scope(monkeypatch):
+def test_pollution_instructions_offer_work_order_before_simulation_and_reuse_the_investigation():
+    skill = read_work_order_skill("water_pollution", include_content=True)
+    text = skill["instructions"]
+    assert text.index("propose_task_action") < text.index("run_contaminant_simulation")
+    assert "include_impact_coverage=true" in text and "不等模拟完成才发起工单" in text
+    assert "同一 task_id" in text and "不覆盖已审核工单或绕过版本校验" in text
+    assert "模型曲线不填观测专用 chart.scada" in text
+    assert "不能把浊度、余氯下降或感官描述直接换算成污染源浓度" 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()
-    reader = AsyncMock(return_value={"task": {"workflow": "burst_report"}})
+    reader = AsyncMock(return_value={"task": {"workflow": workflow}})
     monkeypatch.setattr(query_service.reads, "task_context_summary", reader)
     monkeypatch.setattr(query_service, "permissions_for_context", lambda project: {"operational_task.view"})
     scope = SimpleNamespace(connection=object(), project=object())
@@ -80,7 +92,15 @@ def test_task_skill_uses_persisted_workflow_and_preserves_task_scope(monkeypatch
     body = query("skill")
     assert summary.data["handling_skill"]["content_sha256"] == body.data["skill"]["content_sha256"]
     assert summary.data["display_time_zone"] == "Asia/Shanghai"
-    assert body.data["skill"]["workflow"] == "burst_report"
+    assert body.data["skill"]["workflow"] == workflow
+    for result in (summary, body):
+        assert ("source_trace_capability" in result.data) == (workflow == "water_pollution")
+        if workflow == "water_pollution":
+            assert result.data["source_trace_capability"]["tool"] == "run_contaminant_source_trace"
+            capability = result.data["source_trace_capability"]
+            assert capability["work_order_wait_blocks_analysis"] is False
+            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 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)

tests/unit/test_operational_recovery.py

@@ -22,6 +22,25 @@ def observations():
     return rows
 
 
+def test_pollution_field_feedback_resumes_review_without_pressure_recovery():
+    task = {"workflow": "water_pollution"}
+    for order_id in ("sampling", "flush"):
+        assert track_work_order(task, {"kind": "create_work_order", "action_id": str(uuid4()),
+            "execution_status": "succeeded", "receipt": {"adapter": "peripheral_sample", "receipt_id": order_id}})
+    feedback = {"source_event_id": "peripheral_sample:sampling", "status": "completed", "source_revision": 2,
+        "occurred_at": RETURNED.isoformat(), "evidence": {"evidence_id": "sampling:2"}}
+    accept_completion(task, feedback)
+    assert task["handling_context"]["phase"] == "waiting_work_order"
+    accept_completion(task, {**feedback, "source_event_id": "peripheral_sample:flush", "evidence": {"evidence_id": "flush:2"}})
+    assert task["handling_context"]["phase"] == "field_review"
+    assert "verification_evidence_id" not in task["handling_context"]
+    assert task["handling_context"]["work_orders"]["sampling"]["feedback_evidence_id"] == "sampling:2"
+    task["handling_context"]["phase"] = "completed"
+    accept_completion(task, {**feedback, "source_revision": 3, "evidence": {"evidence_id": "sampling:3"}})
+    assert task["handling_context"]["phase"] == "field_review"
+    assert task["handling_context"]["work_orders"]["sampling"]["feedback_evidence_id"] == "sampling:3"
+
+
 def evaluate(rows):
     return evaluate_recovery(rows, device_ids=["P1", "F1"], returned_at=RETURNED,
                              incident_at=INCIDENT, interval_seconds=900)

tests/unit/test_operational_tasks.py

@@ -675,6 +675,19 @@ async def test_completed_investigation_resumes_once_for_linked_new_field_feedbac
     again = await service.external_feedback(task["task_id"], ExternalFeedbackCommand(request_id=uuid4(), **identity))
     assert again.task.execution_generation == 2
     assert len([event for event in repository.events if event["topic"] == "operational.run.queued"]) == 1
+    notices = [item for item in repository.notifications if item["kind"] == "work_order_feedback"]
+    assert len(notices) == 1
+    assert notices[0]["subject_id"] == str(action.action_id)
+    assert notices[0]["work_order_action"]["receipt"]["receipt_id"] == child_id
+    review_notice = next(item for item in repository.notifications if item["kind"] == "action_approval_pending")
+    assert review_notice["work_order_action"]["approval_status"] == "pending"
+    assert review_notice["work_order_action"]["receipt"] is None
+    assert notices[0]["work_order_reference"] == {
+        "source_event_id": f"peripheral_sample:{child_id}", "source_revision": 2,
+        "evidence_id": None, "status": "completed",
+    }
+    assert any(event["topic"] == "operational.notification.changed"
+               and event["metadata"].get("kind") == "work_order_feedback" for event in repository.events)
     repository.task.update(status="cancelled", lifecycle_status="cancelled")
     with pytest.raises(DomainError) as cancelled:
         await service.external_feedback(task["task_id"], ExternalFeedbackCommand(

tests/unit/test_pollution_event_workflow.py

@@ -0,0 +1,139 @@
+from datetime import datetime, timedelta, timezone
+from unittest.mock import AsyncMock
+from uuid import uuid4
+
+import pytest
+
+from app.domain.domain_errors import DomainError
+from app.domain.schemas.operational_data import TaskDataSnapshot
+from app.domain.schemas.operational_task import (
+    ActionDecisionCommand, ActionExecuteCommand, ActionProposalCommand, ActionReceipt,
+    AgentArtifactsCommand, AgentRunCommand, ExternalFeedbackCommand,
+)
+from app.services.operational_recovery import track_work_order
+from app.services.operational_tasks import OperationalTaskService
+from tests.support.operational_task_repository import TaskRepository, _active_run, _context, _task_document
+
+
+@pytest.mark.anyio
+async def test_pollution_work_order_can_be_approved_and_dispatched_while_simulation_runs():
+    from types import SimpleNamespace
+    from app.auth.permissions import OPERATIONAL_ACTION_WORK_ORDER_APPROVE, OPERATIONAL_ACTION_WORK_ORDER_EXECUTE
+    from app.infra.integrations.operational_actions import ActionDispatchOutcome
+
+    now = datetime(2026, 7, 16, 6, tzinfo=timezone.utc)
+    context = _context()
+    task = _task_document(context, now, lease=now + timedelta(minutes=1))
+    task["workflow"] = "water_pollution"
+    repository = TaskRepository(task, _active_run(task, now))
+    repository.pending_calculation_jobs = AsyncMock(return_value=["pollution-simulation"])
+    service = OperationalTaskService(repository, context, set(), clock=lambda: now)
+    report = {"stage": "interim", "outcome": "concluded", "data_gaps": [], "title": "候选源核验",
+        "current_conclusion": "候选节点已取得，可先派现场核查", "conclusion_boundary": "污染传播计算中",
+        "evidence_refs": [], "recommendations": ["采样核验"], "pending_items": ["传播模拟"]}
+    await service.publish_artifacts(task["task_id"], AgentArtifactsCommand(
+        request_id=uuid4(), source_session_id="session-1", report=report))
+    action = await service.propose_action(task["task_id"], ActionProposalCommand(
+        request_id=uuid4(), kind="create_work_order", reason="先核验候选源",
+        payload={"title": "候选节点现场核查", "description": "核查污染进入路径并采样", "work_order_type": "general"}))
+    with pytest.raises(DomainError) as denied:
+        await service.decide_action(task["task_id"], action.action_id, ActionDecisionCommand(
+            request_id=uuid4(), expected_version=action.version, decision="approved"))
+    assert denied.value.code == "PERMISSION_DENIED"
+    service = OperationalTaskService(repository, context,
+        {OPERATIONAL_ACTION_WORK_ORDER_APPROVE, OPERATIONAL_ACTION_WORK_ORDER_EXECUTE}, clock=lambda: now)
+    approved = await service.decide_action(task["task_id"], action.action_id, ActionDecisionCommand(
+        request_id=uuid4(), expected_version=action.version, decision="approved"))
+    dispatcher = SimpleNamespace(dispatch=AsyncMock(return_value=ActionDispatchOutcome("succeeded", "现场工单已创建",
+        ActionReceipt(adapter="peripheral_sample", receipt_id=str(uuid4()), submitted_at=now,
+            completed_at=now, external_status="open", detail="创建成功"))))
+    with pytest.raises(DomainError) as stale:
+        await service.submit_action(task["task_id"], action.action_id, ActionExecuteCommand(
+            request_id=uuid4(), expected_version=action.version), dispatcher, None)
+    assert stale.value.code == "VERSION_CONFLICT"
+    dispatched = await service.submit_action(task["task_id"], action.action_id, ActionExecuteCommand(
+        request_id=uuid4(), expected_version=approved.version), dispatcher, None)
+    assert dispatched.execution_status == "succeeded"
+    assert repository.task["handling_context"]["phase"] == "waiting_work_order"
+    assert repository.run["status"] == "running" and repository.task["execution_generation"] == 1
+    dispatcher.dispatch.assert_awaited_once()
+    repository.pending_calculation_jobs.assert_not_awaited()  # No simulation dependency on dispatch.
+
+    with pytest.raises(DomainError) as pending:
+        await service.publish_artifacts(task["task_id"], AgentArtifactsCommand(
+            request_id=uuid4(), source_session_id="session-1", report={**report, "stage": "final"}))
+    assert pending.value.code == "PENDING_CALCULATIONS"
+    repository.pending_calculation_jobs.return_value = []
+    evidence, _ = await service.record_data_snapshot(task["task_id"], TaskDataSnapshot(
+        source="run_contaminant_simulation", source_ref="calculation:pollution", queried_at=now,
+        origin="synthetic", truncated=False, data={"result": {"impact": {"affected_node_count": 2}}}))
+    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}"]}))
+    assert len(repository.reports) == 2 and repository.action["version"] == dispatched.version
+    assert repository.action["approval_status"] == "approved" and repository.run["status"] == "running"
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+@pytest.mark.anyio
+async def test_pollution_report_waits_for_field_feedback_and_requires_latest_feedback_evidence(monkeypatch):
+    now = datetime(2026, 7, 16, 6, tzinfo=timezone.utc)
+    context = _context()
+    task = _task_document(context, now, lease=now + timedelta(minutes=1))
+    task.update(workflow="water_pollution", source={"kind": "external_system", "external_id": "scada:quality:1",
+        "external_revision": 1, "event_kind": "water_pollution"})
+    repository = TaskRepository(task, _active_run(task, now))
+    service = OperationalTaskService(repository, context, set(), clock=lambda: now)
+    monkeypatch.setattr(service, "_resolve_investigation_alerts", AsyncMock())
+    scada, _ = await service.record_data_snapshot(task["task_id"], TaskDataSnapshot(
+        source="query_scada_readings", source_ref="mcp:quality", queried_at=now,
+        origin="system_data", truncated=False, data={
+            "window": {"start": (now - timedelta(hours=3)).isoformat(), "end": now.isoformat()},
+            "data": {"devices": [], "readings": [], "coverage": {"mode": "all_scada", "query_complete": True}},
+        }))
+    report = {"stage": "interim", "outcome": "concluded", "data_gaps": [], "title": "水污染事件调查",
+        "current_conclusion": "候选分析完成，待采样与处置反馈", "conclusion_boundary": "未确诊污染源",
+        "evidence_refs": [scada.evidence_id], "recommendations": ["现场采样复测"], "pending_items": ["现场反馈"]}
+    await service.publish_artifacts(task["task_id"], AgentArtifactsCommand(
+        request_id=uuid4(), source_session_id="session-1", report=report))
+    action = await service.propose_action(task["task_id"], ActionProposalCommand(
+        request_id=uuid4(), kind="create_work_order", reason="采样和处置核验",
+        payload={"title": "节点采样复测", "description": "回传位置、时间、数值和判定依据", "work_order_type": "general"}))
+    child = str(uuid4())
+    repository.action = action.model_copy(update={"approval_status": "approved", "execution_status": "succeeded"}).model_dump(mode="json")
+    repository.action["receipt"] = {"adapter": "peripheral_sample", "receipt_id": child,
+        "submitted_at": now.isoformat(), "completed_at": now.isoformat(), "external_status": "open", "detail": "创建成功"}
+    track_work_order(repository.task, repository.action)
+    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 == "waiting_human"
+    service._resolve_investigation_alerts.assert_not_awaited()
+
+    evidence_id = "field-quality:2"
+    await service.external_feedback(task["task_id"], ExternalFeedbackCommand(
+        request_id=uuid4(), source_event_id=f"peripheral_sample:{child}", source_revision=2,
+        status="completed", occurred_at=now, detail="采样、处置与复测反馈已返回",
+        evidence={"evidence_id": evidence_id, "kind": "external_feedback", "summary": "现场记录及复测结果",
+            "source_label": "现场反馈", "source_ref": f"peripheral_sample:{child}", "observed_at": now}))
+    assert repository.task["handling_context"]["phase"] == "field_review"
+    assert "last_check" not in repository.task["handling_context"]  # No pressure recovery shortcut.
+    await service.claim_run(task["task_id"], repository.run["run_id"], AgentRunCommand(
+        request_id=repository.run["execution_request_id"], expected_revision=repository.run["revision"], source_session_id="session-1"))
+    await service.publish_artifacts(task["task_id"], AgentArtifactsCommand(
+        request_id=uuid4(), source_session_id="session-1", report={**report, "stage": "final"}))
+    with pytest.raises(DomainError) as rejected:
+        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 == "POLLUTION_FEEDBACK_REQUIRED"
+    await service.publish_artifacts(task["task_id"], AgentArtifactsCommand(
+        request_id=uuid4(), source_session_id="session-1", report={**report, "stage": "final",
+            "evidence_refs": [scada.evidence_id, evidence_id], "pending_items": []}))
+    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"
+    service._resolve_investigation_alerts.assert_awaited_once_with(task["task_id"], evidence_id)

tests/unit/test_scada_presentation_coverage.py

@@ -85,6 +85,25 @@ async def investigation():
     return task, repository, service, evidence.evidence_id
 
 
+@pytest.mark.anyio
+async def test_pollution_charts_validate_quality_values_and_keep_their_own_workflow():
+    from app.services.scada_presentation_coverage import add_targets, applies, ScadaObservations, verify_chart
+    task, repository, _, evidence_id = await investigation()
+    task["workflow"] = "water_pollution"
+    evidence = repository.evidence[evidence_id]
+    for device in evidence["data_snapshot"]["data"]["data"]["devices"]:
+        device.update(device_type="quality", measurement_unit="mg/L")
+    value = chart(devices=[DEVICES[0]])
+    value["scada"].update(metric="quality", unit="mg/L")
+    assert applies(task)
+    verified = verify_chart(task, value, [evidence], [evidence_id])
+    assert verified["metric"] == "quality" and verified["gaps"] == []
+    observations = ScadaObservations(task, [evidence])
+    assert add_targets(None, [DEVICES[0]], observations)["targets"] == [DEVICES[0]]
+    task["handling_context"] = {"phase": "field_review"}
+    assert not applies(task)
+
+
 def coverage(comparison_devices):
     slots = [(START + timedelta(minutes=15 * i)).isoformat() for i in range(13)]
     common = {"coverage": {device: {"covered": slots} for device in DEVICES}, "metric": "flow", "window_start": START.isoformat(), "window_end": END.isoformat(), "gaps": []}

tests/unit/test_scada_replay_service.py

@@ -807,6 +807,60 @@ def test_combined_monitoring_keeps_rule_evidence_and_quality_notifications(monke
     asyncio.run(run())
 
 
+@pytest.mark.parametrize("create_tasks", [False, True])
+def test_pollution_detection_creates_own_investigation_even_during_period_diagnosis(monkeypatch, create_tasks):
+    from app.services.scada_normal_profile import diagnostic_signal, pollution_signal
+    from app.services.scada_replay_processing import detect_frame
+
+    _, service, *_ = setup(monkeypatch)
+    push = AsyncMock()
+    create_task = AsyncMock(return_value=SimpleNamespace(task_id=uuid4()))
+    advance = AsyncMock()
+    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 *_, **_kwargs: SimpleNamespace(create=create_task, advance_replay_investigation=advance))
+    config = ScadaReplayCreate(request_id=uuid4(), start_time=START, end_time=START + timedelta(hours=1),
+        device_ids=["Q1"], rules=[{"rule_id": "chlorine", "device_ids": ["Q1"],
+            "kind": "bounds", "lower": 0.1, "unit": "mg/L", "metric_name": "余氯"}])
+    document = {"replay_id": str(uuid4()), "config": config.model_dump(mode="json"),
+        "devices": [{"device_id": "Q1", "device_type": "quality", "transmission_frequency": "0:15:00"}]}
+
+    async def run():
+        nonlocal document
+        # One continuous quality episode creates exactly one task, without any
+        # pressure roles, burst score or complaint.
+        for step, value in enumerate((0.05, 0.04, 0.2, 0.3)):
+            at = START + timedelta(minutes=15 * step)
+            document, changes = detect_frame(document, [{"device_id": "Q1", "time": at.isoformat(),
+                "quality": "valid", "cleaned_value": value, "monitored_value": value}], config, at)
+            for event in changes:
+                assert pollution_signal(document, event)
+                assert not diagnostic_signal(document, event)
+            await ScadaReplayService._create_investigations(service, document, changes,
+                {"sequence": step + 1, "at": at.isoformat()}, create_tasks=create_tasks)
+        create_task.assert_awaited_once()
+        command = create_task.await_args.args[0]
+        assert command.workflow == command.source.event_kind == "water_pollution"
+        assert command.task_type == "event_investigation"
+        assert command.source.replay_context.batch_sequence == 1
+        assert command.trigger.kind == "event"
+        assert "投诉" not in command.title and "爆管" not in command.summary
+        assert [call.args[0].event_type for call in push.await_args_list] == ["raised", "resolved"]
+        assert all(call.args[0].condition_id == f"operational-task:{create_task.return_value.task_id}"
+                   for call in push.await_args_list)
+        advance.assert_awaited_once()
+    asyncio.run(run())
+
+
+def test_invalid_quality_data_is_not_a_pollution_detection():
+    from app.services.scada_normal_profile import pollution_signal
+    document = {"devices": [{"device_id": "Q1", "device_type": "quality"}]}
+    assert not pollution_signal(document, {"category": "data_quality", "device_ids": ["Q1"]})
+    assert not pollution_signal(document, {"device_ids": ["unknown"]})
+
+
 def test_failed_incident_publication_rolls_back_state_and_retries_same_task_request(monkeypatch):
     repo, service, _, _, create = setup(monkeypatch)
     create_task = AsyncMock(return_value=SimpleNamespace(task_id=uuid4()))