lambdagent/agentexample/research67/run.py

#!/usr/bin/env python3
"""
research67/run.py — 简化版 Runner
===================================

用本地工具 stub 替代尚不存在的 MCP 服务，直接通过 lambdagent runtime
执行 research67 多智能体科研流程。

使用 Claude Opus 4.6 (1M context) 的原生能力:
  - web_search → 替代 scholar-mcp / arxiv-mcp
  - python_exec → 替代 code-sandbox-mcp / plotting-mcp
  - local fs    → 替代 fs-mcp
  - local latex → 替代 latex-mcp

用法:
  export ANTHROPIC_API_KEY=sk-ant-...
  python run.py                         # 完整流程
  python run.py --phase idea_analysis   # 只跑某个阶段
  python run.py --resume                # 从上次检查点恢复
"""
from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
import time
import shutil
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional

# ── 路径设置 ──────────────────────────────────────────────
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent  # Paper08/
AGENT_DIR = Path(__file__).resolve().parent                   # research67/
sys.path.insert(0, str(PROJECT_ROOT))

from lambdagent.fromconfig import from_config
from lambdagent.agentruntime.runtime import Runtime, RuntimeResult
from lambdagent.agentruntime.config import RuntimeConfig
from lambdagent.core import Context


# ═══════════════════════════════════════════════════════════
# 1. 本地工具 Stubs — 替代 MCP 服务
# ═══════════════════════════════════════════════════════════

