Agent/examples/process_pipeline/run_pipeline.py

"""
V4 Pipeline: Hardcoded Map-Reduce Orchestration for AIGC Process Research
"""
import argparse
import asyncio
import json
import os
import sys
import time
from datetime import datetime
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent

# Force runtime working directory to project root so relative trace/cache paths
# always land in the repo root no matter where this script is launched from.
os.chdir(PROJECT_ROOT)

# Add project root to path
sys.path.insert(0, str(PROJECT_ROOT))

from dotenv import load_dotenv
load_dotenv()

from agent.llm.prompts import SimplePrompt
from agent.core.runner import AgentRunner, RunConfig
from agent.tools.builtin.knowledge import KnowledgeConfig
from agent.trace import FileSystemTraceStore, Trace, Message
from agent.llm import create_qwen_llm_call
from agent.llm.openrouter import create_openrouter_llm_call
from agent.llm.claude import create_claude_llm_call

# config from existing setup
from examples.process_research.config import (
    OUTPUT_DIR, TRACE_STORE_PATH, SKILLS_DIR, LOG_LEVEL, LOG_FILE,
    BROWSER_TYPE, HEADLESS, COORDINATOR_RUN_CONFIG
)
from agent.utils import setup_logging

async def run_agent_task(runner: AgentRunner, prompt_name: str, kwargs: dict, task_name: str, model_name: str, skip_validation: bool = False):
    from examples.process_pipeline.script.validate_schema import validate_case, validate_blueprint, validate_capabilities, validate_strategy
    base_dir = Path(__file__).parent
    prompt_path = base_dir / "prompts" / f"{prompt_name}.prompt"
    prompt = SimplePrompt(prompt_path)

    messages = prompt.build_messages(**kwargs)
    target_tools = []
    if prompt_name == "extract_capabilities":
        target_tools = ["capability_search", "capability_list", "tool_search"]

    # 按 agent 类型配置工具组权限
    tool_groups_map = {
        "researcher": ["core", "content"],           # 搜索+文件，无浏览器
        "filter_and_blueprint": ["core"],             # 只需文件读写
        "extract_capabilities": ["core"],             # 只需文件读写（额外工具由 target_tools 补充）
        "assemble_strategy": ["core"],                # 只需文件读写
    }

    total_task_cost = 0.0
    task_errors = []
    out_file = kwargs.get("output_file")

    max_retries = 3
    last_trace_id = None
    last_validation_error = None
    final_trace_id = None  # 用于返回最终成功的 trace_id

    for attempt in range(max_retries):
        if attempt > 0 and last_trace_id and last_validation_error:
            # 续跑模式：把错误信息告诉之前的 agent，让它修复
            print(f"🔄 [Continue {attempt}/{max_retries-1}] {task_name} - sending fix instructions to existing agent")

            fix_messages = [{
                "role": "user",
                "content": (
                    f"【系统校验失败】你上一次写入的文件 `{out_file}` 未通过 schema 校验。\n"
                    f"错误详情：{last_validation_error}\n\n"
                    f"请立刻读取该文件，根据以上错误信息修复内容，然后重新调用 write_json 写入到同一路径 `{out_file}`。"
                    f"只修复有问题的部分，不要丢弃已有的正确内容。"
                )
            }]

            fix_config = RunConfig(
                model=prompt.config.get("model") or model_name,
                temperature=0.1,
                name=f"{task_name}_Fix{attempt}",
                agent_type=prompt_name,
                tools=target_tools,
                tool_groups=tool_groups_map.get(prompt_name, ["core"]),
                trace_id=last_trace_id,
                knowledge=KnowledgeConfig(enable_completion_extraction=False, enable_extraction=False, enable_injection=False)
            )

            try:
                async for item in runner.run(messages=fix_messages, config=fix_config):
                    if isinstance(item, Trace):
                        last_trace_id = item.trace_id
                        if item.status == "completed":
                            total_task_cost += item.total_cost
                        elif item.status == "failed":
                            task_errors.append(f"{task_name} Fix Failed: {item.error_message}")
                    if isinstance(item, Message) and item.role == "tool":
                        content = item.content if isinstance(item.content, dict) else {}
                        if content.get("tool_name") in ("write_file", "write_json"):
                            print(f"   💾 [Fix File Written by {task_name}]")
            except Exception as e:
                err_msg = f"{type(e).__name__}: {e}"
                print(f"❌ [Exception Fix] {task_name} crashed: {err_msg}")
                task_errors.append(f"{task_name} fix crashed: {err_msg}")

        elif attempt > 0:
            # 没有 trace_id 或没有 validation error，只能完全重跑
            print(f"🔄 [Retry {attempt}/{max_retries-1}] {task_name} - no prior trace, full restart")
            if out_file and Path(out_file).exists():
                Path(out_file).unlink()

            run_config = RunConfig(
                model=prompt.config.get("model") or model_name,
                temperature=prompt.config.get("temperature") or 0.3,
                name=f"{task_name}_A{attempt}",
                agent_type=prompt_name,
                tools=target_tools,
                tool_groups=tool_groups_map.get(prompt_name, ["core"]),
                knowledge=KnowledgeConfig(enable_completion_extraction=False, enable_extraction=False, enable_injection=False)
            )

            print(f"🚀 [Launch] {task_name} (Attempt {attempt+1})")

            try:
                async for item in runner.run(messages=messages, config=run_config):
                    if isinstance(item, Trace):
                        last_trace_id = item.trace_id
                        if item.status == "completed":
                            total_task_cost += item.total_cost
                        elif item.status == "failed":
                            task_errors.append(f"{task_name} Failed: {item.error_message}")
                    if isinstance(item, Message):
                        if item.role == "tool":
                            content = item.content if isinstance(item.content, dict) else {}
                            t_name = content.get("tool_name", "unknown")
                            if t_name in ("write_file", "write_json"):
                                print(f"   💾 [File Written by {task_name}]")
            except Exception as e:
                err_msg = f"{type(e).__name__}: {e}"
                print(f"❌ [Exception] {task_name} crashed: {err_msg}")
                task_errors.append(f"{task_name} crashed: {err_msg}")

        else:
            # 首次执行
            run_config = RunConfig(
                model=prompt.config.get("model") or model_name,
                temperature=prompt.config.get("temperature") or 0.3,
                name=f"{task_name}_A{attempt}",
                agent_type=prompt_name,
                tools=target_tools,
                tool_groups=tool_groups_map.get(prompt_name, ["core"]),
                knowledge=KnowledgeConfig(enable_completion_extraction=False, enable_extraction=False, enable_injection=False)
            )

            print(f"🚀 [Launch] {task_name} (Attempt {attempt+1})")

            try:
                async for item in runner.run(messages=messages, config=run_config):
                    if isinstance(item, Trace):
                        last_trace_id = item.trace_id
                        if item.status == "completed":
                            total_task_cost += item.total_cost
                        elif item.status == "failed":
                            task_errors.append(f"{task_name} Failed: {item.error_message}")
                    if isinstance(item, Message):
                        if item.role == "tool":
                            content = item.content if isinstance(item.content, dict) else {}
                            t_name = content.get("tool_name", "unknown")
                            if t_name in ("write_file", "write_json"):
                                print(f"   💾 [File Written by {task_name}]")
            except Exception as e:
                err_msg = f"{type(e).__name__}: {e}"
                print(f"❌ [Exception] {task_name} crashed: {err_msg}")
                task_errors.append(f"{task_name} crashed: {err_msg}")

        # Verification & Recovery block
        if out_file and not Path(out_file).exists() and last_trace_id:
            print(f"⚠️ [Recovery] {task_name} missing output file. Triggering forced wrap-up continuation...")
            recovery_messages = [{
                "role": "user", 
                "content": f"【系统强制指令】你的任务阶段已终止，但尚未将结果写入文件。请立刻调用 write_json 工具，将你目前已经搜集或处理到的原生结构化内容直接作为 json_data 参数对象写入到绝对路径 `{out_file}`，如果搜集失败也请写入空的总结对象。必须立刻执行写入！"
            }]
            rec_config = RunConfig(
                model=model_name,
                temperature=0.1,
                name=task_name + "_Rec",
                agent_type=prompt_name,
                trace_id=last_trace_id,
                knowledge=KnowledgeConfig(enable_completion_extraction=False, enable_extraction=False, enable_injection=False)
            )
            try:
                async for r_item in runner.run(messages=recovery_messages, config=rec_config):
                    if isinstance(r_item, Trace):
                        if r_item.status == "completed":
                            total_task_cost += r_item.total_cost
                        elif r_item.status == "failed":
                            task_errors.append(f"{task_name} Recovery Failed: {r_item.error_message}")
                    if isinstance(r_item, Message) and r_item.role == "tool":
                        content = r_item.content if isinstance(r_item.content, dict) else {}
                        if content.get("tool_name") in ("write_file", "write_json"):
                            print(f"   💾 [Recovery File Written by {task_name}]")
            except Exception as e:
                err_msg = f"{type(e).__name__}: {e}"
                print(f"❌ [Exception Recovery] {task_name} crashed: {err_msg}")
                task_errors.append(f"{task_name} recovery crashed: {err_msg}")

        # Schema Validation
        if out_file and Path(out_file).exists() and str(out_file).endswith(".json"):
            if skip_validation:
                print(f"   ⏭️ [Schema Validation Skipped] {Path(out_file).name} (Single-Step Mode)")
                return total_task_cost, task_errors, last_trace_id
                
            try:
                with open(out_file, "r", encoding="utf-8") as f:
                    data = json.loads(f.read())

                filename = Path(out_file).name
                err = None
                if filename.startswith("case_"):
                    err = validate_case(data)
                elif filename == "blueprint.json":
                    err = validate_blueprint(data)
                elif filename == "capabilities_extracted.json":
                    err = validate_capabilities(data)
                elif filename == "strategy.json":
                    err = validate_strategy(data)

                if err:
                    raise ValueError(f"Schema Validation Failed: {err}")

                print(f"   ✅ [Schema Validated] {Path(out_file).name}")
                final_trace_id = last_trace_id
                return total_task_cost, task_errors, final_trace_id # Success! Exit retry loop.
            except Exception as e:
                err_msg = f"Invalid JSON or Schema in {Path(out_file).name}: {e}"
                print(f"❌ [Validation Error] {task_name}: {err_msg}")
                task_errors.append(f"{task_name} Error: {e}")
                last_validation_error = str(e)

                if attempt == max_retries - 1:
                    print(f"❌ [Retry Limit] {task_name} exhausted retries.")
                    return total_task_cost, task_errors, last_trace_id
        else:
            print(f"❌ [Missing File] {task_name} did not produce output file after recovery.")
            last_validation_error = None
            if attempt == max_retries - 1:
                return total_task_cost, task_errors, last_trace_id

    return total_task_cost, task_errors, last_trace_id

