PlantAI-dev/tests/unit/test_main_agent_data_preparation_handoff.py

import asyncio
import json

from gfa_core.mas.agents.intent_v2 import IntentAgentV2
from gfa_core.mas.agents.conversation_tools import build_plantai_conversation_tools
from gfa_core.mas.llm import LLMManager
from gfa_core.mas.workflow_models import GoalLedger


def _registry(tmp_path, *, job_id="job-data-preparation-handoff"):
    ledger = GoalLedger(
        conversation_id="conversation-data-preparation-handoff",
        revision=0,
        user_terms={
            "active_job_id": job_id,
            "gene_family": "SPL",
            "target_species": ["Ft"],
            "reference_species": ["At"],
            "config_confirmed": True,
        },
    )
    return build_plantai_conversation_tools(
        llm_manager=LLMManager({}),
        project_root=tmp_path,
        bound_job_id=job_id,
        goal_ledger=ledger,
    )


def test_analysis_handoff_is_not_rewritten_to_genome_prepare(tmp_path):
    resolved = _registry(tmp_path).resolve_analysis_handoff_request(
        "执行亚家族分类",
        workflow_id="gene_family",
        guidance=True,
    )

    assert resolved.startswith("执行亚家族分类\n")
    assert "执行数据准备（genome_prepare）" not in resolved
    assert "目标物种 Ft" in resolved
    assert "参考物种 At" in resolved
    assert "正式 config.xlsx 已确认" in resolved


def test_genome_prepare_manifest_does_not_change_analysis_handoff(tmp_path):
    job_id = "job-manifest-independent-handoff"
    registry = _registry(tmp_path, job_id=job_id)
    before = registry.resolve_analysis_handoff_request(
        "执行亚家族分类",
        workflow_id="gene_family",
        guidance=True,
    )
    final_dir = tmp_path / "workspace" / job_id / "final"
    final_dir.mkdir(parents=True)
    (final_dir / "genome_prepare.genome_prepare.manifest.json").write_text(
        json.dumps({"stage": "genome_prepare", "status": "success"}),
        encoding="utf-8",
    )

    after = registry.resolve_analysis_handoff_request(
        "执行亚家族分类",
        workflow_id="gene_family",
        guidance=True,
    )

    assert after == before


def test_explicit_data_preparation_request_is_preserved(tmp_path):
    resolved = _registry(tmp_path).resolve_analysis_handoff_request(
        "执行数据准备",
        workflow_id="data_preparation",
        guidance=False,
    )

    assert resolved.startswith("执行数据准备\n")
    assert "当前业务动作已由数据准备契约唯一确定" not in resolved


def test_next_step_question_is_not_rewritten_to_genome_prepare(tmp_path):
    resolved = _registry(tmp_path).resolve_analysis_handoff_request(
        "接下来需要做什么？",
        workflow_id="gene_family",
        guidance=True,
    )

    assert resolved.startswith("接下来需要做什么？\n")
    assert "执行数据准备（genome_prepare）" not in resolved


def test_bound_analysis_intent_uses_default_language(monkeypatch, tmp_path):
    captured_queries = []

    async def fake_execute(_self, state, message):
        captured_queries.append(message.payload["query"])
        state.results["results"] = {
            "intent": {
                "gene_family": "SPL",
                "stages": ["ppi", "kaks"],
                "primary_stage": "ppi",
            }
        }
        return state

    monkeypatch.setattr(IntentAgentV2, "execute", fake_execute)
    result = json.loads(
        asyncio.run(
            _registry(tmp_path).dispatch(
                "recognize_analysis_intent",
                json.dumps({"query": "执行Ft的PPI和kaks分析"}, ensure_ascii=False),
            )
        )
    )

    assert result["ok"] is True
    assert result["data"]["stages"] == ["ppi", "kaks"]
    assert captured_queries[0].startswith("执行Ft的PPI和kaks分析\n")