测试：轻量拓展分析 NL 触发两轮测试交付（单测 79 + 集成 23 + 前端契约 20）

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 路由

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"]

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