class LocalToolkit:
    """
    将 6 个 MCP 服务的功能用本地实现替代。
    每个方法签名对齐 MCP tool 的 (input: str) -> str 接口。
    """

    def __init__(self, workspace: Path):
        self.workspace = workspace

    # ── fs-mcp 替代 ──────────────────────────────────────
    def read_file(self, path: str) -> str:
        p = self._resolve(path)
        if not p.exists():
            return f"[ERROR] File not found: {p}"
        return p.read_text(encoding="utf-8")

    def write_file(self, args: str) -> str:
        """args = JSON: {"path": "...", "content": "..."}"""
        d = json.loads(args)
        p = self._resolve(d["path"])
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(d["content"], encoding="utf-8")
        return f"Written {len(d['content'])} bytes to {p}"

    def list_dir(self, path: str) -> str:
        p = self._resolve(path)
        if not p.is_dir():
            return f"[ERROR] Not a directory: {p}"
        entries = sorted(p.iterdir())
        return "\n".join(f"{'[DIR] ' if e.is_dir() else ''}{e.name}" for e in entries)

    def mkdir(self, path: str) -> str:
        p = self._resolve(path)
        p.mkdir(parents=True, exist_ok=True)
        return f"Created: {p}"

    def copy_file(self, args: str) -> str:
        d = json.loads(args)
        src, dst = self._resolve(d["src"]), self._resolve(d["dst"])
        dst.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(src, dst)
        return f"Copied {src} → {dst}"

    # ── scholar-mcp / arxiv-mcp 替代 ─────────────────────
    def semantic_scholar_search(self, query: str) -> str:
        """用 Semantic Scholar 公开 API 搜索论文 (无需 API key)"""
        import urllib.request
        import urllib.parse
        url = f"https://api.semanticscholar.org/graph/v1/paper/search?query={urllib.parse.quote(query)}&limit=10&fields=title,authors,year,citationCount,venue,abstract,externalIds"
        try:
            req = urllib.request.Request(url, headers={"User-Agent": "research67/1.0"})
            with urllib.request.urlopen(req, timeout=30) as resp:
                data = json.loads(resp.read())
            papers = data.get("data", [])
            results = []
            for p in papers:
                results.append({
                    "title": p.get("title", ""),
                    "authors": [a.get("name", "") for a in (p.get("authors") or [])[:3]],
                    "year": p.get("year"),
                    "venue": p.get("venue", ""),
                    "citationCount": p.get("citationCount", 0),
                    "abstract": (p.get("abstract") or "")[:200],
                    "arxivId": (p.get("externalIds") or {}).get("ArXiv", ""),
                })
            return json.dumps({"total": data.get("total", 0), "papers": results}, ensure_ascii=False, indent=2)
        except Exception as e:
            return f"[SEARCH_ERROR] {e}"

    def get_paper_details(self, paper_id: str) -> str:
        import urllib.request
        url = f"https://api.semanticscholar.org/graph/v1/paper/{paper_id}?fields=title,abstract,authors,year,venue,citationCount,references,citations"
        try:
            req = urllib.request.Request(url, headers={"User-Agent": "research67/1.0"})
            with urllib.request.urlopen(req, timeout=30) as resp:
                data = json.loads(resp.read())
            return json.dumps(data, ensure_ascii=False, indent=2)[:3000]
        except Exception as e:
            return f"[ERROR] {e}"

    def get_citations(self, paper_id: str) -> str:
        return self.get_paper_details(paper_id)

    def get_references(self, paper_id: str) -> str:
        return self.get_paper_details(paper_id)

    def arxiv_search(self, query: str) -> str:
        """用 arXiv API 搜索"""
        import urllib.request
        import urllib.parse
        url = f"http://export.arxiv.org/api/query?search_query=all:{urllib.parse.quote(query)}&start=0&max_results=10&sortBy=submittedDate&sortOrder=descending"
        try:
            req = urllib.request.Request(url, headers={"User-Agent": "research67/1.0"})
            with urllib.request.urlopen(req, timeout=30) as resp:
                xml = resp.read().decode("utf-8")
            # Simple XML extraction
            entries = []
            for entry in xml.split("<entry>")[1:]:
                title = _extract_xml(entry, "title").strip().replace("\n", " ")
                summary = _extract_xml(entry, "summary").strip()[:200]
                arxiv_id = _extract_xml(entry, "id").split("/abs/")[-1]
                published = _extract_xml(entry, "published")[:10]
                entries.append({"title": title, "arxiv_id": arxiv_id, "published": published, "summary": summary})
            return json.dumps({"results": entries}, ensure_ascii=False, indent=2)
        except Exception as e:
            return f"[ARXIV_ERROR] {e}"

    def arxiv_download(self, arxiv_id: str) -> str:
        return f"[STUB] PDF download not implemented for {arxiv_id}. Use arxiv_search for metadata."

    def pdf_to_text(self, path: str) -> str:
        return f"[STUB] PDF extraction not implemented for {path}."

    # ── code-sandbox-mcp 替代 ────────────────────────────
    def python_exec(self, code: str) -> str:
        """在本地沙箱执行 Python 代码"""
        try:
            result = subprocess.run(
                [sys.executable, "-c", code],
                capture_output=True, text=True, timeout=120,
                cwd=str(self.workspace),
            )
            output = result.stdout
            if result.returncode != 0:
                output += f"\n[STDERR] {result.stderr}"
            return output[:5000]
        except subprocess.TimeoutExpired:
            return "[TIMEOUT] Python execution exceeded 120s"
        except Exception as e:
            return f"[EXEC_ERROR] {e}"

    def install_package(self, pkg: str) -> str:
        try:
            result = subprocess.run(
                [sys.executable, "-m", "pip", "install", pkg, "-q"],
                capture_output=True, text=True, timeout=60,
            )
            return result.stdout + result.stderr
        except Exception as e:
            return f"[PIP_ERROR] {e}"

    def gpu_status(self, _: str = "") -> str:
        try:
            r = subprocess.run(["nvidia-smi", "--query-gpu=name,memory.used,memory.total", "--format=csv,noheader"],
                               capture_output=True, text=True, timeout=10)
            return r.stdout if r.returncode == 0 else "No GPU detected (CPU mode)"
        except FileNotFoundError:
            return "nvidia-smi not found (CPU mode)"

    def disk_usage(self, path: str = ".") -> str:
        import shutil as sh
        total, used, free = sh.disk_usage(self._resolve(path))
        return f"Total: {total // (1024**3)}GB, Used: {used // (1024**3)}GB, Free: {free // (1024**3)}GB"

    # ── latex-mcp 替代 ───────────────────────────────────
    def compile_latex(self, tex_path: str) -> str:
        p = self._resolve(tex_path)
        if not p.exists():
            return f"[ERROR] {p} not found"
        try:
            r = subprocess.run(
                ["pdflatex", "-interaction=nonstopmode", "-output-directory", str(p.parent), str(p)],
                capture_output=True, text=True, timeout=60, cwd=str(p.parent),
            )
            # Run twice for references
            subprocess.run(
                ["pdflatex", "-interaction=nonstopmode", "-output-directory", str(p.parent), str(p)],
                capture_output=True, text=True, timeout=60, cwd=str(p.parent),
            )
            pdf = p.with_suffix(".pdf")
            if pdf.exists():
                return f"Compiled: {pdf} ({pdf.stat().st_size} bytes)"
            return f"Compilation output:\n{r.stdout[-2000:]}\n{r.stderr[-1000:]}"
        except FileNotFoundError:
            return "[STUB] pdflatex not found. LaTeX compilation skipped."
        except Exception as e:
            return f"[LATEX_ERROR] {e}"

    def latex_lint(self, tex_path: str) -> str:
        return f"[STUB] LaTeX lint for {tex_path}: no issues found."

    def bibtex_format(self, bib_path: str) -> str:
        return f"[STUB] BibTeX formatted: {bib_path}"

    # ── plotting-mcp 替代 ────────────────────────────────
    def create_figure(self, args: str) -> str:
        """用 matplotlib 生成图表，args 是 JSON spec"""
        code = f"""
import json
spec = json.loads('''{args}''')
# Placeholder: actual plotting code would be generated by the agent
print(f"Figure created: {{spec.get('name', 'figure')}}.pdf")
"""
        return self.python_exec(code)

    def create_table(self, args: str) -> str:
        return f"[STUB] Table created from spec"

    # ── 特殊本地工具 ─────────────────────────────────────
    def terminate(self, result: str) -> str:
        return result

    def read_idea(self, path: str = "") -> str:
        idea_path = AGENT_DIR / "paper" / "paper01" / "IDEA.md"
        if not idea_path.exists():
            return f"[ERROR] IDEA.md not found at {idea_path}"
        return idea_path.read_text(encoding="utf-8")

    def save_checkpoint(self, data: str) -> str:
        ckpt_dir = self.workspace / ".checkpoints"
        ckpt_dir.mkdir(parents=True, exist_ok=True)
        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
        path = ckpt_dir / f"checkpoint_{ts}.json"
        path.write_text(data, encoding="utf-8")
        return f"Checkpoint saved: {path}"

    def load_checkpoint(self, pattern: str = "latest") -> str:
        ckpt_dir = self.workspace / ".checkpoints"
        if not ckpt_dir.exists():
            return "[ERROR] No checkpoints found"
        files = sorted(ckpt_dir.glob("checkpoint_*.json"))
        if not files:
            return "[ERROR] No checkpoints found"
        return files[-1].read_text(encoding="utf-8")

    def human_feedback(self, question: str) -> str:
        print(f"\n{'='*60}")
        print(f"[HUMAN FEEDBACK REQUESTED]")
        print(f"{question}")
        print(f"{'='*60}")
        try:
            answer = input("Your response (or press Enter to skip): ").strip()
            return answer if answer else "[NO FEEDBACK]"
        except (EOFError, KeyboardInterrupt):
            return "[NO FEEDBACK]"

    def dispatch_agent(self, args: str) -> str:
        """调度子智能体执行"""
        d = json.loads(args)
        agent_name = d.get("agent", "")
        agent_input = d.get("input", "")
        phase_dir = d.get("phase_dir", "")

        agent_yml = AGENT_DIR / "agents" / f"{agent_name}.yml"
        if not agent_yml.exists():
            return f"[ERROR] Agent config not found: {agent_yml}"

        print(f"\n  >> Dispatching sub-agent: {agent_name}")
        print(f"     Config: {agent_yml}")
        print(f"     Phase dir: {phase_dir}")

        try:
            result = Runtime.execute(str(agent_yml), agent_input)
            return result.result
        except Exception as e:
            return f"[AGENT_ERROR] {agent_name} failed: {e}"

    # ── 内部辅助 ─────────────────────────────────────────
    def _resolve(self, path: str) -> Path:
        p = Path(path)
        if p.is_absolute():
            return p
        return self.workspace / p

    def get_tool_map(self) -> Dict[str, callable]:
        """返回工具名 → 函数的映射，用于注入 runtime"""
        return {
            # fs-mcp
            "read_file": self.read_file,
            "write_file": self.write_file,
            "list_dir": self.list_dir,
            "mkdir": self.mkdir,
            "copy_file": self.copy_file,
            # scholar-mcp
            "semantic_scholar_search": self.semantic_scholar_search,
            "get_paper_details": self.get_paper_details,
            "get_citations": self.get_citations,
            "get_references": self.get_references,
            # arxiv-mcp
            "arxiv_search": self.arxiv_search,
            "arxiv_download": self.arxiv_download,
            "pdf_to_text": self.pdf_to_text,
            # code-sandbox-mcp
            "python_exec": self.python_exec,
            "install_package": self.install_package,
            "gpu_status": self.gpu_status,
            "disk_usage": self.disk_usage,
            # latex-mcp
            "compile_latex": self.compile_latex,
            "latex_lint": self.latex_lint,
            "bibtex_format": self.bibtex_format,
            # plotting-mcp
            "create_figure": self.create_figure,
            "create_table": self.create_table,
            # local
            "terminate": self.terminate,
            "read_idea": self.read_idea,
            "save_checkpoint": self.save_checkpoint,
            "load_checkpoint": self.load_checkpoint,
            "human_feedback": self.human_feedback,
            "dispatch_agent": self.dispatch_agent,
        }


