fix(burst): recognize native pressure units and separate isolation impact

tests/unit/test_burst_impact.py

@@ -195,3 +195,11 @@ async def test_order_notification_precedes_durable_impact_and_keeps_run_availabl
     result.data["result"]["map_payload"]["features"].clear()
     assert len(repository.events[-1]["metadata"]["command"]["data"]["features"]) == 3
     assert repository.run["status"] == "running"
+
+
+@pytest.mark.parametrize("unit", ["MTR", "mtr", " MTR "])
+def test_native_mtr_pressure_generates_all_three_grades(inputs, unit):
+    result = service.estimate_burst_impact("own", START, simulation([1, 2.999, 3, 4.999, 5], unit))
+    assert result["impact"]["status"] == "ready"
+    assert [g["node_ids"] for g in result["impact"]["grades"]] == [["J0", "J1"], ["J2", "J3"], ["J4"]]
+    assert len(result["map_payload"]["features"]) == 3

tests/unit/test_operational_presentation.py

@@ -7,7 +7,7 @@ from pydantic import ValidationError
 from app.domain.domain_errors import DomainError
 from app.domain.schemas.operational_data import TaskDataSnapshot
 from app.domain.schemas.operational_task import TaskPresentationEventCommand, TaskSafeChart
-from app.services.operational_presentation import resolve_overlay
+from app.services.operational_presentation import resolve_overlay, validate_overlay_stage
 
 
 def _snapshot(data):
@@ -58,6 +58,7 @@ def test_presentation_input_rejects_free_form_or_invalid_commands(command):
 
 @pytest.mark.parametrize("command", [
     {"action": "locate_features", "layer": "scada", "feature_ids": ["D-1"]},
+    {"action": "locate_features", "layer": "pipes", "feature_ids": ["P-1"], "event_kind": "burst"},
     {"action": "zoom_to_map", "center": [121.4, 31.1], "zoom": 24},
     {"action": "render_scada_analysis", "items": [{"device_id": "D-1", "level": "unrated"}]},
     {"action": "clear_scada_analysis"},
@@ -110,3 +111,14 @@ def test_safe_chart_preserves_sparse_scatter_and_real_sample_times():
 def test_safe_chart_rejects_invalid_sample_metadata_and_values(series):
     with pytest.raises(ValidationError):
         TaskSafeChart(title="X", chart_type="line", x_data=["t"], series=[series])
+
+
+@pytest.mark.parametrize("phase", ["after_burst_before_intervention", "after_contamination_before_intervention"])
+def test_event_impact_requires_pre_intervention_geometry(phase):
+    data = {"features": [{"properties": {"impact_phase": phase}}]}
+    validate_overlay_stage(data, "event_impact")
+    closure = {"features": [{"properties": {"coverage_group": "supply_affected_nodes"}}]}
+    validate_overlay_stage(closure, "isolation_plan")
+    with pytest.raises(DomainError) as error:
+        validate_overlay_stage(closure, "event_impact")
+    assert error.value.code == "OVERLAY_IMPACT_PHASE_MISMATCH"

tests/unit/test_operational_tasks.py

@@ -580,6 +580,15 @@ async def test_saved_mcp_coverage_envelope_publishes_groups_without_agent_geomet
         }}),
     )
     evidence, _ = await service.record_data_snapshot(task["task_id"], snapshot)
+    with pytest.raises(DomainError) as error:
+        await service.publish_presentation_event(task["task_id"], TaskPresentationEventCommand(
+            request_id=uuid4(), run_id=repository.run["run_id"], source_session_id="session-1",
+            command={"action": "set_overlay", "overlay_id": "wrong-impact", "title": "事件影响",
+                     "mode": "geometry", "evidence_id": evidence.evidence_id},
+            narration={"stage": "event_impact", "summary": "预计影响范围，现场待核实。",
+                       "evidence_refs": [evidence.evidence_id]},
+        ))
+    assert error.value.code == "OVERLAY_IMPACT_PHASE_MISMATCH"
     await service.publish_presentation_event(task["task_id"], TaskPresentationEventCommand(
         request_id=uuid4(), run_id=repository.run["run_id"], source_session_id="session-1",
         command={"action": "set_overlay", "overlay_id": "isolation-coverage", "title": "节点覆盖示意",