lambdagent/scripts/dogfood_agentpack.py

#!/usr/bin/env python3
"""
Phase I dogfood — AgentPack end-to-end script with real LLM.

Usage:
    # Start the server in another terminal first:
    AGENTPAAS_DEPLOYMENT_MODE=desktop uvicorn agentpaas.api.app:app --port 8000

    # Run the dogfood script:
    python scripts/dogfood_agentpack.py

    # Or point at a non-default host/key:
    python scripts/dogfood_agentpack.py --url http://127.0.0.1:8000 --key ap_xxxx

The script:
  1. Zips the built-in research.top-journal-reviewer pack from the repo
  2. Installs it via POST /api/v1/agentpacks/install
  3. Creates an agent via POST /api/v1/agentpacks/{id}/create-agent
  4. Runs the agent with a short test abstract
  5. Prints the workspace path and verifies review_report.md exists

Requirements:
  - Server running with ANTHROPIC_API_KEY or DASHSCOPE_API_KEY set
  - pip install requests (standard, usually present)
"""
from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
import tempfile
import time
import zipfile
from pathlib import Path


# ── Config ────────────────────────────────────────────────────────────────────

REVIEWER_PACK_ID = "research.top-journal-reviewer"

TEST_ABSTRACT = """\
Title: Quantum Speedup for Dense Matrix Multiplication via Phase Estimation.

Abstract: We present a quantum algorithm that computes dense n×n matrix products
in O(n^{1.5} log n) time using quantum phase estimation and amplitude amplification,
outperforming the classical Strassen barrier O(n^{2.37}). We provide a gate-level
circuit of depth O(n^{0.75} polylog n) with error probability δ < 0.01. Experiments
on 16×16 matrices on a simulated 40-qubit processor confirm a 3.2× speedup.

请审阅这篇论文的核心贡献声明、技术正确性和实验充分性，生成完整的审稿报告。
"""

# ── Helpers ───────────────────────────────────────────────────────────────────

def _print(label: str, msg: str = "", ok: bool = True) -> None:
    status = "✓" if ok else "✗"
    color = "\033[32m" if ok else "\033[31m"
    reset = "\033[0m"
    print(f"  {color}{status}{reset}  {label}: {msg}")


def _zip_pack(repo_root: Path, pack_id: str, dest_dir: Path) -> str:
    pack_dir = repo_root / "agentexample" / "agentpacks" / pack_id
    if not pack_dir.is_dir():
        sys.exit(f"Pack not found on disk: {pack_dir}")
    zip_path = str(dest_dir / f"{pack_id}.zip")
    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
        for p in pack_dir.rglob("*"):
            if p.is_file():
                zf.write(p, p.relative_to(pack_dir))
    return zip_path


def _read_api_key_from_config() -> str:
    """Read the auto-bootstrapped API key from ~/.agentpaas/config.json."""
    cfg_path = Path.home() / ".agentpaas" / "config.json"
    try:
        with open(cfg_path) as f:
            return (json.load(f) or {}).get("api_key", "")
    except Exception:
        return ""


# ── Main flow ─────────────────────────────────────────────────────────────────