async def run_anthropic_sdk_task(prompt_name: str, kwargs: dict, task_name: str, model_name: str, skip_validation: bool = False):
    """
    备用：使用纯净的官方 Anthropic SDK 驱动。
    跳过内部大架构的 trace 追踪，但保留对原有 Python 工具库（如 write_file/glob_files）的无缝调用。
    """
    from anthropic import AsyncAnthropic
    from agent.tools.registry import get_tool_registry
    
    base_dir = Path(__file__).parent
    prompt_path = base_dir / "prompts" / f"{prompt_name}.prompt"
    prompt = SimplePrompt(prompt_path)
    
    # 1. 组装输入 Message
    raw_messages = prompt.build_messages(**kwargs)
    system_prompt = ""
    messages = []
    for msg in raw_messages:
        if msg["role"] == "system":
            system_prompt += msg["content"] + "\n\n"
        else:
            messages.append({"role": msg["role"], "content": msg["content"]})
            
    # 2. 映射目标工具
    target_tools = ["write_file", "write_json", "read_file", "glob_files"]
    if prompt_name == "extract_capabilities":
        target_tools.extend(["capability_search", "capability_list", "tool_search"])
        
    registry = get_tool_registry()
    schemas = registry.get_schemas(target_tools)
    anthropic_tools = []
    for s in schemas:
        anthropic_tools.append({
            "name": s["function"]["name"],
            "description": s["function"].get("description", ""),
            "input_schema": s["function"]["parameters"]
        })
        
    # 3. 初始化并开启 Loop
    # 提示：你需要在你的终端中配置好 ANTHROPIC_API_KEY 环境变量
    client = AsyncAnthropic()
    total_task_cost = 0.0
    task_errors = []
    sdk_trace_id = None  # 用于记录 Anthropic SDK 的 response ID
    
    from examples.process_pipeline.script.validate_schema import validate_case, validate_blueprint, validate_capabilities, validate_strategy
    out_file = kwargs.get("output_file")
    
    max_retries = 3
    for attempt in range(max_retries):
        if attempt > 0:
            print(f"🔄 [Retry SDK {attempt}/{max_retries-1}] {task_name}")
            if out_file and Path(out_file).exists():
                Path(out_file).unlink()
                
        print(f"🚀 [Launch Anthropic SDK] {task_name} (Attempt {attempt+1})")
        
        # Reset messages for retry
        messages_copy = list(messages)
        
        max_loops = 50
        for loop_idx in range(max_loops):
            try:
                # 去除前缀（兼容比如 openrouter 传入的名字）
                clean_model = model_name.split("/")[-1] if "/" in model_name else model_name
                
                # 这里专门将实际请求映射到其可用的特殊别名 claude-sonnet-4-5 
                target_model = "claude-sonnet-4-5" if "claude" in clean_model else clean_model

                response = await client.messages.create(
                    model=target_model,
                    max_tokens=4096,
                    temperature=0.2,
                    system=system_prompt,
                    messages=messages_copy,
                    tools=anthropic_tools
                )
                
                # (简略预估，不代表真实官方开销)
                if hasattr(response, 'usage'):
                    step_cost = (response.usage.input_tokens / 1e6 * 3.0) + (response.usage.output_tokens / 1e6 * 15.0)
                    total_task_cost += step_cost
                
                # 加入助手回复
                assistant_content = []
                tool_uses = []
                
                for content_block in response.content:
                    if content_block.type == "text":
                        text_val = content_block.text
                        if text_val:
                            assistant_content.append({"type": "text", "text": text_val})
                            print(f"\n🤖 [{task_name} Output]:\n{text_val}\n")
                    elif content_block.type == "tool_use":
                        assistant_content.append({
                            "type": "tool_use",
                            "id": content_block.id,
                            "name": content_block.name,
                            "input": content_block.input
                        })
                        tool_uses.append(content_block)
                
                if not assistant_content:
                    assistant_content.append({"type": "text", "text": "(Thinking completed but no output)"})
                    
                messages_copy.append({"role": "assistant", "content": assistant_content})
                
                # 出口：没有调用工具说明任务结束
                if not tool_uses:
                    print(f"✅ [Done Anthropic SDK] {task_name} (Cost: ${total_task_cost:.4f})")
                    break
                    
                # 工具执行与回传
                tool_results = []
                for tu in tool_uses:
                    if tu.name in ("write_file", "write_json"):
                        print(f"   💾 [File Written by SDK] {task_name}")
                    
                    print(f"   🛠️ [Tool Exec Debug] name_is={tu.name}, input_is={tu.input}, type_is={type(tu.input)}")
                    # 执行本地环境的函数
                    result_str = await registry.execute(tu.name, tu.input)
                    
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": tu.id,
                        "content": result_str
                    })
                    
                messages_copy.append({"role": "user", "content": tool_results})
                
            except Exception as e:
                err_msg = str(e)
                print(f"❌ [Fail SDK Core] {task_name}: {err_msg}")
                task_errors.append(err_msg)
                break
                
        # Verification & Recovery block for SDK (porting from AgentRunner)
        if out_file and not Path(out_file).exists():
            print(f"⚠️ [Recovery SDK] {task_name} missing output file. Triggering forced wrap-up continuation...")
            messages_copy.append({
                "role": "user", 
                "content": f"【系统强制指令】你的任务阶段已完成分析，但尚未将最终结果写入目标文件。请立刻调用 write_json (或 write_file) 工具，将你的成果数据直接写入到绝对路径 `{out_file}`，务必立刻执行写入动作！"
            })
            try:
                target_model = "claude-sonnet-4-5" if "claude" in clean_model else clean_model
                rec_response = await client.messages.create(
                    model=target_model,
                    max_tokens=4096,
                    temperature=0.1,
                    system=system_prompt,
                    messages=messages_copy,
                    tools=anthropic_tools,
                    tool_choice={"type": "any"}
                )
                for content_block in rec_response.content:
                    if content_block.type == "tool_use":
                        if content_block.name in ("write_file", "write_json"):
                            print(f"   💾 [Recovery File Written by SDK] {task_name}")
                        await registry.execute(content_block.name, content_block.input)
            except Exception as e:
                print(f"❌ [Fail SDK Recovery] {task_name}: {e}")
                task_errors.append(str(e))
                
        # Schema Validation
        if out_file and Path(out_file).exists() and str(out_file).endswith(".json"):
            if skip_validation:
                print(f"   ⏭️ [Schema Validation Skipped] {Path(out_file).name} (Single-Step Mode)")
                return total_task_cost, task_errors, None
                
            try:
                with open(out_file, "r", encoding="utf-8") as f:
                    data = json.loads(f.read())
                    
                filename = Path(out_file).name
                err = None
                if filename.startswith("case_"):
                    err = validate_case(data)
                elif filename == "blueprint.json":
                    err = validate_blueprint(data)
                elif filename == "capabilities_extracted.json":
                    err = validate_capabilities(data)
                elif filename == "strategy.json":
                    err = validate_strategy(data)
                    
                if err:
                    raise ValueError(f"Schema Validation Failed: {err}")
                    
                print(f"   ✅ [Schema Validated] {Path(out_file).name}")
                return total_task_cost, task_errors # Success! Exit retry loop.
            except Exception as e:
                err_msg = f"Invalid JSON or Schema in {Path(out_file).name}: {e}"
                print(f"❌ [Validation Error] {task_name}: {err_msg}")
                task_errors.append(f"{task_name} Error: {e}")
                
                if attempt == max_retries - 1:
                    print(f"❌ [Retry Limit] {task_name} exhausted retries.")
                    return total_task_cost, task_errors
        else:
            print(f"❌ [Missing File] {task_name} did not produce output file after recovery.")
            if attempt == max_retries - 1:
                return total_task_cost, task_errors
                
    return total_task_cost, task_errors

