技能展示样本运行器 + 参考文献来源过滤调整

- skills 路由新增 _run_skill_python_sample，按本地 Python 候选实际执行
  技能脚本，作为展示适配器的真实样本来源
- skill-display 运行时与适配器调整，参考文献收集放宽起始消息约束

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

offline-backend-20260512/backend/app/gateway/routers/skills.py

@@ -2,7 +2,10 @@ import asyncio
 import json
 import logging
 import mimetypes
+import os
 import re
+import subprocess
+import sys
 import tempfile
 from datetime import datetime
 from pathlib import Path
@@ -412,6 +415,102 @@ def _extract_json_from_llm_output(raw: str) -> dict[str, Any] | None:
         return None
 
 
+def _skill_sample_python_candidates() -> list[str]:
+    candidates = [
+        os.environ.get("DEERFLOW_SKILL_SAMPLE_PYTHON"),
+        r"C:\Users\ji\anaconda3\envs\open_manus\python.exe",
+        sys.executable,
+        "python",
+    ]
+    result: list[str] = []
+    for candidate in candidates:
+        if not candidate or candidate in result:
+            continue
+        result.append(candidate)
+    return result
+
+
+def _find_skill_sample_script(skill: Skill) -> Path | None:
+    skill_dir = skill.skill_file.parent
+    preferred = [
+        skill_dir / "scripts" / "search.py",
+        skill_dir / "search.py",
+        skill_dir / "scripts" / "run.py",
+        skill_dir / "run.py",
+    ]
+    for path in preferred:
+        if path.exists() and path.is_file():
+            return path
+    scripts_dir = skill_dir / "scripts"
+    if scripts_dir.exists():
+        for path in sorted(scripts_dir.glob("*.py")):
+            if path.is_file():
+                return path
+    return None
+
+
+async def _run_skill_python_sample(skill: Skill) -> dict[str, Any] | None:
+    """Run a skill's Python script to capture a real JSON sample.
+
+    This is used by the display adapter editor. It intentionally runs the same
+    script the agent would run, instead of guessing from SKILL.md examples.
+    """
+    script = _find_skill_sample_script(skill)
+    if script is None:
+        return None
+
+    query = os.environ.get("DEERFLOW_SKILL_SAMPLE_QUERY") or "参考文献模式测试"
+    env = os.environ.copy()
+    env.setdefault("PYTHONIOENCODING", "utf-8")
+    env.setdefault("DEERFLOW_MOCK_KNOWLEDGE_SEARCH", "1")
+    env.setdefault("KNOWLEDGE_SEARCH_V2_FIXED", "1")
+
+    for python_exe in _skill_sample_python_candidates():
+        try:
+            completed = await asyncio.to_thread(
+                subprocess.run,
+                [python_exe, str(script), query],
+                cwd=str(script.parent),
+                env=env,
+                capture_output=True,
+                text=True,
+                encoding="utf-8",
+                errors="replace",
+                timeout=60,
+            )
+        except (subprocess.TimeoutExpired, FileNotFoundError, OSError) as exc:
+            logger.warning("Skill sample runner failed for %s via %s: %s", skill.name, python_exe, exc)
+            continue
+        except Exception:
+            logger.warning("Skill sample runner crashed for %s via %s", skill.name, python_exe, exc_info=True)
+            continue
+
+        out_text = (completed.stdout or "").strip()
+        err_text = (completed.stderr or "").strip()
+        if completed.returncode != 0:
+            logger.warning(
+                "Skill sample script exited %s for %s via %s: %s",
+                completed.returncode,
+                skill.name,
+                python_exe,
+                err_text[-1000:],
+            )
+            continue
+        parsed = _extract_json_from_llm_output(out_text)
+        if parsed:
+            parsed.setdefault("__sampleSource", "python-script")
+            parsed.setdefault("__sampleScript", str(script))
+            return parsed
+        logger.warning(
+            "Skill sample script returned non-JSON for %s via %s: stdout=%s stderr=%s",
+            skill.name,
+            python_exe,
+            out_text[-1000:],
+            err_text[-1000:],
+        )
+    return None
+
+
 async def _generate_skill_display_sample(skill: Skill, app_config: AppConfig) -> dict[str, Any]:
     """Generate a realistic return-shape sample for ``skill`` via the LLM.
 
@@ -420,6 +519,10 @@ async def _generate_skill_display_sample(skill: Skill, app_config: AppConfig) ->
     when the model is unconfigured, the call times out, or the output cannot
     be parsed as a JSON object.
     """
+    script_sample = await _run_skill_python_sample(skill)
+    if script_sample:
+        return script_sample
+
     try:
         skill_content = skill.skill_file.read_text(encoding="utf-8")
     except OSError as exc:
@@ -461,7 +564,23 @@ async def _generate_skill_display_sample(skill: Skill, app_config: AppConfig) ->
         logger.warning("Skill display sample LLM call timed out for %s; falling back to static mock", skill.name)
     except Exception:
         logger.warning("Skill display sample LLM call failed for %s; falling back to static mock", skill.name, exc_info=True)
-    return _mock_skill_display_sample(skill.name)
+    return _generic_skill_display_sample(skill.name)
+
+
+def _generic_skill_display_sample(skill_name: str) -> dict[str, Any]:
+    return {
+        "success": True,
+        "query": "sample query",
+        "results": [
+            {
+                "id": f"{skill_name}-sample-001",
+                "title": "Sample result",
+                "summary": "Sample structured output used to infer field mapping.",
+                "source": skill_name,
+                "updatedAt": "2026-05-25",
+            }
+        ],
+    }
 
 
 def _mock_skill_display_sample(skill_name: str) -> dict[str, Any]:
@@ -1883,9 +2002,6 @@ async def get_skill_display_sample(
         skill = next((s for s in skills if s.name == skill_name), None)
         if skill is None:
             raise HTTPException(status_code=404, detail=f"Skill '{skill_name}' not found")
-        adapter = await _get_skill_display_adapter(skill_name)
-        if adapter and isinstance(adapter.get("sample"), dict):
-            return SkillDisplaySampleResponse(skill_name=skill_name, sample=adapter["sample"])
         sample = await _generate_skill_display_sample(skill, config)
         await _upsert_skill_display_adapter(skill_name, sample=sample)
         return SkillDisplaySampleResponse(skill_name=skill_name, sample=sample)