fix(investigation): allow autonomous model baseline recovery

tests/unit/test_burst_impact.py

@@ -99,6 +99,27 @@ def test_unknown_units_do_not_draw_an_impact_area(inputs):
     inputs.assert_not_called()
 
 
+def test_missing_online_baseline_offers_recalculation_then_coverage_can_be_retried(inputs):
+    inputs.side_effect = None
+    inputs.return_value = {}
+    native = simulation([2, 4, 6])
+    missing = service.estimate_burst_impact("own", START, native)
+    assert missing["impact"]["reason"] == "no_exact_comparable_online_snapshot"
+    recovery = missing["impact"]["baseline_recovery"]
+    assert recovery["tool"] == "run_simulation"
+    assert recovery["parameters"] == {
+        "start_time": START.isoformat(), "duration": 60, "output_node_ids": ["J0"],
+    }
+    assert "map_payload" not in missing
+
+    inputs.return_value = {
+        f"J{i}": [{"time": START.isoformat(), "value": 30}] for i in range(3)
+    }
+    retried = service.estimate_burst_impact("own", START, native)
+    assert retried["impact"]["status"] == "ready"
+    assert len(retried["map_payload"]["features"]) == 3
+
+
 def test_full_native_network_compared_before_selected_series_filter(monkeypatch, inputs):
     from app.mcp import scenario_service
     from app.services import simulation_scenarios

tests/unit/test_mcp_query.py

@@ -373,7 +373,7 @@ 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):
+def test_replay_burst_scenario_allows_exploration_and_checks_coverage_permission(monkeypatch):
     bound = binding(task=True)
     bound.replay_context = {"knowledge_cutoff": "2026-10-08T01:00:00Z"}
     permissions = PERMISSIONS | {"simulation.run", "scada.view", "simulation.view"}
@@ -386,11 +386,12 @@ def test_replay_burst_scenario_requires_coverage_permission_and_known_start(monk
         "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
+    asyncio.run(query_service.authorize_query(scope, bound, TOOLS["run_burst_simulation"],
+                                              args.model_copy(update={"include_impact_coverage": False})))
+    with pytest.raises(HTTPException) as error:
+        asyncio.run(query_service.authorize_query(scope, bound, TOOLS["run_burst_simulation"],
+                                                  args.model_copy(update={"start_time": datetime(2026, 10, 8, 2, tzinfo=timezone.utc)})))
+    assert error.value.status_code == 422
     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))
@@ -881,7 +882,7 @@ def test_old_task_run_and_reader_parent_are_rejected(monkeypatch):
 
 @pytest.mark.parametrize("purpose", ["task_execution", "task_collaboration"])
 @pytest.mark.parametrize("replay", [False, True])
-def test_work_order_task_excludes_location_but_scada_replay_keeps_it(monkeypatch, purpose, replay):
+def test_work_order_source_does_not_remove_authorized_analysis_tools(monkeypatch, purpose, replay):
     monkeypatch.setattr(contexts, "permissions_for_context", lambda _: PERMISSIONS)
     value = binding(task=True)
     value.purpose = purpose