# ═══════════════════════════════════════════════════════════
# 2. Runner — 驱动整个科研流程
# ═══════════════════════════════════════════════════════════

PHASES = [
    ("01_idea_analysis",       "idea-analyst"),
    ("02_literature_review",   "lit-searcher"),
    ("03_experiment_plan",     "exp-planner"),
    ("04_experiment_execution","exp-executor"),
    ("05_result_analysis",     "result-analyzer"),
    ("06_paper_writing",       "paper-writer"),
    ("07_data_verification",   "data-verifier"),
    ("08_paper_review",        "paper-reviewer"),
]


def create_workspace() -> Path:
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    ws = AGENT_DIR / "workspace" / f"run_{ts}"
    ws.mkdir(parents=True, exist_ok=True)
    return ws


def run_phase(
    phase_name: str,
    agent_name: str,
    workspace: Path,
    round_num: int,
    toolkit: LocalToolkit,
    prev_reports: List[str],
    revision_context: str = "",
) -> Dict[str, Any]:
    """运行单个阶段的子智能体"""
    phase_dir = workspace / f"round_{round_num}" / phase_name
    phase_dir.mkdir(parents=True, exist_ok=True)
    (phase_dir / "artifacts").mkdir(exist_ok=True)

    agent_yml = AGENT_DIR / "agents" / f"{agent_name}.yml"

    # 构建输入上下文
    context_parts = [
        f"workspace: {workspace}",
        f"phase_dir: {phase_dir}",
        f"round: {round_num}",
        f"dependencies: {json.dumps(prev_reports)}",
    ]
    if revision_context:
        context_parts.append(f"revision_context: {revision_context}")

    # 读取 IDEA
    idea_content = toolkit.read_idea()
    context_parts.append(f"\n## IDEA.md Content:\n{idea_content}")

    # 附上前序阶段的关键结果摘要
    for rp in prev_reports[-3:]:  # 最多带最近 3 个阶段
        rp_path = workspace / rp
        if rp_path.exists():
            try:
                data = json.loads(rp_path.read_text(encoding="utf-8"))
                # 只取 _meta + 关键字段，避免太长
                summary = {k: v for k, v in data.items() if k == "_meta" or not isinstance(v, (list, dict))}
                context_parts.append(f"\n## Previous report ({rp}):\n{json.dumps(summary, ensure_ascii=False, indent=2)[:2000]}")
            except Exception:
                pass

    input_text = "\n".join(context_parts)

    print(f"\n{'━'*60}")
    print(f"  Phase: {phase_name}")
    print(f"  Agent: {agent_name} → {agent_yml.name}")
    print(f"  Output: {phase_dir}")
    print(f"{'━'*60}")

    t0 = time.time()
    try:
        result = Runtime.execute(str(agent_yml), input_text)
        duration = time.time() - t0
        print(f"  ✓ Completed in {duration:.1f}s, {len(result.trace)} trace steps")

        # 保存 trace
        trace_path = phase_dir / "trace.json"
        trace_data = [
            {"step": t.step, "term": t.term_name, "type": t.term_type,
             "duration_ms": t.duration_ms, "output": t.output[:200]}
            for t in result.trace
        ]
        trace_path.write_text(json.dumps(trace_data, indent=2), encoding="utf-8")

        # 尝试解析 report.json
        report_path = phase_dir / "report.json"
        if report_path.exists():
            report = json.loads(report_path.read_text(encoding="utf-8"))
        else:
            # Agent 可能直接返回了 JSON
            try:
                report = json.loads(result.result)
                report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
            except (json.JSONDecodeError, TypeError):
                report = {"_meta": {"status": "completed"}, "raw_output": result.result[:3000]}
                report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")

        return report

    except Exception as e:
        duration = time.time() - t0
        print(f"  ✗ Failed after {duration:.1f}s: {e}")
        error_report = {
            "_meta": {"status": "failed", "error": str(e), "duration_seconds": duration},
        }
        (phase_dir / "report.json").write_text(json.dumps(error_report, indent=2), encoding="utf-8")
        return error_report


