146 lines
4.9 KiB
Python
146 lines
4.9 KiB
Python
# ============================================================
|
|
# Mesh 人工审批恢复黄金测试
|
|
# 覆盖：批准后续跑依赖任务、驳回停止、非 Mesh 不命中、重复调用幂等。
|
|
# ============================================================
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from server.agent_core import mesh as mesh_module
|
|
|
|
|
|
@pytest.fixture()
|
|
def mesh_env(tmp_path, monkeypatch):
|
|
store = mesh_module.MeshStore(str(tmp_path / "agent_mesh.json"))
|
|
monkeypatch.setattr(mesh_module, "_MESH", store)
|
|
monkeypatch.setattr(mesh_module, "_DISPATCH_THREADS", {})
|
|
return store
|
|
|
|
|
|
def _approval_goal(store: mesh_module.MeshStore):
|
|
goal = store.create_goal(
|
|
"approval resume",
|
|
template=None,
|
|
tasks=[
|
|
{
|
|
"key": "analyze",
|
|
"title": "analyze",
|
|
"intent": "data.analyze",
|
|
"role": "data-analyst",
|
|
"dependsOn": [],
|
|
},
|
|
{
|
|
"key": "readiness",
|
|
"title": "readiness",
|
|
"intent": "readiness.query",
|
|
"role": "verifier",
|
|
"dependsOn": ["analyze"],
|
|
},
|
|
],
|
|
)
|
|
task = goal["tasks"][0]
|
|
task["status"] = "AWAITING_APPROVAL"
|
|
task["confirmationIds"] = ["confirm-approve"]
|
|
goal["status"] = "ATTENTION"
|
|
store.save()
|
|
return goal, task
|
|
|
|
|
|
def _join_dispatch(goal_id: str) -> None:
|
|
thread = mesh_module._DISPATCH_THREADS[goal_id]
|
|
thread.join(timeout=3)
|
|
assert not thread.is_alive(), "mesh dispatch thread did not finish"
|
|
|
|
|
|
def test_approved_confirmation_resumes_dependent_tasks(mesh_env, monkeypatch):
|
|
calls: list[str] = []
|
|
|
|
def fake_execute(session_id, intent, params, agent, **kwargs):
|
|
calls.append(intent)
|
|
return f"{intent} completed", []
|
|
|
|
monkeypatch.setattr(mesh_module, "_execute_intent", fake_execute)
|
|
goal, task = _approval_goal(mesh_env)
|
|
|
|
first = mesh_module.resume_goal_after_confirmation(
|
|
"confirm-approve", approve=True, actor="alice",
|
|
)
|
|
assert first["matched"] is True
|
|
assert first["resumed"] is True
|
|
assert first["started"] is True
|
|
assert first["alreadyRunning"] is False
|
|
assert task["status"] == "DONE"
|
|
assert task["confirmationIds"] == []
|
|
assert task["finishedAt"]
|
|
assert "人工已批准 confirm-approve by alice" in task["result"]
|
|
|
|
_join_dispatch(goal["id"])
|
|
assert calls == ["readiness.query"]
|
|
assert goal["tasks"][1]["status"] == "DONE"
|
|
assert goal["status"] == "ACHIEVED"
|
|
assert any(
|
|
message.get("kind") == "RESPONSE"
|
|
and "confirm:confirm-approve" in (message.get("evidenceRefs") or [])
|
|
for message in mesh_env.data["messages"]
|
|
)
|
|
|
|
first_thread = mesh_module._DISPATCH_THREADS[goal["id"]]
|
|
repeated = mesh_module.resume_goal_after_confirmation(
|
|
"confirm-approve", approve=True, actor="alice",
|
|
)
|
|
assert repeated["matched"] is True
|
|
assert repeated["resolved"] is True
|
|
assert repeated["resumed"] is False
|
|
assert repeated["reason"] == "ALREADY_RESOLVED"
|
|
assert mesh_module._DISPATCH_THREADS[goal["id"]] is first_thread
|
|
assert calls == ["readiness.query"]
|
|
|
|
|
|
def test_rejected_confirmation_stops_dispatch(mesh_env, monkeypatch):
|
|
calls: list[str] = []
|
|
|
|
def fake_execute(session_id, intent, params, agent, **kwargs):
|
|
calls.append(intent)
|
|
return f"{intent} completed", []
|
|
|
|
monkeypatch.setattr(mesh_module, "_execute_intent", fake_execute)
|
|
goal, task = _approval_goal(mesh_env)
|
|
|
|
result = mesh_module.resume_goal_after_confirmation(
|
|
"confirm-approve", approve=False, actor="bob",
|
|
)
|
|
assert result["matched"] is True
|
|
assert result["resumed"] is False
|
|
assert task["status"] == "FAILED"
|
|
assert task["confirmationIds"] == []
|
|
assert "人工驳回 confirm-approve by bob" in task["error"]
|
|
assert goal["tasks"][1]["status"] == "PENDING"
|
|
assert goal["status"] == "ATTENTION"
|
|
assert mesh_module._DISPATCH_THREADS == {}
|
|
assert calls == []
|
|
assert any(
|
|
message.get("kind") == "ALERT"
|
|
and "confirm:confirm-approve" in (message.get("evidenceRefs") or [])
|
|
for message in mesh_env.data["messages"]
|
|
)
|
|
|
|
|
|
def test_unrelated_confirmation_is_ignored(mesh_env):
|
|
result = mesh_module.resume_goal_after_confirmation(
|
|
"normal-p2-confirm", approve=True, actor="alice",
|
|
)
|
|
assert result == {"matched": False, "resolved": False, "resumed": False}
|
|
assert mesh_module._DISPATCH_THREADS == {}
|
|
|
|
|
|
def test_sop_stage_request_rejects_query_only_payload():
|
|
from server.gateway.app import SopStageRequest
|
|
|
|
with pytest.raises(ValidationError):
|
|
SopStageRequest(sessionId="session-sop", query="换线")
|
|
|
|
structured = SopStageRequest(sessionId="session-sop", assetId="sop-asset-1")
|
|
assert structured.assetId == "sop-asset-1"
|
|
assert structured.pack is None
|