@@ -899,13 +900,9 @@ def test_work_order_task_excludes_location_but_scada_replay_keeps_it(monkeypatch
     monkeypatch.setattr(contexts, "task_repository", lambda _: Tasks())
     scope = SimpleNamespace(project=object(), connection=Connection())
     asyncio.run(contexts.verify_task(scope, value))
-    assert ("run_burst_location" in value.tool_cap) is replay
+    assert "run_burst_location" in value.tool_cap
     assert "analyze_valve_isolation" in value.tool_cap
     assert "analyze_valve_closure" in value.tool_cap
-    if not replay:
-        with pytest.raises(HTTPException) as caught:
-            asyncio.run(query_service.authorize_query(scope, value, TOOLS["run_burst_location"], SimpleNamespace()))
-        assert caught.value.status_code == 403
 
 
 def test_jwt_signature_issuer_audience_lifetime_and_key_cache():

tests/unit/test_mcp_work_order_skills.py

@@ -46,7 +46,7 @@ def test_work_order_handling_and_legacy_scada_replay_receive_separate_skills():
     assert replay["workflow"] == "burst_report"
     assert "影响说明" in handling["instructions"]
     assert "supply_affected_nodes" in handling["instructions"]
-    assert "run_burst_location" not in handling["instructions"]
+    assert "run_burst_location" in handling["instructions"]
     assert "run_burst_location" in replay["instructions"]
 
 
@@ -66,14 +66,14 @@ def test_pollution_instructions_offer_work_order_before_simulation_and_reuse_the
     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 "模型曲线不填观测专用chart.scada" in text
     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 skill["version"] == "1.2.1"
     assert text.index("## 4. 源头控制与用户隔离") < text.index("## 5. 污染物清除与分步冲洗")
     assert "opened_valves" in text and "closed_valves" in text
     assert "不继承上次模拟" in text and "排放去向" in text

tests/unit/test_online_simulation.py

@@ -46,6 +46,17 @@ def test_online_simulation_rejects_invalid_step_before_calculating(monkeypatch):
     run.assert_not_called()
 
 
+def test_online_baseline_preserves_fractional_start_for_exact_impact_comparison(monkeypatch):
+    start = START.replace(microsecond=123456)
+    monkeypatch.setattr(simulation, "get_time", lambda _: {"HYDRAULIC TIMESTEP": "0:15:00"})
+    monkeypatch.setattr(simulation, "query_corresponding_element_id_and_query_id", lambda _: object())
+    run = Mock(return_value={"node_results": [{"node": "J1", "result": [{"pressure": 30}]}]})
+    monkeypatch.setattr(simulation, "run_simulation", run)
+    result = simulation.run_online_simulation("online", start, 60, output_node_ids=["J1"])
+    assert run.call_args.kwargs["modify_pattern_start_time"] == start.isoformat()
+    assert result["times"]["sample_times"] == [start.isoformat()]
+
+
 def test_mcp_online_simulation_does_not_copy_a_database(monkeypatch):
     from app.mcp.scenario_service import calculate_scenario
 

tests/unit/test_replay_online_simulation.py

@@ -0,0 +1,171 @@
+import asyncio
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+from uuid import uuid4
+
+import pytest
+from fastapi import HTTPException
+
+from app.mcp import query_service
+from app.mcp.analysis_models import ScenarioInput
+from app.mcp.catalog import TOOLS
+from app.services import simulation
+
+
+def parameters(**changes):
+    return ScenarioInput.model_validate({
+        "context_handle": "a" * 43, "start_time": "2026-07-16T08:00:00+08:00",
+        "duration": 60, "output_node_ids": ["J1"], **changes,
+    })
+
+
+def context(monkeypatch):
+    permissions = {"simulation.run", "simulation.view", "scada.view", "webgis.view", "optimization.run"}
+    monkeypatch.setattr(query_service, "permissions_for_context", lambda _: permissions)
+
+    @asynccontextmanager
+    async def routing():
+        yield
+
+    scope = SimpleNamespace(
+        connection=SimpleNamespace(commit=AsyncMock()), simulation_routing=routing,
+        project=SimpleNamespace(project_code="owned", project_role="member"),
+        user=SimpleNamespace(username="actor"),
+    )
+    binding = SimpleNamespace(
+        tool_cap=["run_simulation"], permission_cap=sorted(permissions),
+        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_coordinator_can_rebuild_online_baseline_without_live_scada(monkeypatch):
+    scope, binding = context(monkeypatch)
+    monkeypatch.setattr(simulation, "get_time", lambda _: {"HYDRAULIC TIMESTEP": "0:15:00"})
+    live_mappings = Mock(side_effect=AssertionError("Replay must not load live SCADA mappings"))
+    monkeypatch.setattr(simulation, "query_corresponding_element_id_and_query_id", live_mappings)
+    native = Mock(return_value={
+        "units": {"pressure": "MTR"},
+        "node_results": [{"node": "J1", "result": [{"pressure": 30}]}],
+    })
+    monkeypatch.setattr(simulation, "run_simulation", native)
+
+    result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["run_simulation"], parameters()))
+
+    assert result.origin == "synthetic"
+    assert any("不读取实时SCADA" in gap for gap in result.data_gaps)
+    assert result.data["result"]["persisted_baseline"] is True
+    assert result.data["result"]["node_results"][0]["result"] == [{"pressure": 30}]
+    native.assert_called_once()
+    call = native.call_args.kwargs
+    assert call["name"] == "owned" and call["simulation_type"] == "realtime"
+    assert call["modify_pattern_start_time"] == "2026-07-16T08:00:00+08:00"
+    assert all(not values for values in vars(call["scada_mappings"]).values())
+    live_mappings.assert_not_called()
+
+
+@pytest.mark.parametrize("problem", ["future_start", "reader", "permission", "tool_cap"])
+def test_replay_online_baseline_preserves_existing_authorization(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("simulation.run")
+    if problem == "tool_cap":
+        binding.tool_cap.clear()
+    with pytest.raises(HTTPException) as caught:
+        asyncio.run(query_service.authorize_query(scope, binding, TOOLS["run_simulation"], args))
+    assert caught.value.status_code == (422 if problem == "future_start" else 403)
+
+
+@pytest.mark.parametrize("name", ["analyze_valve_closure", "analyze_pressure_regulation"])
+def test_replay_can_compare_hydraulic_alternatives_with_model_only_boundaries(monkeypatch, name):
+    from app.infra.db.postgresql.scada import ScadaElementMappings
+    from app.services import simulation_scenarios as scenarios
+
+    scope, binding = context(monkeypatch)
+    binding.tool_cap = [name]
+    captures = []
+    monkeypatch.setattr(scenarios, "get_option", lambda _: {"UNITS": "LPS"})
+    monkeypatch.setattr(scenarios, "set_option", lambda *_: None)
+    monkeypatch.setattr(simulation, "query_corresponding_element_id_and_query_id",
+                        lambda _: ScadaElementMappings({}, {}, {"P1": "device"}, {}, {}, {}, {}))
+    monkeypatch.setattr(simulation, "run_simulation", lambda **kw: captures.append(kw) or {
+        "node_results": [{"node": "J1", "result": [{"pressure": 30}]}],
+    })
+    native_name = "valve_close_analysis" if name == "analyze_valve_closure" else "pressure_regulation"
+    native = getattr(scenarios, native_name).__wrapped__
+    monkeypatch.setattr(scenarios, native_name,
+                        lambda **kw: native(**kw, _temporary_project="tjw_tmp_owned"))
+    args = TOOLS[name].model.model_validate({
+        **parameters().model_dump(),
+        **({"valves": ["V1"]} if name == "analyze_valve_closure" else {"pump_control": {"P1": [0]}}),
+    })
+    result = asyncio.run(query_service.execute_query(scope, binding, TOOLS[name], args))
+    assert result.origin == "synthetic"
+    assert result.data["result"]["persisted_baseline"] is False
+    assert len(captures) == 1
+    assert captures[0]["name"] == "tjw_tmp_owned" and captures[0]["persist_results"] is False
+    assert all(not values for values in vars(captures[0]["scada_mappings"]).values())
+    if name == "analyze_valve_closure":
+        assert captures[0]["modify_valve_opening"] == {"V1": 0.0}
+    else:
+        assert captures[0]["modify_fixed_pump_pattern"] == {"P1": [0]}
+
+
+@pytest.mark.parametrize("name", [name for name, tool in TOOLS.items()
+                                  if tool.semantics == "asset" and not tool.mutating])
+def test_replay_allows_authorized_asset_queries_for_input_discovery(monkeypatch, name):
+    scope, binding = context(monkeypatch)
+    binding.tool_cap = [name]
+    # Authorization needs only the schema kind; execution validates each tool's inputs.
+    asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], SimpleNamespace(kind="node")))
+    binding.permission_cap = []
+    with pytest.raises(HTTPException) as caught:
+        asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], SimpleNamespace(kind="node")))
+    assert caught.value.status_code == 403
+
+
+@pytest.mark.parametrize("name,fields", [
+    ("query_model_baseline_series", {"element_type": "junction", "element_ids": ["J1"],
+                                    "start": "2026-07-16T07:45:00+08:00", "end": "2026-07-16T08:00:00+08:00"}),
+    ("get_model_baseline_at_time", {"element_type": "junction", "element_id": "J1", "time": "2026-07-16T08:00:00+08:00"}),
+    ("query_model_baseline_field", {"element_type": "junction", "field": "pressure", "time": "2026-07-16T08:00:00+08:00"}),
+])
+def test_replay_can_verify_baseline_without_reading_future_values(monkeypatch, name, fields):
+    scope, binding = context(monkeypatch)
+    binding.tool_cap = [name]
+    args = TOOLS[name].model.model_validate({"context_handle": "a" * 43, **fields})
+    asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], args))
+    future = parameters(start_time="2026-07-16T08:15:00+08:00").start_time
+    with pytest.raises(HTTPException) as caught:
+        asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], args.model_copy(
+            update={"end" if "end" in fields else "time": future},
+        )))
+    assert caught.value.status_code == 422
+
+
+@pytest.mark.parametrize("name", ["calculate_demand_allocation", "optimize_sensor_placement"])
+def test_replay_can_choose_current_model_design_calculations(monkeypatch, name):
+    scope, binding = context(monkeypatch)
+    binding.tool_cap = [name]
+    asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], SimpleNamespace()))
+    binding.purpose = "task_reader"
+    with pytest.raises(HTTPException) as caught:
+        asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], SimpleNamespace()))
+    assert caught.value.status_code == 403
+
+
+def test_replay_can_explore_contamination_without_requiring_map_generation(monkeypatch):
+    scope, binding = context(monkeypatch)
+    binding.tool_cap = ["run_contaminant_simulation"]
+    args = TOOLS["run_contaminant_simulation"].model.model_validate({
+        **parameters().model_dump(), "source": "J1", "concentration": 1,
+        "include_impact_coverage": False,
+    })
+    asyncio.run(query_service.authorize_query(scope, binding, TOOLS["run_contaminant_simulation"], args))