def run_full_pipeline(
    start_phase: Optional[str] = None,
    max_rounds: int = 5,
):
    """运行完整的科研流程"""

    # 检查 API key
    if not os.environ.get("ANTHROPIC_API_KEY"):
        print("ERROR: ANTHROPIC_API_KEY environment variable not set.")
        print("  export ANTHROPIC_API_KEY=sk-ant-...")
        sys.exit(1)

    workspace = create_workspace()
    toolkit = LocalToolkit(workspace)

    print(f"""
╔══════════════════════════════════════════════════════════╗
║         Research-467 Autonomous Research Agent           ║
║         Model: Claude Opus 4.6 (1M context)              ║
║         IDEA: paper/paper01/IDEA.md                      ║
╚══════════════════════════════════════════════════════════╝
  Workspace: {workspace}
  Max rounds: {max_rounds}
""")

    # 写入项目计划
    idea = toolkit.read_idea()
    project_plan = f"""# Research-467 Project Plan

## IDEA Source
paper/paper01/IDEA.md

## Model
Claude Opus 4.6 (1M context) — claude-opus-4-6

## Phases
{chr(10).join(f'{i+1}. {name} → {agent}' for i, (name, agent) in enumerate(PHASES))}

## Target
CCF-A conference acceptance probability ≥ 50%

## Created
{datetime.now(timezone.utc).isoformat()}
"""
    (workspace / "project_plan.md").write_text(project_plan, encoding="utf-8")

    progress = {
        "project_id": workspace.name,
        "current_round": 1,
        "current_phase": PHASES[0][0],
        "max_rounds": max_rounds,
        "best_score": 0.0,
        "rounds": [],
    }

    for round_num in range(1, max_rounds + 1):
        print(f"\n{'═'*60}")
        print(f"  ROUND {round_num} / {max_rounds}")
        print(f"{'═'*60}")

        progress["current_round"] = round_num
        round_reports = []
        prev_reports = []

        # 确定本轮需要跑哪些阶段
        phases_to_run = PHASES
        if round_num > 1 and "revision_target" in progress:
            # 回退到指定阶段
            target = progress["revision_target"]
            start_idx = next((i for i, (name, _) in enumerate(PHASES) if name == target), 0)
            phases_to_run = PHASES[start_idx:]
            print(f"  Revision: starting from {target}")

        if start_phase and round_num == 1:
            start_idx = next((i for i, (name, _) in enumerate(PHASES) if start_phase in name), 0)
            phases_to_run = PHASES[start_idx:]

        for phase_name, agent_name in phases_to_run:
            progress["current_phase"] = phase_name
            _save_progress(workspace, progress)

            report = run_phase(
                phase_name, agent_name, workspace, round_num, toolkit,
                prev_reports,
                revision_context=progress.get("revision_context", ""),
            )

            report_rel = f"round_{round_num}/{phase_name}/report.json"
            prev_reports.append(report_rel)
            round_reports.append({"phase": phase_name, "status": report.get("_meta", {}).get("status", "unknown")})

        progress["rounds"].append({"round": round_num, "phases": round_reports})

        # 检查评审结果
        review_report_path = workspace / f"round_{round_num}" / "08_paper_review" / "report.json"
        if review_report_path.exists():
            review = json.loads(review_report_path.read_text(encoding="utf-8"))
            score = review.get("acceptance_probability", 0)
            progress["best_score"] = max(progress.get("best_score", 0), score)

            print(f"\n  Review Score: {score:.2%}")
            print(f"  Best Score:   {progress['best_score']:.2%}")

            if score >= 0.5:
                print(f"\n  ✓ TARGET REACHED! Acceptance probability: {score:.2%}")
                _finalize(workspace, round_num, toolkit, progress)
                break

            # 决定回退目标
            scores = review.get("scores", {})
            if scores:
                lowest = min(scores.items(), key=lambda x: x[1].get("score", 10) if isinstance(x[1], dict) else 10)
                dimension = lowest[0]
                target_map = {
                    "novelty": "01_idea_analysis",
                    "soundness": "03_experiment_plan",
                    "clarity": "06_paper_writing",
                    "significance": "02_literature_review",
                    "reproducibility": "04_experiment_execution",
                }
                progress["revision_target"] = target_map.get(dimension, "06_paper_writing")
                revision_items = review.get("revision_priority", [])
                progress["revision_context"] = json.dumps(revision_items[:3], ensure_ascii=False)
                print(f"  Lowest dimension: {dimension} → Retrying from {progress['revision_target']}")
        else:
            print(f"\n  [WARN] No review report found, ending pipeline")
            break

    else:
        # 达到 max_rounds
        print(f"\n  Max rounds ({max_rounds}) reached. Finalizing best version.")
        _finalize(workspace, round_num, toolkit, progress)

    _save_progress(workspace, progress)
    print(f"\n{'═'*60}")
    print(f"  Pipeline complete. Workspace: {workspace}")
    print(f"{'═'*60}\n")


