diff --git a/apps/web/vue3/index_ml.html b/apps/web/vue3/index_ml.html
index 4acfb84e4..48a862bcd 100644
--- a/apps/web/vue3/index_ml.html
+++ b/apps/web/vue3/index_ml.html
@@ -23500,16 +23500,20 @@ compare: [[flesh, peel]]`;
let stagesDetail = (rawExecutableDetails || []).filter(stage => !completedSet.has(stage.id || stage.name));
if (stages.length === 0) {
- const plannedStageIds = (plan.stages || [])
+ const plannedStageIds = (
+ (plan.all_stages || []).length > 0
+ ? plan.all_stages
+ : (plan.stages || [])
+ )
.map(stage => typeof stage === 'object' ? stage.id : stage)
.filter(Boolean);
const completedOnlyPlan = plannedStageIds.length === 0
&& completedSet.size > 0
&& ['所有阶段已完成', 'All stages completed'].includes(String(plan.description || ''));
- const requestedCompletedStage = (plannedStageIds.length > 0
- ? plannedStageIds.filter(stageId => completedSet.has(stageId))
- : Array.from(completedSet)
- ).pop() || '';
+ const requestedCompletedStage = plannedStageIds.length === 1
+ && completedSet.has(plannedStageIds[0])
+ ? plannedStageIds[0]
+ : '';
const allPlannedCompleted = (plannedStageIds.length > 0 && plannedStageIds.every(stageId => completedSet.has(stageId)))
|| completedOnlyPlan;
if (allPlannedCompleted && requestedCompletedStage) {
@@ -23520,6 +23524,11 @@ compare: [[flesh, peel]]`;
taskType: 'analysis_completed',
plan: plan
});
+ } else if (allPlannedCompleted) {
+ addAgentMessage('assistant', t('当前请求中的分析阶段均已有同一分析范围的有效结果，无需重复执行。'), {
+ taskType: 'analysis_completed',
+ plan: plan
+ });
} else {
addAgentMessage('assistant', t('未生成可执行阶段。未检测到当前 scope 的完成结果时，不再把空计划解释为已完成；请检查阶段名称或前置依赖。'), {
taskType: 'analysis_plan_empty',
diff --git a/apps/web/vue3/main.py b/apps/web/vue3/main.py
index 084662736..74bdc9a5f 100644
--- a/apps/web/vue3/main.py
+++ b/apps/web/vue3/main.py
@@ -2682,12 +2682,26 @@ def _normalize_full_gene_family_workflow_plan(
stages_detail: list[Dict[str, Any]],
plan_params: Dict[str, Any],
plan_prefix: str,
+ authorized_stage_ids: Optional[list[str]] = None,
) -> tuple[list[str], list[Dict[str, Any]]]:
if not _looks_like_gene_family_workflow_query(message or ""):
return stage_ids, stages_detail
- desired = list(FULL_GENE_FAMILY_WORKFLOW_STAGES)
- if "expression" in stage_ids or any(token in (message or "").lower() for token in ["表达", "expression", "rna"]):
+ authorized = None if not authorized_stage_ids else {
+ str(stage_id).strip()
+ for stage_id in authorized_stage_ids
+ if str(stage_id).strip()
+ }
+ desired = [
+ stage_id
+ for stage_id in FULL_GENE_FAMILY_WORKFLOW_STAGES
+ if authorized is None or stage_id in authorized
+ ]
+ include_expression = (
+ "expression" in stage_ids
+ or any(token in (message or "").lower() for token in ["表达", "expression", "rna"])
+ )
+ if include_expression and (authorized is None or "expression" in authorized):
desired.append("expression")
shared_refs = _param_list(plan_params.get("referenceSpecies") or plan_params.get("reference_species") or plan_params.get("refSpeciesList"))
@@ -2699,7 +2713,7 @@ def _normalize_full_gene_family_workflow_plan(
plan_params["scopeMode"] = "target_plus_comparative"
ordered: list[str] = []
- if "identify" in stage_ids:
+ if "identify" in stage_ids and (authorized is None or "identify" in authorized):
ordered.append("identify")
for sid in desired:
if sid not in ordered:
@@ -2707,6 +2721,8 @@ def _normalize_full_gene_family_workflow_plan(
for sid in stage_ids:
if sid == "ortholog":
continue
+ if authorized is not None and sid not in authorized:
+ continue
if sid and sid not in ordered:
ordered.append(sid)
@@ -3681,13 +3697,38 @@ def _identity_completed_stages_for_plan(
workspace_root=WORKSPACE_ROOT,
)
status = _stage_status_for_plan_identity(job_id, stage_id, identity, strict_identity=True)
+ workflow_phase = str(
+ merged_status_params.get("workflowPhase")
+ or merged_status_params.get("workflow_phase")
+ or ""
+ ).strip()
+ requires_confirmed = bool(
+ merged_status_params.get("requiresConfirmedSubfamily")
+ or merged_status_params.get("requires_confirmed_subfamily")
+ )
+ contextual_kaks_shared_reference = (
+ stage_id == "kaks"
+ and workflow_phase == "post_subfamily_confirm"
+ and requires_confirmed
+ )
technical_reference_keys = (
- "referenceSpecies",
- "refSpeciesList",
- "ppiReferenceSpecies",
- "referenceBundleSpecies",
- "ppiKaksReferenceSpecies",
- "reference_species",
+ (
+ "ppiReferenceSpecies",
+ "referenceBundleSpecies",
+ "ppiKaksReferenceSpecies",
+ "ppi_reference_species",
+ "reference_bundle_species",
+ "ppi_kaks_reference_species",
+ )
+ if contextual_kaks_shared_reference
+ else (
+ "referenceSpecies",
+ "refSpeciesList",
+ "ppiReferenceSpecies",
+ "referenceBundleSpecies",
+ "ppiKaksReferenceSpecies",
+ "reference_species",
+ )
)
explicit_technical_identity = any(
_param_list(merged_status_params.get(key)) for key in technical_reference_keys
@@ -3774,6 +3815,114 @@ def _identity_completed_stages_for_plan(
strict_identity=True,
)
break
+ if (
+ status.completed is not True
+ and stage_id in set(FULL_GENE_FAMILY_WORKFLOW_STAGES) - {"ppi", "kaks"}
+ and workflow_phase == "post_subfamily_confirm"
+ and requires_confirmed
+ ):
+ requested_family = str(
+ merged_status_params.get("geneFamily")
+ or merged_status_params.get("gene_family")
+ or (str(detail_prefix).split(".")[-1] if "." in str(detail_prefix) else "")
+ ).strip()
+ requested_targets = _param_list(
+ merged_status_params.get("targetSpecies")
+ or merged_status_params.get("target_species")
+ )
+ expected_route = str(
+ merged_status_params.get("analysisRoute")
+ or merged_status_params.get("analysis_route")
+ or ""
+ ).strip()
+ plan_references = _param_list(
+ (plan_params or {}).get("referenceSpecies")
+ or (plan_params or {}).get("reference_species")
+ )
+ final_dir = WORKSPACE_ROOT / job_id / "final"
+ latest_confirmation_mtime = 0.0
+ for confirmation_path in final_dir.glob("*.subfamily_confirm.manifest.json") if final_dir.is_dir() else []:
+ try:
+ confirmation = json.loads(confirmation_path.read_text(encoding="utf-8"))
+ except Exception:
+ continue
+ if (
+ confirmation.get("status") == "success"
+ and str(confirmation.get("prefix") or "").strip() == str(detail_prefix)
+ ):
+ latest_confirmation_mtime = max(latest_confirmation_mtime, confirmation_path.stat().st_mtime)
+ candidates = sorted(
+ final_dir.glob(f"*.{stage_id}.manifest.json") if final_dir.is_dir() else [],
+ key=lambda path: path.stat().st_mtime,
+ reverse=True,
+ )
+ for manifest_path in candidates:
+ try:
+ manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+ except Exception:
+ continue
+ stage_context = manifest.get("stage_context") if isinstance(manifest.get("stage_context"), dict) else {}
+ parameters = manifest.get("parameters") if isinstance(manifest.get("parameters"), dict) else {}
+
+ def regular_manifest_value(*keys: str):
+ for source in (manifest, stage_context, parameters):
+ for key in keys:
+ if source.get(key) not in (None, "", []):
+ return source.get(key)
+ return ""
+
+ manifest_prefix = str(manifest.get("prefix") or manifest.get("output_prefix") or "").strip()
+ manifest_family = str(
+ regular_manifest_value("family", "gene_family")
+ or (manifest_prefix.split(".")[-1] if "." in manifest_prefix else "")
+ ).strip()
+ manifest_targets = _param_list(regular_manifest_value("target_species", "targetSpecies"))
+ manifest_references = _param_list(regular_manifest_value("reference_species", "referenceSpecies"))
+ manifest_route = str(regular_manifest_value("analysis_route", "analysisRoute")).strip()
+ if (
+ manifest.get("stage") != stage_id
+ or manifest.get("status") != "success"
+ or manifest_prefix != str(detail_prefix)
+ or manifest_family != requested_family
+ or set(manifest_targets) != set(requested_targets)
+ or manifest_route != expected_route
+ or (
+ manifest_references
+ and set(manifest_references) != set(plan_references)
+ )
+ or (
+ latest_confirmation_mtime
+ and manifest_path.stat().st_mtime < latest_confirmation_mtime
+ )
+ ):
+ continue
+ inherited_params = {
+ **merged_status_params,
+ "prefix": detail_prefix,
+ "targetSpecies": manifest_targets,
+ "referenceSpecies": manifest_references,
+ "scopeMode": regular_manifest_value("scope_mode", "scopeMode"),
+ "analysisRoute": manifest_route,
+ "memberSetId": regular_manifest_value("member_set_id", "memberSetId"),
+ }
+ inherited_params.pop("scopeId", None)
+ inherited_params.pop("scope_id", None)
+ inherited_identity = resolve_stage_identity(
+ stage=stage_id,
+ prefix=detail_prefix,
+ params=inherited_params,
+ job_id=job_id,
+ workspace_root=WORKSPACE_ROOT,
+ )
+ inherited_status = _stage_status_for_plan_identity(
+ job_id,
+ stage_id,
+ inherited_identity,
+ strict_identity=True,
+ )
+ if inherited_status.completed is True:
+ status = inherited_status
+ break
if status.completed is True:
if stage_id == "combo":
required_panels = _combo_panels_from_params(merged_status_params)
@@ -19096,6 +19245,7 @@ def _build_contextual_gene_family_plan_response(request: Any, session_id: str, c
stages_detail=stages_detail,
plan_params=plan_params,
plan_prefix=plan_prefix,
+ authorized_stage_ids=list(request_intent.get("stages") or requested_stage_ids),
)
completed_stages: list[str] = []
@@ -21292,6 +21442,11 @@ async def agent_plan(request: AgentPlanRequest):
"description": "执行完整 GWAS 分析链",
"confidence": 0.98,
})
+ authoritative_intent_stage_ids = [
+ str(stage_id).strip()
+ for stage_id in (incoming_intent.get("stages") or [])
+ if str(stage_id).strip()
+ ]
incoming_params = dict(incoming_intent.get("params") or {})
def _request_list(*keys):
for key in keys:
@@ -22202,6 +22357,11 @@ async def agent_plan(request: AgentPlanRequest):
if result.get("status") == "completed":
plan = result.get("final_output", {}).get("analysis_plan", {})
intent = result.get("final_output", {}).get("intent", {})
+ plan_authorized_stage_ids = authoritative_intent_stage_ids or [
+ str(stage_id).strip()
+ for stage_id in (intent.get("stages") or [])
+ if str(stage_id).strip()
+ ]
if authoritative_full_gwas_stages:
gwas_prefix = (
intent.get("prefix")
@@ -22907,6 +23067,7 @@ async def agent_plan(request: AgentPlanRequest):
stages_detail=stages_detail,
plan_params=plan_params,
plan_prefix=plan_prefix or effective_plan_prefix or request.prefix or "",
+ authorized_stage_ids=plan_authorized_stage_ids,
)
raw_stages = stage_ids
plan["stages"] = stage_ids
@@ -22939,6 +23100,7 @@ async def agent_plan(request: AgentPlanRequest):
stages_detail=stages_detail,
plan_params=plan_params,
plan_prefix=plan_prefix or effective_plan_prefix or request.prefix or "",
+ authorized_stage_ids=plan_authorized_stage_ids,
)
raw_stages = stage_ids
plan["stages"] = stage_ids
diff --git a/apps/web/vue3/static/i18n_en.js b/apps/web/vue3/static/i18n_en.js
index e72f794f4..df1fb06b9 100644
--- a/apps/web/vue3/static/i18n_en.js
+++ b/apps/web/vue3/static/i18n_en.js
@@ -593,6 +593,7 @@ window.__I18N_EN__ = {
" 亚家族分类已完成预测，但确认表尚未保存；后续会继续显示为待确认。": " subfamily classification predicted; confirmation table not yet saved; will show as pending.",
" 到后端；系统会先生成 ": " to backend; system will generate ",
" 前端连接短暂中断，正在按后端产物状态恢复，不会立即判定阶段失败": " frontend connection interrupted; recovering from backend artifact state. Stage will not be marked as failed.",
+ "当前请求中的分析阶段均已有同一分析范围的有效结果，无需重复执行。": "All analysis stages in the current request already have valid results for the same analysis scope. No need to re-execute.",
" 已有同一分析范围的有效结果，无需重复执行。": " already has valid results for the same analysis scope. No need to re-execute.",
" 的依赖尚未完成": " dependency not yet completed",
" 重新提交（自动注释背景）...": " re-submitting (auto-annotation background)...",
diff --git a/gfa_core/mas/agents/intent_v2.py b/gfa_core/mas/agents/intent_v2.py
index d0860a5aa..648c2788e 100755
--- a/gfa_core/mas/agents/intent_v2.py
+++ b/gfa_core/mas/agents/intent_v2.py
@@ -13,6 +13,7 @@ from gfa_core.mas.stage_catalog import (
complete_dependencies as catalog_complete_dependencies,
)
from gfa_core.mas.utils.intent_parser import (
+ extract_negated_target_stages,
is_gwas_fmt_intent,
parse_gene_family,
parse_gene_family_scope_selection,
@@ -154,6 +155,18 @@ class IntentAgentV2(BaseAgent):
def __init__(self, llm_manager: LLMManager):
super().__init__(llm_manager)
+ @staticmethod
+ def _current_request_text(query: str) -> str:
+ """Return the user request without Router-appended confirmed context."""
+ text = str(query or "")
+ boundaries = [
+ index
+ for marker in ("\n已确认上下文：", "\nConfirmed context:")
+ if (index := text.find(marker)) >= 0
+ ]
+ request = text[:min(boundaries)] if boundaries else text
+ return request.strip() or text.strip()
+
@staticmethod
def _apply_structured_goal_context(
intent: Dict[str, Any],
@@ -237,6 +250,7 @@ class IntentAgentV2(BaseAgent):
使用LLM理解用户输入，提取完整意图信息
"""
query = message.payload.get("query", "")
+ request_query = self._current_request_text(query)
# 获取前端传递的 prefix（例如 "At.WRKY"）
provided_prefix = message.payload.get("prefix", "")
@@ -264,7 +278,15 @@ class IntentAgentV2(BaseAgent):
# ===== 关键修改：使用公共模块的硬编码规则解析意图 =====
from gfa_core.mas.utils.intent_parser import parse_intent_from_query
- base_intent = parse_intent_from_query(query)
+ request_intent = parse_intent_from_query(request_query)
+ context_intent = parse_intent_from_query(query)
+ base_intent = dict(request_intent)
+ for key in (
+ "gene_family", "species", "target_species", "reference_species",
+ "comparative_species", "scopeMode", "params", "prefix",
+ ):
+ if context_intent.get(key):
+ base_intent[key] = context_intent[key]
print(f"[IntentAgentV2] Base intent from parser: {base_intent}")
# 如果硬编码规则解析成功（有明确阶段），直接使用
@@ -301,7 +323,22 @@ class IntentAgentV2(BaseAgent):
else:
# 硬编码规则未命中，使用 LLM 细化 stages
print(f"[IntentAgentV2] Hardcoded rule not hit, using LLM")
- intent_result = await self._recognize_intent_with_llm(query, completed_stages, provided_prefix)
+ intent_result = await self._recognize_intent_with_llm(request_query, completed_stages, provided_prefix)
+ excluded_stages = set(extract_negated_target_stages(request_query))
+ if excluded_stages:
+ intent_result["stages"] = [
+ stage for stage in intent_result.get("stages", [])
+ if stage not in excluded_stages
+ ]
+ if intent_result.get("primary_stage") in excluded_stages:
+ intent_result["primary_stage"] = intent_result["stages"][0] if intent_result["stages"] else None
+ if not intent_result["stages"]:
+ intent_result.update({
+ "description": "未检测到需要执行的正向分析阶段",
+ "needs_clarification": True,
+ "clarification_type": "analysis_stage",
+ "clarification_question": "请说明要执行的具体分析阶段",
+ })
# 关键修复：保护 Parser 已解析的 gene_family/species，不被 LLM 的"未知"覆盖
if base_intent.get("gene_family") and base_intent["gene_family"] not in ["未知", "待用户指定"]:
@@ -803,6 +840,22 @@ class IntentAgentV2(BaseAgent):
}
)
+ excluded_stages = set(extract_negated_target_stages(query))
+ if excluded_stages:
+ intent_data["stages"] = [
+ stage for stage in intent_data.get("stages", [])
+ if stage not in excluded_stages
+ ]
+ if intent_data.get("primary_stage") in excluded_stages:
+ intent_data["primary_stage"] = intent_data["stages"][0] if intent_data["stages"] else None
+ if not intent_data["stages"]:
+ intent_data.update({
+ "description": "未检测到需要执行的正向分析阶段",
+ "needs_clarification": True,
+ "clarification_type": "analysis_stage",
+ "clarification_question": "请说明要执行的具体分析阶段",
+ })
+
# 根据依赖关系补充缺失的阶段（关键修复：确保依赖展开）
original_stages = intent_data["stages"]
intent_data["stages"] = self._complete_dependencies(original_stages)
@@ -1069,6 +1122,8 @@ class IntentAgentV2(BaseAgent):
stages = [stage for stage in stages if stage not in {"identify", "prepare_extract_pep", "prepare_extract_cds"}]
if any(stage.startswith("rnaseq") for stage in stages):
stages = [stage for stage in stages if stage not in {"expression", "msa"}]
+ excluded_stages = set(extract_negated_target_stages(query))
+ stages = [stage for stage in stages if stage not in excluded_stages]
return list(dict.fromkeys(stages))
def _fallback_recognition(self, query: str) -> Dict[str, Any]:
@@ -1077,13 +1132,20 @@ class IntentAgentV2(BaseAgent):
# 使用关键词匹配提取阶段
stages = self._extract_stages_from_query(query)
+ excluded_stages = set(extract_negated_target_stages(query))
if not stages:
query_lower = query.lower()
has_rnaseq_keyword = any(
kw in query_lower for kw in ["rnaseq", "rna-seq", "rna seq", "转录组", "fastq"]
)
if has_rnaseq_keyword:
- stages = ["rnaseq_de", "rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea", "rnaseq_wgcna"]
+ stages = [
+ stage for stage in (
+ "rnaseq_de", "rnaseq_go_enrich", "rnaseq_kegg_enrich",
+ "rnaseq_fgsea", "rnaseq_wgcna",
+ )
+ if stage not in excluded_stages
+ ]
# 补充依赖（使用 stage_catalog 的主规划依赖链）
original_stages = stages.copy()
diff --git a/gfa_core/mas/core/fast_intent.py b/gfa_core/mas/core/fast_intent.py
index d32dc204f..626ba34a7 100755
--- a/gfa_core/mas/core/fast_intent.py
+++ b/gfa_core/mas/core/fast_intent.py
@@ -4,7 +4,15 @@
import re
from typing import Optional, Dict, Any, List
from gfa_core.mas.stage_catalog import complete_dependencies as catalog_complete_dependencies
-from gfa_core.mas.utils.intent_parser import is_gwas_fmt_intent, parse_gene_family_scope_selection, resolve_gene_family_report_intent, resolve_gwas_intent, resolve_population_intent
+from gfa_core.mas.utils.intent_parser import (
+ extract_negated_target_stages,
+ is_gwas_fmt_intent,
+ parse_gene_family_scope_selection,
+ parse_intent_from_query,
+ resolve_gene_family_report_intent,
+ resolve_gwas_intent,
+ resolve_population_intent,
+)
# 分析类型关键词映射（注意：key必须与orchestrator.py中的analysis_type匹配）
ANALYSIS_KEYWORDS = {
@@ -180,6 +188,28 @@ def fast_intent_recognition(query: str) -> Optional[Dict[str, Any]]:
}
except Exception:
pass
+ excluded_stages = extract_negated_target_stages(query)
+ if excluded_stages:
+ parsed = parse_intent_from_query(query)
+ target_stages = list(parsed.get("stages") or [])
+ if not target_stages:
+ return None
+ all_stages = _complete_dependencies(target_stages)
+ return {
+ "stage": parsed.get("primary_stage") or target_stages[0],
+ "stages": all_stages,
+ "gene_family": parsed.get("gene_family", ""),
+ "species": parsed.get("species", ""),
+ "prefix": parsed.get("prefix", ""),
+ "target_species": parsed.get("target_species", []),
+ "reference_species": parsed.get("reference_species", []),
+ "comparative_species": parsed.get("comparative_species", []),
+ "scopeMode": parsed.get("scopeMode", "target_only"),
+ "params": parsed.get("params", {}),
+ "confidence": parsed.get("confidence", 0.9),
+ "task_kind": parsed.get("task_kind", "pipeline"),
+ "method": "keyword_negation",
+ }
has_rnaseq_marker = any(marker in query_lower for marker in RNASEQ_MARKERS)
if has_rnaseq_marker and any(
keyword in query_lower
diff --git a/gfa_core/mas/utils/intent_parser.py b/gfa_core/mas/utils/intent_parser.py
index 2afae0fea..d24f9ac4f 100755
--- a/gfa_core/mas/utils/intent_parser.py
+++ b/gfa_core/mas/utils/intent_parser.py
@@ -1344,6 +1344,46 @@ def _build_msa_region_plot_message(params: Dict) -> str:
return "我理解你想对 MSA 指定区间进行局部绘图；请在确认弹窗中填写 alignment 起止列。"
+def _stage_keyword_pattern(keyword: str) -> str:
+ """Build a boundary-safe regex for one existing stage keyword."""
+ escaped = re.escape(keyword.lower())
+ if keyword and keyword[0].isascii() and keyword[0].isalnum():
+ escaped = rf"(? List[str]:
+ """Return explicitly excluded ordinary gene-family/RNA-seq target stages."""
+ query_lower = str(query or "").lower()
+ if any(marker in query_lower for marker in (
+ "亚家族", "subfamily", "reference_classification", "subfamily_confirm",
+ "build_subfamily_evidence", "subfamily_evidence", "subfamily_prediction",
+ )):
+ return []
+ if re.search(r"(?:不是|并非)\s*(?:不执行|不单独执行|不要|不用|无需|不做|跳过|排除|去掉)", query_lower):
+ return []
+
+ eligible = set(CORE_STAGES) | set(RNASEQ_STAGES) | {"expression", "annotation", "msa"}
+ chinese_prefix = r"(?:不单独执行|不执行|不要(?:执行|做|进行)?|不用(?:执行|做|进行)?|无需(?:执行|做|进行)?|不做|跳过|排除|去掉)\s*"
+ english_prefix = r"(?:without(?:\s+(?:running|executing|performing))?|skip|exclude|do\s+not\s+(?:run|execute|perform)?)\s+"
+ excluded = []
+ for stage, keywords in STAGE_KEYWORDS.items():
+ if stage not in eligible:
+ continue
+ for keyword in sorted(keywords, key=len, reverse=True):
+ keyword_pattern = _stage_keyword_pattern(keyword)
+ if (
+ re.search(chinese_prefix + keyword_pattern, query_lower)
+ or re.search(english_prefix + keyword_pattern, query_lower)
+ or re.search(r"(?:^|[\s,，;；。:：])-\s*" + keyword_pattern, query_lower)
+ ):
+ excluded.append(stage)
+ break
+ return list(dict.fromkeys(excluded))
+
+
def _handle_stage_resolution(query: str, query_lower: str, stages: List[str]) -> Dict:
"""
方案B: 处理阶段解析逻辑
@@ -1352,11 +1392,15 @@ def _handle_stage_resolution(query: str, query_lower: str, stages: List[str]) ->
- 有"分析"但不明确 → 引导澄清
- 完全无法识别 → 引导澄清
"""
+ excluded_stages = set(extract_negated_target_stages(query))
+ stages = [stage for stage in stages if stage not in excluded_stages]
+
# 情况1: 明确指定了具体阶段 → 直接执行。
# 如果同时出现“完整/全流程”等词，则以基础基因家族链路为主体，
# 再追加用户明确点名的 Expression/KaKs/PPI/种间共线性等可选阶段。
if stages:
resolved_stages = list(dict.fromkeys([*CORE_STAGES, *stages])) if _is_complete_workflow_query(query_lower) else stages
+ resolved_stages = [stage for stage in resolved_stages if stage not in excluded_stages]
return {
"stages": resolved_stages,
"need_clarify": False,
@@ -1368,7 +1412,13 @@ def _handle_stage_resolution(query: str, query_lower: str, stages: List[str]) ->
if any(marker in query_lower for marker in RNASEQ_MARKERS):
if any(kw in query_lower for kw in ["分析", "执行", "运行", "开始", "流程", "pipeline"]):
return {
- "stages": ["rnaseq_de", "rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea", "rnaseq_wgcna", "rnaseq_candidate_family"],
+ "stages": [
+ stage for stage in (
+ "rnaseq_de", "rnaseq_go_enrich", "rnaseq_kegg_enrich",
+ "rnaseq_fgsea", "rnaseq_wgcna", "rnaseq_candidate_family",
+ )
+ if stage not in excluded_stages
+ ],
"need_clarify": False,
"confidence": 0.9,
"note": "generic_rnaseq_workflow"
@@ -1377,7 +1427,7 @@ def _handle_stage_resolution(query: str, query_lower: str, stages: List[str]) ->
# 情况2: "基因家族分析" = 基础基因家族链路
if _is_complete_workflow_query(query_lower):
return {
- "stages": CORE_STAGES,
+ "stages": [stage for stage in CORE_STAGES if stage not in excluded_stages],
"need_clarify": False,
"confidence": 0.9,
"note": "完整流程"
diff --git a/tests/unit/test_gene_family_empty_plan_frontend.py b/tests/unit/test_gene_family_empty_plan_frontend.py
new file mode 100644
index 000000000..665845ed8
--- /dev/null
+++ b/tests/unit/test_gene_family_empty_plan_frontend.py
@@ -0,0 +1,90 @@
+import json
+import subprocess
+from pathlib import Path
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+FRONTEND_PATH = PROJECT_ROOT / "apps/web/vue3/index_ml.html"
+I18N_PATH = PROJECT_ROOT / "apps/web/vue3/static/i18n_en.js"
+
+
+def _run_empty_plan_branch(tmp_path: Path, plan_data: dict) -> dict:
+ frontend = FRONTEND_PATH.read_text(encoding="utf-8")
+ start = frontend.index("const plan = planData.plan;")
+ end = frontend.index("// 显示智能规划结果", start)
+ branch = frontend[start:end]
+ script = tmp_path / "empty-plan-branch.cjs"
+ script.write_text(
+ "const planData=" + json.dumps(plan_data, ensure_ascii=False) + ";"
+ "const events={messages:[],filters:[],refreshes:[]};"
+ "const t=value=>value;"
+ "const getStageLabel=id=>({genome_prepare:'0️⃣ Genome Prepare',ppi:'PPI分析'}[id]||id);"
+ "const setResultStageFilter=id=>events.filters.push(id);"
+ "const refreshResults=async(jobId,id)=>events.refreshes.push([jobId,id]);"
+ "const addAgentMessage=(role,message,meta)=>events.messages.push({role,message,meta});"
+ "const jobInfo={job_id:'job-test'};"
+ "const nlLoading={value:true};const nlInput={value:'request'};"
+ "async function run(){" + branch + "}"
+ "run().then(()=>process.stdout.write(JSON.stringify(events)));",
+ encoding="utf-8",
+ )
+ completed = subprocess.run(
+ ["/usr/bin/node", str(script)],
+ text=True,
+ capture_output=True,
+ check=False,
+ )
+ assert completed.returncode == 0, completed.stderr
+ return json.loads(completed.stdout)
+
+
+def test_completed_full_workflow_does_not_masquerade_as_genome_prepare(tmp_path: Path) -> None:
+ requested = ["phylogeny", "motif", "domain", "structure", "combo", "promoter"]
+ result = _run_empty_plan_branch(
+ tmp_path,
+ {
+ "success": True,
+ "plan": {
+ "stages": [],
+ "all_stages": requested,
+ "description": "所有阶段已完成",
+ },
+ "completed_stages": [*requested, "genome_prepare"],
+ },
+ )
+
+ assert result["filters"] == []
+ assert result["refreshes"] == []
+ assert [item["message"] for item in result["messages"]] == [
+ "当前请求中的分析阶段均已有同一分析范围的有效结果，无需重复执行。"
+ ]
+
+
+def test_completed_single_stage_keeps_stage_specific_result_handoff(tmp_path: Path) -> None:
+ result = _run_empty_plan_branch(
+ tmp_path,
+ {
+ "success": True,
+ "plan": {
+ "stages": [],
+ "all_stages": ["ppi"],
+ "description": "所有阶段已完成",
+ },
+ "completed_stages": ["ppi"],
+ },
+ )
+
+ assert result["filters"] == ["ppi"]
+ assert result["refreshes"] == [["job-test", "ppi"]]
+ assert [item["message"] for item in result["messages"]] == [
+ "PPI分析 已有同一分析范围的有效结果，无需重复执行。"
+ ]
+
+
+def test_generic_completed_plan_label_has_english_translation() -> None:
+ translations = I18N_PATH.read_text(encoding="utf-8")
+
+ assert (
+ '"当前请求中的分析阶段均已有同一分析范围的有效结果，无需重复执行。": '
+ '"All analysis stages in the current request already have valid results for the same analysis scope. No need to re-execute."'
+ ) in translations
diff --git a/tests/unit/test_gene_family_plan_stage_authority.py b/tests/unit/test_gene_family_plan_stage_authority.py
new file mode 100644
index 000000000..573cad96b
--- /dev/null
+++ b/tests/unit/test_gene_family_plan_stage_authority.py
@@ -0,0 +1,124 @@
+from pathlib import Path
+
+
+_ORIGINAL_PATH_MKDIR = Path.mkdir
+
+
+def _ignore_broken_remote_mount_mkdir(path: Path, *args, **kwargs):
+ if path.is_symlink() and not path.exists():
+ return None
+ return _ORIGINAL_PATH_MKDIR(path, *args, **kwargs)
+
+
+Path.mkdir = _ignore_broken_remote_mount_mkdir
+try:
+ from apps.web.vue3 import main as web_main
+finally:
+ Path.mkdir = _ORIGINAL_PATH_MKDIR
+
+
+def _full_plan_inputs():
+ stage_ids = ["identify", *web_main.FULL_GENE_FAMILY_WORKFLOW_STAGES]
+ details = [web_main._stage_detail_defaults(stage_id) for stage_id in stage_ids]
+ params = {
+ "targetSpecies": ["Ft"],
+ "referenceSpecies": ["At"],
+ "geneFamily": "SPL",
+ }
+ return stage_ids, details, params
+
+
+def test_full_workflow_normalization_respects_intent_stage_exclusions():
+ stage_ids, details, params = _full_plan_inputs()
+
+ for excluded_stage in ("kaks", "karyotype"):
+ authorized = [stage for stage in stage_ids if stage != excluded_stage]
+ normalized, _ = web_main._normalize_full_gene_family_workflow_plan(
+ message=f"执行基因家族分析，但是不要执行{excluded_stage}",
+ stage_ids=stage_ids,
+ stages_detail=details,
+ plan_params=dict(params),
+ plan_prefix="Ft.SPL",
+ authorized_stage_ids=authorized,
+ )
+
+ assert excluded_stage not in normalized
+ assert set(normalized) == set(authorized)
+
+
+def test_full_workflow_normalization_keeps_full_intent_and_required_dependencies():
+ stage_ids, details, params = _full_plan_inputs()
+ normalized, _ = web_main._normalize_full_gene_family_workflow_plan(
+ message="执行基因家族分析",
+ stage_ids=stage_ids,
+ stages_detail=details,
+ plan_params=dict(params),
+ plan_prefix="Ft.SPL",
+ authorized_stage_ids=stage_ids,
+ )
+
+ assert normalized == stage_ids
+ assert normalized.index("ppi") < normalized.index("kaks")
+
+ normalized_without_supplied_intent, _ = web_main._normalize_full_gene_family_workflow_plan(
+ message="执行基因家族分析",
+ stage_ids=stage_ids,
+ stages_detail=details,
+ plan_params=dict(params),
+ plan_prefix="Ft.SPL",
+ authorized_stage_ids=[],
+ )
+ assert normalized_without_supplied_intent == stage_ids
+
+ dependency_stages = ["identify", "synteny", "karyotype"]
+ dependency_details = [web_main._stage_detail_defaults(stage) for stage in dependency_stages]
+ normalized_dependencies, _ = web_main._normalize_full_gene_family_workflow_plan(
+ message="执行核型分析",
+ stage_ids=dependency_stages,
+ stages_detail=dependency_details,
+ plan_params=dict(params),
+ plan_prefix="Ft.SPL",
+ authorized_stage_ids=dependency_stages,
+ )
+ assert normalized_dependencies == dependency_stages
+
+
+def test_contextual_full_workflow_uses_request_intent_as_stage_authority():
+ stage_ids, _, params = _full_plan_inputs()
+ authorized = [stage for stage in stage_ids if stage != "kaks"]
+ request = web_main.AgentPlanRequest(
+ message="执行基因家族分析，但是不要执行kaks分析",
+ prefix="Ft.SPL",
+ intent={"stages": authorized, "params": params},
+ )
+ contextual_request = {
+ "family": "SPL",
+ "prefix": "Ft.SPL",
+ "stages": stage_ids,
+ "targetSpecies": ["Ft"],
+ "referenceSpecies": ["At"],
+ "comparativeSpecies": [],
+ "scopeMode": "target_plus_reference",
+ "requiresConfirmedSubfamily": True,
+ "skipSubfamilyLabels": False,
+ "subfamilyState": {"by_species": {"Ft": {"status": "confirmed"}}},
+ "contextGate": {"status": "ready", "issues": []},
+ }
+
+ response = web_main._build_contextual_gene_family_plan_response(
+ request,
+ "test_intent_stage_authority",
+ contextual_request,
+ )
+
+ assert "kaks" not in response["plan"]["stages"]
+
+
+def test_llm_plan_normalization_calls_keep_original_intent_stage_authority():
+ source = (web_main.PROJECT_ROOT / "apps/web/vue3/main.py").read_text(encoding="utf-8")
+ start = source.index("full_workflow_normalized = False")
+ end = source.index("stage_ids, stages_detail = _apply_standard_combo_panel_plan", start)
+ normalization_block = source[start:end]
+
+ assert "plan_authorized_stage_ids = authoritative_intent_stage_ids or" in source
+ assert normalization_block.count("authorized_stage_ids=plan_authorized_stage_ids") == 2
diff --git a/tests/unit/test_intent_stage_negation.py b/tests/unit/test_intent_stage_negation.py
new file mode 100644
index 000000000..b5c2f1d3a
--- /dev/null
+++ b/tests/unit/test_intent_stage_negation.py
@@ -0,0 +1,144 @@
+import asyncio
+import json
+from types import SimpleNamespace
+
+from gfa_core.mas.agents.base import AgentMessage, AgentState
+from gfa_core.mas.agents.intent_v2 import IntentAgentV2
+from gfa_core.mas.core.fast_intent import fast_intent_recognition
+from gfa_core.mas.utils.intent_parser import CORE_STAGES, parse_intent_from_query
+
+
+def test_complete_gene_family_workflow_excludes_explicitly_negated_ppi():
+ intent = parse_intent_from_query("执行完整流程，但不执行PPI")
+
+ assert intent["stages"] == [stage for stage in CORE_STAGES if stage != "ppi"]
+
+
+def test_symbolic_minus_excludes_stage_from_complete_workflow():
+ intent = parse_intent_from_query("完整流程 - PPI")
+
+ assert intent["stages"] == [stage for stage in CORE_STAGES if stage != "ppi"]
+
+
+def test_mixed_positive_and_negative_stages_keep_only_positive_targets():
+ intent = parse_intent_from_query("做系统发育和Motif，不要做结构域分析")
+
+ assert intent["stages"] == ["phylogeny", "motif"]
+
+
+def test_positive_ppi_and_kaks_request_is_unchanged():
+ intent = parse_intent_from_query("执行PPI和KaKs分析")
+
+ assert intent["stages"] == ["kaks", "ppi"]
+
+
+def test_generic_rnaseq_workflow_excludes_wgcna_target():
+ intent = parse_intent_from_query("执行RNA-seq分析，但不要WGCNA")
+
+ assert intent["stages"] == [
+ "rnaseq_de",
+ "rnaseq_go_enrich",
+ "rnaseq_kegg_enrich",
+ "rnaseq_fgsea",
+ "rnaseq_candidate_family",
+ ]
+
+
+def test_pure_negative_request_never_becomes_positive_in_rule_fallbacks():
+ parsed = parse_intent_from_query("不要执行PPI")
+ fallback = IntentAgentV2(None)._fallback_recognition("不要执行PPI")
+ fast = fast_intent_recognition("不要执行PPI")
+
+ assert parsed["stages"] == []
+ assert parsed["need_clarify"] is True
+ assert fallback["stages"] == []
+ assert fallback["needs_clarification"] is True
+ assert fast is None
+
+
+def test_model_output_cannot_reintroduce_a_negated_stage():
+ class _PpiLLM:
+ async def acall(self, *_args, **_kwargs):
+ return SimpleNamespace(
+ content=json.dumps(
+ {
+ "gene_family": "SPL",
+ "species": "Ft",
+ "stages": ["ppi"],
+ "primary_stage": "ppi",
+ "description": "执行PPI分析",
+ "requirements": [],
+ "confidence": 0.9,
+ }
+ )
+ )
+
+ state = AgentState(session_id="negated-stage", job_id="job-negated-stage")
+ message = AgentMessage(
+ sender="router",
+ receiver="intent_v2",
+ message_type="recognize_intent",
+ payload={"query": "不要执行PPI"},
+ )
+
+ returned = asyncio.run(IntentAgentV2(_PpiLLM()).execute(state, message))
+ intent = returned.results["results"]["intent"]
+
+ assert intent["stages"] == []
+ assert intent["needs_clarification"] is True
+
+
+def test_dependency_completion_can_restore_required_non_target_stage():
+ class _ForbiddenLLM:
+ async def acall(self, *_args, **_kwargs):
+ raise AssertionError("deterministic intent must not call the LLM")
+
+ state = AgentState(session_id="dependency", job_id="job-dependency")
+ message = AgentMessage(
+ sender="router",
+ receiver="intent_v2",
+ message_type="recognize_intent",
+ payload={"query": "执行核型分析，不单独执行种内共线性"},
+ )
+
+ returned = asyncio.run(IntentAgentV2(_ForbiddenLLM()).execute(state, message))
+ intent = returned.results["results"]["intent"]
+
+ assert intent["primary_stage"] == "karyotype"
+ assert intent["stages"] == ["identify", "synteny", "karyotype"]
+
+
+def test_completed_subfamily_context_does_not_override_current_stage_negation():
+ class _ForbiddenLLM:
+ async def acall(self, *_args, **_kwargs):
+ raise AssertionError("deterministic intent must not call the LLM")
+
+ state = AgentState(session_id="context-negation", job_id="job-context-negation")
+ message = AgentMessage(
+ sender="router",
+ receiver="intent_v2",
+ message_type="recognize_intent",
+ payload={
+ "query": (
+ "帮我执行Ft的基因家族分析，但是不要执行kaks分析\n"
+ "已确认上下文：基因家族 SPL；目标物种 Ft；参考物种 At；"
+ "当前 job 的 subfamily_confirm 已成功完成，家族成员和亚家族分类表已确认。"
+ ),
+ },
+ )
+
+ returned = asyncio.run(IntentAgentV2(_ForbiddenLLM()).execute(state, message))
+ intent = returned.results["results"]["intent"]
+
+ assert intent["gene_family"] == "SPL"
+ assert intent["species"] == "Ft"
+ assert "ppi" in intent["stages"]
+ assert "kaks" not in intent["stages"]
+ assert "reference_classification" not in intent["stages"]
+
+
+def test_subfamily_classification_bypasses_generic_stage_negation():
+ intent = parse_intent_from_query("执行亚家族分类，不要结构域分析")
+
+ assert "reference_classification" in intent["stages"]
+ assert "domain" in intent["stages"]
diff --git a/tests/unit/test_task_status_attachment.py b/tests/unit/test_task_status_attachment.py
index 449d9ba00..e1a7e2864 100755
--- a/tests/unit/test_task_status_attachment.py
+++ b/tests/unit/test_task_status_attachment.py
@@ -2,6 +2,7 @@ from __future__ import annotations
import hashlib
import json
+import os
def _write_ft_technical_reference_manifest(final_dir, stage, *, status="success"):
@@ -34,6 +35,50 @@ def _write_ft_technical_reference_manifest(final_dir, stage, *, status="success"
return manifest
+def _write_regular_gene_family_manifest(
+ final_dir,
+ stage,
+ *,
+ scope_id="SPL.target-Ft.ref-At.target_plus_reference",
+ route="multi_species",
+ references=None,
+ member_set_id="",
+ filename="",
+ with_artifact=True,
+):
+ references = ["At"] if references is None else list(references)
+ member_set_id = member_set_id or f"{scope_id}:{route}:all_members"
+ artifact = final_dir / f"Ft.SPL.{stage}.result.tsv"
+ if with_artifact:
+ artifact.write_text(f"{stage}\tresult\n", encoding="utf-8")
+ manifest = final_dir / (filename or f"Ft.SPL.scoped.{stage}.manifest.json")
+ manifest.write_text(
+ json.dumps(
+ {
+ "manifest_version": "1.0",
+ "stage": stage,
+ "status": "success",
+ "prefix": "Ft.SPL",
+ "scope_id": scope_id,
+ "scope_mode": "target_plus_reference" if references else "target_only",
+ "analysis_route": route,
+ "member_set_id": member_set_id,
+ "family": "SPL",
+ "target_species": ["Ft"],
+ "reference_species": references,
+ "artifacts": (
+ {"result": {"path": f"final/{artifact.name}", "type": "tsv"}}
+ if with_artifact
+ else {"result": {"path": "final/missing.tsv", "type": "tsv"}}
+ ),
+ },
+ ensure_ascii=False,
+ ),
+ encoding="utf-8",
+ )
+ return manifest
+
+
def _confirmed_contextual_ppi_fixture(web_main, workspace_root, job_id, intent_params):
from types import SimpleNamespace
@@ -416,6 +461,179 @@ def test_identity_completed_stages_force_rerun_keeps_technical_stages_pending(tm
assert completed == []
+def test_post_confirm_plan_recovers_regular_manifest_identity(tmp_path, monkeypatch):
+ from apps.web.vue3 import main as web_main
+
+ workspace_root = tmp_path / "workspace"
+ final_dir = workspace_root / "job_regular_recovery" / "final"
+ final_dir.mkdir(parents=True)
+ confirm = final_dir / "Ft.SPL.subfamily_confirm.manifest.json"
+ confirm.write_text('{"stage":"subfamily_confirm","status":"success","prefix":"Ft.SPL"}', encoding="utf-8")
+ manifest = _write_regular_gene_family_manifest(final_dir, "phylogeny")
+ os.utime(confirm, (1, 1))
+ os.utime(manifest, (2, 2))
+ monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
+
+ scope_id = "SPL.target-Ft.ref-At.target_plus_reference"
+ plan_params = {
+ "prefix": "Ft.SPL",
+ "geneFamily": "SPL",
+ "targetSpecies": ["Ft"],
+ "referenceSpecies": ["At"],
+ "workflowPhase": "post_subfamily_confirm",
+ "requiresConfirmedSubfamily": True,
+ }
+ stage_params = {
+ **plan_params,
+ "scopeMode": "target_plus_reference",
+ "analysisRoute": "multi_species",
+ "memberSetId": f"{scope_id}:confirmed_subfamily:all_members",
+ }
+
+ completed = web_main._identity_completed_stages_for_plan(
+ "job_regular_recovery",
+ ["phylogeny"],
+ plan_params,
+ [{"id": "phylogeny", "params": stage_params}],
+ request_prefix="Ft.SPL",
+ )
+
+ assert completed == ["phylogeny"]
+
+
+def test_post_confirm_plan_tries_next_manifest_after_incomplete_alias(tmp_path, monkeypatch):
+ from apps.web.vue3 import main as web_main
+
+ workspace_root = tmp_path / "workspace"
+ final_dir = workspace_root / "job_regular_alias" / "final"
+ final_dir.mkdir(parents=True)
+ confirm = final_dir / "Ft.SPL.subfamily_confirm.manifest.json"
+ confirm.write_text('{"stage":"subfamily_confirm","status":"success","prefix":"Ft.SPL"}', encoding="utf-8")
+ canonical = _write_regular_gene_family_manifest(
+ final_dir,
+ "synteny",
+ route="single_species",
+ filename="Ft.SPL.scoped.synteny.manifest.json",
+ )
+ alias = _write_regular_gene_family_manifest(
+ final_dir,
+ "synteny",
+ route="single_species",
+ filename="Ft.SPL.synteny.manifest.json",
+ with_artifact=False,
+ )
+ alias_payload = json.loads(alias.read_text(encoding="utf-8"))
+ for key in ("scope_id", "scope_mode", "analysis_route", "member_set_id"):
+ alias_payload.pop(key, None)
+ alias.write_text(json.dumps(alias_payload), encoding="utf-8")
+ os.utime(confirm, (1, 1))
+ os.utime(canonical, (2, 2))
+ os.utime(alias, (3, 3))
+ monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
+
+ target_scope = "SPL.target-Ft.target_only"
+ plan_params = {
+ "prefix": "Ft.SPL",
+ "geneFamily": "SPL",
+ "targetSpecies": ["Ft"],
+ "referenceSpecies": ["At"],
+ "workflowPhase": "post_subfamily_confirm",
+ "requiresConfirmedSubfamily": True,
+ }
+ stage_params = {
+ **plan_params,
+ "referenceSpecies": [],
+ "scopeMode": "target_only",
+ "analysisRoute": "single_species",
+ "memberSetId": f"{target_scope}:confirmed_subfamily:all_members",
+ }
+
+ completed = web_main._identity_completed_stages_for_plan(
+ "job_regular_alias",
+ ["synteny"],
+ plan_params,
+ [{"id": "synteny", "params": stage_params}],
+ request_prefix="Ft.SPL",
+ )
+
+ assert completed == ["synteny"]
+
+
+def test_post_confirm_plan_rejects_result_older_than_confirmation(tmp_path, monkeypatch):
+ from apps.web.vue3 import main as web_main
+
+ workspace_root = tmp_path / "workspace"
+ final_dir = workspace_root / "job_regular_stale" / "final"
+ final_dir.mkdir(parents=True)
+ manifest = _write_regular_gene_family_manifest(final_dir, "phylogeny")
+ confirm = final_dir / "Ft.SPL.subfamily_confirm.manifest.json"
+ confirm.write_text('{"stage":"subfamily_confirm","status":"success","prefix":"Ft.SPL"}', encoding="utf-8")
+ os.utime(manifest, (1, 1))
+ os.utime(confirm, (2, 2))
+ monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
+
+ scope_id = "SPL.target-Ft.ref-At.target_plus_reference"
+ plan_params = {
+ "prefix": "Ft.SPL",
+ "geneFamily": "SPL",
+ "targetSpecies": ["Ft"],
+ "referenceSpecies": ["At"],
+ "workflowPhase": "post_subfamily_confirm",
+ "requiresConfirmedSubfamily": True,
+ }
+ stage_params = {
+ **plan_params,
+ "scopeMode": "target_plus_reference",
+ "analysisRoute": "multi_species",
+ "memberSetId": f"{scope_id}:confirmed_subfamily:all_members",
+ }
+
+ completed = web_main._identity_completed_stages_for_plan(
+ "job_regular_stale",
+ ["phylogeny"],
+ plan_params,
+ [{"id": "phylogeny", "params": stage_params}],
+ request_prefix="Ft.SPL",
+ )
+
+ assert completed == []
+
+
+def test_post_confirm_kaks_inherits_completed_shared_ppi_reference(tmp_path, monkeypatch):
+ from apps.web.vue3 import main as web_main
+
+ workspace_root = tmp_path / "workspace"
+ final_dir = workspace_root / "job_contextual_kaks" / "final"
+ final_dir.mkdir(parents=True)
+ _write_ft_technical_reference_manifest(final_dir, "kaks")
+ monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
+
+ plan_params = {
+ "prefix": "Ft.SPL",
+ "geneFamily": "SPL",
+ "targetSpecies": ["Ft"],
+ "referenceSpecies": ["At"],
+ "workflowPhase": "post_subfamily_confirm",
+ "requiresConfirmedSubfamily": True,
+ }
+ stage_params = {
+ **plan_params,
+ "scopeMode": "target_plus_reference",
+ "analysisRoute": "multi_species",
+ "memberSetId": "SPL.target-Ft.ref-At.target_plus_reference:confirmed_subfamily:all_members",
+ }
+
+ completed = web_main._identity_completed_stages_for_plan(
+ "job_contextual_kaks",
+ ["kaks"],
+ plan_params,
+ [{"id": "kaks", "params": stage_params}],
+ request_prefix="Ft.SPL",
+ )
+
+ assert completed == ["kaks"]
+
+
def test_contextual_plan_returns_only_pending_stages(tmp_path, monkeypatch):