async def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--index", type=int, required=True, help="Index of requirement in db_requirements.json")
    parser.add_argument("--skip-research", action="store_true", help="Skip Phase 1 and use existing raw cases")
    parser.add_argument("--research-only", action="store_true", help="Only run research phases, skip Phase 2 and 3")
    parser.add_argument("--platforms", type=str, default="xhs,youtube,bili,x", help="Comma-separated list of platforms to search")
    parser.add_argument("--use-claude-sdk", action="store_true", help="Use pure Anthropic SDK (run_anthropic_sdk_task) instead of internal AgentRunner for Phase 2/3")
    parser.add_argument("--restart-mode", type=str, default="smart", help="Granular restart mode for cascading deletions")
    args = parser.parse_args()

    base_dir = Path(__file__).parent
    # Load requirements locally
    req_path = base_dir / "db_requirements.json"
    
    with open(req_path, encoding='utf-8') as f:
        reqs = json.load(f)
    
    if args.index < 0 or args.index >= len(reqs):
        print("Index out of bounds")
        sys.exit(1)
        
    requirement = reqs[args.index]
    
    # 0. Setup directories
    output_dir = base_dir / "output" / f"{(args.index+1):03d}"
    output_dir.mkdir(parents=True, exist_ok=True)
    raw_cases_dir = output_dir / "raw_cases"
    raw_cases_dir.mkdir(parents=True, exist_ok=True)
    
    setup_logging(level=LOG_LEVEL, file=LOG_FILE)
    
    print("=" * 60)
    print(f"V4 Hardcoded Pipeline | Demand: [{args.index+1:03d}] {requirement[:40]}...")
    print("=" * 60)

    # Load presets
    presets_path = base_dir / "presets.json"
    if presets_path.exists():
        from agent.core.presets import load_presets_from_json
        load_presets_from_json(str(presets_path))
        print("✅ Configured Agent Presets (Skills Boundaries)")

    # Browser initialization removed to save resources
    
    store = FileSystemTraceStore(base_path=TRACE_STORE_PATH)

    # Instantiate two distinct LLM orchestrators
    qwen_model = "qwen3.5-plus"  # maps to qwen3.5-plus via Qwen interface
    
    # 用户指示使用 OpenRouter 代理的 Claude 4.5
    claude_model = "anthropic/claude-4.5-sonnet"
    args.use_claude_sdk = False  # 禁用纯 Native SDK 模式，走内部通用 AgentRunner (即可对接 OpenRouter)
    from agent.llm.openrouter import create_openrouter_llm_call
    claude_llm_call = create_openrouter_llm_call(model=claude_model)
    
    runner_qwen = AgentRunner(
        trace_store=store,
        llm_call=create_qwen_llm_call(model=qwen_model),
        skills_dir=SKILLS_DIR
    )
    
    runner_claude = AgentRunner(
        trace_store=store,
        llm_call=claude_llm_call,
        skills_dir=SKILLS_DIR
    )

    try:
        start_time = time.time()
        total_cost = 0.0
        costs_breakdown = {}
        global_errors = []
        strategy_file = None
        
        existing_platforms = []
        if raw_cases_dir.exists():
            for f in raw_cases_dir.glob("case_*.json"):
                plat = f.stem.replace("case_", "")
                if plat in ["xhs", "youtube", "bili", "x", "zhihu", "gzh"]:
                    existing_platforms.append(plat)
                    
        platforms = [p.strip() for p in args.platforms.split(",") if p.strip()]
        needed_count = max(0, 4 - len(existing_platforms))
        
        # Phase 0: Dynamic Routing
        if not args.skip_research:
            if needed_count == 0:
                print(f"\n--- Phase 0: Skipping Routing (Already have {len(existing_platforms)} existing cases: {existing_platforms}) ---")
                platforms = []
            else:
                print(f"\n--- Phase 0: Dynamic Platform Routing ({qwen_model}) ---")
                print(f"📡 Found existing cases: {existing_platforms}. Requesting {needed_count} new platforms...")
                try:
                    router_prompt = SimplePrompt(base_dir / "prompts" / "router.prompt")
                    rmessages = router_prompt.build_messages(
                        requirement=requirement,
                        existing_platforms=",".join(existing_platforms) if existing_platforms else "无",
                        needed_count=str(needed_count)
                    )
                    rconfig = RunConfig(
                        model=qwen_model,
                        temperature=0.1,
                        name="P0_Router",
                        agent_type="router",
                        knowledge=KnowledgeConfig(enable_completion_extraction=False, enable_extraction=False, enable_injection=False)
                    )
                    print(f"🚀 [Launch] P0_Router calculating optimal platforms...")
                    
                    router_response = ""
                    async for item in runner_qwen.run(messages=rmessages, config=rconfig):
                        if isinstance(item, Message) and item.role == "assistant" and isinstance(item.content, dict):
                            text = item.content.get("text", "")
                            if text and not item.content.get("tool_calls"):
                                router_response = text
                        if isinstance(item, Trace) and item.status == "completed":
                            total_cost += item.total_cost
                            costs_breakdown["P0_Router"] = round(item.total_cost, 4)
                            
                    if router_response:
                        import re
                        # Extract all alphabetic words from the response to handle extra text or markdown
                        words = set(re.findall(r'\b[a-z]+\b', router_response.lower()))
                        valid_platforms = ["xhs", "youtube", "bili", "x", "zhihu", "gzh"]
                        # Intersect words with valid platforms (direct exact word matching, exclude existing)
                        final_platforms = [p for p in valid_platforms if p in words and p not in existing_platforms]
                        
                        if final_platforms:
                            platforms = final_platforms[:needed_count]
                            print(f"🎯 [Router Decision] Selected {len(platforms)} new platforms: {platforms}")
                        else:
                            platforms = [p for p in platforms if p not in existing_platforms][:needed_count]
                            print(f"⚠️ [Router Fallback] Invalid output '{router_response}'. Using delta default: {platforms}")
                except Exception as e:
                    platforms = [p for p in platforms if p not in existing_platforms][:needed_count]
                    print(f"⚠️ [Router Logic Failed] Using delta default platforms ({platforms}). Error: {e}")

        is_single_step = args.restart_mode.startswith("single_")
        
        # Phase 1: MAP (Parallel Search) uses Qwen
        if not args.skip_research:
            print(f"\n--- Phase 1: Distributed Research Map ({qwen_model}) ---")
            phase1_tasks = []
            for p in platforms:
                task_desc = f"渠道：{p.upper()}。核心需求：{requirement}"
                out_file = str(raw_cases_dir / f"case_{p}.json")
                kwargs = {
                    "task": task_desc,
                    "output_file": out_file
                }
                phase1_tasks.append(run_agent_task(runner_qwen, "researcher", kwargs, f"P1_Research_{p}", qwen_model, skip_validation=is_single_step))
                
            phase1_results = await asyncio.gather(*phase1_tasks)
            phase1_trace_ids = {}
            for (task_cost, task_errors, trace_id), p in zip(phase1_results, platforms):
                total_cost += task_cost
                costs_breakdown[f"P1_Research_{p}"] = round(task_cost, 4)
                phase1_trace_ids[f"P1_Research_{p}"] = trace_id
                global_errors.extend(task_errors)

                # Check if cases actually got written
                expected_file = Path(raw_cases_dir / f"case_{p}.json")
                if not expected_file.exists():
                    err_msg = f"Missing case file for {p}! Agent likely hit max_iterations without saving."
                    print(f"⚠️ [Warning] {err_msg}")
                    global_errors.append(err_msg)

            # Phase 1.5: Extract raw post data from cache → raw_cases/source.json
            try:
                from examples.process_pipeline.script.extract_sources import extract_sources_to_json
                # 使用本次 pipeline 的 trace_ids 精确定位 cache 文件
                trace_id_list = [tid for tid in phase1_trace_ids.values() if tid]
                src_stats = extract_sources_to_json(raw_cases_dir, trace_ids=trace_id_list)
                print(
                    f"📎 [Source Extraction] "
                    f"existing={src_stats['total_existing']} "
                    f"new={src_stats['total_matched']} "
                    f"total={src_stats['total_existing'] + src_stats['total_matched']} "
                    f"→ {raw_cases_dir / 'source.json'}"
                )
            except Exception as e:
                err_msg = f"Source extraction failed: {type(e).__name__}: {e}"
                print(f"⚠️ [Warning] {err_msg}")
                global_errors.append(err_msg)

            # Phase 1.6: Process sources with Claude → case_detailed.json
            source_file = raw_cases_dir / "source.json"
            if source_file.exists():
                try:
                    from examples.process_pipeline.script.process_sources import process_sources
                    print(f"\n--- Phase 1.6: Workflow Extraction ({claude_model}) ---")

                    detailed_file = raw_cases_dir / "case_detailed.json"
                    workflow_stats = await process_sources(
                        source_file,
                        detailed_file,
                        claude_llm_call,
                        model=claude_model,
                        max_concurrent=3
                    )

                    total_cost += workflow_stats.get("total_cost", 0.0)
                    costs_breakdown["P1.6_WorkflowExtraction"] = round(workflow_stats.get("total_cost", 0.0), 4)

                    print(
                        f"🔍 [Workflow Extraction] "
                        f"success={workflow_stats['success']} "
                        f"failed={workflow_stats['failed']} "
                        f"→ {detailed_file}"
                    )
                except Exception as e:
                    err_msg = f"Workflow extraction failed: {type(e).__name__}: {e}"
                    print(f"⚠️ [Warning] {err_msg}")
                    global_errors.append(err_msg)
        else:
            print("\n⏭️  [Skip] Phase 1 Skipped via --skip-research. Using existing cases...")

        # Phase 2: REDUCE 1 (Parallel Distillation) uses Claude
        if not args.research_only:
            # Phase 2: REDUCE 1 (Parallel Distillation) uses Claude
            print(f"\n--- Phase 2: Parallel Distillation ({claude_model}) ---")
            blueprint_file = str(output_dir / "blueprint.json")
            capabilities_file = str(output_dir / "capabilities_extracted.json")
            raw_glob = str(raw_cases_dir / "case_*.json").replace("\\", "/")
            
            task_a = None
            task_b = None
            force_strategy_rerun = False
            
            if Path(blueprint_file).exists():
                print(f"⏭️  [Skip P2] blueprint.json already exists. Skipping P2_FilterBlueprint.")
            else:
                force_strategy_rerun = True
                if args.use_claude_sdk:
                    print("   > Using [Anthropic SDK Core] for P2_FilterBlueprint")
                    task_a = run_anthropic_sdk_task("filter_and_blueprint", {
                        "requirement": requirement,
                        "raw_files_glob": raw_glob,
                        "output_file": blueprint_file
                    }, "P2_FilterBlueprint", claude_model, skip_validation=is_single_step)
                else:
                    print("   > Using [AgentRunner Core] for P2_FilterBlueprint")
                    task_a = run_agent_task(runner_claude, "filter_and_blueprint", {
                        "requirement": requirement,
                        "raw_files_glob": raw_glob,
                        "output_file": blueprint_file
                    }, "P2_FilterBlueprint", claude_model, skip_validation=is_single_step)
                    
            if Path(capabilities_file).exists():
                print(f"⏭️  [Skip P2] capabilities_extracted.json already exists. Skipping P2_ExtractCaps.")
            else:
                force_strategy_rerun = True
                if args.use_claude_sdk:
                    print("   > Using [Anthropic SDK Core] for P2_ExtractCaps")
                    task_b = run_anthropic_sdk_task("extract_capabilities", {
                        "requirement": requirement,
                        "raw_files_glob": raw_glob,
                        "output_file": capabilities_file
                    }, "P2_ExtractCaps", claude_model, skip_validation=is_single_step)
                else:
                    print("   > Using [AgentRunner Core] for P2_ExtractCaps")
                    task_b = run_agent_task(runner_claude, "extract_capabilities", {
                        "requirement": requirement,
                        "raw_files_glob": raw_glob,
                        "output_file": capabilities_file
                    }, "P2_ExtractCaps", claude_model, skip_validation=is_single_step)
            
            to_await = []
            names_await = []
            if task_a:
                to_await.append(task_a)
                names_await.append("P2_FilterBlueprint")
            if task_b:
                to_await.append(task_b)
                names_await.append("P2_ExtractCaps")
                
            if to_await:
                phase2_results = await asyncio.gather(*to_await)
                phase2_trace_ids = {}
                for (cost, errs, trace_id), t_name in zip(phase2_results, names_await):
                    total_cost += cost
                    costs_breakdown[t_name] = round(cost, 4)
                    phase2_trace_ids[t_name] = trace_id
                    global_errors.extend(errs)

            # Phase 3: REDUCE 2 (Final Assembly) uses Claude
            print(f"\n--- Phase 3: Final Strategy Assembly ({claude_model}) ---")
            strategy_file = str(output_dir / "strategy.json")
            
            if args.restart_mode == "single":
                force_strategy_rerun = False
            
            if Path(strategy_file).exists() and not force_strategy_rerun:
                print(f"⏭️  [Skip P3] strategy.json already exists. Skipping P3_Assembler.")
            else:
                if Path(strategy_file).exists():
                    print(f"⚠️  [Force P3] Upstream dependencies were regenerated. Forcing re-run of P3_Assembler...")
                if args.use_claude_sdk:
                    print("   > Using [Anthropic SDK Core]")
                    phase3_cost, phase3_errs, phase3_trace_id = await run_anthropic_sdk_task("assemble_strategy", {
                        "requirement": requirement,
                        "blueprint_file": blueprint_file,
                        "capabilities_file": capabilities_file,
                        "output_file": strategy_file
                    }, "P3_Assembler", claude_model, skip_validation=is_single_step)
                else:
                    print("   > Using [AgentRunner Core]")
                    phase3_cost, phase3_errs, phase3_trace_id = await run_agent_task(runner_claude, "assemble_strategy", {
                        "requirement": requirement,
                        "blueprint_file": blueprint_file,
                        "capabilities_file": capabilities_file,
                        "output_file": strategy_file
                    }, "P3_Assembler", claude_model, skip_validation=is_single_step)
                total_cost += phase3_cost
                costs_breakdown["P3_Assembler"] = round(phase3_cost, 4)
                global_errors.extend(phase3_errs)
        else:
            print("\n--- [Research Only] Stopping early. Skipping Phase 2 and Phase 3 ---")
        
        end_time = time.time()
        elapsed_sec = end_time - start_time
        
        # Save Metrics
        metrics_file = base_dir / "run_metrics.json"
        metrics_data = []
        if metrics_file.exists():
            with open(metrics_file, "r", encoding="utf-8") as f:
                try:
                    metrics_data = json.load(f)
                except json.JSONDecodeError:
                    pass
                    
        # Collect trace_ids from all phases
        trace_ids = {}
        if not args.skip_research and 'phase1_trace_ids' in dir():
            trace_ids.update(phase1_trace_ids)
        if not args.research_only:
            if 'phase2_trace_ids' in dir():
                trace_ids.update(phase2_trace_ids)
            if 'phase3_trace_id' in dir() and phase3_trace_id:
                trace_ids["P3_Assembler"] = phase3_trace_id

        metrics_data.append({
            "index": args.index,
            "requirement": requirement[:80] + "...",
            "duration_seconds": round(elapsed_sec, 2),
            "total_cost_usd": round(total_cost, 4),
            "costs_breakdown": costs_breakdown,
            "trace_ids": trace_ids,
            "errors": global_errors,
            "timestamp": datetime.now().isoformat()
        })
        
        with open(metrics_file, "w", encoding="utf-8") as f:
            json.dump(metrics_data, f, indent=2, ensure_ascii=False)
            
        print(f"\n📊 [Metrics] Pipeline completed in {elapsed_sec:.1f}s. Total Cost: ${total_cost:.4f}")
        
    finally:
        pass
        
    print("✅ Pipeline run finished.")
    if strategy_file:
        print("✅ Strategy saved to:", strategy_file)


if __name__ == "__main__":
    asyncio.run(main())