988 lines
30 KiB
Python
988 lines
30 KiB
Python
from __future__ import annotations
|
||
|
||
import importlib
|
||
import json
|
||
|
||
import pytest
|
||
|
||
from apps.web.vue3 import main as WEB_MAIN
|
||
from gfa_core.mas.agent_state_repository import SQLiteAgentStateRepository
|
||
from gfa_core.mas.persistent_conversation import PersistentConversationStore
|
||
|
||
|
||
@pytest.fixture
|
||
def web_main(monkeypatch):
|
||
monkeypatch.setattr(WEB_MAIN, "llm_manager", object())
|
||
return WEB_MAIN
|
||
|
||
|
||
def _install_router(monkeypatch, outcome, captured):
|
||
router_module = importlib.import_module("gfa_core.mas.agents.router")
|
||
|
||
async def fake_execute(_self, state, message):
|
||
captured.append(message.payload)
|
||
state.results["router_outcome"] = outcome
|
||
return state
|
||
|
||
monkeypatch.setattr(router_module.RouterAgent, "execute", fake_execute)
|
||
|
||
|
||
def test_agent_conversation_store_initializes_from_env(
|
||
monkeypatch,
|
||
web_main,
|
||
tmp_path,
|
||
):
|
||
database_path = tmp_path / "state" / "agent-state.sqlite3"
|
||
monkeypatch.setenv("PLANTAI_AGENT_STATE_DB", str(database_path))
|
||
monkeypatch.setattr(web_main, "agent_conversation_store", None, raising=False)
|
||
|
||
web_main.initialize_agent_conversation_store()
|
||
|
||
assert database_path.is_file()
|
||
assert isinstance(web_main.agent_conversation_store, PersistentConversationStore)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.parametrize(
|
||
("action", "task_type", "expected_intent"),
|
||
[
|
||
("reply", "faq", None),
|
||
("clarify", "clarification", None),
|
||
("show_result", "result_view", None),
|
||
("propose_analysis", "analysis", {"stages": ["motif"], "stage": "motif"}),
|
||
(
|
||
"propose_utility",
|
||
"utility",
|
||
{
|
||
"taskType": "utility",
|
||
"utility_tool": "fasta_header_clean",
|
||
"stages": ["sequence_extract_by_id"],
|
||
"params": {"sequenceExtractFasta": "test.fa"},
|
||
},
|
||
),
|
||
],
|
||
)
|
||
async def test_agent_chat_maps_router_actions_and_passes_context(
|
||
monkeypatch, web_main, action, task_type, expected_intent
|
||
):
|
||
captured = []
|
||
outcome = {
|
||
"status": "ok",
|
||
"action": action,
|
||
"message": f"message for {action}",
|
||
"intent": expected_intent,
|
||
"sources": [{"path": "final/result.tsv"}],
|
||
"tool_trace": [{"name": "inspect_analysis_results", "result": "large"}] * 12,
|
||
}
|
||
_install_router(monkeypatch, outcome, captured)
|
||
history = [{"role": "user", "content": "上一轮"}]
|
||
|
||
result = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(
|
||
message="请接着处理",
|
||
session_id="session-7",
|
||
job_id="job-7",
|
||
history=history,
|
||
)
|
||
)
|
||
|
||
assert captured == [
|
||
{
|
||
"query": "请接着处理",
|
||
"history": history,
|
||
"session_id": "session-7",
|
||
"job_id": "job-7",
|
||
"language": "zh",
|
||
}
|
||
]
|
||
assert result["success"] is True
|
||
assert result["status"] == "ok"
|
||
assert result["action"] == action
|
||
assert result["intent"]["taskType"] == task_type
|
||
assert result["intent"]["message"] == f"message for {action}"
|
||
assert result["sources"] == [{"path": "final/result.tsv"}]
|
||
assert len(result["tool_trace"]) <= 5
|
||
assert all(len(item["result"]) <= 1000 for item in result["tool_trace"])
|
||
if action in {"propose_analysis", "propose_utility"}:
|
||
assert result["intent"]["stages"] == expected_intent["stages"]
|
||
else:
|
||
assert result["intent"]["stages"] == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_agent_chat_keeps_english_language_separate_from_business_query(
|
||
monkeypatch,
|
||
web_main,
|
||
):
|
||
captured = []
|
||
_install_router(
|
||
monkeypatch,
|
||
{
|
||
"status": "ok",
|
||
"action": "reply",
|
||
"message": "I can help with that.",
|
||
"intent": None,
|
||
"sources": [],
|
||
"tool_trace": [],
|
||
},
|
||
captured,
|
||
)
|
||
request = web_main.AgentChatRequest(
|
||
message="执行接下来的分析",
|
||
language="en",
|
||
session_id="english-business-query",
|
||
job_id="job-english-business-query",
|
||
history=[],
|
||
)
|
||
|
||
result = await web_main.agent_chat(request)
|
||
|
||
assert result["intent"]["message"] == "I can help with that."
|
||
assert request.message == "执行接下来的分析"
|
||
assert captured[0]["query"] == "执行接下来的分析"
|
||
assert captured[0]["language"] == "en"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_agent_chat_persists_router_proposal_without_analysis_intent(
|
||
monkeypatch,
|
||
web_main,
|
||
tmp_path,
|
||
):
|
||
repository = SQLiteAgentStateRepository(tmp_path / "agent-state.sqlite3")
|
||
repository.initialize()
|
||
store = PersistentConversationStore(repository)
|
||
monkeypatch.setattr(web_main, "agent_conversation_store", store)
|
||
captured = []
|
||
proposal = {
|
||
"proposal_id": "proposal-123",
|
||
"request": "执行 Ft 的数据准备",
|
||
"original_request": "执行数据准备",
|
||
"subject": "执行 Ft 的数据准备",
|
||
}
|
||
_install_router(
|
||
monkeypatch,
|
||
{
|
||
"status": "ok",
|
||
"action": "clarify",
|
||
"message": "我理解为只执行 Ft 数据准备，是否确认？",
|
||
"intent": None,
|
||
"proposal": proposal,
|
||
"sources": [],
|
||
"tool_trace": [],
|
||
},
|
||
captured,
|
||
)
|
||
|
||
result = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(
|
||
message="执行 Ft 的数据准备",
|
||
session_id="proposal-session",
|
||
job_id="job-ft",
|
||
history=[],
|
||
)
|
||
)
|
||
|
||
assert result["action"] == "clarify"
|
||
assert result["intent"]["stages"] == []
|
||
ledger = repository.get_goal_ledger(result["session_id"])
|
||
assert ledger is not None
|
||
assert ledger.user_terms["pending_proposal"] == {
|
||
**proposal,
|
||
"job_id": "job-ft",
|
||
}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_agent_chat_exposes_and_persists_current_recommendation(
|
||
monkeypatch,
|
||
web_main,
|
||
tmp_path,
|
||
):
|
||
repository = SQLiteAgentStateRepository(tmp_path / "agent-state.sqlite3")
|
||
repository.initialize()
|
||
store = PersistentConversationStore(repository)
|
||
monkeypatch.setattr(web_main, "agent_conversation_store", store)
|
||
recommendation = {
|
||
"recommendation_id": "rec-build-subfamily-evidence",
|
||
"job_id": "job-recommendation",
|
||
"workflow_id": "gene_family",
|
||
"transition_id": "build_subfamily_evidence",
|
||
"from_state_id": "family_identified",
|
||
"to_state_id": "subfamily_evidence_ready",
|
||
"target_stage": "combo",
|
||
"authorized_stages": [
|
||
"domain",
|
||
"motif",
|
||
"phylogeny",
|
||
"structure",
|
||
"combo",
|
||
],
|
||
"evidence_digest": "a" * 64,
|
||
"family": "SPL",
|
||
"scope_id": "SPL.target-Ft.target_only",
|
||
}
|
||
_install_router(
|
||
monkeypatch,
|
||
{
|
||
"status": "ok",
|
||
"action": "show_result",
|
||
"message": "下一步是构建亚家族证据。",
|
||
"intent": None,
|
||
"recommendation": recommendation,
|
||
"sources": [],
|
||
"tool_trace": [],
|
||
},
|
||
[],
|
||
)
|
||
|
||
result = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(
|
||
message="接下来的分析有哪些？",
|
||
session_id="recommendation-session",
|
||
job_id="job-recommendation",
|
||
history=[],
|
||
)
|
||
)
|
||
|
||
assert result["action"] == "show_result"
|
||
assert result["recommendation"] == recommendation
|
||
ledger = repository.get_goal_ledger(result["session_id"])
|
||
assert ledger is not None
|
||
assert (
|
||
ledger.model_dump(mode="json")["user_terms"]["current_recommendation"]
|
||
== recommendation
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_agent_chat_exposes_confirmed_request_for_original_planner(
|
||
monkeypatch,
|
||
web_main,
|
||
):
|
||
captured = []
|
||
proposal = {
|
||
"proposal_id": "proposal-next",
|
||
"request": "我已确认 config.xlsx，接下来应该做什么分析",
|
||
"original_request": "执行kaks分析",
|
||
"subject": "启动苦荞SPL基因家族分析",
|
||
}
|
||
_install_router(
|
||
monkeypatch,
|
||
{
|
||
"status": "ok",
|
||
"action": "propose_analysis",
|
||
"message": "已交给现有分析系统识别。",
|
||
"intent": {"stages": ["genome_prepare"]},
|
||
"proposal": proposal,
|
||
"sources": [],
|
||
"tool_trace": [],
|
||
},
|
||
captured,
|
||
)
|
||
|
||
result = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(
|
||
message="确认",
|
||
session_id="session-next",
|
||
job_id="job-next",
|
||
history=[],
|
||
)
|
||
)
|
||
|
||
assert result["action"] == "propose_analysis"
|
||
assert result["handoff_request"] == proposal["original_request"]
|
||
assert "尚未执行" in result["intent"]["message"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_agent_chat_rejects_job_change_before_analysis_handoff(
|
||
monkeypatch,
|
||
web_main,
|
||
tmp_path,
|
||
):
|
||
from fastapi import HTTPException
|
||
|
||
repository = SQLiteAgentStateRepository(tmp_path / "agent-state.sqlite3")
|
||
repository.initialize()
|
||
store = PersistentConversationStore(repository)
|
||
monkeypatch.setattr(web_main, "agent_conversation_store", store)
|
||
turn = store.begin_turn(
|
||
session_id="bound-session",
|
||
job_id="job-ft",
|
||
query="分析 Ft.SPL",
|
||
client_history=[],
|
||
)
|
||
|
||
with pytest.raises(HTTPException) as exc_info:
|
||
await web_main.agent_chat(
|
||
web_main.AgentChatRequest(
|
||
message="分析 Os.SPL",
|
||
session_id=turn.conversation_id,
|
||
job_id="job-os",
|
||
history=[],
|
||
)
|
||
)
|
||
|
||
assert exc_info.value.status_code == 409
|
||
assert exc_info.value.detail == "conversation job binding conflict"
|
||
|
||
|
||
def test_main_agent_normalization_uses_confirmed_config_family_as_authority(
|
||
monkeypatch,
|
||
web_main,
|
||
):
|
||
monkeypatch.setattr(
|
||
web_main,
|
||
"_confirmed_config_species_scope",
|
||
lambda _job_id: {
|
||
"targetSpecies": ["Ft"],
|
||
"referenceSpecies": ["At"],
|
||
"comparativeSpecies": [],
|
||
},
|
||
)
|
||
monkeypatch.setattr(
|
||
web_main,
|
||
"_infer_family_from_config_or_final",
|
||
lambda _job_id: "SPL",
|
||
)
|
||
|
||
normalized = web_main._normalize_main_agent_analysis_intent(
|
||
{
|
||
"stages": ["genome_prepare"],
|
||
"gene_family": "xlsx，接下来应该做什么分析",
|
||
"prefix": "Ft.xlsx，接下来应该做什么分析",
|
||
},
|
||
job_id="job-ft",
|
||
)
|
||
|
||
assert normalized["gene_family"] == "SPL"
|
||
assert normalized["prefix"] == "Ft.SPL"
|
||
assert normalized["params"]["geneFamily"] == "SPL"
|
||
|
||
|
||
def test_main_agent_normalization_uses_hydrated_job_identity_without_config(
|
||
monkeypatch,
|
||
web_main,
|
||
):
|
||
monkeypatch.setattr(web_main, "_confirmed_config_species_scope", lambda _job: {})
|
||
|
||
normalized = web_main._normalize_main_agent_analysis_intent(
|
||
{"stages": ["identify"], "params": {}},
|
||
job_id="job-hkl",
|
||
context_terms={
|
||
"gene_family": "HKL",
|
||
"target_species": ["Sv"],
|
||
"reference_species": ["At"],
|
||
},
|
||
)
|
||
|
||
assert normalized["gene_family"] == "HKL"
|
||
assert normalized["prefix"] == "Sv.HKL"
|
||
assert normalized["target_species"] == ["Sv"]
|
||
assert normalized["reference_species"] == ["At"]
|
||
assert normalized["stages"] == ["identify"]
|
||
|
||
|
||
def test_rnaseq_family_member_normalization_keeps_species_prefix(
|
||
monkeypatch,
|
||
web_main,
|
||
):
|
||
monkeypatch.setattr(web_main, "_confirmed_config_species_scope", lambda _job: {})
|
||
|
||
normalized = web_main._normalize_main_agent_analysis_intent(
|
||
{
|
||
"gene_family": "HKL",
|
||
"rna_route": "family_member",
|
||
"stages": ["rnaseq_de"],
|
||
"params": {"analysisRoute": "family_member"},
|
||
},
|
||
job_id="job-os",
|
||
context_terms={"target_species": ["Os"]},
|
||
)
|
||
|
||
assert normalized["prefix"] == "Os"
|
||
assert normalized["params"]["prefix"] == "Os"
|
||
assert normalized["params"]["geneFamily"] == "HKL"
|
||
|
||
|
||
def test_current_job_hydration_fills_identity_without_planning_stages(
|
||
monkeypatch,
|
||
web_main,
|
||
tmp_path,
|
||
):
|
||
final_dir = tmp_path / "workspace" / "job-hkl" / "final"
|
||
final_dir.mkdir(parents=True)
|
||
(final_dir / "Sv.HKL.subfamily_confirm.manifest.json").write_text(
|
||
json.dumps(
|
||
{
|
||
"stage": "subfamily_confirm",
|
||
"status": "success",
|
||
"prefix": "Sv.HKL",
|
||
"target_species": ["Sv"],
|
||
"reference_species": ["At"],
|
||
"analysis_route": "confirmed_subfamily",
|
||
}
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
(final_dir / "Sv.HKL.ppi.manifest.json").write_text(
|
||
json.dumps(
|
||
{
|
||
"stage": "ppi",
|
||
"status": "success",
|
||
"prefix": "Sv.HKL",
|
||
"target_species": ["Sv"],
|
||
"reference_species": ["string_taxon_4555"],
|
||
}
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
monkeypatch.setattr(web_main, "_confirmed_config_species_scope", lambda _job: {})
|
||
monkeypatch.setattr(web_main, "_infer_family_from_config_or_final", lambda _job: "HKL")
|
||
|
||
hydrated = web_main._hydrate_goal_terms_from_current_job(
|
||
{"reference_species": ["ExplicitRef"]},
|
||
"job-hkl",
|
||
)
|
||
|
||
assert hydrated["gene_family"] == "HKL"
|
||
assert hydrated["target_species"] == ["Sv"]
|
||
assert hydrated["reference_species"] == ["ExplicitRef"]
|
||
assert "stages" not in hydrated
|
||
assert "requested_stages" not in hydrated
|
||
|
||
canonical = web_main._hydrate_goal_terms_from_current_job({}, "job-hkl")
|
||
assert canonical["reference_species"] == ["At"]
|
||
|
||
|
||
def test_whole_genome_rna_job_does_not_inherit_later_gene_family_context(
|
||
monkeypatch,
|
||
web_main,
|
||
tmp_path,
|
||
):
|
||
final_dir = tmp_path / "workspace" / "job-rna" / "final"
|
||
final_dir.mkdir(parents=True)
|
||
for name, payload in {
|
||
"Os.HKL.subfamily_confirm.manifest.json": {
|
||
"stage": "subfamily_confirm",
|
||
"status": "success",
|
||
"prefix": "Os.HKL",
|
||
"target_species": ["Os"],
|
||
"reference_species": ["At"],
|
||
},
|
||
"Os.rnaseq_de.manifest.json": {
|
||
"stage": "rnaseq_de",
|
||
"status": "success",
|
||
"prefix": "Os",
|
||
"foreground_mode": "global",
|
||
},
|
||
}.items():
|
||
(final_dir / name).write_text(json.dumps(payload), encoding="utf-8")
|
||
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
monkeypatch.setattr(web_main, "_confirmed_config_species_scope", lambda _job: {})
|
||
monkeypatch.setattr(web_main, "_infer_family_from_config_or_final", lambda _job: "HKL")
|
||
|
||
hydrated = web_main._hydrate_goal_terms_from_current_job({}, "job-rna")
|
||
|
||
assert hydrated["rna_route"] == "whole_genome"
|
||
assert hydrated["target_species"] == ["Os"]
|
||
assert "gene_family" not in hydrated
|
||
assert "reference_species" not in hydrated
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_agent_chat_provider_failure_has_no_faq_fallback(monkeypatch, web_main):
|
||
captured = []
|
||
_install_router(
|
||
monkeypatch,
|
||
{
|
||
"status": "agent_unavailable",
|
||
"action": None,
|
||
"message": "The Router model is currently unavailable.",
|
||
"intent": None,
|
||
"sources": [],
|
||
"tool_trace": [],
|
||
},
|
||
captured,
|
||
)
|
||
|
||
result = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(message="请做 motif 分析", job_id="job-7")
|
||
)
|
||
|
||
assert result["success"] is False
|
||
assert result["status"] == "agent_unavailable"
|
||
assert result["action"] is None
|
||
assert result["intent"] is None
|
||
assert "知识库" not in result["message"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_agent_chat_manager_initialization_failure_is_agent_unavailable(
|
||
monkeypatch, web_main
|
||
):
|
||
from gfa_core.mas.config import config as mas_config
|
||
|
||
monkeypatch.setattr(web_main, "llm_manager", None)
|
||
monkeypatch.setattr(
|
||
mas_config,
|
||
"load_llm_config",
|
||
lambda: (_ for _ in ()).throw(RuntimeError("missing provider config")),
|
||
)
|
||
|
||
result = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(message="请做 motif 分析", job_id="job-7")
|
||
)
|
||
|
||
assert result["success"] is False
|
||
assert result["status"] == "agent_unavailable"
|
||
assert result["action"] is None
|
||
assert result["intent"] is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_only_propose_analysis_can_return_plannable_intent(monkeypatch, web_main):
|
||
captured = []
|
||
_install_router(
|
||
monkeypatch,
|
||
{
|
||
"status": "ok",
|
||
"action": "reply",
|
||
"message": "这是解释，不是分析请求。",
|
||
"intent": {"stages": ["phylogeny"], "stage": "phylogeny"},
|
||
"sources": [],
|
||
"tool_trace": [],
|
||
},
|
||
captured,
|
||
)
|
||
|
||
result = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(message="这个树为什么这样", job_id="job-7")
|
||
)
|
||
|
||
assert result["action"] == "reply"
|
||
assert result["intent"]["taskType"] == "faq"
|
||
assert result["intent"]["stages"] == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_proposed_analysis_normalizes_scope_from_the_current_user_action(
|
||
monkeypatch,
|
||
web_main,
|
||
):
|
||
captured = []
|
||
_install_router(
|
||
monkeypatch,
|
||
{
|
||
"status": "ok",
|
||
"action": "propose_analysis",
|
||
"message": "已识别当前数据准备操作。",
|
||
"intent": {
|
||
"gene_family": "SPL",
|
||
"target_species": ["At"],
|
||
"reference_species": [],
|
||
"stages": ["genome_prepare"],
|
||
"primary_stage": "genome_prepare",
|
||
"params": {
|
||
"targetSpecies": ["Ft"],
|
||
"referenceSpecies": [],
|
||
"comparativeSpecies": [],
|
||
"scopeMode": "target_only",
|
||
},
|
||
},
|
||
"sources": [],
|
||
"tool_trace": [],
|
||
},
|
||
captured,
|
||
)
|
||
|
||
result = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(
|
||
message="执行Ft的数据准备",
|
||
job_id="job-ft",
|
||
)
|
||
)
|
||
|
||
assert result["intent"]["target_species"] == ["Ft"]
|
||
assert result["intent"]["reference_species"] == []
|
||
assert result["intent"]["params"]["targetSpecies"] == ["Ft"]
|
||
assert result["intent"]["prefix"] == "Ft.SPL"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_proposed_analysis_uses_intent_params_as_the_scope_authority(
|
||
monkeypatch,
|
||
web_main,
|
||
):
|
||
captured = []
|
||
_install_router(
|
||
monkeypatch,
|
||
{
|
||
"status": "ok",
|
||
"action": "propose_analysis",
|
||
"message": "已识别目标与参考物种。",
|
||
"intent": {
|
||
"gene_family": "SPL",
|
||
"species": "At",
|
||
"target_species": ["At"],
|
||
"reference_species": [],
|
||
"stages": ["identify"],
|
||
"primary_stage": "identify",
|
||
"params": {
|
||
"targetSpecies": ["Ft"],
|
||
"referenceSpecies": ["At"],
|
||
"comparativeSpecies": [],
|
||
"scopeMode": "target_plus_reference",
|
||
},
|
||
},
|
||
"sources": [],
|
||
"tool_trace": [],
|
||
},
|
||
captured,
|
||
)
|
||
|
||
result = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(
|
||
message=(
|
||
"我想开展苦荞的SPL基因家族分析，"
|
||
"Ft是目标物种，At是参考物种"
|
||
),
|
||
job_id="job-ft",
|
||
)
|
||
)
|
||
|
||
assert result["intent"]["target_species"] == ["Ft"]
|
||
assert result["intent"]["reference_species"] == ["At"]
|
||
assert result["intent"]["params"]["targetSpecies"] == ["Ft"]
|
||
assert result["intent"]["params"]["referenceSpecies"] == ["At"]
|
||
assert result["intent"]["prefix"] == "Ft.SPL"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_confirmed_config_species_roles_override_polluted_intent(
|
||
monkeypatch,
|
||
web_main,
|
||
tmp_path,
|
||
):
|
||
import pandas as pd
|
||
from gfa_core.mas import confirmation_snapshot
|
||
|
||
input_root = tmp_path / "input"
|
||
job_dir = input_root / "job-ft"
|
||
job_dir.mkdir(parents=True)
|
||
pd.DataFrame(
|
||
[
|
||
{"Species_Code": "Ft", "Species_Role": "target", "Ref": False},
|
||
{"Species_Code": "At", "Species_Role": "reference", "Ref": True},
|
||
]
|
||
).to_excel(job_dir / "config.xlsx", index=False)
|
||
monkeypatch.setattr(web_main, "INPUT_ROOT", input_root)
|
||
monkeypatch.setattr(
|
||
confirmation_snapshot,
|
||
"build_confirmation_snapshot",
|
||
lambda *_args, **_kwargs: {
|
||
"config_confirmed": True,
|
||
"samples_confirmed": False,
|
||
"subfamily_confirmed": False,
|
||
"config": {"artifact_path": str(job_dir / "config.xlsx")},
|
||
"samples": None,
|
||
"subfamily": None,
|
||
"errors": (),
|
||
},
|
||
)
|
||
captured = []
|
||
_install_router(
|
||
monkeypatch,
|
||
{
|
||
"status": "ok",
|
||
"action": "propose_analysis",
|
||
"message": "已识别目标与参考物种。",
|
||
"intent": {
|
||
"gene_family": "SPL",
|
||
"species": "unknown",
|
||
"target_species": [],
|
||
"reference_species": ["Ft", "At"],
|
||
"stages": ["identify"],
|
||
"primary_stage": "identify",
|
||
"params": {
|
||
"targetSpecies": [],
|
||
"referenceSpecies": ["Ft", "At"],
|
||
"scopeMode": "target_plus_reference",
|
||
},
|
||
},
|
||
"sources": [],
|
||
"tool_trace": [],
|
||
},
|
||
captured,
|
||
)
|
||
|
||
result = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(
|
||
message=(
|
||
"苦荞数据在 /data/ref_data/Tartary_genome，"
|
||
"拟南芥数据在 /data/ref_data/ath"
|
||
),
|
||
job_id="job-ft",
|
||
)
|
||
)
|
||
|
||
assert result["intent"]["target_species"] == ["Ft"]
|
||
assert result["intent"]["reference_species"] == ["At"]
|
||
assert result["intent"]["params"]["targetSpecies"] == ["Ft"]
|
||
assert result["intent"]["params"]["referenceSpecies"] == ["At"]
|
||
assert result["intent"]["prefix"] == "Ft.SPL"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_explicit_species_role_conflict_returns_actionable_clarification(
|
||
monkeypatch,
|
||
web_main,
|
||
tmp_path,
|
||
):
|
||
import pandas as pd
|
||
from gfa_core.mas import confirmation_snapshot
|
||
|
||
input_root = tmp_path / "input"
|
||
job_dir = input_root / "job-os"
|
||
job_dir.mkdir(parents=True)
|
||
pd.DataFrame(
|
||
[
|
||
{"Species_Code": "Os", "Species_Role": "target", "Ref": False},
|
||
{"Species_Code": "At", "Species_Role": "reference", "Ref": True},
|
||
]
|
||
).to_excel(job_dir / "config.xlsx", index=False)
|
||
monkeypatch.setattr(web_main, "INPUT_ROOT", input_root)
|
||
monkeypatch.setattr(
|
||
confirmation_snapshot,
|
||
"build_confirmation_snapshot",
|
||
lambda *_args, **_kwargs: {
|
||
"config_confirmed": True,
|
||
"samples_confirmed": False,
|
||
"subfamily_confirmed": False,
|
||
"config": {"artifact_path": str(job_dir / "config.xlsx")},
|
||
"samples": None,
|
||
"subfamily": None,
|
||
"errors": (),
|
||
},
|
||
)
|
||
captured = []
|
||
_install_router(
|
||
monkeypatch,
|
||
{
|
||
"status": "ok",
|
||
"action": "propose_analysis",
|
||
"message": "已识别分析需求。",
|
||
"intent": {
|
||
"gene_family": "PEBP",
|
||
"target_species": ["At"],
|
||
"reference_species": ["Os"],
|
||
"stages": ["identify"],
|
||
"primary_stage": "identify",
|
||
"params": {
|
||
"targetSpecies": ["At"],
|
||
"referenceSpecies": ["Os"],
|
||
},
|
||
},
|
||
"sources": [],
|
||
"tool_trace": [],
|
||
},
|
||
captured,
|
||
)
|
||
|
||
result = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(
|
||
message="At是目标物种，Os是参考物种，请执行PEBP基因家族分析",
|
||
job_id="job-os",
|
||
)
|
||
)
|
||
|
||
assert result["action"] == "clarify"
|
||
assert result["intent"]["taskType"] == "clarification"
|
||
assert "At" in result["intent"]["message"]
|
||
assert "Os" in result["intent"]["message"]
|
||
assert "打开 config.xlsx" in result["intent"]["message"]
|
||
assert result["intent"]["stages"] == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.parametrize(
|
||
"candidate",
|
||
[
|
||
None,
|
||
{},
|
||
[],
|
||
"not-an-intent",
|
||
{"foo": "bar"},
|
||
{"stages": []},
|
||
{"stages": [""]},
|
||
{"stages": "motif"},
|
||
],
|
||
)
|
||
async def test_propose_analysis_without_verified_intent_fails_closed(
|
||
monkeypatch, web_main, candidate
|
||
):
|
||
captured = []
|
||
_install_router(
|
||
monkeypatch,
|
||
{
|
||
"status": "ok",
|
||
"action": "propose_analysis",
|
||
"message": "分析建议",
|
||
"intent": candidate,
|
||
"sources": [],
|
||
"tool_trace": [],
|
||
},
|
||
captured,
|
||
)
|
||
|
||
result = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(message="继续分析", job_id="job-7")
|
||
)
|
||
|
||
assert result["success"] is False
|
||
assert result["status"] == "invalid_response"
|
||
assert result["action"] is None
|
||
assert result["intent"] is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.parametrize("query", ["那两个", "继续后面的", "刚才的结果"])
|
||
async def test_coreference_journeys_reach_router_with_prior_history(
|
||
monkeypatch, web_main, query
|
||
):
|
||
captured = []
|
||
_install_router(
|
||
monkeypatch,
|
||
{
|
||
"status": "ok",
|
||
"action": "clarify",
|
||
"message": "请确认你指的是哪一项。",
|
||
"intent": None,
|
||
"sources": [],
|
||
"tool_trace": [],
|
||
},
|
||
captured,
|
||
)
|
||
history = [
|
||
{"role": "user", "content": "比较 motif 和 domain"},
|
||
{"role": "assistant", "content": "我找到了两个结果。"},
|
||
]
|
||
|
||
result = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(message=query, history=history, job_id="job-7")
|
||
)
|
||
|
||
assert captured[0]["query"] == query
|
||
assert captured[0]["history"] == history
|
||
assert result["action"] == "clarify"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_agent_chat_uses_persisted_history_and_goal_ledger(
|
||
monkeypatch,
|
||
web_main,
|
||
tmp_path,
|
||
):
|
||
repository = SQLiteAgentStateRepository(tmp_path / "agent-state.sqlite3")
|
||
repository.initialize()
|
||
store = PersistentConversationStore(repository)
|
||
monkeypatch.setattr(web_main, "agent_conversation_store", store, raising=False)
|
||
captured = []
|
||
_install_router(
|
||
monkeypatch,
|
||
{
|
||
"status": "ok",
|
||
"action": "propose_analysis",
|
||
"message": "已记录初始目标。",
|
||
"intent": {"stages": ["motif"], "stage": "motif"},
|
||
"sources": [],
|
||
"tool_trace": [],
|
||
"interaction_kind": "independent",
|
||
},
|
||
captured,
|
||
)
|
||
first = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(
|
||
message="进行 motif 分析",
|
||
session_id="legacy-session",
|
||
job_id="job-ft",
|
||
)
|
||
)
|
||
_install_router(
|
||
monkeypatch,
|
||
{
|
||
"status": "ok",
|
||
"action": "propose_analysis",
|
||
"message": "已按修正后的目标处理。",
|
||
"intent": {"stages": ["domain"], "stage": "domain"},
|
||
"sources": [],
|
||
"tool_trace": [],
|
||
"interaction_kind": "correction",
|
||
},
|
||
captured,
|
||
)
|
||
|
||
second = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(
|
||
message="不是 motif，改成 domain",
|
||
session_id=first["session_id"],
|
||
job_id="",
|
||
history=[{"role": "user", "content": "客户端伪造历史"}],
|
||
)
|
||
)
|
||
|
||
assert captured[1]["job_id"] == "job-ft"
|
||
assert captured[1]["history"] == [
|
||
{"role": "user", "content": "进行 motif 分析"},
|
||
{"role": "assistant", "content": "已记录初始目标。"},
|
||
]
|
||
assert captured[1]["goal_ledger"].active_goal == "进行 motif 分析"
|
||
ledger = repository.get_goal_ledger(second["session_id"])
|
||
assert ledger is not None
|
||
assert ledger.active_goal == "不是 motif，改成 domain"
|
||
assert ledger.corrections == ("不是 motif，改成 domain",)
|
||
assert [item.content for item in repository.list_messages(second["session_id"])] == [
|
||
"进行 motif 分析",
|
||
"已记录初始目标。",
|
||
"不是 motif，改成 domain",
|
||
"已按修正后的目标处理。",
|
||
]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_deterministic_utility_turn_uses_same_persistent_conversation(
|
||
monkeypatch,
|
||
web_main,
|
||
tmp_path,
|
||
):
|
||
repository = SQLiteAgentStateRepository(tmp_path / "agent-state.sqlite3")
|
||
repository.initialize()
|
||
store = PersistentConversationStore(repository)
|
||
monkeypatch.setattr(web_main, "agent_conversation_store", store, raising=False)
|
||
monkeypatch.setattr(
|
||
web_main.extension_stage_agent,
|
||
"list_extensions",
|
||
lambda include_drafts=False: [],
|
||
)
|
||
message = "请从 proteins.fa 按 ids.txt 提取蛋白序列"
|
||
|
||
result = await web_main.agent_chat(
|
||
web_main.AgentChatRequest(
|
||
message=message,
|
||
session_id="browser-session",
|
||
job_id="job-ft",
|
||
)
|
||
)
|
||
|
||
assert result["action"] == "propose_utility"
|
||
assert result["status"] == "ok"
|
||
assert [item.content for item in repository.list_messages(result["session_id"])] == [
|
||
message,
|
||
result["intent"]["message"],
|
||
]
|
||
ledger = repository.get_goal_ledger(result["session_id"])
|
||
assert ledger is not None
|
||
assert ledger.active_goal == message
|