diff --git a/docs/nl_trigger_test_rounds/round_1.md b/docs/nl_trigger_test_rounds/round_1.md new file mode 100644 index 000000000..4cf6461eb --- /dev/null +++ b/docs/nl_trigger_test_rounds/round_1.md @@ -0,0 +1,98 @@ +# 轻量版拓展分析自然语言触发 — 第一轮测试报告 + +- 时间：2026-08-27 +- 被测提交：a60f45754（轻量拓展分析自然语言触发，4 文件 +813/-8，未 push） +- 测试环境：/home/xzq/miniconda3/envs/plantai-bff/bin/python（Python 3.11.15 / pytest 9.1.1 / pytest-cov 7.1.0；本轮补充安装 pyyaml/openai/pandas/requests/openpyxl 后方可跑通 main.py 相关测试） + +## 测试文件清单 + +| 文件 | 新增/追加 | 用例数 | 结果 | +|---|---|---|---| +| tests/unit/test_extension_stage_lightweight_nl_intent.py | 新增 | 58 | 全绿 | +| tests/unit/test_extension_stage_lightweight_nl_backend.py | 新增 | 18 | 全绿 | +| tests/unit/test_extension_stage_lightweight_frontend_contract.py | 追加 5 个测试（未删任何现有断言，原 15 → 现 20） | 20 | 全绿 | + +## 通过/失败数 + +| 组 | 结果 | +|---|---| +| A 组（nl_intent 单元） | 58 passed | +| B 组（main.py 后端集成） | 18 passed | +| C 组（前端契约） | 20 passed | +| 新增三件套合计 | 96 passed | +| 回归：Phase A 三件套（agent/input_binding/golden_runner_semantic） | 6 passed | +| 回归：test_agent_chat_actions + deepseek_semantics | 46 passed | +| 回归：全套 test_extension_stage_lightweight*.py | 312 passed + **6 既有失败** + 1 skip | + +**6 个既有失败**（已在父提交 90f51ed60 干净 worktree 上逐一复现，与本改动无关，不拦路）： +1. test_extension_stage_lightweight_fact_contract.py::test_build_execution_facts_keeps_unmatched_outputs_without_guessing +2. test_extension_stage_lightweight_runtime_binding.py::test_shell_patch_targets_record_hash_literal_kind_and_exact_count +3. test_extension_stage_lightweight_runtime_binding.py::test_same_uploaded_path_in_two_steps_forms_one_safe_path_binding +4. test_extension_stage_lightweight_runtime_binding.py::test_repeated_parameter_literal_across_facts_is_not_patchable +5. test_extension_stage_lightweight_runtime_package.py::test_unrelated_implicit_output_remains_blocked +6. test_extension_stage_lightweight_target_launchers.py::test_target_launchers_share_lightweight_extension_runtime_roots + +## 覆盖率 + +`python3 -m pytest --cov=gfa_core.extension_stage_lightweight.nl_intent --cov-report=term-missing tests/unit/test_extension_stage_lightweight_nl_intent.py` + +- gfa_core/extension_stage_lightweight/nl_intent.py：**187 stmts / 0 miss / 100%** +- 无未覆盖行（补测过程：93% → 98% → 99% → 100%，最终 0 遗漏） + +## 失败用例+关键报错 + +本轮无功能缺陷导致的测试失败（测试自身两处断言问题已修复：集合类型断言、reason 前缀断言）。 + +## 重点验证结论（对应 developer 标记风险 R1-R4 + 测试要求 1-10） + +### R1 目录截断（P0 缺陷确认） +真实 store 实测：61 个 published 记录，`_build_catalog` 只放前 11 个即达 11276 字符，truncated=True。 +「拓展分析」强触发词下 `_rough_candidates` 返回全部 61 个，但目录按目录名排序截断后 **12_/13_/14_/15_/16_/17_/18_ 共 50 个扩展（含 manifestqc、tfbs、kaks_cis、motif、pav_cnv 等 18 案例中的后 7 个）LLM 完全不可见**，任何消息都无法命中这些扩展。 +测试固化：test_real_store_catalog_truncation_smoke、test_build_catalog_truncates_when_over_max_chars。 + +### R2 中文关键词粗筛（缺陷确认） +真实 store 实测：「帮我做样本表校验」→ 0 候选、不调 LLM（function_summary 全英文，纯中文功能词无法命中 `_summary_keywords` 英文 token 与中文子串）。只有强触发词（拓展分析/扩展分析/extension）或工具词（hmmsearch/hmmer/blast/blastp）才能进 LLM。**注意：工具词路径会把全部 61 个记录都当候选（无法按工具区分），再叠加 R1 截断，实际能解析的只有前 11 个。** +测试固化：test_real_store_pure_chinese_function_words_miss_rough_candidates。 + +### R3 prefill 字段位置（实现一致） +已验证：`LightweightIntentResolution.prefill` 在 dataclass 顶层；`_match_lightweight_extension_intent` 返回 dict 中 `params.prefill`（顶层副本）与 `params.llm_slots`；前端 `openLightweightNlRunDialog` 读 `intentParams.llm_slots` 的 input_overrides/parameter_overrides/explicit_prefix，不依赖顶层。C 组断言按实际实现写，无错位。 + +### R4 extension_name 英文（产品缺陷候选，记录待验收） +已确认：`extension_name` 取 function_summary 英文首句（≤60 字符），后端 message 与前端向导标题会显示英文（如 "This tool searches sequences with hmmsearch."）。测试固化：test_resolution_uses_first_sentence_as_extension_name。建议后续加中文 display_name 映射。 + +### 测试要求 1-10 覆盖情况 +1. 无触发词 → extension_id None 且 client_factory 计数为 0（参数化 3 条消息）✅ +2. 触发词 + 无 records → 不调 LLM（reason=no_published_records）✅ +3. _build_catalog：只含 replaceable inputs 的 {path,role,description}、剔 locked input、parameter 无 value、含 function_summary、超长截断置 truncated ✅ +4. _sanitize_resolution：编造 path/fact_id/extension_id/非 str value/confidence 非法 → None+原因（参数化 7 种非法输入 + 4 种非对象）；合法 → 正确映射 ✅ +5. LLM 抛异常/坏 JSON/缺 choices → None resolution 不抛（reason=llm_unavailable_or_invalid 或 invalid_resolution: 前缀）✅ +6. happy path：「使用 HMM 拓展分析对刚上传的文件进行分析，参数 e-value=1e-5」→ input_overrides/parameter_overrides/confidence/prefill 全对 ✅ +7. load_published_lightweight_records：published/confirmation_pending/baseline_failed/坏 JSON/非对象 → 只返回 published；坏 store 根返回 [] ✅ +8. job_file_hints 进入 LLM user 消息 JSON（并测 MAX_JOB_FILE_HINTS 截断 200）✅ +9. 参数值约束：sanitize 只做结构校验（fact_id 合法即收），不校验"参数是否在用户消息中出现"（prompt 层约束），测试固化该行为 ✅ +10. 覆盖率 100%（见上）✅ + +### B 组关键用例 +- _match_lightweight_extension_intent：命中返回 lightweight:True/llm_slots/extension_store/prefill/confidence/parse_reason；空 store、resolution None、extension_id None → 均返回 {} ✅ +- 回退三处：Phase A 命中不标 lightweight；Phase A 空列表 → 回退轻量（三个空返回点中 candidates 为空与 trigger 为空两处直接测到；rnaseq 短路点静态确认回退存在，未单独构造 rnaseq 消息，因为该消息本身也不命中轻量粗筛）✅ +- _lightweight_job_file_hints：input+workspace 相对路径、空 job_id → []、200 上限 ✅ +- import-paths 端点：合法 /home/xzq/ 下文件复制进 _nl_imports、同名冲突加 _1 后缀、前缀外路径 400、不存在路径 400、空 job_id 400 ✅（注：source 必须在 /home/xzq/ 下构造，tmp_path 在 /tmp 会被前缀校验拒绝） +- _run_agent_chat 短路：预闸命中 + 轻量 intent → action="propose_extension_stage"、model_used="rule"，且 monkeypatch 证明未触碰 LLMManager/RouterAgent（会抛异常）；预闸不命中 → 走正常 MainAgent 流程 ✅ +- /api/agent/plan 守卫端到端：轻量 intent 不调 _extension_stage_plan_response；Phase A 命中正常调用 ✅ + +## 遗留问题清单 + +| 编号 | 位置 | 现象 | 建议修复 | +|---|---|---|---| +| 1（P0） | gfa_core/extension_stage_lightweight/nl_intent.py `_build_catalog`（约 128-173 行） | 真实 store 61 个 published 记录只进前 11 个即截断（11276/12000 字符），目录按目录名排序，12_ 及以后的 50 个扩展（含 18 案例中 12-18 的 7 类）LLM 不可见，「拓展分析」强触发词下也永远无法命中 | 截断策略需改为"全部进目录、按相关性排序"或"先粗筛再目录化"（按关键词把候选缩到相关子集），至少保证 18 案例的功能词能进目录；12_ 以后的扩展在截断前应有兜底（如只放 extension_id+summary 的轻量条目） | +| 2（P1） | 同文件 `_rough_candidates`（约 108-122 行） | 纯中文功能词（如"样本表校验"）0 候选不调 LLM，只能靠强触发词或英文工具词进入；工具词 hmmsearch/hmmer/blast 路径会把全部 61 个记录都当候选（不按工具区分） | 中文 summary 子串匹配对 18 案例 prompt 影响需产品确认；工具词路径建议按工具名映射到对应扩展（如 hmmsearch → 03_ 系列），或至少接受"全候选+目录排序优化" | +| 3（P1） | gfa_core/extension_stage_lightweight/nl_intent.py `_rough_candidates`（约 113-114 行） | `str(record.get("extension_id"))` 对缺失 id 的记录产出 "None" 字符串候选 | 改为 `record.get("extension_id") and str(...)` 过滤，或直接 `[r["extension_id"] for r in records if r.get("extension_id")]` | +| 4（P2） | 同文件 `resolve_lightweight_extension_intent`（约 265-346 行） | `extension_name` 取 function_summary 英文首句（≤60），后端 message 与前端向导标题显示英文 | 产品验收时确认是否需中文 display_name 映射；若需，在 record 或 catalog 层加中文名字段 | +| 5（P2） | apps/web/vue3/main.py `nl_import_paths`（约 4449-4512 行） | 目录导入用 `shutil.copytree(dirs_exist_ok=True)` 合并进目标目录，且同名目录无冲突后缀（与文件路径的 _1 后缀策略不一致）；多路径循环中前一个成功、后一个 400 时已复制的文件不回滚 | 目录同名冲突建议同样加后缀；整体事务性按需评估（低风险，向导场景可接受部分成功） | +| 6（P3/信息） | 全套 lightweight 回归 | 6 个既有失败（见上清单），父提交 90f51ed60 复现，与本改动无关 | 不拦路；建议另开任务修复 | +| 7（P3/信息） | 测试环境 | plantai-bff 环境缺 pyyaml/openai/pandas/requests/openpyxl，main.py 相关测试此前跑不通（test_agent_chat_actions 修复前 27 failed） | 已补充安装；建议把依赖清单写进仓库 requirements 或测试文档 | + +## 测试环境备注 +- 本轮在 plantai-bff conda 环境补充安装了：pyyaml、openai、pandas、requests、openpyxl（均为项目既有依赖，纯 pip 安装，未动仓库文件）。 +- 验证既有失败时在父提交 90f51ed60 建了临时 worktree，验证后已删除（git worktree remove --force）。 +- 工作区改动仅限测试文件：tests/unit/test_extension_stage_lightweight_nl_intent.py（新增）、tests/unit/test_extension_stage_lightweight_nl_backend.py（新增）、tests/unit/test_extension_stage_lightweight_frontend_contract.py（追加 5 个测试，现有断言未删）。开发代码未触碰，未 push。 diff --git a/docs/nl_trigger_test_rounds/round_2.md b/docs/nl_trigger_test_rounds/round_2.md new file mode 100644 index 000000000..cd120b65d --- /dev/null +++ b/docs/nl_trigger_test_rounds/round_2.md @@ -0,0 +1,87 @@ +# 轻量版拓展分析自然语言触发 — 第二轮测试报告 + +- 时间：2026-08-27 +- 被测提交：db610031c + afdb1614a（修复第一轮 7 项反馈 + PM 裁断修正，均未 push） +- 测试环境：/home/xzq/miniconda3/envs/plantai-bff/bin/python（Python 3.11.15 / pytest 9.1.1 / pytest-cov 7.1.0） + +## 测试文件清单 + +| 文件 | 状态 | 用例数 | 结果 | +|---|---|---|---| +| tests/unit/test_extension_stage_lightweight_nl_intent.py | 更新+新增 | 79（上轮 58） | 全绿 | +| tests/unit/test_extension_stage_lightweight_nl_backend.py | 更新+新增 | 23（上轮 18） | 全绿 | +| tests/unit/test_extension_stage_lightweight_frontend_contract.py | 不变（与新实现一致，未删断言） | 20 | 全绿 | + +## 通过/失败数 + +| 组 | 结果 | +|---|---| +| A 组（nl_intent 单元） | 79 passed，覆盖率 100%（223 stmts / 0 miss） | +| B 组（main.py 后端集成） | 23 passed | +| C 组（前端契约） | 20 passed | +| 相关四文件回归（A+B+C+deepseek_semantics） | 131 passed | +| Phase A 三件套（agent/input_binding/golden_runner_semantic） | 6 passed | +| test_agent_chat_actions | 36 passed | +| 全套 test_extension_stage_lightweight*.py | 337 passed + 6 既有失败（与第一轮相同 6 个，父提交复现，不拦路）+ 1 skip | + +## 覆盖率 + +`nl_intent.py`：**223 stmts / 0 miss / 100%**。第一轮后新增代码（union 粗筛、_extension_id_tokens、_parameter_flag_tokens、_record_keyword_sources、compact_extension_name、catalog 24000、truncated reason、resolve 排序）全部覆盖。补测过程中 99%→100% 的一次补齐为 `_parameter_flag_tokens` 非 Mapping 元素跳过分支。 + +## 失败用例+关键报错 + +本轮无功能缺陷导致的失败；5 个上轮用例因行为有意变更而更新（工具词不再全量、extension_name 改紧凑形式、blast/hmmer 用例 fixture 隔离）。 + +## 第二轮修复验证结论（对应 main 要求的 5 项） + +### 1. union 粗筛策略（afdb1614a）— 通过 +- 带工具词+强触发词 → 关键词子集（不因强触发词全量）：test_rough_candidates_union_strategy_with_strong_trigger_and_keyword ✅ +- 纯强触发词无关键词命中 → 全量兜底：test_rough_candidates_pure_strong_trigger_falls_back_to_all ✅ +- 纯工具词无强触发词 → 子集：参数化用例 "用 hmmsearch 搜索结构域" → ["ext_hmm"] ✅ +- 无任何命中 → []：test_rough_candidates_no_match_returns_empty ✅ +- 命中数降序（同分按 id）：test_rough_candidates_sorted_by_hits_desc ✅ +- 缺失 extension_id 过滤不产出 None：test_rough_candidates_skips_missing_extension_id ✅ +- resolve 按候选分排序目录：test_resolve_orders_records_by_candidate_score ✅ + +### 2. compact_extension_name — 通过 +参数化 6 例：去尾缀时间戳（`03___hmm_1787130993548`→`03___hmm`）、超长截 30、空串、无时间戳原样 ✅ + +### 3. catalog 24000 上限与 truncated reason — 通过 +- MAX_CATALOG_CHARS == 24000 断言 ✅ +- 60 条长 summary 记录截断、总长 ≤ 上限、截断点后首条追加超限 ✅ +- 去重 seen_ids ✅ +- 截断时 reason="ok; 候选过多已截断，请补充工具名" ✅ + +### 4. _extension_id_tokens / _parameter_flag_tokens — 通过 +- id token：`03___hmm_1787130993548` → 含 hmm、剔 len<3 ✅ +- flag token：`--samples` 保留、单字母剔除、停用词剔除、无 contract 返回 []、非 Mapping 元素跳过 ✅ +- _record_keyword_sources 三源合并（summary + id token + flag）✅ + +### 5. B 组 message 中文模板 — 通过 +- extension_id 为主、summary 首句截 60 字括号补充："已识别为轻量拓展分析：ext_hmm（This tool searches sequences with hmmsearch）。请按引导上传或选择输入文件并核对参数后执行。" ✅ +- 无 summary 时无括号补充 ✅ +- params 新增 function_summary 字段 ✅ + +### 6. import-paths 第二轮修复 — 通过 +- 目录同名冲突 _1 后缀（原目录不被覆盖）✅ +- 多路径先整体校验后复制：合法+非法混合 → 400 且合法文件未复制 ✅ +- 不存在路径混入 → 400 且未复制 ✅ +- 多合法路径全部复制 ✅ + +### 7. PM 实测对齐（真实 store 固化）— 通过 +test_real_store_pm_alignment_counts：HMM→9、DIAMOND→8、OrthoFinder→3、KaKs→2、ManifestQC→5、纯拓展分析→61，与 PM 实测完全一致；且 HMM 子集（9 条）构建 catalog 不截断（truncated=False），P0 截断问题在带工具词场景已解决。 + +### 8. C 组一致性 — 通过 +index.html 两提交间零改动，现有 20 个契约测试（含上轮追加 5 个）全部保持绿色，无需调整。 + +## 遗留问题清单 + +| 编号 | 位置 | 现象 | 建议修复 | +|---|---|---|---| +| 1（P2） | nl_intent.py `_rough_candidates` / `_summary_keywords` | 纯中文功能词（如「样本表校验」）仍 0 候选不调 LLM——manifestqc 等中文场景只能靠「拓展分析」强触发词全量兜底（61 候选 + 24000 截断后 13 条进 catalog，排序靠目录名，manifestqc 18_ 系列仍在截断线外）。实测「用 ManifestQC 拓展分析」走 id token 命中可解，但用户不知工具英文名的场景仍受限 | 产品层面确认是否需要中文功能词→工具映射表；或截断时按 18 案例优先级排序 | +| 2（P3） | nl_intent.py `_extension_id_tokens` | extension_id 的通用英文 token（如 "ext"、数字时间戳）也进关键词源，实测 "ext" 已在 _ENGLISH_STOPWORDS 之外（是 "extension" 的子串场景会误命中）；上轮实测 blast 消息误命中 "against" 属 summary 普通词干扰，union 策略下仍可能存在这类弱命中把无关记录顶进子集 | 低风险；可考虑给关键词源加权重（id token > summary > flag），弱命中不参与排序竞争 | +| 3（P3） | 全套 lightweight 回归 | 6 个既有失败（与第一轮相同清单），父提交 90f51ed60 复现，与本改动无关 | 不拦路；建议另开任务 | + +## 测试环境备注 +- 环境同第一轮（plantai-bff + 已补装的 pyyaml/openai/pandas/requests/openpyxl）。 +- 工作区改动仅测试文件：tests/unit/test_extension_stage_lightweight_nl_intent.py（更新+新增）、tests/unit/test_extension_stage_lightweight_nl_backend.py（更新+新增）。开发代码未触碰，未 push。 diff --git a/tests/unit/test_extension_stage_lightweight_frontend_contract.py b/tests/unit/test_extension_stage_lightweight_frontend_contract.py index 9f44a6fed..3e686844d 100644 --- a/tests/unit/test_extension_stage_lightweight_frontend_contract.py +++ b/tests/unit/test_extension_stage_lightweight_frontend_contract.py @@ -248,3 +248,55 @@ assert.strictEqual(parameterRows.find((row) => row.item === '可改参数').valu intent = source[source.index("const runExtensionStageFromIntent"):source.index("const extensionOutputIcon")] assert "if (isLightweightExtensionId(extensionId))" in intent assert intent.index("openLightweightRuntimeCard(extensionId)") < intent.index("executeExtensionStageRun(") + + +# ---------- 自然语言触发（轻量拓展分析）追加断言 ---------- + +def test_lightweight_nl_run_wizard_and_state_exist_in_source() -> None: + source = _source() + + assert "lightweightNlRun" in source + assert "const lightweightNlRun = reactive({" in source + assert "let lightweightNlRunResolve = null;" in source + assert "openLightweightNlRunDialog" in source + assert "executeLightweightNlRun" in source + assert "handleNlRunFileDrop" in source + assert "importNlRunServerPaths" in source + assert "cancelLightweightNlRun" in source + assert "confirmLightweightNlRun" in source + assert "nextLightweightNlRunStep" in source + + +def test_lightweight_nl_run_uploads_with_progress_and_imports_server_paths() -> None: + source = _source() + + assert "xhr.upload.onprogress" in source + assert "el-progress" in source + assert "/api/nl-import-paths" in source + assert "input/" in source + + +def test_lightweight_nl_run_prefills_from_llm_slots() -> None: + source = _source() + + assert "llm_slots" in source + assert "input_overrides" in source + assert "parameter_overrides" in source + assert "explicit_prefix" in source + assert "parse_reason" in source + + +def test_lightweight_nl_run_intent_context_and_conversation_replies() -> None: + source = _source() + + assert "intent_context" in source + assert "fromIntent" in source + assert "addAgentMessage" in source + assert "taskType === 'extension_stage'" in source + + +def test_lightweight_nl_run_prefill_flag_wired() -> None: + source = _source() + + assert "prefill" in source + assert "已识别为轻量拓展分析" not in source # 消息文案由后端生成，前端按 taskType 路由 diff --git a/tests/unit/test_extension_stage_lightweight_nl_backend.py b/tests/unit/test_extension_stage_lightweight_nl_backend.py new file mode 100644 index 000000000..dd0a7442f --- /dev/null +++ b/tests/unit/test_extension_stage_lightweight_nl_backend.py @@ -0,0 +1,562 @@ +"""轻量拓展分析自然语言触发的后端集成测试（main.py 侧）。 + +monkeypatch BBAV5_LIGHTWEIGHT_EXTENSION_STORE 指向 tmp store， +monkeypatch nl_intent.resolve_lightweight_extension_intent 返回固定 resolution。 +风格对齐 tests/unit/test_agent_chat_actions.py。 +""" +from __future__ import annotations + +import json +import shutil +import tempfile +from pathlib import Path + +import pytest + +from apps.web.vue3 import main as WEB_MAIN +from gfa_core.extension_stage_lightweight import nl_intent + +STORE_ENV = "BBAV5_LIGHTWEIGHT_EXTENSION_STORE" + + +# ---------- fixtures / helpers ---------- + +def _write(root: Path, relative: str, content: str) -> Path: + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + return path + + +def _record(extension_id: str) -> dict: + return { + "extension_id": extension_id, + "status": "published", + "semantic_candidate": {"function_summary": "This tool searches sequences with hmmsearch."}, + "runtime_binding_contract": { + "inputs": [ + {"path": "family.hmm", "role": "HMM profile input", "description": "profile", "replaceable": True}, + {"path": "01.col.primary.pep.fa", "role": "Protein sequence input", "description": "seqs", "replaceable": True}, + ], + "parameters": [ + {"fact_id": "step_002:parameter:003", "flag": "-E", "display_name": "E-value threshold", "description": "evalue"}, + ], + }, + } + + +def _write_published_record(store_root: Path, extension_id: str) -> None: + _write(store_root, f"extensions/{extension_id}/extension.json", json.dumps(_record(extension_id))) + + +def _resolution(extension_id: str = "ext_hmm") -> nl_intent.LightweightIntentResolution: + return nl_intent.LightweightIntentResolution( + extension_id=extension_id, + extension_name="This tool searches sequences with hmmsearch.", + input_overrides={"family.hmm": "input/job-nl-1/_nl_imports/family.hmm"}, + parameter_overrides={"step_002:parameter:003": "1e-5"}, + explicit_prefix="outputs/nl", + confidence=0.9, + prefill=True, + reason="ok", + candidates_considered=[extension_id], + ) + + +def _patch_resolver(monkeypatch, resolution): + monkeypatch.setattr( + nl_intent, + "resolve_lightweight_extension_intent", + lambda message, records, environ=None, client_factory=None, job_file_hints=None: resolution, + ) + + +@pytest.fixture +def lightweight_store(monkeypatch, tmp_path: Path) -> Path: + store_root = tmp_path / "lw-store" + monkeypatch.setenv(STORE_ENV, str(store_root)) + return store_root + + +# ---------- 1. _match_lightweight_extension_intent ---------- + +def test_match_lightweight_intent_returns_full_contract( + monkeypatch, lightweight_store: Path +) -> None: + _write_published_record(lightweight_store, "ext_hmm") + _patch_resolver(monkeypatch, _resolution()) + + intent = WEB_MAIN._match_lightweight_extension_intent("使用 HMM 拓展分析", "job-nl-1") + + assert intent.get("lightweight") is True + assert intent["stage"] == "extension_stage" + params = intent["params"] + assert params["extension_id"] == "ext_hmm" + assert params["extensionName"] == "This tool searches sequences with hmmsearch." + assert params["extension_store"] == "lightweight" + assert params["llm_slots"] == { + "input_overrides": {"family.hmm": "input/job-nl-1/_nl_imports/family.hmm"}, + "parameter_overrides": {"step_002:parameter:003": "1e-5"}, + "explicit_prefix": "outputs/nl", + } + assert params["prefill"] is True + assert params["confidence"] == 0.9 + assert params["parse_reason"] == "ok" + assert intent["confidence"] == 0.9 + # message 中文模板：extension_id 为主、summary 首句截 60 字括号补充 + assert "已识别为轻量拓展分析" in intent["message"] + assert "ext_hmm" in intent["message"] + assert "This tool searches sequences with hmmsearch" in intent["message"] + assert "请按引导上传或选择输入文件并核对参数后执行" in intent["message"] + + +def test_match_lightweight_intent_empty_store_returns_empty_dict( + monkeypatch, lightweight_store: Path +) -> None: + # store 无任何 published record + intent = WEB_MAIN._match_lightweight_extension_intent("使用拓展分析", "job-x") + assert intent == {} + + +def test_match_lightweight_intent_none_resolution_returns_empty_dict( + monkeypatch, lightweight_store: Path +) -> None: + _write_published_record(lightweight_store, "ext_hmm") + _patch_resolver(monkeypatch, None) # resolution None（模拟 LLM 降级） + + intent = WEB_MAIN._match_lightweight_extension_intent("使用拓展分析", "job-x") + assert intent == {} + + +def test_match_lightweight_intent_none_extension_id_returns_empty_dict( + monkeypatch, lightweight_store: Path +) -> None: + _write_published_record(lightweight_store, "ext_hmm") + resolution = _resolution() + resolution = nl_intent.LightweightIntentResolution( + extension_id=None, + extension_name=None, + input_overrides={}, + parameter_overrides={}, + explicit_prefix=None, + confidence=0.0, + prefill=False, + reason="no_trigger", + candidates_considered=[], + ) + _patch_resolver(monkeypatch, resolution) + + intent = WEB_MAIN._match_lightweight_extension_intent("使用拓展分析", "job-x") + assert intent == {} + + +# ---------- 2. _match_extension_stage_intent 回退 ---------- + +def _phase_a_ext(extension_id: str = "legacy_ext", status: str = "active", nl_triggers=None) -> dict: + return { + "extension_id": extension_id, + "name": extension_id, + "display_name": f"{extension_id} display", + "status": status, + "nl_triggers": nl_triggers or ["hmmsearch"], + "requirements": {"system": ["hmmsearch"]}, + "analysis_type": "hmmer_search", + "parameters": {}, + } + + +def test_extension_stage_intent_keeps_phase_a_when_published( + monkeypatch, lightweight_store: Path +) -> None: + """Phase A 命中（构造 nl_triggers 命中）时不回退轻量匹配。""" + monkeypatch.setattr( + WEB_MAIN.extension_stage_agent, + "list_extensions", + lambda include_drafts=False: [_phase_a_ext()], + ) + _write_published_record(lightweight_store, "ext_hmm") + _patch_resolver(monkeypatch, _resolution()) + + intent = WEB_MAIN._match_extension_stage_intent("使用 hmmsearch 做结构域分析", "job-x") + + assert intent.get("lightweight") is None + assert intent.get("params", {}).get("extension_id") == "legacy_ext" + + +def test_extension_stage_intent_falls_back_to_lightweight_when_phase_a_empty( + monkeypatch, lightweight_store: Path +) -> None: + monkeypatch.setattr( + WEB_MAIN.extension_stage_agent, + "list_extensions", + lambda include_drafts=False: [], + ) + _write_published_record(lightweight_store, "ext_hmm") + _patch_resolver(monkeypatch, _resolution()) + + intent = WEB_MAIN._match_extension_stage_intent("使用 HMM 拓展分析我的文件", "job-x") + + assert intent.get("lightweight") is True + assert intent["params"]["extension_id"] == "ext_hmm" + + +# ---------- 3. _lightweight_job_file_hints ---------- + +def test_job_file_hints_collects_input_and_workspace_relative_paths( + monkeypatch, tmp_path: Path +) -> None: + input_root = tmp_path / "input" + workspace_root = tmp_path / "workspace" + _write(input_root, "job-h1/data/seq.fa", ">a\nAAAA\n") + _write(input_root, "job-h1/notes.txt", "notes") + _write(workspace_root, "job-h1/out/table.tsv", "a\tb\n") + _write(workspace_root, "job-h1/stray.tsv", "x") + _write(input_root, "job-other/other.fa", ">o\n") # 其他 job 不应出现 + + monkeypatch.setattr(WEB_MAIN, "INPUT_ROOT", input_root) + monkeypatch.setattr(WEB_MAIN, "WORKSPACE_ROOT", workspace_root) + + hints = WEB_MAIN._lightweight_job_file_hints("job-h1") + + assert "input/job-h1/data/seq.fa" in hints + assert "input/job-h1/notes.txt" in hints + assert "workspace/job-h1/out/table.tsv" in hints + assert "workspace/job-h1/stray.tsv" in hints + assert "input/job-other/other.fa" not in hints + assert len(hints) == 4 + + +def test_job_file_hints_empty_job_id_returns_empty(monkeypatch) -> None: + assert WEB_MAIN._lightweight_job_file_hints("") == [] + + +def test_job_file_hints_capped_at_max(monkeypatch, tmp_path: Path) -> None: + input_root = tmp_path / "input" + for i in range(nl_intent.MAX_JOB_FILE_HINTS + 20): + _write(input_root, f"job-cap/f{i:03d}.txt", "x") + monkeypatch.setattr(WEB_MAIN, "INPUT_ROOT", input_root) + monkeypatch.setattr(WEB_MAIN, "WORKSPACE_ROOT", tmp_path / "workspace") + + hints = WEB_MAIN._lightweight_job_file_hints("job-cap") + + assert len(hints) == nl_intent.MAX_JOB_FILE_HINTS + + +# ---------- 4. import-paths 端点 ---------- + +def _call_nl_import_paths(job_id: str, paths: list[str]): + request = WEB_MAIN.NlImportPathsRequest(job_id=job_id, paths=paths) + import asyncio + + return asyncio.run(WEB_MAIN.nl_import_paths(request)) + + +@pytest.fixture +def allowed_source(tmp_path: Path) -> Path: + """在端点允许前缀 /home/xzq/ 下创建临时 source 文件。""" + workdir = Path(tempfile.mkdtemp(dir="/home/xzq", prefix="nl-import-test-")) + yield workdir + shutil.rmtree(workdir, ignore_errors=True) + + +def test_nl_import_paths_copies_file_into_nl_imports(monkeypatch, tmp_path: Path, allowed_source: Path) -> None: + source = _write(allowed_source, "family.hmm", "HMM profile content") + input_root = tmp_path / "input" + monkeypatch.setattr(WEB_MAIN, "INPUT_ROOT", input_root) + + result = _call_nl_import_paths("job-imp-1", [str(source)]) + + assert result["success"] is True + assert result["imported"] == {str(source): "input/job-imp-1/_nl_imports/family.hmm"} + copied = input_root / "job-imp-1" / "_nl_imports" / "family.hmm" + assert copied.is_file() + assert copied.read_text(encoding="utf-8") == "HMM profile content" + + +def test_nl_import_paths_name_conflict_gets_suffix(monkeypatch, tmp_path: Path, allowed_source: Path) -> None: + source = _write(allowed_source, "family.hmm", "new content") + input_root = tmp_path / "input" + existing = _write(input_root, "job-imp-2/_nl_imports/family.hmm", "old content") + monkeypatch.setattr(WEB_MAIN, "INPUT_ROOT", input_root) + + result = _call_nl_import_paths("job-imp-2", [str(source)]) + + assert result["imported"] == {str(source): "input/job-imp-2/_nl_imports/family_1.hmm"} + assert existing.read_text(encoding="utf-8") == "old content" + copied = input_root / "job-imp-2" / "_nl_imports" / "family_1.hmm" + assert copied.is_file() + assert copied.read_text(encoding="utf-8") == "new content" + + +def test_nl_import_paths_rejects_path_outside_allowed_prefixes( + monkeypatch, tmp_path: Path +) -> None: + monkeypatch.setattr(WEB_MAIN, "INPUT_ROOT", tmp_path / "input") + from fastapi import HTTPException + + with pytest.raises(HTTPException) as exc: + _call_nl_import_paths("job-imp-3", ["/etc/passwd"]) + assert exc.value.status_code == 400 + assert "路径必须位于" in exc.value.detail + + +def test_nl_import_paths_rejects_missing_path(monkeypatch, tmp_path: Path) -> None: + monkeypatch.setattr(WEB_MAIN, "INPUT_ROOT", tmp_path / "input") + from fastapi import HTTPException + + with pytest.raises(HTTPException) as exc: + _call_nl_import_paths("job-imp-4", ["/home/xzq/definitely-not-exists-xyz.tsv"]) + assert exc.value.status_code == 400 + assert "路径不存在" in exc.value.detail + + +def test_nl_import_paths_rejects_empty_job_id(monkeypatch, tmp_path: Path) -> None: + monkeypatch.setattr(WEB_MAIN, "INPUT_ROOT", tmp_path / "input") + from fastapi import HTTPException + + with pytest.raises(HTTPException) as exc: + _call_nl_import_paths("", []) + assert exc.value.status_code == 400 + assert "job_id 不能为空" in exc.value.detail + + +# ---------- 5. _run_agent_chat 预闸短路 ---------- + +def _call_run_agent_chat(request): + import asyncio + + return asyncio.run(WEB_MAIN._run_agent_chat(request)) + + +def test_run_agent_chat_short_circuits_on_lightweight_intent( + monkeypatch, lightweight_store: Path, tmp_path: Path +) -> None: + """预闸命中 + 轻量 intent 时 _run_agent_chat 短路返回 propose_extension_stage， + 且绝不触碰 LLM manager / RouterAgent。""" + _write_published_record(lightweight_store, "ext_hmm") + monkeypatch.setattr( + nl_intent, + "resolve_lightweight_extension_intent", + lambda message, records, environ=None, client_factory=None, job_file_hints=None: _resolution(), + ) + monkeypatch.setattr(WEB_MAIN, "llm_manager", None, raising=False) + from gfa_core.mas.config import config as mas_config + + monkeypatch.setattr( + mas_config, + "load_llm_config", + lambda: (_ for _ in ()).throw(RuntimeError("LLM manager must not be touched")), + ) + from gfa_core.mas.agents.router import RouterAgent + + def _boom(_self, _llm_manager, **kwargs): + raise RuntimeError("RouterAgent must not be constructed") + + monkeypatch.setattr(RouterAgent, "__init__", _boom) + + result = _call_run_agent_chat( + WEB_MAIN.AgentChatRequest(message="使用 HMM 拓展分析我的文件", job_id="job-nl-1") + ) + + assert result["success"] is True + assert result["status"] == "ok" + assert result["action"] == "propose_extension_stage" + assert result["intent"].get("lightweight") is True + assert result["intent"]["params"]["extension_id"] == "ext_hmm" + assert result["model_used"] == "rule" + + +def test_run_agent_chat_does_not_short_circuit_without_lightweight_match( + monkeypatch, lightweight_store: Path +) -> None: + """预闸不命中（消息无触发词）时不查轻量 store、正常走 MainAgent 流程。""" + _write_published_record(lightweight_store, "ext_hmm") + monkeypatch.setattr( + nl_intent, + "resolve_lightweight_extension_intent", + lambda message, records, environ=None, client_factory=None, job_file_hints=None: nl_intent.LightweightIntentResolution( + None, None, {}, {}, None, 0.0, False, "no_trigger", [] + ), + ) + monkeypatch.setattr(WEB_MAIN, "llm_manager", None, raising=False) + from gfa_core.mas.config import config as mas_config + + monkeypatch.setattr( + mas_config, + "load_llm_config", + lambda: (_ for _ in ()).throw(RuntimeError("LLM manager unavailable (expected path)")), + ) + + result = _call_run_agent_chat( + WEB_MAIN.AgentChatRequest(message="帮我看看这株水稻的基因家族", job_id="job-nl-2") + ) + + assert result["status"] == "agent_unavailable" # 说明走完了预闸后的 MainAgent 流程 + + +# ---------- 6. /api/agent/plan 守卫 ---------- + +def _call_agent_plan(request): + import asyncio + + return asyncio.run(WEB_MAIN.agent_plan(request)) + + +def test_agent_plan_guard_skips_extension_response_for_lightweight( + monkeypatch, lightweight_store: Path +) -> None: + """plan 里 extension_intent 带 lightweight=True 时不走 _extension_stage_plan_response。""" + _write_published_record(lightweight_store, "ext_hmm") + _patch_resolver(monkeypatch, _resolution()) + captured = {} + + def _fake_extension_stage_plan_response(extension_intent, session_id, job_id=""): + captured["called"] = True + return {"_extension_plan_response": True} + + monkeypatch.setattr(WEB_MAIN, "_extension_stage_plan_response", _fake_extension_stage_plan_response) + + result = _call_agent_plan( + WEB_MAIN.AgentPlanRequest( + message="使用 HMM 拓展分析我的文件", + job_id="job-guard", + intent={}, + ) + ) + + assert "called" not in captured # 守卫拦截，未进入扩展规划响应 + assert result.get("_extension_plan_response") is not True + + +def test_agent_plan_guard_allows_phase_a_extension_response( + monkeypatch, lightweight_store: Path +) -> None: + """Phase A 命中（非轻量）时 plan 正常进入扩展规划响应。""" + monkeypatch.setattr( + WEB_MAIN.extension_stage_agent, + "list_extensions", + lambda include_drafts=False: [_phase_a_ext()], + ) + _write_published_record(lightweight_store, "ext_hmm") + captured = {} + + def _fake_extension_stage_plan_response(extension_intent, session_id, job_id=""): + captured["called"] = True + return {"_extension_plan_response": True} + + monkeypatch.setattr(WEB_MAIN, "_extension_stage_plan_response", _fake_extension_stage_plan_response) + + result = _call_agent_plan( + WEB_MAIN.AgentPlanRequest( + message="使用 hmmsearch 做结构域分析", + job_id="job-guard-phase-a", + intent={}, + ) + ) + + assert captured.get("called") is True + assert result.get("_extension_plan_response") is True + + +# ---------- 8. import-paths 第二轮修复 ---------- + +def test_nl_import_paths_directory_name_conflict_gets_suffix( + monkeypatch, tmp_path: Path, allowed_source: Path +) -> None: + source_dir = allowed_source / "mydata" + source_dir.mkdir(parents=True) + _write(source_dir, "a.txt", "content-a") + input_root = tmp_path / "input" + existing_dir = input_root / "job-imp-dir" / "_nl_imports" / "mydata" + existing_dir.mkdir(parents=True) + _write(existing_dir, "old.txt", "old") + + monkeypatch.setattr(WEB_MAIN, "INPUT_ROOT", input_root) + + result = _call_nl_import_paths("job-imp-dir", [str(source_dir)]) + + assert result["success"] is True + assert result["imported"] == {str(source_dir): "input/job-imp-dir/_nl_imports/mydata_1"} + copied = input_root / "job-imp-dir" / "_nl_imports" / "mydata_1" + assert copied.is_dir() + assert (copied / "a.txt").read_text(encoding="utf-8") == "content-a" + assert (existing_dir / "old.txt").read_text(encoding="utf-8") == "old" # 原目录未被覆盖 + + +def test_nl_import_paths_validates_all_before_copying( + monkeypatch, tmp_path: Path, allowed_source: Path +) -> None: + """多路径先整体校验后复制：任一非法直接 400，前面的合法路径不被复制。""" + good = _write(allowed_source, "good.fa", ">g\nAAAA\n") + input_root = tmp_path / "input" + monkeypatch.setattr(WEB_MAIN, "INPUT_ROOT", input_root) + from fastapi import HTTPException + + with pytest.raises(HTTPException) as exc: + _call_nl_import_paths("job-imp-valid", [str(good), "/etc/passwd"]) + assert exc.value.status_code == 400 + assert "路径必须位于" in exc.value.detail + # 整体校验失败时 good 未被复制 + assert not (input_root / "job-imp-valid" / "_nl_imports" / "good.fa").exists() + + +def test_nl_import_paths_missing_path_validated_before_copying( + monkeypatch, tmp_path: Path, allowed_source: Path +) -> None: + good = _write(allowed_source, "good2.fa", ">g\nAAAA\n") + input_root = tmp_path / "input" + monkeypatch.setattr(WEB_MAIN, "INPUT_ROOT", input_root) + from fastapi import HTTPException + + with pytest.raises(HTTPException) as exc: + _call_nl_import_paths("job-imp-valid2", [str(good), "/home/xzq/definitely-not-exists-xyz.tsv"]) + assert exc.value.status_code == 400 + assert "路径不存在" in exc.value.detail + assert not (input_root / "job-imp-valid2" / "_nl_imports" / "good2.fa").exists() + + +def test_nl_import_paths_multiple_valid_paths_all_copied( + monkeypatch, tmp_path: Path, allowed_source: Path +) -> None: + good1 = _write(allowed_source, "one.fa", ">1\n") + good2 = _write(allowed_source, "two.fa", ">2\n") + input_root = tmp_path / "input" + monkeypatch.setattr(WEB_MAIN, "INPUT_ROOT", input_root) + + result = _call_nl_import_paths("job-imp-multi", [str(good1), str(good2)]) + + assert result["success"] is True + assert set(result["imported"].values()) == { + "input/job-imp-multi/_nl_imports/one.fa", + "input/job-imp-multi/_nl_imports/two.fa", + } + + +def test_match_lightweight_intent_message_without_summary_hint( + monkeypatch, lightweight_store: Path +) -> None: + """record 无 summary 时 message 不带括号补充。""" + _write_published_record(lightweight_store, "ext_nosum") + resolution = nl_intent.LightweightIntentResolution( + extension_id="ext_nosum", + extension_name="ext_nosum", + input_overrides={}, + parameter_overrides={}, + explicit_prefix=None, + confidence=0.5, + prefill=False, + reason="ok", + candidates_considered=["ext_nosum"], + ) + # 用无 semantic_candidate 的 record 覆盖 fixture + record_path = lightweight_store / "extensions" / "ext_nosum" / "extension.json" + record = json.loads(record_path.read_text(encoding="utf-8")) + record["semantic_candidate"] = {} + record_path.write_text(json.dumps(record), encoding="utf-8") + _patch_resolver(monkeypatch, resolution) + + intent = WEB_MAIN._match_lightweight_extension_intent("使用拓展分析", "job-x") + + assert intent.get("lightweight") is True + assert "已识别为轻量拓展分析：ext_nosum。" in intent["message"] + assert "（" not in intent["message"] diff --git a/tests/unit/test_extension_stage_lightweight_nl_intent.py b/tests/unit/test_extension_stage_lightweight_nl_intent.py new file mode 100644 index 000000000..faf24fca6 --- /dev/null +++ b/tests/unit/test_extension_stage_lightweight_nl_intent.py @@ -0,0 +1,908 @@ +"""轻量版拓展分析自然语言触发 nl_intent 模块单元测试。 + +mock 风格对齐 tests/unit/test_extension_stage_lightweight_deepseek_semantics.py： +client_factory 注入 + SimpleNamespace 伪造 choices[0].message.content。 +""" +from __future__ import annotations + +import json +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import pytest + +from gfa_core.extension_stage_lightweight import nl_intent + + +def _module(): + return nl_intent + + +def _write(root: Path, relative: str, content: str) -> Path: + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + return path + + +# ---------- 契约构造 ---------- + +def _contract(*, inputs: list[dict] | None = None, parameters: list[dict] | None = None) -> dict: + inputs = inputs if inputs is not None else [ + { + "path": "family.hmm", + "role": "HMM profile input", + "description": "HMM profile file used to search the protein sequences.", + "replaceable": True, + "locked": False, + }, + { + "path": "01.col.primary.pep.fa", + "role": "Protein sequence input", + "description": "Protein FASTA file containing the sequences to search against the HMM profile.", + "replaceable": True, + "locked": False, + }, + { + "path": "02_hmm/fixed.tsv", + "role": "Locked intermediate", + "description": "Intermediate produced inside the run, must not be overridable.", + "replaceable": False, + "locked": True, + }, + ] + parameters = parameters if parameters is not None else [ + { + "fact_id": "step_001:parameter:001", + "flag": "-p", + "display_name": "Output directory", + "description": "Directory path to create for HMM search outputs.", + "value": "outputs/02_hmm", + }, + { + "fact_id": "step_002:parameter:003", + "flag": "-E", + "display_name": "E-value threshold", + "description": "E-value threshold for full sequence hits in hmmsearch.", + "value": "1e-3", + }, + ] + return {"inputs": inputs, "parameters": parameters} + + +def _record( + extension_id: str, + *, + status: str = "published", + summary: str = "This tool searches a protein FASTA file against an HMM profile using hmmsearch.", + contract: dict | None = None, +) -> dict: + return { + "extension_id": extension_id, + "status": status, + "semantic_candidate": { + "function_summary": summary, + }, + "runtime_binding_contract": contract if contract is not None else _contract(), + } + + +def _good_raw() -> dict: + return { + "extension_id": "ext_hmm", + "input_overrides": [ + {"path": "family.hmm", "value": "input/job-1/_nl_imports/family.hmm"}, + {"path": "01.col.primary.pep.fa", "value": "input/job-1/01.col.primary.pep.fa"}, + ], + "parameter_overrides": [ + {"fact_id": "step_002:parameter:003", "value": "1e-5"}, + ], + "explicit_prefix": "outputs/nl_run", + "confidence": 0.85, + } + + +def _catalog() -> dict: + return _module()._build_catalog([_record("ext_hmm")]) + + +def _response(content: str): + return SimpleNamespace( + id="request-123", + model="deepseek-test", + choices=[SimpleNamespace(message=SimpleNamespace(content=content), finish_reason="stop")], + usage=SimpleNamespace(prompt_tokens=120, completion_tokens=80, total_tokens=200), + ) + + +class _FakeCompletions: + def __init__(self, response=None, error: Exception | None = None) -> None: + self.response = response + self.error = error + self.requests: list[dict] = [] + + def create(self, **kwargs): + self.requests.append(kwargs) + if self.error is not None: + raise self.error + return self.response + + +class _FakeClient: + def __init__(self, response=None, error: Exception | None = None) -> None: + self.completions = _FakeCompletions(response=response, error=error) + self.chat = SimpleNamespace(completions=self.completions) + + +class _CountingFactory: + def __init__(self, client) -> None: + self.client = client + self.calls = 0 + self.configs: list[Any] = [] + + def __call__(self, config): + self.calls += 1 + self.configs.append(config) + return self.client + + +# ---------- 1. 粗筛不中不调 LLM ---------- + +@pytest.mark.parametrize( + "message", + [ + "帮我看看这株水稻的基因家族有哪些", + "请绘制进化树并做共线性分析", + "这个蛋白序列应该怎么注释", + ], +) +def test_no_trigger_message_resolves_none_and_never_calls_client(message: str) -> None: + records = [_record("ext_hmm")] + fake = _FakeClient(response=_response(json.dumps(_good_raw()))) + factory = _CountingFactory(fake) + + resolution = _module().resolve_lightweight_extension_intent( + message, records, environ={}, client_factory=factory + ) + + assert resolution.extension_id is None + assert resolution.reason == "no_trigger" + assert factory.calls == 0 + assert fake.completions.requests == [] + + +# ---------- 2. 触发词命中但无 records ---------- + +def test_trigger_without_records_never_calls_llm() -> None: + fake = _FakeClient(response=_response(json.dumps(_good_raw()))) + factory = _CountingFactory(fake) + + resolution = _module().resolve_lightweight_extension_intent( + "使用拓展分析处理我的文件", [], environ={}, client_factory=factory + ) + + assert resolution.extension_id is None + assert resolution.reason == "no_published_records" + assert factory.calls == 0 + + +# ---------- 3. _build_catalog ---------- + +def test_build_catalog_only_replaceable_inputs_without_locked_and_parameter_values() -> None: + catalog = _module()._build_catalog([_record("ext_hmm")]) + + assert catalog["truncated"] is False + entry = catalog["extensions"][0] + assert entry["extension_id"] == "ext_hmm" + assert entry["function_summary"].startswith("This tool searches a protein FASTA") + + input_keys = {tuple(sorted(item.keys())) for item in entry["inputs"]} + assert input_keys == {("description", "path", "role")} + paths = {item["path"] for item in entry["inputs"]} + assert "family.hmm" in paths + assert "01.col.primary.pep.fa" in paths + assert "02_hmm/fixed.tsv" not in paths # locked 输入被剔除 + + parameter_keys = {tuple(sorted(item.keys())) for item in entry["parameters"]} + assert parameter_keys == {("description", "display_name", "fact_id", "flag")} + assert all("value" not in item for item in entry["parameters"]) + flags = {item["flag"] for item in entry["parameters"]} + assert flags == {"-p", "-E"} + + +def test_build_catalog_truncates_when_over_max_chars() -> None: + module = _module() + long_summary = "x" * 500 + record = _record("ext_long", summary=long_summary) + single_size = len(json.dumps(module._build_catalog([record])["extensions"][0], ensure_ascii=False)) + + assert single_size < module.MAX_CATALOG_CHARS # 单条不足以触发截断 + + records = [_record(f"ext_{i:02d}", summary=long_summary) for i in range(60)] + catalog = module._build_catalog(records) + + assert catalog["truncated"] is True + included = len(catalog["extensions"]) + assert included < len(records) + assert included >= 1 + total = sum(len(json.dumps(item, ensure_ascii=False)) for item in catalog["extensions"]) + assert total <= module.MAX_CATALOG_CHARS + # 截断点之后的首条记录（若追加）必然超限 + first_excluded = records[included]["extension_id"] + extra = len(json.dumps(module._build_catalog([records[included]])["extensions"][0], ensure_ascii=False)) + assert total + extra > module.MAX_CATALOG_CHARS + assert first_excluded not in [item["extension_id"] for item in catalog["extensions"]] + + +# ---------- 4. _sanitize_resolution ---------- + +@pytest.mark.parametrize( + ("mutate", "fragment"), + [ + (lambda raw: raw.update({"input_overrides": [{"path": "invented.fa", "value": "input/job-1/invented.fa"}]}), + "input_overrides 引用了不存在的 path"), + (lambda raw: raw.update({"parameter_overrides": [{"fact_id": "step_099:parameter:099", "value": "x"}]}), + "parameter_overrides 引用了不存在的 fact_id"), + (lambda raw: raw.update({"extension_id": "ext_invented"}), "extension_id 不在可选目录中"), + (lambda raw: raw.update({"parameter_overrides": [{"fact_id": "step_002:parameter:003", "value": 7}]}), + "的 value 必须是非空字符串"), + (lambda raw: raw.update({"confidence": "high"}), "confidence 不是数字"), + (lambda raw: raw.update({"input_overrides": "family.hmm"}), "input_overrides 必须是数组"), + (lambda raw: raw.update({"input_overrides": [{"path": "family.hmm", "value": ""}]}), + "的 value 必须是非空字符串"), + ], +) +def test_sanitize_rejects_invalid_resolution(mutate, fragment: str) -> None: + raw = _good_raw() + mutate(raw) + + sanitized, reason = _module()._sanitize_resolution(raw, _catalog()) + + assert sanitized is None + assert reason is not None + assert fragment in reason + + +@pytest.mark.parametrize( + "raw", + [ + "not a json object", + ["extension_id"], + None, + 42, + ], +) +def test_sanitize_rejects_non_object_payloads(raw) -> None: + sanitized, reason = _module()._sanitize_resolution(raw, _catalog()) + assert sanitized is None + assert reason is not None + + +def test_sanitize_accepts_valid_resolution_and_maps_fields() -> None: + sanitized, reason = _module()._sanitize_resolution(_good_raw(), _catalog()) + + assert reason is None + assert sanitized is not None + assert sanitized["extension_id"] == "ext_hmm" + assert sanitized["input_overrides"] == { + "family.hmm": "input/job-1/_nl_imports/family.hmm", + "01.col.primary.pep.fa": "input/job-1/01.col.primary.pep.fa", + } + assert sanitized["parameter_overrides"] == {"step_002:parameter:003": "1e-5"} + assert sanitized["explicit_prefix"] == "outputs/nl_run" + assert sanitized["confidence"] == 0.85 + + +def test_sanitize_accepts_parameter_override_even_when_user_message_omits_it() -> None: + """sanitize 只做结构校验；'参数值必须来自用户消息' 的约束在 prompt 层，不做程序级校验。""" + raw = _good_raw() # 参数 1e-5 未在用户消息中出现（消息由调用方构造） + + sanitized, reason = _module()._sanitize_resolution(raw, _catalog()) + + assert reason is None + assert sanitized is not None + assert sanitized["parameter_overrides"] == {"step_002:parameter:003": "1e-5"} + + +def test_sanitize_allows_missing_optional_fields() -> None: + raw = {"extension_id": "ext_hmm", "confidence": None} + + sanitized, reason = _module()._sanitize_resolution(raw, _catalog()) + + assert reason is None + assert sanitized is not None + assert sanitized["input_overrides"] == {} + assert sanitized["parameter_overrides"] == {} + assert sanitized["explicit_prefix"] is None + assert sanitized["confidence"] == 0.0 + + +# ---------- 5. LLM 异常 / 坏 JSON / 缺 choices ---------- + +@pytest.mark.parametrize( + ("content", "error"), + [ + ("this is not json", None), # 坏 JSON + ("{\"hello\": \"world\"}", None), # 合法 JSON 但字段缺失 + (None, RuntimeError("connection reset")), # LLM 抛异常 + ], +) +def test_llm_failures_degrade_to_none_resolution(content, error) -> None: + fake = _FakeClient(response=_response(content) if content is not None else None, error=error) + factory = _CountingFactory(fake) + records = [_record("ext_hmm")] + + resolution = _module().resolve_lightweight_extension_intent( + "使用 HMM 拓展分析对刚上传的文件进行分析", + records, + environ={}, + client_factory=factory, + ) + + assert resolution.extension_id is None + assert resolution.reason == "llm_unavailable_or_invalid" or resolution.reason.startswith("invalid_resolution:") + assert factory.calls == 1 + + +class _NoChoicesClient: + def __init__(self) -> None: + self.chat = SimpleNamespace( + completions=SimpleNamespace( + create=lambda **kwargs: SimpleNamespace(choices=[]), + ) + ) + + +def test_missing_choices_degrades_to_none_resolution() -> None: + factory = _CountingFactory(_NoChoicesClient()) + + resolution = _module().resolve_lightweight_extension_intent( + "使用 HMM 拓展分析对刚上传的文件进行分析", + [_record("ext_hmm")], + environ={}, + client_factory=factory, + ) + + assert resolution.extension_id is None + assert resolution.reason == "llm_unavailable_or_invalid" + + +# ---------- 6. happy path ---------- + +def test_happy_path_resolves_all_fields_and_prefills() -> None: + fake = _FakeClient(response=_response(json.dumps(_good_raw()))) + factory = _CountingFactory(fake) + records = [_record("ext_hmm")] + message = "使用 HMM 拓展分析对刚上传的文件进行分析，参数 e-value=1e-5" + + resolution = _module().resolve_lightweight_extension_intent( + message, records, environ={}, client_factory=factory + ) + + assert resolution.extension_id == "ext_hmm" + assert resolution.extension_name == "ext_hmm" + assert resolution.input_overrides == { + "family.hmm": "input/job-1/_nl_imports/family.hmm", + "01.col.primary.pep.fa": "input/job-1/01.col.primary.pep.fa", + } + assert resolution.parameter_overrides == {"step_002:parameter:003": "1e-5"} + assert resolution.explicit_prefix == "outputs/nl_run" + assert resolution.confidence == 0.85 + assert resolution.prefill is True + assert resolution.reason == "ok" + assert resolution.candidates_considered == ["ext_hmm"] + assert factory.calls == 1 + + +def test_high_confidence_without_overrides_does_not_prefill() -> None: + raw = _good_raw() + raw["input_overrides"] = [] + raw["parameter_overrides"] = [] + fake = _FakeClient(response=_response(json.dumps(raw))) + + resolution = _module().resolve_lightweight_extension_intent( + "使用 HMM 拓展分析", [_record("ext_hmm")], environ={}, client_factory=lambda config: fake + ) + + assert resolution.extension_id == "ext_hmm" + assert resolution.confidence >= _module().PREFILL_THRESHOLD + assert resolution.prefill is False + + +def test_low_confidence_does_not_prefill() -> None: + raw = _good_raw() + raw["confidence"] = 0.3 + fake = _FakeClient(response=_response(json.dumps(raw))) + + resolution = _module().resolve_lightweight_extension_intent( + "使用 HMM 拓展分析", [_record("ext_hmm")], environ={}, client_factory=lambda config: fake + ) + + assert resolution.extension_id == "ext_hmm" + assert resolution.prefill is False + + +# ---------- 7. load_published_lightweight_records ---------- + +def test_load_published_records_only_returns_published(monkeypatch, tmp_path: Path) -> None: + _write(tmp_path, "extensions/ext_pub/extension.json", json.dumps(_record("ext_pub", status="published"))) + _write(tmp_path, "extensions/ext_pending/extension.json", json.dumps(_record("ext_pending", status="confirmation_pending"))) + _write(tmp_path, "extensions/ext_failed/extension.json", json.dumps(_record("ext_failed", status="baseline_failed"))) + _write(tmp_path, "extensions/ext_bad/extension.json", "{bad json") + _write(tmp_path, "extensions/ext_notdict/extension.json", "[1, 2, 3]") + _write(tmp_path, "extensions/ext_other/list.txt", "{}") + + records = _module().load_published_lightweight_records(tmp_path) + + ids = [str(r["extension_id"]) for r in records] + assert ids == ["ext_pub"] + + +# ---------- 8. job_file_hints 进入 LLM user 消息 ---------- + +def test_job_file_hints_reach_llm_user_message() -> None: + fake = _FakeClient(response=_response(json.dumps(_good_raw()))) + factory = _CountingFactory(fake) + + _module().resolve_lightweight_extension_intent( + "使用 HMM 拓展分析对刚上传的文件进行分析", + [_record("ext_hmm")], + environ={}, + client_factory=factory, + job_file_hints=["input/job-1/01.col.primary.pep.fa", "workspace/job-1/out/table.tsv"], + ) + + request = fake.completions.requests[0] + user_message = next(item["content"] for item in request["messages"] if item["role"] == "user") + payload = json.loads(user_message) + assert payload["user_message"] == "使用 HMM 拓展分析对刚上传的文件进行分析" + assert payload["job_files"] == ["input/job-1/01.col.primary.pep.fa", "workspace/job-1/out/table.tsv"] + assert payload["extensions"][0]["extension_id"] == "ext_hmm" + + +def test_job_file_hints_are_capped() -> None: + fake = _FakeClient(response=_response(json.dumps(_good_raw()))) + + _module().resolve_lightweight_extension_intent( + "使用 HMM 拓展分析", + [_record("ext_hmm")], + environ={}, + client_factory=lambda config: fake, + job_file_hints=[f"input/job-1/file_{i}.txt" for i in range(500)], + ) + + request = fake.completions.requests[0] + user_message = next(item["content"] for item in request["messages"] if item["role"] == "user") + payload = json.loads(user_message) + assert len(payload["job_files"]) == _module().MAX_JOB_FILE_HINTS + + +# ---------- 9. 粗筛行为 ---------- + +def test_rough_candidates_without_records_is_empty() -> None: + assert _module()._rough_candidates("拓展分析我的文件", []) == [] + + +def test_rough_trigger_without_records_is_false() -> None: + assert _module()._rough_trigger("拓展分析我的文件", []) is False + + +@pytest.mark.parametrize( + ("message", "expected_ids"), + [ + # 强触发词且无关键词命中：全量兜底 + ("帮我做拓展分析", ["ext_hmm", "ext_dia"]), + ("帮我做扩展分析", ["ext_hmm", "ext_dia"]), + ("do an extension analysis", ["ext_hmm", "ext_dia"]), + # 工具词命中子集（不再全量） + ("用 hmmsearch 搜索结构域", ["ext_hmm"]), + # 工具词无关键词命中：空（不再全量）；用不含 blast/hmmer 相关词的记录验证 + ("run blast against the library", []), + ("run hmmer on my file", []), + # 英文关键词命中单个 + ("diamond align my sequences", ["ext_dia"]), + ("evalue cutoff please", ["ext_dia"]), + # 无关消息 + ("帮我看看这株水稻的基因家族", []), + ], +) +def test_rough_candidates(message: str, expected_ids: list[str]) -> None: + if message.startswith("run "): + # 工具词无关键词命中：用不含相关词的记录，确保结果为 [] + records = [ + _record("x9_zz", summary="Sorts entries in parallel for a table."), + _record("x9_yy", summary="Counts rows in a matrix."), + ] + assert _module()._rough_candidates(message, records) == expected_ids + return + records = [ + _record("ext_hmm", summary="This tool searches a protein FASTA file against an HMM profile using hmmsearch."), + _record("ext_dia", summary="This tool aligns sequences with diamond using an e-value cutoff."), + ] + assert _module()._rough_candidates(message, records) == expected_ids + + +def test_rough_candidates_reject_summary_stopwords_only() -> None: + """summary 只含停用词时，无关消息不应命中该记录。""" + records = [_record("ext_generic", summary="This tool analyzes sequence files using a data tool.")] + assert _module()._rough_candidates("帮我看看这株水稻的基因家族", records) == [] + + +def test_rough_candidates_chinese_summary_substrings() -> None: + record = _record("ext_zh", summary="使用样本表校验工具对测序数据进行质控。") + assert "ext_zh" in _module()._rough_candidates("帮我做样本表校验", [record]) + + +# ---------- 10. resolve 对空 catalog 的容错 ---------- + +def test_resolve_degrades_when_catalog_empty_after_rough_match() -> None: + record = _record("ext_broken") + record["runtime_binding_contract"] = None # 粗筛命中（有 summary 关键词）但目录构造为空 + fake = _FakeClient(response=_response(json.dumps(_good_raw()))) + factory = _CountingFactory(fake) + + resolution = _module().resolve_lightweight_extension_intent( + "hmmsearch tool analysis", [record], environ={}, client_factory=factory + ) + + assert resolution.extension_id is None + assert factory.calls == 0 + + +def test_resolution_uses_compact_extension_name() -> None: + raw = _good_raw() + raw["extension_id"] = "03___hmm_1787130993548" + fake = _FakeClient(response=_response(json.dumps(raw))) + summary = "Searches profiles. It then parses the domain table and reports hits." + record = _record("03___hmm_1787130993548", summary=summary) + + resolution = _module().resolve_lightweight_extension_intent( + "使用拓展分析", [record], environ={}, client_factory=lambda config: fake + ) + + assert resolution.extension_name == "03___hmm" + + +# ---------- 11. 真实 store 冒烟 ---------- + +def test_real_store_catalog_truncation_smoke() -> None: + """用远端真实 store 验证目录截断现象（数据源随仓库存在）。""" + store_root = Path(__file__).resolve().parents[2] / ".runtime/lightweight-extension/store" + records = _module().load_published_lightweight_records(store_root) + if not records: + pytest.skip("真实 store 不可用（无 published 记录）") + catalog = _module()._build_catalog(records) + all_ids = sorted(str(r.get("extension_id")) for r in records) + included_ids = [item["extension_id"] for item in catalog["extensions"]] + assert catalog["truncated"] is True + assert set(included_ids) < set(all_ids) + + +def test_rough_trigger_accepts_real_published_records() -> None: + store_root = Path(__file__).resolve().parents[2] / ".runtime/lightweight-extension/store" + records = _module().load_published_lightweight_records(store_root) + if not records: + pytest.skip("真实 store 不可用（无 published 记录）") + assert _module()._rough_trigger("帮我做拓展分析", records) is True + + +# ---------- 12. 分支补测（覆盖率补充） ---------- + +def test_load_published_records_bad_store_root_returns_empty() -> None: + # Path 构造带 NUL 字节会抛 ValueError，走 except 分支 + assert _module().load_published_lightweight_records(5) == [] + + +def test_function_summary_text_missing_semantic_candidate_returns_empty() -> None: + record = _record("ext_none") + record["semantic_candidate"] = None + assert _module()._function_summary_text(record) == "" + + +def test_build_catalog_skips_record_without_extension_id() -> None: + record = _record("") + catalog = _module()._build_catalog([record]) + assert catalog["extensions"] == [] + assert catalog["truncated"] is False + + +def test_sanitize_rejects_non_mapping_input_override_element() -> None: + raw = _good_raw() + raw["input_overrides"] = [42] + + sanitized, reason = _module()._sanitize_resolution(raw, _catalog()) + + assert sanitized is None + assert reason == "input_overrides 元素必须是对象" + + +def test_sanitize_rejects_parameter_overrides_not_list() -> None: + raw = _good_raw() + raw["parameter_overrides"] = {"step_002:parameter:003": "1e-5"} + + sanitized, reason = _module()._sanitize_resolution(raw, _catalog()) + + assert sanitized is None + assert reason == "parameter_overrides 必须是数组" + + +def test_sanitize_rejects_non_mapping_parameter_element() -> None: + raw = _good_raw() + raw["parameter_overrides"] = ["1e-5"] + + sanitized, reason = _module()._sanitize_resolution(raw, _catalog()) + + assert sanitized is None + assert reason == "parameter_overrides 元素必须是对象" + + +def test_sanitize_rejects_non_string_explicit_prefix() -> None: + raw = _good_raw() + raw["explicit_prefix"] = 42 + + sanitized, reason = _module()._sanitize_resolution(raw, _catalog()) + + assert sanitized is None + assert reason == "explicit_prefix 必须是字符串或 null" + + +def test_sanitize_rejects_confidence_out_of_range() -> None: + raw = _good_raw() + raw["confidence"] = 1.5 + + sanitized, reason = _module()._sanitize_resolution(raw, _catalog()) + + assert sanitized is None + assert reason == "confidence 超出 [0,1]" + + +def test_default_client_path_uses_configured_env_file(monkeypatch, tmp_path: Path) -> None: + """environ 提供有效 env 文件且未注入 client_factory 时走 _default_client 分支。""" + from gfa_core.extension_stage_lightweight import deepseek_semantics + + env_file = _write( + tmp_path, + ".env", + "NEW_API_BASE_URL=https://api.example.test/v1\nNEW_API_MODEL=deepseek-test\nNEW_API_KEY=secret-test-key\n", + ) + fake = _FakeClient(response=_response(json.dumps(_good_raw()))) + monkeypatch.setattr(deepseek_semantics, "_default_client", lambda config: fake) + + resolution = _module().resolve_lightweight_extension_intent( + "使用 HMM 拓展分析", + [_record("ext_hmm")], + environ={"BBAV5_PROJECT_ENV_FILE": str(env_file)}, + client_factory=None, + ) + + assert resolution.extension_id == "ext_hmm" + assert fake.completions.requests + + +def test_default_client_without_config_degrades(monkeypatch) -> None: + """environ 无有效配置且未注入 client_factory 时，默认客户端分支因 config=None 降级。""" + from gfa_core.extension_stage_lightweight import deepseek_semantics + + monkeypatch.setattr(deepseek_semantics, "load_deepseek_config", lambda environ: None) + + resolution = _module().resolve_lightweight_extension_intent( + "使用 HMM 拓展分析", + [_record("ext_hmm")], + environ={}, + client_factory=None, + ) + + assert resolution.extension_id is None + assert resolution.reason == "llm_unavailable_or_invalid" + + +def test_real_store_pure_chinese_function_words_miss_rough_candidates() -> None: + """R2 实测固化：function_summary 全英文，纯中文功能词（无强触发词）不命中粗筛、不调 LLM。""" + store_root = Path(__file__).resolve().parents[2] / ".runtime/lightweight-extension/store" + records = _module().load_published_lightweight_records(store_root) + if not records: + pytest.skip("真实 store 不可用（无 published 记录）") + + assert _module()._rough_candidates("帮我做样本表校验", records) == [] + assert _module()._rough_trigger("帮我做样本表校验", records) is False + # 强触发词下全部记录进候选（61 个），目录截断导致 LLM 看不到后段记录 + assert len(_module()._rough_candidates("用拓展分析做分析", records)) == len(records) + + +# ---------- 13. 第二轮：union 粗筛 / compact 名称 / catalog 24000 ---------- + +@pytest.mark.parametrize( + ("extension_id", "expected"), + [ + ("03___hmm_1787130993548", "03___hmm"), + ("10_pangene__pangene_1786213304494", "10_pangene__pangene"), + ("ext_no_timestamp", "ext_no_timestamp"), + ("", ""), + ("x" * 40, "x" * 30), + ("12_1787211341321", "12"), + ], +) +def test_compact_extension_name(extension_id: str, expected: str) -> None: + assert _module().compact_extension_name(extension_id) == expected + + +def test_extension_id_tokens() -> None: + record = _record("03___hmm_1787130993548") + tokens = _module()._extension_id_tokens(record) + assert "hmm" in tokens + assert "03" not in tokens # len<3 剔除 + + +def test_parameter_flag_tokens() -> None: + record = _record("ext_hmm") + record["runtime_binding_contract"]["parameters"] = [ + {"flag": "--samples", "fact_id": "p1"}, + {"flag": "-e", "fact_id": "p2"}, # len<3 剔除 + {"flag": "--E", "fact_id": "p3"}, # 单字母剔除 + {"flag": "output", "fact_id": "p4"}, # 英文停用词剔除 + ] + tokens = _module()._parameter_flag_tokens(record) + assert "samples" in tokens + assert "e" not in tokens + assert "output" not in tokens + + +def test_parameter_flag_tokens_no_contract() -> None: + record = _record("ext_hmm") + record["runtime_binding_contract"] = None + assert _module()._parameter_flag_tokens(record) == [] + + +def test_record_keyword_sources_merges_all_sources() -> None: + record = _record("03___hmm_1787130993548", summary="Aligns with diamond using e-value.") + record["runtime_binding_contract"]["parameters"] = [{"flag": "--samples"}] + sources = _module()._record_keyword_sources(record) + assert "hmm" in sources # extension_id token + assert "diamond" in sources # summary token + assert "samples" in sources # parameter flag + + +def test_rough_candidates_union_strategy_with_strong_trigger_and_keyword() -> None: + """带工具词 + 强触发词 → 关键词子集（不因强触发词全量）。""" + records = [ + _record("ext_hmm", summary="This tool searches a protein FASTA file against an HMM profile using hmmsearch."), + _record("ext_dia", summary="This tool aligns sequences with diamond using an e-value cutoff."), + ] + cands = _module()._rough_candidates("用 hmmsearch 拓展分析我的文件", records) + assert cands == ["ext_hmm"] + + +def test_rough_candidates_pure_strong_trigger_falls_back_to_all() -> None: + records = [ + _record("ext_hmm", summary="This tool searches a protein FASTA file against an HMM profile using hmmsearch."), + _record("ext_dia", summary="This tool aligns sequences with diamond using an e-value cutoff."), + ] + cands = _module()._rough_candidates("帮我做拓展分析", records) + assert sorted(cands) == ["ext_dia", "ext_hmm"] + + +def test_rough_candidates_no_match_returns_empty() -> None: + records = [_record("ext_hmm", summary="This tool searches a protein FASTA file against an HMM profile using hmmsearch.")] + assert _module()._rough_candidates("帮我看看这株水稻的基因家族", records) == [] + + +def test_rough_candidates_skips_missing_extension_id() -> None: + """缺失 extension_id 的记录被过滤，不产出 None 候选。""" + no_id = _record("") + no_id["extension_id"] = None + records = [ + no_id, + _record("ext_hmm", summary="This tool searches a protein FASTA file against an HMM profile using hmmsearch."), + ] + cands = _module()._rough_candidates("hmmsearch 结构域", records) + assert cands == ["ext_hmm"] + assert "None" not in cands + assert None not in cands + + +def test_rough_candidates_sorted_by_hits_desc() -> None: + """命中关键词数降序（同分按 id 排序）。""" + records = [ + _record("ext_a", summary="diamond e-value cutoff search"), + _record("ext_b", summary="diamond align"), + ] + cands = _module()._rough_candidates("diamond e-value", records) + assert cands == ["ext_a", "ext_b"] + + +def test_catalog_uses_max_24000() -> None: + module = _module() + assert module.MAX_CATALOG_CHARS == 24000 + long_summary = "x" * 500 + records = [_record(f"ext_{i:02d}", summary=long_summary) for i in range(60)] + catalog = module._build_catalog(records) + assert catalog["truncated"] is True + included = len(catalog["extensions"]) + assert included < len(records) + total = sum(len(json.dumps(item, ensure_ascii=False)) for item in catalog["extensions"]) + assert total <= module.MAX_CATALOG_CHARS + first_excluded = records[included]["extension_id"] + extra = len(json.dumps(module._build_catalog([records[included]])["extensions"][0], ensure_ascii=False)) + assert total + extra > module.MAX_CATALOG_CHARS + assert first_excluded not in [item["extension_id"] for item in catalog["extensions"]] + + +def test_catalog_dedupes_duplicate_ids() -> None: + module = _module() + records = [_record("ext_dup"), _record("ext_dup")] + catalog = module._build_catalog(records) + assert len(catalog["extensions"]) == 1 + + +def test_truncated_catalog_sets_reason_hint() -> None: + """目录截断时 resolution.reason 注明请补充工具名。""" + module = _module() + records = [_record(f"ext_{i:02d}", summary="x" * 500) for i in range(60)] + raw = _good_raw() + raw["extension_id"] = "ext_00" # catalog 首个 id（命中数降序后） + # 纯强触发词全量兜底 → catalog 截断 → reason 带提示 + resolution = module.resolve_lightweight_extension_intent( + "帮我做拓展分析", + records, + environ={}, + client_factory=lambda config: _FakeClient(response=_response(json.dumps(raw))), + ) + assert resolution.reason.startswith("ok; 候选过多已截断") + + +def test_resolve_orders_records_by_candidate_score() -> None: + """catalog 顺序跟随 _rough_candidates 的命中数降序。""" + module = _module() + records = [ + _record("ext_a", summary="diamond e-value cutoff search"), + _record("ext_b", summary="diamond align"), + ] + raw = _good_raw() + raw["extension_id"] = "ext_a" + fake = _FakeClient(response=_response(json.dumps(raw))) + + resolution = module.resolve_lightweight_extension_intent( + "diamond e-value", records, environ={}, client_factory=lambda config: fake + ) + assert resolution.extension_id == "ext_a" + request = fake.completions.requests[0] + user_message = next(item["content"] for item in request["messages"] if item["role"] == "user") + payload = json.loads(user_message) + ids = [e["extension_id"] for e in payload["extensions"]] + assert ids[0] == "ext_a" + + +def test_parameter_flag_tokens_skips_non_mapping_items() -> None: + """parameters 列表含非 dict 元素时跳过。""" + record = _record("ext_hmm") + record["runtime_binding_contract"]["parameters"] = [ + {"flag": "--samples", "fact_id": "p1"}, + "not-a-dict", + 42, + None, + ] + tokens = _module()._parameter_flag_tokens(record) + assert tokens == ["samples"] + + +def test_real_store_pm_alignment_counts() -> None: + """PM 实测对齐固化：HMM→9、DIAMOND→8、OrthoFinder→3、KaKs→2、ManifestQC→5、纯拓展→61。""" + store_root = Path(__file__).resolve().parents[2] / ".runtime/lightweight-extension/store" + records = _module().load_published_lightweight_records(store_root) + if len(records) != 61: + pytest.skip(f"真实 store 记录数变化（当前 {len(records)}，基线 61），跳过 PM 对齐断言") + + assert len(_module()._rough_candidates("帮我用 HMM 拓展分析我的文件", records)) == 9 + assert len(_module()._rough_candidates("帮我用 DIAMOND 拓展分析", records)) == 8 + assert len(_module()._rough_candidates("用 OrthoFinder 拓展分析", records)) == 3 + assert len(_module()._rough_candidates("用 KaKs 拓展分析", records)) == 2 + assert len(_module()._rough_candidates("用 ManifestQC 拓展分析", records)) == 5 + assert len(_module()._rough_candidates("帮我做拓展分析", records)) == 61 + # HMM 子集全为 03_ 系列（hmm 关键词），且截断在子集内不发生 + hmm = _module()._rough_candidates("帮我用 HMM 拓展分析我的文件", records) + assert all(c.startswith("03___hmm") for c in hmm[:4]) + catalog = _module()._build_catalog( + [r for r in records if str(r.get("extension_id")) in hmm] + ) + assert catalog["truncated"] is False