def run_dogfood(base_url: str, api_key: str, verbose: bool = False) -> bool:
    try:
        import requests
    except ImportError:
        sys.exit("Missing dependency: pip install requests")

    session = requests.Session()
    session.headers.update({"Authorization": f"Bearer {api_key}"})

    repo_root = Path(__file__).parent.parent.resolve()
    ok = True

    print(f"\n{'─'*60}")
    print(f"  AgentPack Dogfood  —  {base_url}")
    print(f"{'─'*60}")

    # Step 0: health check
    print("\n[0] 服务健康检查")
    try:
        r = session.get(f"{base_url}/health", timeout=5)
        r.raise_for_status()
        _print("health", r.json().get("status", "?"))
    except Exception as e:
        _print("health", str(e), ok=False)
        print("\n  服务未响应。请先启动:\n  uvicorn agentpaas.api.app:app --port 8000\n")
        return False

    # Step 1: zip + install pack
    print(f"\n[1] 安装内置 Pack: {REVIEWER_PACK_ID}")
    with tempfile.TemporaryDirectory() as td:
        zip_path = _zip_pack(repo_root, REVIEWER_PACK_ID, Path(td))
        _print("zip", f"{Path(zip_path).stat().st_size // 1024} KB")

        # install is loopback-only — works when server is on localhost
        r = session.post(
            f"{base_url}/api/v1/agentpacks/install",
            json={"zip_path": zip_path},
            timeout=30,
        )
        if r.status_code == 200:
            pack = r.json()["installed"]
            _print("install", f"{pack['id']} v{pack['version']}")
        elif r.status_code == 400 and "already installed" in r.text.lower():
            _print("install", "已安装，跳过")
        else:
            _print("install", f"HTTP {r.status_code}: {r.text[:200]}", ok=False)
            ok = False

    # Step 2: verify pack is listed
    print(f"\n[2] 列出已安装 packs")
    r = session.get(f"{base_url}/api/v1/agentpacks", timeout=10)
    packs = r.json().get("agentpacks", [])
    found = any(p["id"] == REVIEWER_PACK_ID for p in packs)
    _print("list", f"{len(packs)} pack(s)，reviewer={'✓' if found else '✗'}", ok=found)
    if not found:
        ok = False

    # Step 3: create agent from pack
    print(f"\n[3] 从 Pack 创建智能体")
    r = session.post(
        f"{base_url}/api/v1/agentpacks/{REVIEWER_PACK_ID}/create-agent",
        json={"name": "Dogfood 审稿助手", "description": "Phase I dogfood"},
        timeout=15,
    )
    if r.status_code == 201:
        body = r.json()
        agent_id = body["agent_id"]
        _print("create-agent", f"id={agent_id}")
    else:
        _print("create-agent", f"HTTP {r.status_code}: {r.text[:200]}", ok=False)
        ok = False
        return ok

    # Step 4: verify agent in list
    print(f"\n[4] 验证智能体出现在 /agents")
    r = session.get(f"{base_url}/api/v1/agents", timeout=10)
    agent_ids = [a["id"] for a in r.json().get("agents", [])]
    found = agent_id in agent_ids
    _print("agents list", f"agent_id in list: {'✓' if found else '✗'}", ok=found)
    if not found:
        ok = False

    # Step 5: verify config._config_dir
    print(f"\n[5] 验证 agent config._config_dir")
    r = session.get(f"{base_url}/api/v1/agents/{agent_id}", timeout=10)
    agent = r.json()
    config_dir = agent.get("config", {}).get("_config_dir", "")
    agent_dir = agent.get("agent_dir", "")
    match = config_dir == agent_dir
    _print("_config_dir", f"{'matches agent_dir ✓' if match else f'MISMATCH: {config_dir!r} != {agent_dir!r}'}", ok=match)
    if not match:
        ok = False

    # Step 6: run agent
    print(f"\n[6] 运行审稿智能体（LLM 调用，可能需要 1-3 分钟）")
    t0 = time.time()
    r = session.post(
        f"{base_url}/api/v1/agents/{agent_id}/run",
        json={"input": TEST_ABSTRACT},
        timeout=300,
    )
    elapsed = int(time.time() - t0)

    if r.status_code != 200:
        _print("run", f"HTTP {r.status_code}: {r.text[:300]}", ok=False)
        ok = False
        return ok

    run_body = r.json()
    workspace = run_body.get("workspace_path", "")
    _print("run", f"完成 ({elapsed}s), status={run_body.get('status', '?')}")
    _print("workspace", workspace or "(未设置)")

    # Step 7: verify workspace files
    print(f"\n[7] 验证 workspace 产出物")
    if not workspace:
        _print("workspace_path", "run response 中未包含", ok=False)
        ok = False
    else:
        for fname in ("review_report.md", "review_result.json"):
            fpath = os.path.join(workspace, fname)
            exists = os.path.isfile(fpath)
            size = os.path.getsize(fpath) if exists else 0
            _print(fname, f"{'存在' if exists else '不存在'} ({size} bytes)", ok=exists)
            if not exists:
                ok = False

        if verbose and workspace and os.path.isfile(os.path.join(workspace, "review_report.md")):
            print("\n  ── review_report.md (前 20 行) ──")
            with open(os.path.join(workspace, "review_report.md"), encoding="utf-8") as f:
                lines = f.readlines()[:20]
            for line in lines:
                print("  ", line, end="")

    # Step 8: cleanup
    print(f"\n[8] 清理（删除 dogfood agent）")
    r = session.delete(f"{base_url}/api/v1/agents/{agent_id}", timeout=10)
    _print("delete agent", f"HTTP {r.status_code}")

    print(f"\n{'─'*60}")
    result_str = "PASS ✓" if ok else "FAIL ✗"
    color = "\033[32m" if ok else "\033[31m"
    print(f"  {color}Dogfood result: {result_str}\033[0m")
    print(f"{'─'*60}\n")
    return ok


# ── CLI ───────────────────────────────────────────────────────────────────────

def main() -> None:
    parser = argparse.ArgumentParser(description="AgentPack dogfood E2E script")
    parser.add_argument(
        "--url",
        default="http://127.0.0.1:8000",
        help="Base URL of the running agentpaas server (default: http://127.0.0.1:8000)",
    )
    parser.add_argument(
        "--key",
        default="",
        help="API key (Bearer). Reads from ~/.agentpaas/config.json if omitted.",
    )
    parser.add_argument(
        "--verbose", "-v",
        action="store_true",
        help="Print first 20 lines of review_report.md",
    )
    args = parser.parse_args()

    api_key = args.key or _read_api_key_from_config()
    if not api_key:
        sys.exit(
            "No API key found. Pass --key ap_xxx or run the server in desktop mode "
            "to auto-generate one at ~/.agentpaas/config.json"
        )

    success = run_dogfood(args.url, api_key, verbose=args.verbose)
    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()