From 35fbee8c853dbfab44b1127278d31a93c6f96122 Mon Sep 17 00:00:00 2001 From: potaser Date: Wed, 5 Aug 2026 15:42:32 +0800 Subject: [PATCH] fix: preserve confirmed species roles in agent planning --- apps/web/vue3/main.py | 376 ++++++-- gfa_core/mas/utils/intent_parser.py | 44 + tests/unit/test_agent_chat_actions.py | 888 +++++++++++++++++++ tests/unit/test_intent_species_roles.py | 32 + tests/unit/test_subfamily_prediction_plan.py | 176 ++++ 5 files changed, 1465 insertions(+), 51 deletions(-) create mode 100644 tests/unit/test_agent_chat_actions.py create mode 100644 tests/unit/test_intent_species_roles.py diff --git a/apps/web/vue3/main.py b/apps/web/vue3/main.py index 11a08cfba..d8499c7c5 100644 --- a/apps/web/vue3/main.py +++ b/apps/web/vue3/main.py @@ -711,6 +711,211 @@ def _confirmed_config_species_scope(job_id: str) -> Dict[str, list[str]]: return _configured_species_scope(job_id) +def _reconcile_confirmed_gene_family_species( + job_id: str, + message: str, + intent: Optional[Dict[str, Any]] = None, +) -> Dict[str, Any]: + """Keep confirmed config roles separate from the current run selection.""" + config_scope = _confirmed_config_species_scope(job_id) + if not config_scope: + return {"applicable": False, "conflicts": []} + + from gfa_core.mas.utils.intent_parser import parse_explicit_species_role_mentions + + role_keys = { + "targetSpecies": "target", + "referenceSpecies": "reference", + "comparativeSpecies": "comparative", + } + canonical_config = { + key: _canonicalize_species_codes_for_job(job_id, config_scope.get(key) or []) + for key in role_keys + } + config_role_by_species: Dict[str, str] = {} + conflicts: list[Dict[str, str]] = [] + for key, role in role_keys.items(): + for species in canonical_config[key]: + existing = config_role_by_species.get(species) + if existing and existing != role: + conflicts.append({ + "code": "config_species_role_conflict", + "species": species, + "config_role": f"{existing}/{role}", + "requested_role": "", + }) + else: + config_role_by_species[species] = role + + explicit_raw = parse_explicit_species_role_mentions(message or "") + explicit = { + key: _canonicalize_species_codes_for_job(job_id, explicit_raw.get(key) or []) + for key in (*role_keys, "unassigned") + } + for key, requested_role in role_keys.items(): + for species in explicit[key]: + config_role = config_role_by_species.get(species) + if config_role != requested_role: + conflicts.append({ + "code": "user_config_species_role_conflict", + "species": species, + "config_role": config_role or "not_configured", + "requested_role": requested_role, + }) + + routed_intent = intent or {} + routed_params = routed_intent.get("params") or {} + structured_targets = _canonicalize_species_codes_for_job( + job_id, + _param_list(routed_params.get("targetSpecies") or routed_intent.get("target_species")), + ) + structured_refs = _canonicalize_species_codes_for_job( + job_id, + _param_list(routed_params.get("referenceSpecies") or routed_intent.get("reference_species")), + ) + structured_comps = _canonicalize_species_codes_for_job( + job_id, + _param_list(routed_params.get("comparativeSpecies") or routed_intent.get("comparative_species")), + ) + explicit_reference_set = set(explicit["referenceSpecies"]) + explicit_assigned_set = set( + explicit["targetSpecies"] + + explicit["referenceSpecies"] + + explicit["comparativeSpecies"] + ) + for species in structured_targets: + config_role = config_role_by_species.get(species) + if ( + config_role in {None, "reference"} + and species not in explicit_reference_set + and species not in explicit_assigned_set + ): + conflicts.append({ + "code": "analysis_target_config_role_conflict", + "species": species, + "config_role": config_role or "not_configured", + "requested_role": "analysis_target", + }) + + mentioned = list(dict.fromkeys( + species + for key in (*role_keys, "unassigned") + for species in explicit[key] + )) + target_candidates = list(dict.fromkeys([ + *structured_targets, + *explicit["targetSpecies"], + *explicit["unassigned"], + ])) + comparative_candidates = list(dict.fromkeys([ + *structured_comps, + *structured_targets, + *explicit["comparativeSpecies"], + *explicit["unassigned"], + ])) + targets = [ + species for species in canonical_config["targetSpecies"] + if species in target_candidates + ] or list(canonical_config["targetSpecies"]) + comparatives = [ + species for species in canonical_config["comparativeSpecies"] + if species in comparative_candidates + ] + references = [ + species for species in canonical_config["referenceSpecies"] + if not structured_refs or species in structured_refs or species in mentioned + ] or list(canonical_config["referenceSpecies"]) + classification_species = list(dict.fromkeys([*targets, *comparatives])) + + unique_conflicts: list[Dict[str, str]] = [] + seen_conflicts: set[tuple[str, str, str, str]] = set() + for conflict in conflicts: + identity = ( + conflict["code"], + conflict["species"], + conflict["config_role"], + conflict["requested_role"], + ) + if identity not in seen_conflicts: + seen_conflicts.add(identity) + unique_conflicts.append(conflict) + + return { + "applicable": True, + "config_scope": canonical_config, + "target_species": targets, + "reference_species": references, + "comparative_species": comparatives, + "classification_species": classification_species, + "conflicts": unique_conflicts, + } + + +def _species_role_conflict_message(conflicts: list[Dict[str, str]]) -> str: + role_labels = { + "target": "目标物种", + "reference": "参考物种", + "comparative": "比较物种", + "analysis_target": "本轮分析对象", + "not_configured": "未登记", + } + details = [] + for conflict in conflicts: + species = conflict.get("species") or "未知物种" + config_role = role_labels.get(conflict.get("config_role") or "", conflict.get("config_role") or "未知") + requested_role = role_labels.get(conflict.get("requested_role") or "", conflict.get("requested_role") or "配置内部重复角色") + details.append(f"{species}：config.xlsx 为{config_role}，本轮为{requested_role}") + return ( + "检测到物种角色不一致：" + + "；".join(details) + + "。为避免按错误范围执行，请发送“打开 config.xlsx”，核对并重新确认物种角色后再继续。" + ) + + +def _apply_reconciled_species_to_intent(intent: Dict[str, Any], resolution: Dict[str, Any]) -> None: + if not resolution.get("applicable") or resolution.get("conflicts"): + return + targets = list(resolution.get("target_species") or []) + references = list(resolution.get("reference_species") or []) + comparatives = list(resolution.get("comparative_species") or []) + params = intent.setdefault("params", {}) + params.update({ + "targetSpecies": targets, + "referenceSpecies": references, + "comparativeSpecies": comparatives, + }) + intent["target_species"] = targets + intent["reference_species"] = references + intent["comparative_species"] = comparatives + scope_mode = ( + "target_plus_reference_plus_comparative" + if references and comparatives + else "target_plus_reference" + if references + else "target_plus_comparative" + if comparatives + else "target_only" + ) + params["scopeMode"] = scope_mode + intent["scopeMode"] = scope_mode + + +def _is_gene_family_analysis_intent(intent: Optional[Dict[str, Any]]) -> bool: + candidate = intent or {} + stages = [str(stage or "").strip() for stage in candidate.get("stages") or []] + if stages and all(stage.startswith("rnaseq_") or stage.startswith("wgs_") for stage in stages): + return False + params = candidate.get("params") or {} + family = str(candidate.get("gene_family") or params.get("geneFamily") or "").strip() + gene_family_stages = { + "genome_prepare", "identify", "phylogeny", "motif", "domain", + "structure", "combo", "reference_classification", "subfamily_confirm", + "promoter", "tfbs", "synteny", "interspecies_synteny", "karyotype", + "pep", "ortholog", "kaks", "ppi", "expression", + } + return bool(family or set(stages) & gene_family_stages) + + def _normalize_main_agent_analysis_intent( intent: Dict[str, Any], *, @@ -1356,6 +1561,14 @@ def _resolve_subfamily_request_species_roles( intent: Optional[Dict[str, Any]] = None, ) -> Dict[str, list[str]]: """Preserve config roles while resolving species mentioned in a subfamily request.""" + confirmed_resolution = _reconcile_confirmed_gene_family_species(job_id, message, intent) + if confirmed_resolution.get("applicable") and not confirmed_resolution.get("conflicts"): + return { + "target_species": list(confirmed_resolution.get("target_species") or []), + "reference_species": list(confirmed_resolution.get("reference_species") or []), + "comparative_species": list(confirmed_resolution.get("comparative_species") or []), + "classification_species": list(confirmed_resolution.get("classification_species") or []), + } routed_intent = intent or {} params = routed_intent.get("params") or {} structured_targets = _canonicalize_species_codes_for_job( @@ -1826,8 +2039,19 @@ def _resolve_contextual_gene_family_request( or contextual_stage_request ): return {} + role_resolution = _reconcile_confirmed_gene_family_species( + job_id, + text, + routed_intent, + ) selection = _load_identify_species_selection(job_id) - config_targets, config_refs, config_comps = _species_groups_from_selection(selection) + if role_resolution.get("applicable") and not role_resolution.get("conflicts"): + confirmed_scope = role_resolution.get("config_scope") or {} + config_targets = list(confirmed_scope.get("targetSpecies") or []) + config_refs = list(confirmed_scope.get("referenceSpecies") or []) + config_comps = list(confirmed_scope.get("comparativeSpecies") or []) + else: + config_targets, config_refs, config_comps = _species_groups_from_selection(selection) family = _nonempty_prefix(_gene_family_from_config(INPUT_ROOT / job_id / "config.xlsx")) if job_id else "" if not family and request_prefix and "." in request_prefix: family = request_prefix.split(".", 1)[1] @@ -1839,37 +2063,38 @@ def _resolve_contextual_gene_family_request( prefix_species = request_prefix.split(".", 1)[0].strip() if prefix_species in {"Target", "Multi", "Ref", "Reference"}: prefix_species = "" - mentioned_targets = [sp for sp in mentioned if sp in config_targets or sp == prefix_species or sp in confirmed_species] - mentioned_refs = [sp for sp in mentioned if sp in config_refs] - mentioned_comps = [sp for sp in mentioned if sp in config_comps] - unassigned_mentions = [sp for sp in mentioned if sp not in mentioned_targets and sp not in mentioned_refs and sp not in mentioned_comps] - if unassigned_mentions: + is_all_species_request = _looks_like_all_species_query(text) or _looks_like_all_subfamily_species_query(text) + is_intraspecies_synteny_request = _looks_like_intraspecies_synteny_query(text) + if role_resolution.get("applicable") and not role_resolution.get("conflicts"): + targets = list(role_resolution.get("target_species") or []) + refs = list(role_resolution.get("reference_species") or []) + comps = list(role_resolution.get("comparative_species") or []) + else: + mentioned_targets = [sp for sp in mentioned if sp in config_targets or sp == prefix_species or sp in confirmed_species] + mentioned_refs = [sp for sp in mentioned if sp in config_refs] + mentioned_comps = [sp for sp in mentioned if sp in config_comps] + unassigned_mentions = [sp for sp in mentioned if sp not in mentioned_targets and sp not in mentioned_refs and sp not in mentioned_comps] for sp in unassigned_mentions: if sp not in mentioned_targets: mentioned_targets.append(sp) - - has_explicit_scope = bool(mentioned or structured_targets or structured_refs or structured_comps) - is_all_species_request = _looks_like_all_species_query(text) or _looks_like_all_subfamily_species_query(text) - is_intraspecies_synteny_request = _looks_like_intraspecies_synteny_query(text) - if mentioned: - targets = mentioned_targets or [sp for sp in mentioned if sp not in mentioned_refs and sp not in mentioned_comps] - elif structured_targets: - targets = list(structured_targets) - elif is_all_species_request and config_targets: - targets = list(config_targets) - elif is_intraspecies_synteny_request and config_targets: - targets = list(config_targets) - elif prefix_species: - targets = [prefix_species] - elif config_targets: - targets = [config_targets[0]] if len(config_targets) == 1 else list(config_targets) - elif confirmed_species: - targets = [confirmed_species[0]] - else: - targets = [] - - refs = mentioned_refs or list(structured_refs) or (list(config_refs) if is_all_species_request else []) - comps = mentioned_comps or list(structured_comps) or (list(config_comps) if is_all_species_request else []) + if mentioned: + targets = mentioned_targets or [sp for sp in mentioned if sp not in mentioned_refs and sp not in mentioned_comps] + elif structured_targets: + targets = list(structured_targets) + elif is_all_species_request and config_targets: + targets = list(config_targets) + elif is_intraspecies_synteny_request and config_targets: + targets = list(config_targets) + elif prefix_species: + targets = [prefix_species] + elif config_targets: + targets = [config_targets[0]] if len(config_targets) == 1 else list(config_targets) + elif confirmed_species: + targets = [confirmed_species[0]] + else: + targets = [] + refs = mentioned_refs or list(structured_refs) or (list(config_refs) if is_all_species_request else []) + comps = mentioned_comps or list(structured_comps) or (list(config_comps) if is_all_species_request else []) if refs and comps: scope_mode = "target_plus_reference_plus_comparative" elif refs: @@ -1895,7 +2120,14 @@ def _resolve_contextual_gene_family_request( for sp, data in (subfamily_state.get("by_species") or {}).items() } all_targets_confirmed = bool(classifiable_species_for_labels) and bool(subfamily_state.get("all_confirmed")) - explicit_custom_scope = bool(refs or comps or len(targets) > 1) + if role_resolution.get("applicable") and not role_resolution.get("conflicts"): + explicit_custom_scope = ( + targets != config_targets + or refs != config_refs + or comps != config_comps + ) + else: + explicit_custom_scope = bool(refs or comps or len(targets) > 1) skip_subfamily_labels = _skip_subfamily_label_query(text) if skip_subfamily_labels: context_route = "plain_phylogeny" if _looks_like_phylogeny_query(text) else "gene_family_without_subfamily" @@ -3959,9 +4191,6 @@ def _attach_agent_manifest_status(task: Dict) -> None: print(f"[API DEBUG] Failed to attach agent manifest status: {e}") app = FastAPI(title="BotanicalBaseAgent API") -from gfa_core.extension_stage_lightweight.api import create_lightweight_extension_router - -app.include_router(create_lightweight_extension_router()) # MainAgent runtime state. Analysis planning and execution remain owned by # the original IntentAgent/PlanningAgent/frontend flow below. llm_manager = None @@ -12701,7 +12930,7 @@ def _get_celery_overview() -> Dict[str, Any]: resource = snapshot(include_terminal=False) overview["resource_usage"] = resource.get("usage") - overview["resource_tasks"] = resource.get("tasks", []) + overview["resource_tasks"] = resource.get("tasks", [])[:20] except Exception as exc: overview["errors"].append(f"Resource governor snapshot failed: {exc}") overview["thresholds"] = { @@ -14302,8 +14531,9 @@ async def get_results( except Exception: pass - # Results are a persisted job view. Apply manifests by execution time so - # the latest attempt determines the visible stage status. + # Result browsing is a persisted job view, so apply manifests in execution + # order. The newest attempt for a stage becomes its visible status instead + # of depending on filesystem iteration order. manifest_payloads.sort( key=lambda item: (item[0].stat().st_mtime, item[0].name) ) @@ -16754,11 +16984,27 @@ async def _run_agent_chat( verified_intent = outcome.get("intent") if action == "propose_analysis" and isinstance(verified_intent, dict): - verified_intent = _normalize_main_agent_analysis_intent( - verified_intent, - job_id=request.job_id, - context_terms=router_context_terms, + role_resolution = ( + _reconcile_confirmed_gene_family_species( + request.job_id, + request.message, + verified_intent, + ) + if _is_gene_family_analysis_intent(verified_intent) + else {"applicable": False, "conflicts": []} ) + if role_resolution.get("conflicts"): + action = "clarify" + outcome["message"] = _species_role_conflict_message(role_resolution["conflicts"]) + outcome["proposal"] = None + verified_intent = None + else: + verified_intent = _normalize_main_agent_analysis_intent( + verified_intent, + job_id=request.job_id, + context_terms=router_context_terms, + ) + _apply_reconciled_species_to_intent(verified_intent, role_resolution) verified_stages = ( verified_intent.get("stages") if isinstance(verified_intent, dict) @@ -19732,6 +19978,28 @@ async def agent_plan(request: AgentPlanRequest): extension_intent = _match_extension_stage_intent(request.message, request.job_id or "") if extension_intent: return _extension_stage_plan_response(extension_intent, session_id, request.job_id or "") + plan_role_resolution = ( + _reconcile_confirmed_gene_family_species( + request.job_id, + request.message, + request.intent, + ) + if _is_gene_family_analysis_intent(request.intent) + else {"applicable": False, "conflicts": []} + ) + if plan_role_resolution.get("conflicts"): + return { + "success": False, + "status": "species_role_conflict", + "message": _species_role_conflict_message(plan_role_resolution["conflicts"]), + "issues": plan_role_resolution["conflicts"], + "suggested_command": "打开 config.xlsx", + "session_id": session_id, + } + if plan_role_resolution.get("applicable"): + request.intent = dict(request.intent or {}) + request.intent["params"] = dict(request.intent.get("params") or {}) + _apply_reconciled_species_to_intent(request.intent, plan_role_resolution) query_lower_for_context = (request.message or "").lower() subfamily_confirm_query_for_context = any( kw in query_lower_for_context @@ -21042,18 +21310,24 @@ async def agent_plan(request: AgentPlanRequest): intent["prefix"] = request.prefix intent.setdefault("params", {})["prefix"] = request.prefix plan_params["prefix"] = request.prefix - _apply_scope_selection_to_intent(intent, request.message) + if plan_role_resolution.get("applicable"): + _apply_reconciled_species_to_intent(intent, plan_role_resolution) + else: + _apply_scope_selection_to_intent(intent, request.message) if subfamily_prediction_query: - explicit_species = _extract_species_codes_from_text_for_job(request.job_id, request.message or "") - if explicit_species: - intent["target_species"] = explicit_species - intent.setdefault("params", {})["targetSpecies"] = explicit_species - intent["species"] = explicit_species[0] if len(explicit_species) == 1 else "Multi" - ref_defaults = [ - item.get("species") - for item in ((_load_identify_species_selection(request.job_id).get("groups") or {}).get("reference") or []) - if item.get("species") - ] if request.job_id else [] + if not plan_role_resolution.get("applicable"): + explicit_species = _extract_species_codes_from_text_for_job(request.job_id, request.message or "") + if explicit_species: + intent["target_species"] = explicit_species + intent.setdefault("params", {})["targetSpecies"] = explicit_species + intent["species"] = explicit_species[0] if len(explicit_species) == 1 else "Multi" + ref_defaults = list(plan_role_resolution.get("reference_species") or []) + if not ref_defaults and request.job_id: + ref_defaults = [ + item.get("species") + for item in ((_load_identify_species_selection(request.job_id).get("groups") or {}).get("reference") or []) + if item.get("species") + ] if ref_defaults: intent["reference_species"] = ref_defaults intent.setdefault("params", {})["referenceSpecies"] = ref_defaults diff --git a/gfa_core/mas/utils/intent_parser.py b/gfa_core/mas/utils/intent_parser.py index 97c93b089..f75329b21 100755 --- a/gfa_core/mas/utils/intent_parser.py +++ b/gfa_core/mas/utils/intent_parser.py @@ -415,6 +415,50 @@ def parse_gene_family_scope_selection(query: str) -> Dict[str, object]: } +def parse_explicit_species_role_mentions(query: str) -> Dict[str, List[str]]: + """Return explicit roles without promoting unassigned species to targets.""" + text = str(query or "") + + def mentioned_species(value: str) -> List[str]: + parsed = _species_mentions_with_roles(value) + return _unique_keep_order([ + species + for key in ("targetSpecies", "referenceSpecies", "comparativeSpecies", "unassigned") + for species in parsed.get(key) or [] + ]) + + explicit_patterns = { + "targetSpecies": r"目标物种|目标基因组|作为目标|是目标|\btarget(?:\s+species|\s+genome)?\b", + "referenceSpecies": r"参考物种|参考基因组|作为参考|是参考|\breference(?:\s+species|\s+genome)?\b|\bseed_reference\b|\bknown_family_reference\b", + "comparativeSpecies": r"比较物种|比较基因组|作为比较|是比较|作为外群|是外群|\bcomparative(?:\s+species|\s+genome)?\b|\boutgroup\b", + } + roles: Dict[str, List[str]] = { + "targetSpecies": [], + "referenceSpecies": [], + "comparativeSpecies": [], + "unassigned": [], + } + for clause in re.split(r"[,，;；。\n]+", text): + clause_species = mentioned_species(clause) + matched_roles = [ + key + for key, pattern in explicit_patterns.items() + if re.search(pattern, clause, flags=re.IGNORECASE) + ] + if len(matched_roles) == 1: + roles[matched_roles[0]].extend(clause_species) + + assigned = set( + roles["targetSpecies"] + + roles["referenceSpecies"] + + roles["comparativeSpecies"] + ) + roles["unassigned"] = [ + species for species in mentioned_species(text) if species not in assigned + ] + return {key: _unique_keep_order(value) for key, value in roles.items()} + + def parse_promoter_plot_revise_request(query: str) -> Dict: """Extract parameters for custom promoter cis-element plotting.""" text = str(query or "").strip() diff --git a/tests/unit/test_agent_chat_actions.py b/tests/unit/test_agent_chat_actions.py new file mode 100644 index 000000000..5aee446d4 --- /dev/null +++ b/tests/unit/test_agent_chat_actions.py @@ -0,0 +1,888 @@ +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", + } + ] + 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_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 的数据准备", + "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_confirmed_request_for_original_planner( + monkeypatch, + web_main, +): + captured = [] + proposal = { + "proposal_id": "proposal-next", + "request": "我已确认 config.xlsx，接下来应该做什么分析", + "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["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 diff --git a/tests/unit/test_intent_species_roles.py b/tests/unit/test_intent_species_roles.py new file mode 100644 index 000000000..bd3c2046d --- /dev/null +++ b/tests/unit/test_intent_species_roles.py @@ -0,0 +1,32 @@ +from gfa_core.mas.utils.intent_parser import ( + parse_explicit_species_role_mentions, + parse_gene_family_scope_selection, +) + + +def test_ref_data_path_does_not_assign_species_roles() -> None: + query = ( + "苦荞基因组数据在 " + "/home/xia-lab/wutx_old/wutx/plantAnalysisAgent/ref_data/Tartary_genome，" + "拟南芥基因组数据在 " + "/home/xia-lab/wutx_old/wutx/plantAnalysisAgent/ref_data/ath，" + "Ft是目标物种，At是参考物种" + ) + + scope = parse_gene_family_scope_selection(query) + + assert scope["targetSpecies"] == ["Ft"] + assert scope["referenceSpecies"] == ["At"] + + +def test_explicit_role_parser_keeps_unassigned_mentions_separate() -> None: + roles = parse_explicit_species_role_mentions( + "Os和At的数据已经准备好；Os是目标物种，At是参考物种，Ppy稍后再分析" + ) + + assert roles == { + "targetSpecies": ["Os"], + "referenceSpecies": ["At"], + "comparativeSpecies": [], + "unassigned": ["Ppy"], + } diff --git a/tests/unit/test_subfamily_prediction_plan.py b/tests/unit/test_subfamily_prediction_plan.py index f8f77efad..d2e6f6167 100755 --- a/tests/unit/test_subfamily_prediction_plan.py +++ b/tests/unit/test_subfamily_prediction_plan.py @@ -3,6 +3,8 @@ import json from pathlib import Path from types import SimpleNamespace +import pytest + from gfa_core.gene_family_stage_context import ( enrich_stage_params_with_context, infer_analysis_route, @@ -330,3 +332,177 @@ def test_direct_stage_preserves_species_roles_from_intent_when_selector_is_empty assert details["phylogeny"]["params"]["analysisRoute"] == "subfamily_evidence" assert details["motif"]["params"]["analysisRoute"] == "subfamily_evidence" assert details["phylogeny"]["params"]["memberSetId"] == details["motif"]["params"]["memberSetId"] + + +def test_contextual_handoff_uses_confirmed_config_roles_when_selector_is_empty(monkeypatch): + main = _load_vue3_main() + monkeypatch.setattr(main, "_load_identify_species_selection", lambda _job_id: {}) + monkeypatch.setattr( + main, + "_confirmed_config_species_scope", + lambda _job_id: { + "targetSpecies": ["Os"], + "referenceSpecies": ["At"], + "comparativeSpecies": [], + }, + ) + monkeypatch.setattr(main, "_gene_family_from_config", lambda _path: "PEBP") + monkeypatch.setattr(main, "_confirmed_subfamily_species", lambda _job_id, _family: []) + monkeypatch.setattr( + main, + "_extract_species_codes_from_text_for_job", + lambda _job_id, _text: ["Os", "At"], + ) + monkeypatch.setattr( + main, + "_subfamily_state_for_targets", + lambda _job_id, species, _family: { + "all_confirmed": species == ["Os"], + "by_species": {"Os": {"status": "confirmed"}}, + "issues": [], + }, + ) + monkeypatch.setattr( + main, + "_validate_gene_family_stage_context", + lambda **_kwargs: {"valid": True, "issues": [], "evidence": {}}, + ) + + contextual = main._resolve_contextual_gene_family_request( + "job-real", + ( + "执行水稻PEBP基因家族分析。基因家族 PEBP；" + "目标物种 Os；参考物种 At；亚家族分类已经确认。" + ), + "Os.PEBP", + requested_stages=["phylogeny", "motif", "domain", "structure", "combo"], + request_intent={ + "gene_family": "PEBP", + "target_species": ["Os"], + "reference_species": ["At"], + "params": { + "targetSpecies": ["Os"], + "referenceSpecies": ["At"], + }, + }, + ) + + assert contextual["targetSpecies"] == ["Os"] + assert contextual["referenceSpecies"] == ["At"] + assert contextual["prefix"] == "Os.PEBP" + assert contextual["context_route"] == "post_subfamily_gene_family_analysis" + + +def test_confirmed_multi_species_roles_remain_separate(monkeypatch): + main = _load_vue3_main() + monkeypatch.setattr( + main, + "_confirmed_config_species_scope", + lambda _job_id: { + "targetSpecies": ["Ft"], + "referenceSpecies": ["At"], + "comparativeSpecies": ["Ppy"], + }, + ) + + resolved = main._reconcile_confirmed_gene_family_species( + "job-multi", + "Ft是目标物种，Ppy是比较物种，At是参考物种，执行联合亚家族分类", + { + "gene_family": "SPL", + "target_species": ["Ft", "Ppy"], + "reference_species": ["At"], + "params": { + "targetSpecies": ["Ft", "Ppy"], + "referenceSpecies": ["At"], + }, + }, + ) + + assert resolved["conflicts"] == [] + assert resolved["target_species"] == ["Ft"] + assert resolved["comparative_species"] == ["Ppy"] + assert resolved["reference_species"] == ["At"] + assert resolved["classification_species"] == ["Ft", "Ppy"] + + +def test_confirmed_config_duplicate_roles_are_blocked(monkeypatch): + main = _load_vue3_main() + monkeypatch.setattr( + main, + "_confirmed_config_species_scope", + lambda _job_id: { + "targetSpecies": ["Ft"], + "referenceSpecies": ["Ft", "At"], + "comparativeSpecies": [], + }, + ) + + resolved = main._reconcile_confirmed_gene_family_species( + "job-invalid-config", + "执行SPL基因家族分析", + {"gene_family": "SPL", "stages": ["identify"]}, + ) + + assert any( + conflict["code"] == "config_species_role_conflict" + and conflict["species"] == "Ft" + for conflict in resolved["conflicts"] + ) + assert "打开 config.xlsx" in main._species_role_conflict_message(resolved["conflicts"]) + + +@pytest.mark.asyncio +async def test_plan_api_blocks_confirmed_config_role_conflict(monkeypatch, tmp_path): + import pandas as pd + from gfa_core.mas import confirmation_snapshot + + main = _load_vue3_main() + 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(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": (), + }, + ) + + result = await main.agent_plan( + main.AgentPlanRequest( + message="At是目标物种，Os是参考物种，请执行PEBP基因家族分析", + session_id="role-conflict", + job_id="job-os", + prefix="Os.PEBP", + intent={ + "gene_family": "PEBP", + "stages": ["identify"], + "target_species": ["At"], + "reference_species": ["Os"], + "params": { + "geneFamily": "PEBP", + "targetSpecies": ["At"], + "referenceSpecies": ["Os"], + }, + }, + ) + ) + + assert result["success"] is False + assert result["status"] == "species_role_conflict" + assert "打开 config.xlsx" in result["message"] + assert "plan" not in result