def _save_progress(workspace: Path, progress: dict):
    (workspace / "progress.json").write_text(
        json.dumps(progress, ensure_ascii=False, indent=2), encoding="utf-8"
    )


def _finalize(workspace: Path, last_round: int, toolkit: LocalToolkit, progress: dict):
    """归档最终产出到 final/ 目录"""
    final_dir = workspace / "final"
    final_dir.mkdir(exist_ok=True)

    round_dir = workspace / f"round_{last_round}"

    # 复制论文
    for src_name, dst_name in [
        ("06_paper_writing/artifacts/paper.tex", "paper.tex"),
        ("06_paper_writing/artifacts/paper.pdf", "paper.pdf"),
        ("06_paper_writing/artifacts/references.bib", "references.bib"),
    ]:
        src = round_dir / src_name
        if src.exists():
            shutil.copy2(src, final_dir / dst_name)

    # 复制图表
    figs_src = round_dir / "05_result_analysis" / "artifacts" / "figures"
    if figs_src.exists():
        shutil.copytree(figs_src, final_dir / "figures", dirs_exist_ok=True)

    # 复制实验数据
    data_src = round_dir / "04_experiment_execution" / "artifacts" / "raw_results"
    if data_src.exists():
        shutil.copytree(data_src, final_dir / "experiment_data", dirs_exist_ok=True)

    # 生成投稿建议
    recommendation = {
        "final_paper_path": "final/paper.pdf",
        "total_rounds": last_round,
        "best_score": progress.get("best_score", 0),
        "workspace": str(workspace),
        "completed_at": datetime.now(timezone.utc).isoformat(),
    }
    (final_dir / "submission_recommendation.json").write_text(
        json.dumps(recommendation, ensure_ascii=False, indent=2), encoding="utf-8"
    )

    progress["final_output"] = str(final_dir)
    print(f"\n  Final output archived to: {final_dir}")


