merge: resolve .gitignore conflict, keep remote memory ignore pattern

tests/unit/test_gene_family_report.py

@@ -0,0 +1,844 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import re
+
+import pytest
+
+import gfa_core.gene_family_report as report_module
+from gfa_core.gene_family_report import (
+    NoSuccessfulGeneFamilyEvidence,
+    generate_gene_family_report,
+)
+
+
+def _write_manifest(
+    workspace: Path,
+    *,
+    prefix: str,
+    stage: str,
+    status: str = "success",
+    scope_id: str = "scope-a",
+    analysis_route: str = "post_subfamily_confirm",
+    member_set_id: str = "members-a",
+    metric_value: int = 1,
+) -> Path:
+    final_dir = workspace / "final"
+    final_dir.mkdir(parents=True, exist_ok=True)
+    token = f"{scope_id}.{analysis_route}.{member_set_id}"
+    table_path = final_dir / f"{prefix}.{token}.{stage}.tsv"
+    table_path.write_text(f"gene\tvalue\n{scope_id}_gene\t{metric_value}\n", encoding="utf-8")
+    manifest_path = final_dir / f"{prefix}.{token}.{stage}.manifest.json"
+    manifest = {
+        "manifest_version": "1.0",
+        "stage": stage,
+        "status": status,
+        "prefix": prefix,
+        "scope_id": scope_id,
+        "analysis_route": analysis_route,
+        "member_set_id": member_set_id,
+        "parameters": {
+            "scope_id": scope_id,
+            "analysis_route": analysis_route,
+            "member_set_id": member_set_id,
+        },
+        "metrics": {"member_count": metric_value},
+        "artifacts": {
+            "result_table": {
+                "path": f"final/{table_path.name}",
+                "full_path": str(table_path),
+                "type": "table",
+            }
+        },
+    }
+    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
+    return manifest_path
+
+
+def _generate(
+    workspace: Path,
+    *,
+    expected_stages: list[str],
+    stage_statuses: dict[str, str],
+    use_model: bool = False,
+    action: str = "generate",
+):
+    return generate_gene_family_report(
+        workspace,
+        "Ft.SPL",
+        expected_stages=expected_stages,
+        stage_statuses=stage_statuses,
+        scope_id="scope-a",
+        analysis_route="post_subfamily_confirm",
+        member_set_id="members-a",
+        use_model=use_model,
+        action=action,
+        job_id="job-report",
+    )
+
+
+def _stage_section(markdown: str, stage: str) -> str:
+    heading = re.search(rf"(?m)^## (?!#).*?\(`{re.escape(stage)}`\)", markdown)
+    if heading is None:
+        raise AssertionError(f"missing report chapter for {stage}")
+    next_heading = re.search(r"(?m)^## (?!#)", markdown[heading.end():])
+    end = heading.end() + next_heading.start() if next_heading else len(markdown)
+    return markdown[heading.start():end]
+
+
+def _subsection(markdown: str, number: str) -> str:
+    heading = re.search(rf"(?m)^### {re.escape(number)}(?:\s|$)", markdown)
+    if heading is None:
+        raise AssertionError(f"missing report subsection {number}")
+    next_heading = re.search(r"(?m)^#{2,3} ", markdown[heading.end():])
+    end = heading.end() + next_heading.start() if next_heading else len(markdown)
+    return markdown[heading.start():end]
+
+
+def _valid_api_arrangement(payload: dict, *, reverse: bool = False) -> dict:
+    sections = []
+    for section in payload["sections"]:
+        sentence_ids = [sentence["sentence_id"] for sentence in section["sentences"]]
+        if reverse:
+            sentence_ids.reverse()
+        sections.append(
+            {
+                "section_id": section["section_id"],
+                "paragraphs": [
+                    {
+                        "sentence_ids": sentence_ids,
+                        "connector_ids": ["also"] * max(0, len(sentence_ids) - 1),
+                    }
+                ],
+            }
+        )
+    return {"sections": sections}
+
+
+def test_report_uses_only_planned_stages_and_current_identity(tmp_path: Path) -> None:
+    workspace = tmp_path / "job"
+    _write_manifest(workspace, prefix="Ft.SPL", stage="identify", metric_value=3)
+    other = _write_manifest(
+        workspace,
+        prefix="Ft.SPL",
+        stage="identify",
+        scope_id="scope-b",
+        analysis_route="target_only",
+        member_set_id="members-b",
+        metric_value=99,
+    )
+
+    result = _generate(
+        workspace,
+        expected_stages=["identify"],
+        stage_statuses={"identify": "success"},
+    )
+
+    report_text = result.report_path.read_text(encoding="utf-8")
+    assert "最终成员数：3" in report_text
+    assert other.name not in report_text
+    generated = {path.name for path in (workspace / "final").glob("Ft.SPL.gene_family_report.*")}
+    assert generated == {
+        "Ft.SPL.gene_family_report.md",
+        "Ft.SPL.gene_family_report.manifest.json",
+    }
+
+
+def test_report_renders_template_subsections_for_only_planned_stages(tmp_path: Path) -> None:
+    workspace = tmp_path / "job"
+    for stage in ("identify", "motif", "pep"):
+        _write_manifest(workspace, prefix="Ft.SPL", stage=stage)
+
+    result = _generate(
+        workspace,
+        expected_stages=["identify", "motif", "pep"],
+        stage_statuses={"identify": "success", "motif": "success", "pep": "success"},
+    )
+
+    markdown = result.report_path.read_text(encoding="utf-8")
+    expected_subsections = {
+        "identify": "### 4.3 最终成员数量与命名",
+        "motif": "### 7.2 Motif 数量与出现频率",
+        "pep": "### 16.1 蛋白理化性质分布图",
+    }
+    for stage, subsection in expected_subsections.items():
+        assert subsection in _stage_section(markdown, stage)
+    assert "(`synteny`)" not in markdown
+    assert "【填写】" not in markdown
+
+
+def test_report_renders_nested_motif_metrics_without_placeholder_values(tmp_path: Path) -> None:
+    workspace = tmp_path / "job"
+    manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="motif")
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    manifest["metrics"] = {
+        "motif_count": 10,
+        "total_sites": 105,
+        "motifs": [
+            {"width": width}
+            for width in (78, 76, 80, 113, 57, 29, 52, 138, 27, 29)
+        ],
+    }
+    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
+
+    result = _generate(
+        workspace,
+        expected_stages=["motif"],
+        stage_statuses={"motif": "success"},
+    )
+
+    motif_section = _stage_section(result.report_path.read_text(encoding="utf-8"), "motif")
+    assert re.search(r"识别\s*10\s*个\s*motif", motif_section)
+    assert re.search(r"总位点数为\s*105", motif_section)
+    assert re.search(r"宽度范围为\s*27–138", motif_section)
+    assert "未解析" not in motif_section
+    assert "N/A" not in motif_section
+    assert "None" not in motif_section
+
+
+def test_identify_uses_real_manifest_member_count_aliases(tmp_path: Path) -> None:
+    workspace = tmp_path / "job"
+    manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="identify")
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    manifest["metrics"] = {"identified_members_count": 24, "final_members": 24}
+    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
+
+    result = _generate(
+        workspace,
+        expected_stages=["identify"],
+        stage_statuses={"identify": "success"},
+    )
+
+    section = _stage_section(result.report_path.read_text(encoding="utf-8"), "identify")
+    assert "最终成员数：24" in section
+    assert re.search(r"形成\s*24\s*个\s*SPL家族", section)
+    assert "源结果未提供此指标" not in section
+
+
+def test_phylogeny_renders_real_manifest_schema_with_explicit_labels(tmp_path: Path) -> None:
+    workspace = tmp_path / "job"
+    manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="phylogeny")
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    manifest["metrics"] = {"sequences_count": 40, "msa_length": 1449}
+    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
+
+    result = _generate(
+        workspace,
+        expected_stages=["phylogeny"],
+        stage_statuses={"phylogeny": "success"},
+    )
+
+    section = _stage_section(result.report_path.read_text(encoding="utf-8"), "phylogeny")
+    assert "序列数：40" in section
+    assert "多序列比对长度：1449" in section
+    assert re.search(r"使用\s*40\s*条序列.*比对长度为\s*1449", section)
+
+
+def test_interspecies_synteny_reads_only_named_real_schema_artifacts(tmp_path: Path) -> None:
+    workspace = tmp_path / "job"
+    manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="interspecies_synteny")
+    final_dir = workspace / "final"
+    gene_to_species = final_dir / "Ft.SPL.gene_to_species.tsv"
+    gene_to_species.write_text(
+        "gene_id\tspecies\tgroup\n"
+        "FtGene1.01.T01\tFt\ttarget\n"
+        "FtGene2.01.T01\tFt\ttarget\n"
+        "AT1G00010\tAT\tref\n"
+        "AT1G00020\tAT\tref\n",
+        encoding="utf-8",
+    )
+    collinearity = final_dir / "Ft.At.collinearity"
+    lines = []
+    for index in range(521):
+        lines.append(f"## Alignment {index}: score=1 e_value=1e-10 N=1 Ft&At plus")
+        if index == 0:
+            lines.append("  0-  0:\tAT1G00010.1\tFtGene1.01.T01\t1e-20")
+        elif index == 1:
+            lines.append("  1-  0:\tAT1G00020.1\tFtGene2.01.T01\t1e-20")
+    collinearity.write_text("\n".join(lines) + "\n", encoding="utf-8")
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    manifest["metrics"] = {"pair_success_count": 1, "pair_total_count": 1}
+    manifest["artifacts"] = {
+        "gene_to_species": {"path": f"final/{gene_to_species.name}", "type": "tsv"},
+        "Ft.At_collinearity": {"path": f"final/{collinearity.name}", "type": "data"},
+    }
+    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
+
+    result = _generate(
+        workspace,
+        expected_stages=["interspecies_synteny"],
+        stage_statuses={"interspecies_synteny": "success"},
+    )
+
+    section = _stage_section(result.report_path.read_text(encoding="utf-8"), "interspecies_synteny")
+    assert "共线性区块数：521" in section
+    assert "家族成员目标—参考配对数：2" in section
+    assert re.search(r"全基因组\s*block\s*521\s*个.*目标-参考家族.*?2\s*对", section)
+
+
+def test_pep_renders_real_manifest_task_summary_with_explicit_labels(tmp_path: Path) -> None:
+    workspace = tmp_path / "job"
+    manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="pep")
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    manifest["metrics"] = {"protein_count": 40, "tasks_success": 3, "tasks_total": 3}
+    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
+
+    result = _generate(
+        workspace,
+        expected_stages=["pep"],
+        stage_statuses={"pep": "success"},
+    )
+
+    section = _stage_section(result.report_path.read_text(encoding="utf-8"), "pep")
+    assert "蛋白数：40" in section
+    assert "子任务成功数：3/3" in section
+    assert re.search(r"覆盖\s*40\s*个蛋白，子任务成功数为\s*3/3", section)
+
+
+def test_report_has_no_empty_third_level_subsections_or_repeated_generic_interpretation(
+    tmp_path: Path,
+) -> None:
+    workspace = tmp_path / "job"
+    for stage in ("identify", "phylogeny", "motif"):
+        manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage=stage)
+        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+        if stage == "phylogeny":
+            manifest["metrics"] = {"sequences_count": 40, "msa_length": 1449}
+        elif stage == "motif":
+            manifest["metrics"] = {
+                "sequence_count": 40,
+                "motif_count": 10,
+                "total_sites": 105,
+                "motifs": [{"width": 27}, {"width": 138}],
+            }
+        manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
+
+    result = _generate(
+        workspace,
+        expected_stages=["identify", "phylogeny", "motif"],
+        stage_statuses={stage: "success" for stage in ("identify", "phylogeny", "motif")},
+    )
+
+    markdown = result.report_path.read_text(encoding="utf-8")
+    for stage in ("identify", "phylogeny", "motif"):
+        section = _stage_section(markdown, stage)
+        chunks = re.split(r"(?m)(?=^### )", section)[1:]
+        assert chunks
+        for chunk in chunks:
+            body = "\n".join(chunk.splitlines()[1:]).strip()
+            assert body, f"empty subsection: {chunk.splitlines()[0]}"
+    assert markdown.count("本节解释仅限上述可追溯结果，不据此断言真实功能或因果关系。") <= 1
+    assert markdown.count("结果范围受当前输入、参数和源文件完整性限制。") <= 1
+
+
+def test_front_summary_contains_project_numbers_in_six_to_ten_sentences(tmp_path: Path) -> None:
+    workspace = tmp_path / "job"
+    identify_path = _write_manifest(workspace, prefix="Ft.SPL", stage="identify")
+    identify = json.loads(identify_path.read_text(encoding="utf-8"))
+    identify["metrics"] = {"identified_members_count": 24, "final_members": 24}
+    identify_path.write_text(json.dumps(identify, ensure_ascii=False), encoding="utf-8")
+    phylogeny_path = _write_manifest(workspace, prefix="Ft.SPL", stage="phylogeny")
+    phylogeny = json.loads(phylogeny_path.read_text(encoding="utf-8"))
+    phylogeny["metrics"] = {"sequences_count": 40, "msa_length": 1449}
+    phylogeny_path.write_text(json.dumps(phylogeny, ensure_ascii=False), encoding="utf-8")
+
+    result = _generate(
+        workspace,
+        expected_stages=["identify", "phylogeny"],
+        stage_statuses={"identify": "success", "phylogeny": "success"},
+    )
+
+    summary = _subsection(result.report_path.read_text(encoding="utf-8"), "1.3")
+    assert re.search(r"目标物种.*24.*家族成员", summary)
+    assert re.search(r"目标.*参考.*40.*1449", summary)
+    assert 6 <= summary.count("。") <= 10
+
+
+def test_stage_specific_key_facts_land_in_their_semantic_subsections(tmp_path: Path) -> None:
+    workspace = tmp_path / "job"
+    final_dir = workspace / "final"
+    stages = [
+        "reference_classification", "domain", "structure", "combo",
+        "promoter", "tfbs", "synteny",
+    ]
+    manifests = {stage: _write_manifest(workspace, prefix="Ft.SPL", stage=stage) for stage in stages}
+
+    assignment = final_dir / "Ft.SPL.reference_classification.assignment.tsv"
+    assignment.write_text(
+        "target_gene\tassigned_subfamily\tconfidence\n"
+        + "\n".join(
+            f"FtGene{i}\t{'I' if i < 2 else 'II'}\t{confidence}"
+            for i, confidence in enumerate(("high", "high", "medium", "low"), start=1)
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    reference = json.loads(manifests["reference_classification"].read_text(encoding="utf-8"))
+    reference["metrics"] = {"target_gene_count": 4, "subfamily_rows": 6}
+    reference["artifacts"] = {"assignment_table": {"path": f"final/{assignment.name}", "type": "tsv"}}
+    manifests["reference_classification"].write_text(json.dumps(reference), encoding="utf-8")
+
+    domain = json.loads(manifests["domain"].read_text(encoding="utf-8"))
+    domain["metrics"] = {
+        "proteins_total": 40, "proteins_with_domains": 40, "domain_coverage_rate": 1.0,
+        "top_domains": [["SBP", 40], ["Ig_SPL7", 12]],
+    }
+    manifests["domain"].write_text(json.dumps(domain), encoding="utf-8")
+
+    structure = json.loads(manifests["structure"].read_text(encoding="utf-8"))
+    structure["metrics"] = {
+        "genes_total": 40, "genes_with_structure": 24, "structure_coverage_rate": 0.6,
+        "long_gene_outlier_count": 1,
+        "long_gene_outliers": [{"gene_id": "FtLong1", "length_bp": 12454, "cds_count": 10}],
+    }
+    manifests["structure"].write_text(json.dumps(structure), encoding="utf-8")
+
+    combo = json.loads(manifests["combo"].read_text(encoding="utf-8"))
+    combo["metrics"] = None
+    combo["parameters"] = {"panels": ["motif", "domain", "structure"]}
+    combo["artifacts"] = {
+        "combo_plot_png": {"path": "final/Ft.SPL.combo.png", "type": "image"},
+        "combo_plot_pdf": {"path": "final/Ft.SPL.combo.pdf", "type": "image"},
+    }
+    manifests["combo"].write_text(json.dumps(combo), encoding="utf-8")
+
+    element_counts = final_dir / "Ft.SPL.element_counts.txt"
+    element_counts.write_text("1460\tTATA-box\n852\tCAAT-box\n72\tMYB\n", encoding="utf-8")
+    category_summary = final_dir / "Ft.SPL.promoter_category_summary.tsv"
+    category_summary.write_text("big7\telement_instances\nCore promoter\t2600\nLight response\t188\n", encoding="utf-8")
+    enrichment = final_dir / "Ft.SPL.enrichment_analysis.txt"
+    enrichment.write_text(
+        "Subfamily\tElement\tCount\tSubfamilyTotal\tGlobalCount\tEnrichmentRatio\nI\tAACA_motif\t1\t222\t1\t16.66\n",
+        encoding="utf-8",
+    )
+    cooccurrence = final_dir / "Ft.SPL.cooccurrence_analysis.txt"
+    cooccurrence.write_text(
+        "Element1\tElement2\tCo-occurrenceCount\tGenes\nCAAT-box\tMYB\t23\tFtGene1,FtGene2\n",
+        encoding="utf-8",
+    )
+    complexity = final_dir / "Ft.SPL.complexity_analysis.txt"
+    complexity.write_text(
+        "Gene\tTotalElements\tUniqueElements\tComplexityScore\tComplexityClass\nFtComplex1\t278\t33\t361.44\tVery High\n",
+        encoding="utf-8",
+    )
+    promoter = json.loads(manifests["promoter"].read_text(encoding="utf-8"))
+    promoter["metrics"] = {"total_elements": 3699, "unique_elements": 87, "genes_with_elements": 23}
+    promoter["artifacts"] = {
+        "element_counts": {"path": f"final/{element_counts.name}"},
+        "viz_promoter_position_distribution_summary": {"path": f"final/{category_summary.name}"},
+        "enrichment_analysis": {"path": f"final/{enrichment.name}"},
+        "cooccurrence_analysis": {"path": f"final/{cooccurrence.name}"},
+        "complexity_analysis": {"path": f"final/{complexity.name}"},
+    }
+    manifests["promoter"].write_text(json.dumps(promoter), encoding="utf-8")
+
+    motif_summary = final_dir / "Ft.SPL.tfbs.motif_summary.tsv"
+    motif_summary.write_text(
+        "motif_id\tmotif_alt_id\tfamily_sites\tfamily_genes_with_site\tbackground_sites\tbackground_genes_with_site\n"
+        "MOTIF_A\t\t153\t18\t0\t0\n",
+        encoding="utf-8",
+    )
+    tfbs = json.loads(manifests["tfbs"].read_text(encoding="utf-8"))
+    tfbs["metrics"] = {
+        "family_promoter_count": 24, "background_promoter_count": 0,
+        "family_site_count": 3734, "family_motif_count": 230, "motif_database": "PlantTFDB",
+    }
+    tfbs["artifacts"] = {"motif_summary": {"path": f"final/{motif_summary.name}"}}
+    manifests["tfbs"].write_text(json.dumps(tfbs), encoding="utf-8")
+
+    synteny = json.loads(manifests["synteny"].read_text(encoding="utf-8"))
+    synteny["metrics"] = {"gene_count": 31679, "synteny_blocks": 316, "gene_pairs": 3267, "circos_chromosomes_plotted": 8}
+    manifests["synteny"].write_text(json.dumps(synteny), encoding="utf-8")
+
+    result = _generate(
+        workspace,
+        expected_stages=stages,
+        stage_statuses={stage: "success" for stage in stages},
+    )
+    markdown = result.report_path.read_text(encoding="utf-8")
+    assert re.search(r"4\s*个目标成员.*2\s*个小组.*高信心\s*2", _subsection(markdown, "6.2"))
+    assert re.search(r"40/40.*SBP.*40.*Ig_SPL7.*12", _subsection(markdown, "8.1"))
+    assert re.search(r"24/40.*FtLong1.*12454", _subsection(markdown, "9.1"))
+    assert "Motif、Domain、Structure" in _subsection(markdown, "10.1")
+    promoter_text = _subsection(markdown, "11.1")
+    assert re.search(r"23\s*个目标基因.*3699.*TATA-box.*1460", promoter_text)
+    assert "FtComplex1" in promoter_text and "CAAT-box—MYB" in promoter_text
+    assert re.search(r"24\s*条目标启动子.*3734.*MOTIF_A.*153", _subsection(markdown, "12.1"))
+    assert re.search(r"背景启动子为\s*0.*不能.*富集", _subsection(markdown, "12.1"))
+    assert re.search(r"全基因组.*316.*3267.*8\s*条染色体", _subsection(markdown, "13.1"))
+    assert "源结果未提供家族专属复制基因对" in _subsection(markdown, "13.1")
+
+
+def test_all_planned_success_is_final(tmp_path: Path) -> None:
+    workspace = tmp_path / "job"
+    _write_manifest(workspace, prefix="Ft.SPL", stage="identify")
+    _write_manifest(workspace, prefix="Ft.SPL", stage="motif")
+
+    result = _generate(
+        workspace,
+        expected_stages=["identify", "motif"],
+        stage_statuses={"identify": "success", "motif": "success"},
+    )
+
+    manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
+    assert result.report_status == "final"
+    assert manifest["metrics"]["report_status"] == "final"
+    assert manifest["metrics"]["included_stages"] == ["identify", "motif"]
+
+
+def test_unsupported_successful_stage_is_visible_but_not_included_or_final(tmp_path: Path) -> None:
+    workspace = tmp_path / "job"
+    _write_manifest(workspace, prefix="Ft.SPL", stage="identify")
+    _write_manifest(workspace, prefix="Ft.SPL", stage="ppi")
+
+    result = _generate(
+        workspace,
+        expected_stages=["identify", "ppi"],
+        stage_statuses={"identify": "success", "ppi": "success"},
+    )
+
+    markdown = result.report_path.read_text(encoding="utf-8")
+    manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
+    assert result.report_status == "partial"
+    assert result.included_stages == ("identify",)
+    assert "ppi" in markdown
+    assert "不支持" in markdown
+    assert "(`ppi`)" not in markdown
+    assert manifest["metrics"]["included_stages"] == ["identify"]
+    assert manifest["metrics"]["unsupported_stages"] == ["ppi"]
+
+
+@pytest.mark.parametrize("expected_stages", [None, []])
+def test_generate_rejects_missing_plan_instead_of_inferring_historical_manifests(
+    tmp_path: Path, expected_stages: list[str] | None
+) -> None:
+    workspace = tmp_path / "job"
+    _write_manifest(workspace, prefix="Ft.SPL", stage="identify")
+
+    with pytest.raises(ValueError, match="expected_stages"):
+        generate_gene_family_report(
+            workspace,
+            "Ft.SPL",
+            expected_stages=expected_stages,
+            stage_statuses={"identify": "success"},
+            scope_id="scope-a",
+            analysis_route="post_subfamily_confirm",
+            member_set_id="members-a",
+            use_model=False,
+            action="generate",
+        )
+
+    assert not list((workspace / "final").glob("*.gene_family_report.*"))
+
+
+def test_report_selects_key_facts_without_dumping_manifest_or_stale_paths(tmp_path: Path) -> None:
+    workspace = tmp_path / "job"
+    manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="identify", metric_value=3)
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    manifest["metrics"].update({"total_proteins": 31679, "blast_enabled": True})
+    manifest["artifacts"]["result_table"]["path"] = "/home/xzq/old/job/final/Ft.SPL.identify.tsv"
+    manifest["artifacts"]["result_table"]["full_path"] = "/home/xzq/old/job/final/Ft.SPL.identify.tsv"
+    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
+
+    result = _generate(
+        workspace,
+        expected_stages=["identify"],
+        stage_statuses={"identify": "success"},
+    )
+
+    markdown = result.report_path.read_text(encoding="utf-8")
+    assert "最终成员数：3" in _subsection(markdown, "4.3")
+    assert not re.search(r"(?m)^- `(?:member_count|total_proteins|blast_enabled)\s*`:", markdown)
+    assert "/home/xzq/" not in markdown
+
+
+def test_fallback_uses_nested_karyotype_metrics(tmp_path: Path) -> None:
+    workspace = tmp_path / "job"
+    manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="karyotype")
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    manifest["metrics"] = {
+        "species": {
+            "Ft": {
+                "gene_count": 24,
+                "chromosome_count": 8,
+                "chromosomes": [f"Ft{i}" for i in range(1, 9)],
+                "tandem_count": 0,
+            }
+        }
+    }
+    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
+
+    result = _generate(
+        workspace,
+        expected_stages=["karyotype"],
+        stage_statuses={"karyotype": "success"},
+    )
+
+    markdown = result.report_path.read_text(encoding="utf-8")
+    assert "24个家族成员定位到8条染色体" in markdown
+    assert "核型分析将N/A" not in markdown
+
+
+def test_artifact_path_remaps_remote_job_to_current_mount(tmp_path: Path) -> None:
+    from gfa_core.mas.agents.interpretation_evidence import _artifact_path
+
+    final_dir = tmp_path / "job" / "final"
+    final_dir.mkdir(parents=True)
+    manifest_path = final_dir / "Ft.SPL.stage.manifest.json"
+    local_artifact = final_dir / "Ft.SPL.result.tsv"
+    local_artifact.write_text("gene\tvalue\nFtSPL1\t1\n", encoding="utf-8")
+    manifest = {
+        "_manifest_path": str(manifest_path),
+        "artifacts": {
+            "result": {
+                "full_path": "/home/xzq/old/PlantAI-dev/workspace/job-old/final/Ft.SPL.result.tsv",
+                "path": "/home/xzq/old/PlantAI-dev/workspace/job-old/final/Ft.SPL.result.tsv",
+            }
+        },
+    }
+
+    assert _artifact_path(manifest, "result") == local_artifact
+
+
+def test_failed_or_missing_planned_stage_is_partial(tmp_path: Path) -> None:
+    workspace = tmp_path / "job"
+    _write_manifest(workspace, prefix="Ft.SPL", stage="identify")
+    _write_manifest(workspace, prefix="Ft.SPL", stage="motif", status="failed")
+
+    result = _generate(
+        workspace,
+        expected_stages=["identify", "motif"],
+        stage_statuses={"identify": "success", "motif": "failed"},
+    )
+
+    markdown = result.report_path.read_text(encoding="utf-8")
+    manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
+    assert result.report_status == "partial"
+    assert "motif" in markdown.lower()
+    assert "failed" in markdown.lower()
+    assert manifest["metrics"]["failed_or_missing_stages"] == ["motif"]
+
+
+def test_model_failure_creates_labeled_fallback_report(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    workspace = tmp_path / "job"
+    _write_manifest(workspace, prefix="Ft.SPL", stage="identify")
+
+    def fail_model(_evidence: dict, _stages: list[str]):
+        raise TimeoutError("provider timeout")
+
+    monkeypatch.setattr(report_module, "_generate_api_narrative", fail_model)
+    result = _generate(
+        workspace,
+        expected_stages=["identify"],
+        stage_statuses={"identify": "success"},
+        use_model=True,
+    )
+
+    markdown = result.report_path.read_text(encoding="utf-8")
+    manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
+    assert result.generation_mode == "fallback_script"
+    assert "生成模式：候补脚本" in markdown
+    assert manifest["metrics"]["generation_mode"] == "fallback_script"
+    assert manifest["model"]["reason"] == "api_unavailable_or_boundary_violation"
+    assert "{'action':" not in markdown
+
+
+def test_valid_model_sections_create_api_report(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    workspace = tmp_path / "job"
+    _write_manifest(workspace, prefix="Ft.SPL", stage="identify")
+
+    seen_payload: dict = {}
+
+    def valid_model(payload: dict, _stages: list[str]):
+        seen_payload.update(payload)
+        return (
+            _valid_api_arrangement(payload, reverse=True),
+            {"status": "generated", "provider": "test", "model": "test-model"},
+        )
+
+    monkeypatch.setattr(report_module, "_generate_api_narrative", valid_model)
+    result = _generate(
+        workspace,
+        expected_stages=["identify"],
+        stage_statuses={"identify": "success"},
+        use_model=True,
+    )
+
+    markdown = result.report_path.read_text(encoding="utf-8")
+    manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
+    assert result.generation_mode == "api"
+    assert set(seen_payload) == {"sections", "allowed_connectors"}
+    assert all(set(section) == {"section_id", "sentences"} for section in seen_payload["sections"])
+    assert all(
+        set(sentence) == {"sentence_id", "text"}
+        for section in seen_payload["sections"]
+        for sentence in section["sentences"]
+    )
+    assert "最终成员数：1" in markdown
+    assert manifest["metrics"]["generation_mode"] == "api"
+    assert "【填写】" not in markdown
+
+
+def test_unknown_model_sentence_id_falls_back_for_the_whole_report(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    workspace = tmp_path / "job"
+    _write_manifest(workspace, prefix="Ft.SPL", stage="identify", metric_value=3)
+
+    def invalid_model(payload: dict, _stages: list[str]):
+        response = _valid_api_arrangement(payload)
+        response["sections"][0]["paragraphs"][0]["sentence_ids"][0] = "unknown.sentence"
+        return response, {"status": "generated"}
+
+    monkeypatch.setattr(report_module, "_generate_api_narrative", invalid_model)
+    result = _generate(
+        workspace,
+        expected_stages=["identify"],
+        stage_statuses={"identify": "success"},
+        use_model=True,
+    )
+
+    markdown = result.report_path.read_text(encoding="utf-8")
+    manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
+    assert result.generation_mode == "fallback_script"
+    assert "最终成员数：3" in markdown
+    assert manifest["model"]["reason"] == "api_unavailable_or_boundary_violation"
+
+
+def test_missing_and_duplicate_model_sentence_ids_fall_back_for_the_whole_report(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    workspace = tmp_path / "job"
+    _write_manifest(workspace, prefix="Ft.SPL", stage="identify", metric_value=3)
+
+    def invalid_model(payload: dict, _stages: list[str]):
+        response = _valid_api_arrangement(payload)
+        paragraph = next(
+            item["paragraphs"][0]
+            for item in response["sections"]
+            if len(item["paragraphs"][0]["sentence_ids"]) >= 2
+        )
+        paragraph["sentence_ids"][-1] = paragraph["sentence_ids"][0]
+        return response, {"status": "generated"}
+
+    monkeypatch.setattr(report_module, "_generate_api_narrative", invalid_model)
+    result = _generate(
+        workspace,
+        expected_stages=["identify"],
+        stage_statuses={"identify": "success"},
+        use_model=True,
+    )
+
+    markdown = result.report_path.read_text(encoding="utf-8")
+    manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
+    assert result.generation_mode == "fallback_script"
+    assert "最终成员数：3" in markdown
+    assert manifest["model"]["reason"] == "api_unavailable_or_boundary_violation"
+
+
+@pytest.mark.parametrize(
+    "violation",
+    ["unknown_connector", "extra_field", "free_text_field", "changed_section"],
+)
+def test_model_schema_or_connector_boundary_violation_falls_back_for_the_whole_report(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, violation: str
+) -> None:
+    workspace = tmp_path / "job"
+    _write_manifest(workspace, prefix="Ft.SPL", stage="identify", metric_value=3)
+
+    def invalid_model(payload: dict, _stages: list[str]):
+        response = _valid_api_arrangement(payload)
+        if violation == "unknown_connector":
+            paragraph = next(
+                item["paragraphs"][0]
+                for item in response["sections"]
+                if item["paragraphs"][0]["connector_ids"]
+            )
+            paragraph["connector_ids"][0] = "invented"
+        elif violation == "extra_field":
+            response["unexpected"] = True
+        elif violation == "free_text_field":
+            response["sections"][0]["paragraphs"][0]["text"] = "API 自由文本"
+        else:
+            response["sections"][0]["section_id"] = "changed.section"
+        return response, {"status": "generated"}
+
+    monkeypatch.setattr(report_module, "_generate_api_narrative", invalid_model)
+    result = _generate(
+        workspace,
+        expected_stages=["identify"],
+        stage_statuses={"identify": "success"},
+        use_model=True,
+    )
+
+    markdown = result.report_path.read_text(encoding="utf-8")
+    manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
+    assert result.generation_mode == "fallback_script"
+    assert "最终成员数：3" in markdown
+    assert "API 自由文本" not in markdown
+    assert manifest["model"]["reason"] == "api_unavailable_or_boundary_violation"
+
+
+def test_open_reuses_existing_report(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    workspace = tmp_path / "job"
+    manifest_path = _write_manifest(workspace, prefix="Ft.SPL", stage="identify")
+    calls = {"count": 0}
+
+    def valid_model(payload: dict, _stages: list[str]):
+        calls["count"] += 1
+        return (
+            _valid_api_arrangement(payload),
+            {"status": "generated", "provider": "test", "model": "test-model"},
+        )
+
+    monkeypatch.setattr(report_module, "_generate_api_narrative", valid_model)
+    first = _generate(
+        workspace,
+        expected_stages=["identify"],
+        stage_statuses={"identify": "success"},
+        use_model=True,
+    )
+    reopened = _generate(
+        workspace,
+        expected_stages=["identify"],
+        stage_statuses={"identify": "success"},
+        use_model=True,
+        action="open",
+    )
+    assert first.reused is False
+    assert reopened.reused is True
+    assert calls["count"] == 1
+
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    manifest["metrics"]["member_count"] = 2
+    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8")
+    reopened_after_source_change = _generate(
+        workspace,
+        expected_stages=["identify"],
+        stage_statuses={"identify": "success"},
+        use_model=True,
+        action="open",
+    )
+    assert reopened_after_source_change.reused is True
+    assert calls["count"] == 1
+
+
+def test_no_successful_evidence_creates_no_empty_report(tmp_path: Path) -> None:
+    workspace = tmp_path / "job"
+    _write_manifest(workspace, prefix="Ft.SPL", stage="identify", status="failed")
+
+    with pytest.raises(NoSuccessfulGeneFamilyEvidence, match="暂无可生成"):
+        _generate(
+            workspace,
+            expected_stages=["identify"],
+            stage_statuses={"identify": "failed"},
+        )
+
+    assert not list((workspace / "final").glob("*.gene_family_report.*"))

tests/unit/test_gene_family_report_intent.py

@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+import asyncio
+import json
+
+import pytest
+
+from gfa_core.mas.core.fast_intent import fast_intent_recognition
+from gfa_core.mas.utils.intent_parser import parse_intent_from_query
+
+
+@pytest.mark.parametrize(
+    ("query", "action"),
+    [
+        ("生成结果报告", "generate"),
+        ("生成基因家族结果报告", "generate"),
+        ("打开基因家族结果报告", "open"),
+    ],
+)
+def test_explicit_gene_family_report_intent_is_provider_independent(query: str, action: str) -> None:
+    for parsed in (parse_intent_from_query(query), fast_intent_recognition(query)):
+        assert parsed is not None
+        assert parsed["stages"] == ["gene_family_report"]
+        assert parsed["params"]["geneFamilyReportAction"] == action
+        assert parsed["params"]["reportAction"] == action
+        assert parsed["method"] == "gene_family_report_deterministic"
+
+
+def test_rnaseq_report_intent_is_not_stolen() -> None:
+    for parsed in (
+        parse_intent_from_query("打开 RNA-seq 结果报告"),
+        fast_intent_recognition("打开 RNA-seq 结果报告"),
+    ):
+        assert not parsed or parsed.get("stages") != ["gene_family_report"]
+
+
+@pytest.mark.parametrize(
+    ("query", "action"),
+    [
+        ("生成结果报告", "generate"),
+        ("生成基因家族结果报告", "generate"),
+        ("打开基因家族结果报告", "open"),
+    ],
+)
+def test_gene_family_report_chat_bypasses_router(
+    monkeypatch,
+    tmp_path,
+    query: str,
+    action: str,
+) -> None:
+    from apps.web.vue3 import main as web_main
+    from gfa_core.mas.agents import router as router_module
+
+    final_dir = tmp_path / "workspace" / "job-report" / "final"
+    final_dir.mkdir(parents=True)
+    (final_dir / "Ft.SPL.gene_family_report.manifest.json").write_text(
+        json.dumps(
+            {
+                "stage": "gene_family_report",
+                "status": "success",
+                "prefix": "Ft.SPL",
+                "parameters": {
+                    "expected_stages": ["identify", "motif"],
+                    "stage_statuses": {"identify": "success", "motif": "success"},
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    async def forbidden_router(*_args, **_kwargs):
+        raise AssertionError("report intent must return before Router/LLM")
+
+    class ForbiddenOrchestrator:
+        async def plan(self, *_args, **_kwargs):
+            raise AssertionError("report plan must not call the LLM orchestrator")
+
+    monkeypatch.setattr(web_main, "PROJECT_ROOT", tmp_path)
+    monkeypatch.setattr(web_main, "WORKSPACE_ROOT", tmp_path / "workspace")
+    monkeypatch.setattr(web_main, "agent_conversation_store", None, raising=False)
+    monkeypatch.setattr(web_main, "llm_manager", object())
+    monkeypatch.setattr(web_main, "orchestrator", ForbiddenOrchestrator())
+    monkeypatch.setattr(web_main, "MAS_AVAILABLE", True)
+    monkeypatch.setattr(router_module.RouterAgent, "execute", forbidden_router)
+
+    result = asyncio.run(
+        web_main._run_agent_chat(
+            web_main.AgentChatRequest(message=query, job_id="job-report")
+        )
+    )
+
+    assert result["action"] == "propose_analysis"
+    assert result["model_used"] == "rule"
+    assert result["intent"]["stages"] == ["gene_family_report"]
+    assert result["intent"]["prefix"] == "Ft.SPL"
+    assert result["intent"]["params"]["reportAction"] == action
+    assert result["intent"]["params"]["expectedStages"] == ["identify", "motif"]
+
+    plan = asyncio.run(
+        web_main.agent_plan(
+            web_main.AgentPlanRequest(
+                message=query,
+                job_id="job-report",
+                prefix="Ft.SPL",
+                intent=result["intent"],
+            )
+        )
+    )
+    assert plan["success"] is True
+    assert plan["mode"] == "gene_family_report_deterministic"
+    assert plan["plan"]["stages"] == ["gene_family_report"]
+    assert plan["plan"]["params"]["expectedStages"] == ["identify", "motif"]

tests/unit/test_gene_family_report_triggers.py

@@ -0,0 +1,181 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+
+def _request_and_stages():
+    from apps.web.vue3.main import ConfirmedPlanExecutionRequest, ConfirmedPlanStage
+
+    stages = [
+        ConfirmedPlanStage(
+            id="identify",
+            params={
+                "scopeId": "scope-a",
+                "analysisRoute": "post_subfamily_confirm",
+                "memberSetId": "members-a",
+            },
+        ),
+        ConfirmedPlanStage(id="motif", dependencies=["identify"]),
+        ConfirmedPlanStage(id="pep", dependencies=["motif"]),
+    ]
+    request = ConfirmedPlanExecutionRequest(
+        session_id="session-a",
+        job_id="job-a",
+        prefix="Ft.SPL",
+        plan_id="plan-a",
+        stages=stages,
+    )
+    return request, stages
+
+
+def test_confirmed_plan_terminal_success_requests_final_report() -> None:
+    from apps.web.vue3.main import _confirmed_plan_report_request
+
+    request, stages = _request_and_stages()
+    report = _confirmed_plan_report_request(
+        request,
+        stages,
+        {"status": "success", "completed_stages": ["identify", "motif", "pep"]},
+    )
+
+    assert report is not None
+    assert report.stage == "gene_family_report"
+    assert report.expectedStages == ["identify", "motif", "pep"]
+    assert report.stageStatuses == {"identify": "success", "motif": "success", "pep": "success"}
+    assert report.reportAction == "generate"
+
+
+def test_confirmed_plan_failure_requests_partial_report() -> None:
+    from apps.web.vue3.main import _confirmed_plan_report_request
+
+    request, stages = _request_and_stages()
+    report = _confirmed_plan_report_request(
+        request,
+        stages,
+        {
+            "status": "failed",
+            "completed_stages": ["identify"],
+            "failure": {"stage": "motif"},
+        },
+    )
+
+    assert report is not None
+    assert report.stageStatuses == {"identify": "success", "motif": "failed", "pep": "not_run"}
+    assert report.scopeId == "scope-a"
+    assert report.analysisRoute == "post_subfamily_confirm"
+    assert report.memberSetId == "members-a"
+
+
+def test_paused_plan_does_not_auto_finalize_report() -> None:
+    from apps.web.vue3.main import _confirmed_plan_report_request
+
+    request, stages = _request_and_stages()
+    report = _confirmed_plan_report_request(
+        request,
+        stages,
+        {"status": "paused", "completed_stages": ["identify"]},
+    )
+    assert report is None
+
+
+def test_frontend_batch_terminal_path_calls_report_stage_with_planned_statuses() -> None:
+    source = (Path(__file__).resolve().parents[2] / "apps/web/vue3/index.html").read_text(encoding="utf-8")
+
+    assert "submitGeneFamilyReportForBatch" in source
+    assert "executeSingleStageWithParams('gene_family_report'" in source
+    assert "expectedStages: plannedIds" in source
+    assert "stageStatuses" in source
+    assert "if (pausedCount > 0) return" in source
+    assert "continuationOpened" in source
+    assert "name: 'gene_family_report'" in source
+    assert "openGeneratedGeneFamilyReport" in source
+
+
+def test_report_context_prefers_terminal_confirmed_plan(monkeypatch, tmp_path: Path) -> None:
+    from apps.web.vue3 import main as web_main
+
+    final_dir = tmp_path / "job-a" / "final"
+    confirmed_dir = final_dir / ".confirmed_plans"
+    confirmed_dir.mkdir(parents=True)
+    (final_dir / "Ft.SPL.gene_family_report.manifest.json").write_text(
+        json.dumps(
+            {
+                "stage": "gene_family_report",
+                "status": "success",
+                "prefix": "Ft.SPL",
+                "parameters": {
+                    "expected_stages": ["identify"],
+                    "stage_statuses": {"identify": "success"},
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+    (confirmed_dir / "0123456789abcdefabcd.json").write_text(
+        json.dumps(
+            {
+                "status": "failed",
+                "job_id": "job-a",
+                "prefix": "Ft.SPL",
+                "stages": ["identify", "motif", "pep", "gene_family_report"],
+                "completed_stages": ["identify", "motif"],
+                "failure": {"stage": "pep"},
+                "scopeId": "scope-a",
+                "memberSetId": "members-a",
+            }
+        ),
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(web_main, "WORKSPACE_ROOT", tmp_path)
+
+    context = web_main._gene_family_report_plan_context("job-a")
+
+    assert context == {
+        "prefix": "Ft.SPL",
+        "expectedStages": ["identify", "motif", "pep"],
+        "stageStatuses": {"identify": "success", "motif": "success", "pep": "failed"},
+        "scopeId": "scope-a",
+        "analysisRoute": "",
+        "memberSetId": "members-a",
+        "rnaseqScopeId": "",
+    }
+
+
+def test_report_context_reuses_formal_report_plan_only(monkeypatch, tmp_path: Path) -> None:
+    from apps.web.vue3 import main as web_main
+
+    final_dir = tmp_path / "job-a" / "final"
+    final_dir.mkdir(parents=True)
+    (final_dir / "Ft.SPL.motif.manifest.json").write_text(
+        json.dumps({"stage": "motif", "status": "success", "prefix": "Ft.SPL"}),
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(web_main, "WORKSPACE_ROOT", tmp_path)
+    assert web_main._gene_family_report_plan_context("job-a") is None
+
+    (final_dir / "Ft.SPL.gene_family_report.manifest.json").write_text(
+        json.dumps(
+            {
+                "stage": "gene_family_report",
+                "status": "success",
+                "prefix": "Ft.SPL",
+                "scope_id": "scope-old",
+                "parameters": {
+                    "expected_stages": ["identify", "motif"],
+                    "stage_statuses": {"identify": "success", "motif": "success"},
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    context = web_main._gene_family_report_plan_context("job-a", "Ft.SPL")
+
+    assert context is not None
+    assert context["expectedStages"] == ["identify", "motif"]
+    assert context["stageStatuses"] == {"identify": "success", "motif": "success"}
+    assert context["scopeId"] == ""
+    assert context["analysisRoute"] == ""
+    assert context["memberSetId"] == ""
+    assert context["rnaseqScopeId"] == ""

tests/unit/test_gene_family_report_wiring.py

@@ -0,0 +1,347 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import subprocess
+import sys
+
+
+def _write_identify_manifest(workspace: Path, prefix: str = "Ft.SPL") -> Path:
+    final_dir = workspace / "final"
+    final_dir.mkdir(parents=True, exist_ok=True)
+    table = final_dir / f"{prefix}.identify.tsv"
+    table.write_text("gene\tvalue\nFtSPL1\t1\n", encoding="utf-8")
+    manifest = final_dir / f"{prefix}.identify.manifest.json"
+    manifest.write_text(
+        json.dumps(
+            {
+                "manifest_version": "1.0",
+                "stage": "identify",
+                "status": "success",
+                "prefix": prefix,
+                "metrics": {"member_count": 1},
+                "artifacts": {
+                    "members": {
+                        "path": f"final/{table.name}",
+                        "full_path": str(table),
+                        "type": "table",
+                    }
+                },
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+    return manifest
+
+
+def _flag_value(command: list[str], flag: str) -> str:
+    return command[command.index(flag) + 1]
+
+
+def test_report_stage_is_registered_without_analysis_dependencies() -> None:
+    from gfa_core.mas.stage_capabilities import CAPABILITIES
+    from gfa_core.mas.stage_catalog import STAGE_ORDER, is_stage_executable
+    from gfa_core.mas.tools.stage_tools import create_stage_registry
+
+    capability = CAPABILITIES["gene_family_report"]
+    stage_tool = create_stage_registry().get("gene_family_report")
+
+    assert "gene_family_report" in STAGE_ORDER
+    assert is_stage_executable("gene_family_report")
+    assert capability.consumes == []
+    assert stage_tool is not None
+    assert stage_tool.required_inputs == []
+    assert stage_tool.script == "run_gene_family_report.py"
+
+
+def test_report_cli_can_start_as_a_script() -> None:
+    root = Path(__file__).resolve().parents[2]
+    completed = subprocess.run(
+        [sys.executable, "apps/cli/run_gene_family_report.py", "--help"],
+        cwd=root,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    assert completed.returncode == 0, completed.stderr
+
+
+def test_report_cli_writes_report_and_manifest(tmp_path: Path) -> None:
+    from apps.cli.run_gene_family_report import main
+
+    workspace = tmp_path / "job"
+    _write_identify_manifest(workspace)
+    result = main(
+        [
+            "--workspace",
+            str(workspace),
+            "--prefix",
+            "Ft.SPL",
+            "--job-id",
+            "job-wiring",
+            "--expected-stages",
+            "identify",
+            "--stage-statuses-json",
+            '{"identify":"success"}',
+            "--no-model",
+        ]
+    )
+
+    manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
+    assert result.report_path.is_file()
+    assert result.manifest_path.is_file()
+    assert "gene_family_report" in manifest["artifacts"]
+
+
+def test_pipeline_backend_passes_plan_status_and_identity_to_report_cli(
+    tmp_path: Path, monkeypatch
+) -> None:
+    from apps.web import pipeline_backend
+
+    workspace = tmp_path / "job"
+    input_dir = workspace / "input"
+    logs_dir = workspace / "logs"
+    input_dir.mkdir(parents=True)
+    logs_dir.mkdir(parents=True)
+    _write_identify_manifest(workspace)
+    captured: dict[str, list[str]] = {}
+
+    def capture(command, _log_file, cwd=None):
+        captured["command"] = [str(value) for value in command]
+        return 0
+
+    monkeypatch.setattr(pipeline_backend, "can_start_new_task", lambda: (True, ""))
+    monkeypatch.setattr(pipeline_backend, "run_cmd_logged", capture)
+    job = pipeline_backend.JobInfo(
+        job_id="job-backend",
+        input_dir=str(input_dir),
+        workspace_dir=str(workspace),
+        logs_dir=str(logs_dir),
+        created_at="2026-08-08 00:00:00",
+    )
+
+    pipeline_backend.run_stage(
+        "gene_family_report",
+        job,
+        "Ft.SPL",
+        expected_stages=["identify", "motif"],
+        stage_statuses={"identify": "success", "motif": "failed"},
+        report_action="open",
+        report_use_model=False,
+        scope_id="scope-a",
+        analysis_route="post_subfamily_confirm",
+        member_set_id="members-a",
+        rnaseq_scope_id="rna-a",
+    )
+
+    command = captured["command"]
+    assert command[1:3] == ["apps/cli/run_gene_family_report.py", "--workspace"]
+    assert _flag_value(command, "--expected-stages") == "identify,motif"
+    assert json.loads(_flag_value(command, "--stage-statuses-json")) == {
+        "identify": "success",
+        "motif": "failed",
+    }
+    assert _flag_value(command, "--action") == "open"
+    assert _flag_value(command, "--job-id") == "job-backend"
+    assert _flag_value(command, "--scope-id") == "scope-a"
+    assert _flag_value(command, "--analysis-route") == "post_subfamily_confirm"
+    assert _flag_value(command, "--member-set-id") == "members-a"
+    assert _flag_value(command, "--rnaseq-scope-id") == "rna-a"
+    assert "--no-model" in command
+    assert not any("run_identify.py" in value or "run_motif.py" in value for value in command)
+
+
+def test_pipeline_backend_rejects_empty_plan_before_starting_report_cli(
+    tmp_path: Path, monkeypatch
+) -> None:
+    from apps.web import pipeline_backend
+
+    workspace = tmp_path / "job"
+    input_dir = workspace / "input"
+    logs_dir = workspace / "logs"
+    input_dir.mkdir(parents=True)
+    logs_dir.mkdir(parents=True)
+    invoked = False
+
+    def capture(*_args, **_kwargs):
+        nonlocal invoked
+        invoked = True
+        return 0
+
+    monkeypatch.setattr(pipeline_backend, "can_start_new_task", lambda: (True, ""))
+    monkeypatch.setattr(pipeline_backend, "run_cmd_logged", capture)
+    job = pipeline_backend.JobInfo(
+        job_id="job-empty-plan",
+        input_dir=str(input_dir),
+        workspace_dir=str(workspace),
+        logs_dir=str(logs_dir),
+        created_at="2026-08-08 00:00:00",
+    )
+
+    message, _manifest = pipeline_backend.run_stage(
+        "gene_family_report",
+        job,
+        "Ft.SPL",
+        expected_stages=[],
+        report_action="generate",
+        report_use_model=False,
+    )
+
+    assert "expected_stages" in message
+    assert invoked is False
+
+
+def test_stage_tool_executor_forwards_gene_family_report_context(
+    tmp_path: Path, monkeypatch
+) -> None:
+    import asyncio
+    import types
+    from dataclasses import dataclass
+
+    from gfa_core.mas.tools.stage_tools import StageTool, StageToolExecutor
+
+    captured = {}
+
+    @dataclass
+    class FakeJobInfo:
+        job_id: str
+        input_dir: str
+        workspace_dir: str
+        logs_dir: str
+        created_at: str
+
+    def fake_run_stage(**kwargs):
+        captured.update(kwargs)
+        manifest = Path(kwargs["job"].workspace_dir) / "final" / "Ft.SPL.gene_family_report.manifest.json"
+        manifest.parent.mkdir(parents=True, exist_ok=True)
+        manifest.write_text(
+            json.dumps({"stage": "gene_family_report", "status": "success"}),
+            encoding="utf-8",
+        )
+        return "报告已生成", manifest
+
+    fake_backend = types.ModuleType("apps.web.pipeline_backend")
+    fake_backend.JobInfo = FakeJobInfo
+    fake_backend.run_stage = fake_run_stage
+    monkeypatch.setitem(sys.modules, "apps.web.pipeline_backend", fake_backend)
+
+    result = asyncio.run(
+        StageToolExecutor(project_root=tmp_path).execute(
+            StageTool(
+                "gene_family_report",
+                "Gene Family Report",
+                "run_gene_family_report.py",
+                "report",
+                "test",
+                timeout=5,
+            ),
+            "job-report",
+            "Ft.SPL",
+            {
+                "expectedStages": ["identify", "motif"],
+                "stageStatuses": {"identify": "success", "motif": "failed"},
+                "reportAction": "open",
+                "reportUseModel": False,
+                "scopeId": "scope-a",
+                "rnaseqScopeId": "rna-a",
+            },
+        )
+    )
+
+    assert result["status"] == "success"
+    assert captured["expected_stages"] == ["identify", "motif"]
+    assert captured["stage_statuses"] == {"identify": "success", "motif": "failed"}
+    assert captured["report_action"] == "open"
+    assert captured["report_use_model"] is False
+    assert captured["scope_id"] == "scope-a"
+    assert captured["rnaseq_scope_id"] == "rna-a"
+
+
+def test_execute_report_does_not_invent_scope_identity(monkeypatch) -> None:
+    import asyncio
+    from unittest.mock import Mock
+
+    from apps.web.vue3 import main as web_main
+    from gfa_core.mas import resource_governor
+    from tasks import execute_single_stage_task
+
+    submitted = Mock()
+    submitted.return_value.id = "report-task"
+    monkeypatch.setattr(execute_single_stage_task, "apply_async", submitted)
+    monkeypatch.setattr(
+        web_main,
+        "_ensure_celery_queue_health",
+        lambda *_args, **_kwargs: ({"available": True}, None),
+    )
+    monkeypatch.setattr(
+        web_main,
+        "_ensure_gene_family_scope_file",
+        lambda **_kwargs: ("", ""),
+    )
+    monkeypatch.setattr(resource_governor, "record_task", lambda **_kwargs: {})
+
+    request = web_main.ExecuteRequest(
+        stage="gene_family_report",
+        job_id="job-report-no-invented-identity",
+        prefix="Ft.SPL",
+        expectedStages=["identify", "motif"],
+        stageStatuses={"identify": "success", "motif": "success"},
+    )
+    result = asyncio.run(web_main.execute_stage(request))
+
+    assert result["success"] is True
+    params = submitted.call_args.kwargs["kwargs"]["params"]
+    assert params["scopeId"] == ""
+    assert params["analysisRoute"] == ""
+    assert params["memberSetId"] == ""
+    assert params["rnaseqScopeId"] == ""
+
+
+def test_report_artifact_validation_does_not_require_invented_identity(monkeypatch) -> None:
+    from types import SimpleNamespace
+
+    import tasks
+    from gfa_core.mas.tools import stage_identity
+
+    captured = {}
+    identity = SimpleNamespace(
+        scope_id="",
+        analysis_route="custom",
+        member_set_id="Ft.SPL:custom:all_members",
+        wgs_sample_set_id="",
+        wgs_reference_id="",
+        wgs_analysis_id="",
+        genotype_scope_id="",
+        phenotype_scope_id="",
+        covariate_scope_id="",
+        model_scope_id="",
+        source_model_scope_id="",
+        analysis_scope_id="",
+    )
+    monkeypatch.setattr(stage_identity, "resolve_stage_identity", lambda **_kwargs: identity)
+
+    def capture_status(**kwargs):
+        captured.update(kwargs)
+        return SimpleNamespace(completed=True)
+
+    monkeypatch.setattr(stage_identity, "get_stage_status_for_identity", capture_status)
+
+    failure = tasks.validate_single_stage_artifacts(
+        "job-report",
+        "Ft.SPL",
+        "gene_family_report",
+        {
+            "manifest": {
+                "stage": "gene_family_report",
+                "status": "success",
+                "prefix": "Ft.SPL",
+                "analysis_route": "",
+                "member_set_id": "",
+            }
+        },
+    )
+
+    assert failure is None
+    assert captured["strict_identity"] is False