# ═══════════════════════════════════════════════════════════
# 3. XML 辅助
# ═══════════════════════════════════════════════════════════

def _extract_xml(text: str, tag: str) -> str:
    start = text.find(f"<{tag}>")
    end = text.find(f"</{tag}>")
    if start == -1 or end == -1:
        return ""
    return text[start + len(f"<{tag}>"):end]


# ═══════════════════════════════════════════════════════════
# 4. CLI
# ═══════════════════════════════════════════════════════════

def main():
    parser = argparse.ArgumentParser(description="Research-467 Simplified Runner")
    parser.add_argument("--phase", type=str, default=None,
                        help="Start from a specific phase (e.g., idea_analysis, paper_writing)")
    parser.add_argument("--max-rounds", type=int, default=5,
                        help="Maximum iteration rounds (default: 5)")
    parser.add_argument("--test-tools", action="store_true",
                        help="Test local toolkit connectivity and exit")
    args = parser.parse_args()

    if args.test_tools:
        _test_tools()
        return

    run_full_pipeline(start_phase=args.phase, max_rounds=args.max_rounds)


def _test_tools():
    """测试本地工具是否可用"""
    ws = Path("/tmp/research67_test")
    ws.mkdir(exist_ok=True)
    toolkit = LocalToolkit(ws)

    tests = [
        ("read_idea", lambda: toolkit.read_idea()),
        ("semantic_scholar_search", lambda: toolkit.semantic_scholar_search("spectral analysis pretrained features")),
        ("arxiv_search", lambda: toolkit.arxiv_search("label-free feature quality")),
        ("python_exec", lambda: toolkit.python_exec("import sys; print(f'Python {sys.version}')")),
        ("write_file", lambda: toolkit.write_file(json.dumps({"path": "test.txt", "content": "hello"}))),
        ("read_file", lambda: toolkit.read_file("test.txt")),
        ("list_dir", lambda: toolkit.list_dir(".")),
        ("gpu_status", lambda: toolkit.gpu_status()),
    ]

    print("Testing local toolkit...\n")
    for name, fn in tests:
        try:
            t0 = time.time()
            result = fn()
            dt = time.time() - t0
            ok = "[ERROR]" not in result and "[STUB]" not in result
            status = "✓" if ok else "⚠"
            print(f"  {status} {name:30s} ({dt:.1f}s) {result[:80]}")
        except Exception as e:
            print(f"  ✗ {name:30s} ERROR: {e}")

    shutil.rmtree(ws, ignore_errors=True)
    print("\nDone.")


if __name__ == "__main__":
    main()