OpenSpace/benchmarks/gdpval/run_benchmark.py

#!/usr/bin/env python3
"""
GDPVal Benchmark — Two-phase experiment for OpenSpace skill-driven token savings.

Design:
  Phase 1 (Cold → Warm):  Run 220 tasks, skills accumulate across tasks.
  Phase 2 (Full Warm):    Run the same 220 tasks again with all Phase 1 skills.

For each task, records:
  - Token usage (prompt, completion, total, cost)
  - Execution metrics (iterations, tool calls, time, status)
  - Skills state (count before/after, which skills were used)

Produces:
  results/<run_name>/
    ├── phase1_results.jsonl     # Per-task results from Phase 1
    ├── phase2_results.jsonl     # Per-task results from Phase 2
    ├── skills_snapshot.json     # All skills after Phase 1
    ├── comparison.jsonl         # Per-task Phase 1 vs Phase 2 comparison
    ├── summary.json             # Aggregate statistics
    └── recordings/              # OpenSpace recordings per task
        ├── phase1/
        └── phase2/

Usage:
    # Full run (220 tasks × 2 phases)
    python -m benchmarks.gdpval.run_benchmark

    # Quick test (5 tasks)
    python -m benchmarks.gdpval.run_benchmark --max-tasks 5

    # Concurrent (3 workers per phase)
    python -m benchmarks.gdpval.run_benchmark --concurrency 3

    # Resume from where you left off
    python -m benchmarks.gdpval.run_benchmark --resume

    # Phase 2 only (requires Phase 1 completed)
    python -m benchmarks.gdpval.run_benchmark --phase2-only

    # Custom config
    python -m benchmarks.gdpval.run_benchmark --config benchmarks/gdpval/config.json

Fair comparison with ClawWork:
    Set "use_clawwork_productivity": true in config (or pass --use-clawwork-productivity when added).
    This adds the same productivity tools as ClawWork (web_search, web_fetch, create_file, read,
    execute_code_sandbox, create_video) so benchmark results are comparable. Requires livebench (ClawWork)
    to be installed in the same environment.
"""

from __future__ import annotations

import argparse
import asyncio
import importlib
import importlib.util
import json
import os
import shutil
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional

# ── Ensure OpenSpace is importable ──
_OPENSPACE_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(_OPENSPACE_ROOT))

# ── Load .env (same logic as openspace/llm/client.py) ──
try:
    from dotenv import load_dotenv
    _pkg_env = _OPENSPACE_ROOT / "openspace" / ".env"
    if _pkg_env.is_file():
        load_dotenv(_pkg_env)
    load_dotenv()  # also try CWD/.env
except ImportError:
    pass  # dotenv not installed; rely on shell env vars

from .token_tracker import TokenTracker, TokenStats
from .task_loader import load_tasks, prepare_task_workspace

# ── Default paths ──
_DEFAULT_CONFIG = Path(__file__).parent / "config.json"
_DEFAULT_RESULTS = Path(__file__).parent / "results"
_OPENSPACE_DB_DIR = _OPENSPACE_ROOT / ".openspace"

# ── Evaluation constants (aligned with ClawWork) ──
# Artifact extensions that ClawWork considers for evaluation
_ARTIFACT_EXTENSIONS = {
    '.pdf', '.docx', '.xlsx', '.pptx',      # documents
    '.txt', '.csv', '.json', '.md',          # text
    '.py', '.js', '.html', '.css',           # code
    '.png', '.jpg', '.jpeg', '.gif', '.webp', # images
}
# Minimum evaluation score to receive payment (ClawWork cliff)
_MIN_EVALUATION_THRESHOLD = 0.6


# ═══════════════════════════════════════════════════════════════════
# Configuration
# ═══════════════════════════════════════════════════════════════════

def load_config(config_path: Optional[str] = None) -> Dict[str, Any]:
    """Load experiment config with sensible defaults."""
    defaults = {
        "clawwork_root": str(_OPENSPACE_ROOT.parent / "ClawWork"),
        "gdpval_path": None,
        "model": "openrouter/qwen/qwen3.5-plus-02-15",
        "max_iterations": 20,
        "backend_scope": ["shell"],
        "use_clawwork_productivity": False,
        "run_name": datetime.now().strftime("gdpval_%Y%m%d_%H%M%S"),
        "max_tasks": None,
        "per_occupation": None,
        "sectors": None,
        "occupations": None,
        "task_ids": None,
        "record_call_details": True,
    }

    if config_path and Path(config_path).exists():
        with open(config_path, "r") as f:
            overrides = json.load(f)
        defaults.update(overrides)

    return defaults


# ═══════════════════════════════════════════════════════════════════
# Result I/O
# ═══════════════════════════════════════════════════════════════════

def _results_dir(cfg: Dict) -> Path:
    name = cfg.get("run_name")
    if not name:
        model = cfg.get("model", "unknown")
        # e.g. "openrouter/qwen/qwen3.5-plus-02-15" → "qwen3.5-plus-02-15"
        short_model = model.rsplit("/", 1)[-1] if "/" in model else model
        name = f"{short_model}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        cfg["run_name"] = name
    return _DEFAULT_RESULTS / name


def _append_jsonl(path: Path, record: Dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False, default=str) + "\n")


def _load_jsonl(path: Path) -> List[Dict]:
    if not path.exists():
        return []
    records = []
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                records.append(json.loads(line))
    return records


def _completed_task_ids(results_file: Path) -> set:
    """Return set of task_ids already completed (for resume)."""
    return {r["task_id"] for r in _load_jsonl(results_file) if r.get("status")}


# ═══════════════════════════════════════════════════════════════════
# Skill snapshot
# ═══════════════════════════════════════════════════════════════════

def _snapshot_skills(skill_store) -> List[Dict[str, Any]]:
    """Dump all active skills from the SkillStore."""
    try:
        records = skill_store.load_active()  # Returns Dict[str, SkillRecord]
        return [r.to_dict() for r in records.values()]
    except Exception as e:
        print(f"⚠️  Could not snapshot skills: {e}")
        return []


def _count_skills_by_origin(skills: List[Dict]) -> Dict[str, int]:
    counts: Dict[str, int] = {}
    for s in skills:
        origin = s.get("lineage", {}).get("origin", "unknown")
        counts[origin] = counts.get(origin, 0) + 1
    return counts


# ═══════════════════════════════════════════════════════════════════
# Evaluation — aligned with ClawWork's LLMEvaluator
# ═══════════════════════════════════════════════════════════════════

def _discover_artifacts(
    workspace_dir: str,
    reference_filenames: List[str],
) -> List[str]:
    """Discover agent-created artifacts in the workspace.

    Scans ``workspace_dir`` for files matching ``_ARTIFACT_EXTENSIONS`` that
    are NOT the downloaded reference files (to avoid evaluating input files).

    This mirrors how ClawWork's ``list_artifacts`` discovers work products.

    Returns:
        Sorted list of absolute paths to artifact files.
    """
    ws = Path(workspace_dir)
    if not ws.exists():
        return []

    ref_names = set(reference_filenames)
    artifacts: List[str] = []

    for f in ws.rglob("*"):
        if not f.is_file():
            continue
        if f.suffix.lower() not in _ARTIFACT_EXTENSIONS:
            continue
        if f.name in ref_names:
            continue
        # Skip zero-byte files
        if f.stat().st_size == 0:
            continue
        artifacts.append(str(f))

    return sorted(artifacts)


def _get_evaluator(cfg: Dict):
    """Lazily create and cache a ClawWork-compatible LLMEvaluator.

    Uses the same initialization logic as ClawWork:
    - Model: gpt-4o (or ``EVALUATION_MODEL`` env override)
    - API key: ``EVALUATION_API_KEY`` > ``OPENAI_API_KEY``
    - API base: ``EVALUATION_API_BASE`` > ``OPENAI_API_BASE``
    - Meta-prompts: ``ClawWork/eval/meta_prompts/``

    The evaluator is created once and reused for all tasks.
    """
    if hasattr(_get_evaluator, "_instance"):
        return _get_evaluator._instance

    clawwork_root = Path(cfg.get("clawwork_root", ""))
    meta_prompts_dir = clawwork_root / "eval" / "meta_prompts"

    if not meta_prompts_dir.exists():
        print(f"  ⚠️  Meta-prompts dir not found: {meta_prompts_dir}")
        print(f"       Evaluation will be skipped.")
        _get_evaluator._instance = None
        return None

    # Import from ClawWork
    clawwork_livebench = clawwork_root / "livebench"
    if str(clawwork_livebench.parent) not in sys.path:
        sys.path.insert(0, str(clawwork_livebench.parent))

    try:
        from livebench.work.llm_evaluator import LLMEvaluator

        evaluator = LLMEvaluator(
            meta_prompts_dir=str(meta_prompts_dir),
            max_payment=50.0,  # default; will be overridden per-task
        )
        _get_evaluator._instance = evaluator
        return evaluator
    except Exception as e:
        print(f"  ⚠️  Could not initialize LLMEvaluator: {e}")
        _get_evaluator._instance = None
        return None


def _evaluate_task(
    task: Dict,
    workspace_dir: str,
    cfg: Dict,
) -> Dict[str, Any]:
    """Evaluate agent artifacts using ClawWork's LLMEvaluator.

    Fully aligned with ClawWork's evaluation pipeline:
      1. Discover artifacts in workspace (same extensions as ClawWork)
      2. Call ``LLMEvaluator.evaluate_artifact`` (same model, system prompt,
         meta-prompt rubric, and evaluation prompt template)
      3. Apply 0.6 payment cliff (same as ``EconomicTracker.add_work_income``)
      4. Return structured result matching ClawWork's output format

    Returns dict with keys:
        evaluation_score, score_10, payment, actual_payment,
        artifact_paths, description, feedback (truncated), has_evaluation
    """
    # ── Discover artifacts ──
    ref_filenames = [
        Path(rf).name for rf in (task.get("reference_files", []) or [])
    ]
    artifact_paths = _discover_artifacts(workspace_dir, ref_filenames)

    if not artifact_paths:
        return {
            "has_evaluation": False,
            "evaluation_score": 0.0,
            "score_10": 0,
            "payment": 0.0,
            "actual_payment": 0.0,
            "artifact_count": 0,
            "artifact_paths": [],
            "description": "",
            "feedback": "No artifacts found in workspace.",
        }

    evaluator = _get_evaluator(cfg)
    if evaluator is None:
        return {
            "has_evaluation": False,
            "evaluation_score": 0.0,
            "score_10": 0,
            "payment": 0.0,
            "actual_payment": 0.0,
            "artifact_count": len(artifact_paths),
            "artifact_paths": [os.path.basename(p) for p in artifact_paths],
            "description": "",
            "feedback": "Evaluator not available (missing meta-prompts or API key).",
        }

    # ── Build description (same format as ClawWork's submit_work) ──
    description = f"Work submission with {len(artifact_paths)} artifact(s)"

    # ── Task-specific max payment ──
    max_payment = task.get("task_value_usd", 0.0) or 50.0

    try:
        evaluation_score, feedback, payment = evaluator.evaluate_artifact(
            task=task,
            artifact_paths=artifact_paths,
            description=description,
            max_payment=max_payment,
        )
    except Exception as e:
        err_msg = str(e)
        print(f"  ⚠️  Evaluation failed: {e}")
        if "langchain" in err_msg.lower():
            print(f"     Fix: pip install langchain_core  (livebench evaluator dependency)")
        return {
            "has_evaluation": False,
            "evaluation_score": 0.0,
            "score_10": 0,
            "payment": 0.0,
            "actual_payment": 0.0,
            "artifact_count": len(artifact_paths),
            "artifact_paths": [os.path.basename(p) for p in artifact_paths],
            "description": description,
            "feedback": f"Evaluation error: {e}",
        }

    # ── Apply 0.6 cliff (same as ClawWork EconomicTracker) ──
    if evaluation_score < _MIN_EVALUATION_THRESHOLD:
        actual_payment = 0.0
    else:
        actual_payment = payment

    # Truncate feedback for JSONL storage (keep first 500 chars)
    feedback_short = feedback[:500] + "..." if len(feedback) > 500 else feedback

    return {
        "has_evaluation": True,
        "evaluation_score": round(evaluation_score, 4),
        "score_10": round(evaluation_score * 10, 1),
        "payment": round(payment, 2),
        "actual_payment": round(actual_payment, 2),
        "max_payment": round(max_payment, 2),
        "artifact_count": len(artifact_paths),
        "artifact_paths": [os.path.basename(p) for p in artifact_paths],
        "description": description,
        "feedback": feedback_short,
        "cliff_applied": evaluation_score < _MIN_EVALUATION_THRESHOLD,
    }


# ═══════════════════════════════════════════════════════════════════
# Helper: create OpenSpaceConfig
# ═══════════════════════════════════════════════════════════════════

def _make_config(cfg: Dict, phase: str, worker_id: int = 0):
    """Create a OpenSpaceConfig for one worker."""
    from openspace import OpenSpaceConfig

    rd = _results_dir(cfg)
    # Each worker gets its own recording dir to avoid collisions
    rec_suffix = f"w{worker_id}" if worker_id > 0 else ""
    rec_dir = str(rd / "recordings" / phase / rec_suffix) if rec_suffix else str(rd / "recordings" / phase)

    return OpenSpaceConfig(
        llm_model=cfg["model"],
        workspace_dir=str(rd / "workspace" / phase),
        recording_log_dir=rec_dir,
        recording_backends=cfg.get("backend_scope", ["shell", "web"]),
        backend_scope=cfg.get("backend_scope"),
        use_clawwork_productivity=cfg.get("use_clawwork_productivity", False),
        grounding_max_iterations=cfg.get("max_iterations", 20),
        enable_recording=True,
        enable_screenshot=False,
        enable_video=False,
        enable_conversation_log=True,
    )


# ═══════════════════════════════════════════════════════════════════
# Helper: execute a single task and build the result record
# ═══════════════════════════════════════════════════════════════════

async def _execute_one_task(
    cs,
    task: Dict,
    phase: str,
    cfg: Dict,
    idx: int,
    total: int,
    token_stats: TokenStats,
    elapsed: float,
) -> Dict[str, Any]:
    """Build a result record (called after cs.execute finishes)."""
    # This is a helper only for record building; see _run_single_task for execution.
    raise NotImplementedError("Use _run_single_task instead")


async def _run_single_task(
    cs,
    task: Dict,
    phase: str,
    cfg: Dict,
    idx: int,
    total: int,
    tracker: TokenTracker,
    results_file: Path,
    results_list: List[Dict],
    *,
    concurrent: bool = False,
) -> Dict[str, Any]:
    """Execute one task on the given OpenSpace instance and persist the result.

    In concurrent mode, uses tracker.begin_task/end_task with ContextVar.
    In serial mode, uses tracker.start/stop.
    """
    rd = _results_dir(cfg)
    tid = task["task_id"]

    print(f"\n{'='*60}")
    print(f"[{phase.upper()}] Task {idx}/{total}: {task['occupation']}")
    print(f"  ID: {tid}")
    print(f"  Prompt: {task['prompt'][:120]}...")
    print(f"{'='*60}")

    # Skill count before execution
    skills_before = 0
    if cs._skill_store:
        try:
            skills_before = len(cs._skill_store.load_active())
        except Exception:
            pass

    # ── Token tracking setup ──
    ctx_token = None
    if concurrent: 
        ctx_token = tracker.begin_task(tid)
    else:
        tracker.start()

    t0 = time.monotonic()

    try:
        exec_task_id = f"{tid}_{phase}"
        task_workspace = str(rd / "workspace" / phase / tid)

        # Download reference files & augment prompt with file locations
        augmented_prompt = prepare_task_workspace(task, task_workspace)
        n_refs = len(task.get("reference_files", []) or [])
        if n_refs:
            print(f"  📎 {n_refs} reference file(s) → {task_workspace}")

        from openspace.runtime import ExecutionRequest

        execution_result = await cs.execute(
            ExecutionRequest(
                prompt=augmented_prompt,
                task_id=exec_task_id,
                workspace_dir=task_workspace,
            )
        )
        result = {
            "status": execution_result.status,
            "response": execution_result.text,
            "iterations": execution_result.iterations,
            "tool_executions": list(execution_result.tool_executions),
            "skills_used": list(execution_result.skills_used),
            "evolved_skills": list(execution_result.evolved_skills),
            "task_id": execution_result.task_id,
            "session_id": execution_result.session_id,
            "execution_time": execution_result.execution_time,
        }
        exec_status = result.get("status", "unknown")
    except Exception as e:
        result = {"status": "error", "error": str(e)}
        exec_status = "error"
        print(f"  ❌ Error: {e}")

    elapsed = time.monotonic() - t0

    # ── Token tracking teardown ──
    if concurrent:
        stats = tracker.end_task(tid, ctx_token)
    else:
        stats = tracker.stop()

    # Skill count after execution
    skills_after = 0
    evolved_skills = result.get("evolved_skills", [])
    if cs._skill_store:
        try:
            skills_after = len(cs._skill_store.load_active())
        except Exception:
            pass

    # ── Evaluate artifacts (aligned with ClawWork) ──
    eval_result: Dict[str, Any] = {"has_evaluation": False}
    if cfg.get("enable_evaluation", True) and exec_status != "error":
        try:
            eval_result = _evaluate_task(task, task_workspace, cfg)
            if eval_result.get("has_evaluation"):
                score_10 = eval_result["score_10"]
                actual_pay = eval_result["actual_payment"]
                cliff = " (⚠️ below 0.6 cliff)" if eval_result.get("cliff_applied") else ""
                print(f"  📝 Evaluation: {score_10}/10 → ${actual_pay:.2f}{cliff}")
                print(f"     Artifacts: {eval_result['artifact_paths']}")
            else:
                print(f"  📝 Evaluation skipped: {eval_result.get('feedback', 'N/A')}")
        except Exception as e:
            print(f"  ⚠️  Evaluation error: {e}")

    # Build result record
    record = {
        "task_id": tid,
        "phase": phase,
        "occupation": task.get("occupation", ""),
        "sector": task.get("sector", ""),
        "task_value_usd": task.get("task_value_usd", 0.0),
        "status": exec_status,
        "tokens": stats.to_dict(include_details=cfg.get("record_call_details", False)),
        "execution": {
            "iterations": result.get("iterations", 0),
            "tool_calls": len(result.get("tool_executions", [])),
            "time_sec": round(elapsed, 2),
        },
        "skills": {
            "before": skills_before,
            "after": skills_after,
            "new_this_task": skills_after - skills_before,
            "evolved": [
                {
                    "name": s.get("name", ""),
                    "origin": s.get("origin", ""),
                    "path": s.get("path", ""),
                }
                for s in evolved_skills
            ],
            "used": result.get("skills_used", []),
        },
        "evaluation": eval_result,
        "timestamp": datetime.now().isoformat(),
    }

    # Persist immediately (crash-safe)
    _append_jsonl(results_file, record)
    results_list.append(record)

    # Print summary
    _print_task_summary(record)

    return record


# ═══════════════════════════════════════════════════════════════════
# Phase runner — serial (concurrency=1)
# ═══════════════════════════════════════════════════════════════════

async def _run_phase_serial(
    phase: str,
    tasks: List[Dict],
    cfg: Dict,
    tracker: TokenTracker,
    completed_ids: set,
) -> List[Dict[str, Any]]:
    """Run one phase sequentially. A single OpenSpace instance is reused
    so that skills accumulate within the phase."""
    from openspace import OpenSpace

    rd = _results_dir(cfg)
    results_file = rd / f"{phase}_results.jsonl"
    config = _make_config(cfg, phase)

    cs = OpenSpace(config=config)
    await cs.initialize()

    results: List[Dict] = []
    total = len(tasks)
    skipped = 0

    try:
        for idx, task in enumerate(tasks, 1):
            if task["task_id"] in completed_ids:
                skipped += 1
                continue

            await _run_single_task(
                cs, task, phase, cfg, idx, total,
                tracker, results_file, results,
                concurrent=False,
            )
    finally:
        await cs.cleanup()

    if skipped:
        print(f"\n📌 Skipped {skipped} already-completed tasks (resume mode)")

    return results


# ═══════════════════════════════════════════════════════════════════
# Phase runner — concurrent (concurrency>1)
# ═══════════════════════════════════════════════════════════════════

async def _run_phase_concurrent(
    phase: str,
    tasks: List[Dict],
    cfg: Dict,
    tracker: TokenTracker,
    completed_ids: set,
    concurrency: int,
) -> List[Dict[str, Any]]:
    """Run one phase with N concurrent OpenSpace workers.

    Trade-offs vs serial:
      - Skills from concurrent tasks may not be visible to each other (reduced
        accumulation within a batch), but completed tasks' skills are visible
        to later tasks.
      - skills.before / skills.after counts are approximate when workers overlap.
      - Significantly faster wall-clock time.

    Architecture:
      - N OpenSpace instances created up front (worker pool via asyncio.Queue)
      - Each asyncio.Task grabs a worker, executes, returns it to the pool
      - Token tracking uses ContextVar so litellm callbacks route to the
        correct per-task bucket automatically.
    """
    from openspace import OpenSpace

    rd = _results_dir(cfg)
    results_file = rd / f"{phase}_results.jsonl"

    # ── Filter pending tasks ──
    pending = [t for t in tasks if t["task_id"] not in completed_ids]
    if not pending:
        print(f"  All {len(tasks)} tasks already completed (resume mode)")
        return []

    skipped = len(tasks) - len(pending)
    if skipped:
        print(f"📌 Skipped {skipped} already-completed tasks (resume mode)")

    # ── Create worker pool ──
    actual_concurrency = min(concurrency, len(pending))
    print(f"🔧 Creating {actual_concurrency} OpenSpace workers...")

    pool: asyncio.Queue = asyncio.Queue()
    workers: List[OpenSpace] = []

    for i in range(actual_concurrency):
        config = _make_config(cfg, phase, worker_id=i)
        cs = OpenSpace(config=config)
        await cs.initialize()
        workers.append(cs)
        pool.put_nowait(cs)
        print(f"  ✅ Worker {i} ready")

    # ── Install persistent callback for concurrent token tracking ──
    tracker.install()

    results: List[Dict] = []
    total_tasks = len(tasks)
    completed = 0
    errors = 0

    async def process_task(idx: int, task: Dict) -> None:
        nonlocal completed, errors
        cs = await pool.get()  # blocks until a worker is available
        try:
            record = await _run_single_task(
                cs, task, phase, cfg, idx, total_tasks,
                tracker, results_file, results,
                concurrent=True,
            )
            if record.get("status") == "error":
                errors += 1
        except Exception as e:
            errors += 1
            print(f"  ❌ [{task['task_id']}] Unhandled error: {e}")
        finally:
            pool.put_nowait(cs)  # return worker to pool
            completed += 1
            remaining = len(pending) - completed
            if remaining > 0 and completed % 5 == 0:
                print(f"\n  📊 Progress: {completed}/{len(pending)} done, "
                      f"{remaining} remaining, {errors} errors")

    # ── Dispatch all tasks as asyncio Tasks ──
    # The pool.get() naturally limits concurrency to N workers
    async_tasks = []
    for idx, task in enumerate(pending, 1 + skipped):
        async_tasks.append(asyncio.create_task(process_task(idx, task)))

    await asyncio.gather(*async_tasks, return_exceptions=True)

    # ── Cleanup ──
    tracker.uninstall()

    print(f"\n🔧 Cleaning up {len(workers)} workers...")
    for i, cs in enumerate(workers):
        try:
            await cs.cleanup()
        except Exception as e:
            print(f"  ⚠️ Worker {i} cleanup error: {e}")

    print(f"  ✅ All workers cleaned up")
    return results


# ═══════════════════════════════════════════════════════════════════
# Phase runner — unified entry point
# ═══════════════════════════════════════════════════════════════════

async def run_phase(
    phase: str,
    tasks: List[Dict],
    cfg: Dict,
    tracker: TokenTracker,
    completed_ids: set,
    concurrency: int = 1,
) -> List[Dict[str, Any]]:
    """Run one phase (phase1 or phase2) over all tasks.

    Args:
        concurrency: Number of parallel OpenSpace workers.
                     1 = serial (default, best for skill accumulation).
                     >1 = concurrent (faster, reduced skill accumulation).
    """
    if concurrency <= 1:
        return await _run_phase_serial(phase, tasks, cfg, tracker, completed_ids)
    else:
        return await _run_phase_concurrent(
            phase, tasks, cfg, tracker, completed_ids, concurrency
        )


def _print_task_summary(record: Dict) -> None:
    tokens = record["tokens"]
    skills = record["skills"]
    exe = record["execution"]
    evaluation = record.get("evaluation", {})
    status_icon = "✅" if record["status"] == "success" else "❌"

    agent_prompt = tokens.get('agent_prompt_tokens', tokens['prompt_tokens'])
    agent_comp = tokens.get('agent_completion_tokens', tokens['completion_tokens'])
    agent_total = tokens.get('agent_total_tokens', tokens['total_tokens'])
    overhead = tokens['total_tokens'] - agent_total

    print(f"\n  {status_icon} Status: {record['status']}")
    print(f"  📊 Tokens (total): {tokens['total_tokens']:,} "
          f"(prompt: {tokens['prompt_tokens']:,}, "
          f"completion: {tokens['completion_tokens']:,})")
    print(f"  📊 Tokens (agent): {agent_total:,} "
          f"(prompt: {agent_prompt:,}, "
          f"completion: {agent_comp:,}, "
          f"overhead: {overhead:,})")
    print(f"  💰 Cost: ${tokens['cost_usd']:.4f}")
    print(f"  🔧 Iterations: {exe['iterations']}, "
          f"Tool calls: {exe['tool_calls']}, "
          f"Time: {exe['time_sec']:.1f}s")
    print(f"  🧬 Skills: {skills['before']} → {skills['after']} "
          f"(+{skills['new_this_task']} new)")
    if skills["used"]:
        print(f"  📎 Skills used: {skills['used']}")
    if evaluation.get("has_evaluation"):
        cliff_mark = " ⚠️cliff" if evaluation.get("cliff_applied") else ""
        print(f"  📝 Quality: {evaluation['score_10']}/10 → "
              f"${evaluation['actual_payment']:.2f}"
              f"/{evaluation.get('max_payment', '?')}{cliff_mark}")


# ═══════════════════════════════════════════════════════════════════
# Comparison & Summary
# ═══════════════════════════════════════════════════════════════════

def _pct(a: int, b: int) -> float:
    """Percentage savings: (a - b) / a * 100.  Positive = saved."""
    return (a - b) / a * 100 if a > 0 else 0.0


def build_comparison(cfg: Dict) -> None:
    """Build per-task comparison and aggregate summary from both phases.

    Reports prompt tokens, completion tokens, and total tokens **separately**
    so that the impact of skill-augmented prompts vs. agent output efficiency
    can be analyzed independently.
    """
    rd = _results_dir(cfg)
    p1 = {r["task_id"]: r for r in _load_jsonl(rd / "phase1_results.jsonl")}
    p2 = {r["task_id"]: r for r in _load_jsonl(rd / "phase2_results.jsonl")}

    comparisons = []
    # Accumulators — separate prompt / completion / total
    agg = {k: 0 for k in (
        "p1_prompt", "p1_completion", "p1_total",
        "p2_prompt", "p2_completion", "p2_total",
        "p1_cost", "p2_cost", "p1_calls", "p2_calls",
        "p1_agent_prompt", "p1_agent_completion", "p1_agent_total",
        "p2_agent_prompt", "p2_agent_completion", "p2_agent_total",
    )}
    savings_total = []
    savings_prompt = []
    savings_completion = []
    savings_agent_total = []
    savings_agent_prompt = []
    savings_agent_completion = []

    # Evaluation accumulators
    eval_scores_p1 = []
    eval_scores_p2 = []
    eval_payments_p1 = []
    eval_payments_p2 = []
    eval_cliffed_p1 = 0
    eval_cliffed_p2 = 0
    eval_count = 0

    for tid in p1:
        r1 = p1[tid]
        r2 = p2.get(tid)
        if r2 is None:
            continue

        t1 = r1["tokens"]
        t2 = r2["tokens"]

        sv_total = _pct(t1["total_tokens"], t2["total_tokens"])
        sv_prompt = _pct(t1["prompt_tokens"], t2["prompt_tokens"])
        sv_completion = _pct(t1["completion_tokens"], t2["completion_tokens"])
        sv_cost = _pct(t1["cost_usd"], t2["cost_usd"]) if t1["cost_usd"] > 0 else 0.0

        # Agent-only tokens (fallback to total if not available)
        a1_prompt = t1.get("agent_prompt_tokens", t1["prompt_tokens"])
        a1_comp = t1.get("agent_completion_tokens", t1["completion_tokens"])
        a1_total = t1.get("agent_total_tokens", t1["total_tokens"])
        a2_prompt = t2.get("agent_prompt_tokens", t2["prompt_tokens"])
        a2_comp = t2.get("agent_completion_tokens", t2["completion_tokens"])
        a2_total = t2.get("agent_total_tokens", t2["total_tokens"])
        sv_agent_total = _pct(a1_total, a2_total)
        sv_agent_prompt = _pct(a1_prompt, a2_prompt)
        sv_agent_completion = _pct(a1_comp, a2_comp)

        # ── Evaluation data ──
        e1 = r1.get("evaluation", {})
        e2 = r2.get("evaluation", {})
        eval_comp = {}
        if e1.get("has_evaluation") and e2.get("has_evaluation"):
            eval_comp = {
                "phase1_score": e1["score_10"],
                "phase2_score": e2["score_10"],
                "score_change": round(e2["score_10"] - e1["score_10"], 1),
                "phase1_actual_payment": e1["actual_payment"],
                "phase2_actual_payment": e2["actual_payment"],
                "phase1_cliff_applied": e1.get("cliff_applied", False),
                "phase2_cliff_applied": e2.get("cliff_applied", False),
            }
            eval_scores_p1.append(e1["score_10"])
            eval_scores_p2.append(e2["score_10"])
            eval_payments_p1.append(e1["actual_payment"])
            eval_payments_p2.append(e2["actual_payment"])
            if e1.get("cliff_applied"):
                eval_cliffed_p1 += 1
            if e2.get("cliff_applied"):
                eval_cliffed_p2 += 1
            eval_count += 1

        comp = {
            "task_id": tid,
            "occupation": r1.get("occupation", ""),
            "sector": r1.get("sector", ""),
            # ── Total tokens (all calls) ──
            "phase1_tokens": t1["total_tokens"],
            "phase2_tokens": t2["total_tokens"],
            "token_savings_pct": round(sv_total, 2),
            # ── Prompt tokens (all calls) ──
            "phase1_prompt_tokens": t1["prompt_tokens"],
            "phase2_prompt_tokens": t2["prompt_tokens"],
            "prompt_savings_pct": round(sv_prompt, 2),
            # ── Completion tokens (all calls) ──
            "phase1_completion_tokens": t1["completion_tokens"],
            "phase2_completion_tokens": t2["completion_tokens"],
            "completion_savings_pct": round(sv_completion, 2),
            # ── Agent-only tokens (excludes skill engine overhead) ──
            "phase1_agent_prompt": a1_prompt,
            "phase2_agent_prompt": a2_prompt,
            "agent_prompt_savings_pct": round(sv_agent_prompt, 2),
            "phase1_agent_completion": a1_comp,
            "phase2_agent_completion": a2_comp,
            "agent_completion_savings_pct": round(sv_agent_completion, 2),
            "phase1_agent_total": a1_total,
            "phase2_agent_total": a2_total,
            "agent_total_savings_pct": round(sv_agent_total, 2),
            # ── LLM calls ──
            "phase1_llm_calls": t1["llm_calls"],
            "phase2_llm_calls": t2["llm_calls"],
            # ── Cost ──
            "phase1_cost": round(t1["cost_usd"], 6),
            "phase2_cost": round(t2["cost_usd"], 6),
            "cost_savings_pct": round(sv_cost, 2),
            # ── Execution ──
            "phase1_iterations": r1["execution"]["iterations"],
            "phase2_iterations": r2["execution"]["iterations"],
            "phase1_time": r1["execution"]["time_sec"],
            "phase2_time": r2["execution"]["time_sec"],
            "phase1_status": r1["status"],
            "phase2_status": r2["status"],
            "skills_used_p2": r2["skills"].get("used", []),
            "skills_total_after_p1": r1["skills"]["after"],
            # ── Evaluation (ClawWork-aligned) ──
            **eval_comp,
        }
        comparisons.append(comp)
        _append_jsonl(rd / "comparison.jsonl", comp)

        # Accumulate
        agg["p1_prompt"] += t1["prompt_tokens"]
        agg["p1_completion"] += t1["completion_tokens"]
        agg["p1_total"] += t1["total_tokens"]
        agg["p2_prompt"] += t2["prompt_tokens"]
        agg["p2_completion"] += t2["completion_tokens"]
        agg["p2_total"] += t2["total_tokens"]
        agg["p1_cost"] += t1["cost_usd"]
        agg["p2_cost"] += t2["cost_usd"]
        agg["p1_calls"] += t1["llm_calls"]
        agg["p2_calls"] += t2["llm_calls"]
        agg["p1_agent_prompt"] += a1_prompt
        agg["p1_agent_completion"] += a1_comp
        agg["p1_agent_total"] += a1_total
        agg["p2_agent_prompt"] += a2_prompt
        agg["p2_agent_completion"] += a2_comp
        agg["p2_agent_total"] += a2_total
        savings_total.append(sv_total)
        savings_prompt.append(sv_prompt)
        savings_completion.append(sv_completion)
        savings_agent_total.append(sv_agent_total)
        savings_agent_prompt.append(sv_agent_prompt)
        savings_agent_completion.append(sv_agent_completion)

    # ── Aggregate summary ──
    n = len(comparisons)
    if n == 0:
        print("⚠️  No matching task pairs found for comparison.")
        return

    def _stats(lst: list) -> dict:
        s = sorted(lst)
        return {
            "mean": round(sum(s) / len(s), 2),
            "median": round(s[len(s) // 2], 2),
            "min": round(min(s), 2),
            "max": round(max(s), 2),
        }

    # Skills snapshot after Phase 1
    skills_file = rd / "skills_snapshot.json"
    skills_data = []
    if skills_file.exists():
        with open(skills_file) as f:
            skills_data = json.load(f)

    # ── Evaluation summary ──
    eval_summary = {}
    if eval_count > 0:
        mean_p1 = sum(eval_scores_p1) / eval_count
        mean_p2 = sum(eval_scores_p2) / eval_count
        improved = sum(1 for i in range(eval_count) if eval_scores_p2[i] > eval_scores_p1[i])
        same = sum(1 for i in range(eval_count) if eval_scores_p2[i] == eval_scores_p1[i])
        regressed = sum(1 for i in range(eval_count) if eval_scores_p2[i] < eval_scores_p1[i])
        eval_summary = {
            "tasks_evaluated": eval_count,
            "phase1": {
                "mean_score": round(mean_p1, 2),
                "total_actual_payment": round(sum(eval_payments_p1), 2),
                "cliffed_count": eval_cliffed_p1,
            },
            "phase2": {
                "mean_score": round(mean_p2, 2),
                "total_actual_payment": round(sum(eval_payments_p2), 2),
                "cliffed_count": eval_cliffed_p2,
            },
            "score_change": round(mean_p2 - mean_p1, 2),
            "improved": improved,
            "same": same,
            "regressed": regressed,
            "cliff_threshold": _MIN_EVALUATION_THRESHOLD,
        }

    summary = {
        "run_name": cfg["run_name"],
        "model": cfg["model"],
        "total_tasks": n,
        "timestamp": datetime.now().isoformat(),
        "token_savings": {
            "total": {
                "overall_pct": round(_pct(agg["p1_total"], agg["p2_total"]), 2),
                "per_task": _stats(savings_total),
                "phase1": agg["p1_total"],
                "phase2": agg["p2_total"],
            },
            "prompt": {
                "overall_pct": round(_pct(agg["p1_prompt"], agg["p2_prompt"]), 2),
                "per_task": _stats(savings_prompt),
                "phase1": agg["p1_prompt"],
                "phase2": agg["p2_prompt"],
            },
            "completion": {
                "overall_pct": round(_pct(agg["p1_completion"], agg["p2_completion"]), 2),
                "per_task": _stats(savings_completion),
                "phase1": agg["p1_completion"],
                "phase2": agg["p2_completion"],
            },
        },
        "agent_token_savings": {
            "total": {
                "overall_pct": round(_pct(agg["p1_agent_total"], agg["p2_agent_total"]), 2),
                "per_task": _stats(savings_agent_total),
                "phase1": agg["p1_agent_total"],
                "phase2": agg["p2_agent_total"],
            },
            "prompt": {
                "overall_pct": round(_pct(agg["p1_agent_prompt"], agg["p2_agent_prompt"]), 2),
                "per_task": _stats(savings_agent_prompt),
                "phase1": agg["p1_agent_prompt"],
                "phase2": agg["p2_agent_prompt"],
            },
            "completion": {
                "overall_pct": round(_pct(agg["p1_agent_completion"], agg["p2_agent_completion"]), 2),
                "per_task": _stats(savings_agent_completion),
                "phase1": agg["p1_agent_completion"],
                "phase2": agg["p2_agent_completion"],
            },
        },
        "llm_calls": {
            "phase1": agg["p1_calls"],
            "phase2": agg["p2_calls"],
        },
        "cost_savings": {
            "total_phase1_usd": round(agg["p1_cost"], 4),
            "total_phase2_usd": round(agg["p2_cost"], 4),
            "saved_usd": round(agg["p1_cost"] - agg["p2_cost"], 4),
        },
        "evaluation": eval_summary,
        "skills": {
            "total_after_phase1": len(skills_data),
            "by_origin": _count_skills_by_origin(skills_data),
        },
    }

    with open(rd / "summary.json", "w") as f:
        json.dump(summary, f, indent=2, ensure_ascii=False)

    # ── Print summary ──
    ts = summary["token_savings"]
    print("\n" + "=" * 60)
    print("📊 BENCHMARK SUMMARY")
    print("=" * 60)
    print(f"  Tasks compared: {n}")
    print(f"  Model: {cfg['model']}")

    ats = summary.get("agent_token_savings", {})

    print(f"\n  🔑 Token Savings — All Calls (Total = Prompt + Completion):")
    print(f"     {'':20s} {'Overall':>10s} {'Mean':>10s} {'Median':>10s}")
    for label, key in [("Total tokens", "total"),
                       ("↳ Prompt tokens", "prompt"),
                       ("↳ Completion tokens", "completion")]:
        s = ts[key]
        print(f"     {label:<20s} {s['overall_pct']:>+9.1f}% "
              f"{s['per_task']['mean']:>+9.1f}% "
              f"{s['per_task']['median']:>+9.1f}%")

    if ats:
        print(f"\n  🎯 Token Savings — Agent Only (excludes skill engine overhead):")
        print(f"     {'':20s} {'Overall':>10s} {'Mean':>10s} {'Median':>10s}")
        for label, key in [("Total tokens", "total"),
                           ("↳ Prompt tokens", "prompt"),
                           ("↳ Completion tokens", "completion")]:
            s = ats[key]
            print(f"     {label:<20s} {s['overall_pct']:>+9.1f}% "
                  f"{s['per_task']['mean']:>+9.1f}% "
                  f"{s['per_task']['median']:>+9.1f}%")

    print(f"\n     Phase 1: {agg['p1_total']:>10,} total "
          f"({agg['p1_prompt']:,} prompt + {agg['p1_completion']:,} completion)")
    print(f"     Phase 2: {agg['p2_total']:>10,} total "
          f"({agg['p2_prompt']:,} prompt + {agg['p2_completion']:,} completion)")
    if ats:
        print(f"     Phase 1 (agent): {agg['p1_agent_total']:>10,} total "
              f"({agg['p1_agent_prompt']:,} prompt + {agg['p1_agent_completion']:,} completion)")
        print(f"     Phase 2 (agent): {agg['p2_agent_total']:>10,} total "
              f"({agg['p2_agent_prompt']:,} prompt + {agg['p2_agent_completion']:,} completion)")
    print(f"     LLM calls: {agg['p1_calls']} → {agg['p2_calls']}")

    # ── Print evaluation summary ──
    if eval_summary:
        print(f"\n  📝 Quality Evaluation (ClawWork-aligned, cliff={_MIN_EVALUATION_THRESHOLD}):")
        print(f"     Evaluated: {eval_summary['tasks_evaluated']} tasks")
        print(f"     Phase 1: mean {eval_summary['phase1']['mean_score']}/10"
              f" | ${eval_summary['phase1']['total_actual_payment']:.2f} earned"
              f" | {eval_summary['phase1']['cliffed_count']} cliffed")
        print(f"     Phase 2: mean {eval_summary['phase2']['mean_score']}/10"
              f" | ${eval_summary['phase2']['total_actual_payment']:.2f} earned"
              f" | {eval_summary['phase2']['cliffed_count']} cliffed")
        print(f"     Change:  {eval_summary['score_change']:+.1f} pts"
              f" | ↑{eval_summary['improved']} ={eval_summary['same']} ↓{eval_summary['regressed']}")

    print(f"\n  🧬 Skills accumulated: {len(skills_data)}")
    print(f"     By origin: {_count_skills_by_origin(skills_data)}")
    print(f"\n  📁 Results: {rd}")
    print("=" * 60)


# ═══════════════════════════════════════════════════════════════════
# DB management
# ═══════════════════════════════════════════════════════════════════

def _wipe_skill_db() -> None:
    """Delete the shared .openspace/openspace.db for a fresh start."""
    db_file = _OPENSPACE_DB_DIR / "openspace.db"
    for f in [db_file, db_file.with_suffix(".db-wal"), db_file.with_suffix(".db-shm")]:
        if f.exists():
            f.unlink()
    print("🗑️  Wiped skill database for fresh start")


def _backup_skill_db(dest: Path) -> None:
    """Copy the current .openspace/openspace.db to dest."""
    db_file = _OPENSPACE_DB_DIR / "openspace.db"
    if db_file.exists():
        dest.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(str(db_file), str(dest))
        # Also copy WAL if exists (for consistency)
        wal = db_file.with_suffix(".db-wal")
        if wal.exists():
            shutil.copy2(str(wal), str(dest.with_suffix(".db-wal")))
        print(f"💾 Backed up skill DB to {dest}")


# ═══════════════════════════════════════════════════════════════════
# Main
# ═══════════════════════════════════════════════════════════════════

async def main(args: argparse.Namespace) -> None:
    cfg = load_config(args.config)

    # CLI overrides
    if args.max_tasks is not None:
        cfg["max_tasks"] = args.max_tasks
    if args.per_occupation is not None:
        cfg["per_occupation"] = args.per_occupation
    if args.model:
        cfg["model"] = args.model
    if args.run_name:
        cfg["run_name"] = args.run_name
    if args.clawwork_root:
        cfg["clawwork_root"] = args.clawwork_root
    if getattr(args, "use_clawwork_productivity", False):
        cfg["use_clawwork_productivity"] = True

    # When using ClawWork productivity tools, ensure livebench is importable
    # before OpenSpace.initialize() (ShellSession loads productivity_tools which imports livebench)
    if cfg.get("use_clawwork_productivity"):
        clawwork_root = Path(cfg.get("clawwork_root", "") or "").resolve()
        if clawwork_root.is_dir():
            parent = str(clawwork_root)  # ClawWork repo root; livebench is livebench/
            if parent not in sys.path:
                sys.path.insert(0, parent)
                print(f"📎 ClawWork on sys.path for productivity tools: {parent}")
        else:
            print(f"⚠️  use_clawwork_productivity is True but clawwork_root not found: {clawwork_root}")

    # Fixed task list overrides individual filters
    if args.task_list:
        tl_path = Path(args.task_list)
        if not tl_path.is_absolute():
            # Try CWD first, then relative to this package directory
            if tl_path.exists():
                tl_path = tl_path.resolve()
            else:
                tl_path = (Path(__file__).parent / tl_path).resolve()
        if not tl_path.exists():
            print(f"❌ Task list file not found: {tl_path}")
            print(f"   (also tried relative to CWD and {Path(__file__).parent})")
            return
        with open(tl_path, "r") as f:
            tl_data = json.load(f)
        fixed_ids = tl_data.get("task_ids", [])
        if not fixed_ids:
            print(f"❌ No 'task_ids' array found in {tl_path}")
            return
        cfg["task_ids"] = fixed_ids
        # Disable filters that conflict with the fixed list
        cfg["max_tasks"] = None
        cfg["per_occupation"] = None
        cfg["sectors"] = None
        cfg["occupations"] = None
        print(f"📋 Using fixed task list: {tl_path.name} ({len(fixed_ids)} tasks)")
        if "_selection_principle" in tl_data:
            desc = tl_data["_selection_principle"].get("description", "")
            if desc:
                print(f"   {desc}")

    concurrency = args.concurrency or 1

    # Evaluation toggle
    if args.no_eval:
        cfg["enable_evaluation"] = False
        print("📝 Evaluation disabled (--no-eval)")
    else:
        cfg.setdefault("enable_evaluation", True)

    # Propagate model to sub-agents (BashAgentTool reads OPENSPACE_MODEL)
    os.environ["OPENSPACE_MODEL"] = cfg.get("model", "")

    # Pre-flight checks
    env_ok = _check_environment(cfg)
    if not env_ok and not args.dry_run:
        print("❌ Environment checks failed. Fix the issues above and retry.")
        print("   Or use --dry-run to just check without executing.")
        return

    # Load tasks (needed by --prefetch-only and --dry-run before results dir)
    tasks = load_tasks(
        clawwork_root=cfg.get("clawwork_root", ""),
        gdpval_path=cfg.get("gdpval_path"),
        task_ids=cfg.get("task_ids"),
        max_tasks=cfg.get("max_tasks"),
        sectors=cfg.get("sectors"),
        occupations=cfg.get("occupations"),
        per_occupation=cfg.get("per_occupation"),
    )

    # ── Prefetch reference files ──
    from .task_loader import prefetch_reference_files

    if args.prefetch_only:
        print("\n📦 Prefetch-only mode: downloading all reference files …")
        result = prefetch_reference_files(tasks)
        n_tasks_with_files = sum(1 for v in result.values() if v)
        total_files = sum(len(v) for v in result.values())
        print(f"\n✅ Prefetch complete: {total_files} files cached for "
              f"{n_tasks_with_files} tasks.")
        return

    if args.dry_run:
        print(f"\n🏁 Dry run complete. {len(tasks)} tasks ready.")
        mode = f"concurrent ({concurrency} workers)" if concurrency > 1 else "serial"
        print(f"   Mode: {mode}")
        print(f"   To execute: remove --dry-run flag")
        if tasks:
            print(f"\n   Sample task:")
            t = tasks[0]
            print(f"     ID: {t['task_id']}")
            print(f"     Occupation: {t['occupation']}")
            print(f"     Prompt ({len(t['prompt'])} chars): {t['prompt'][:150]}...")
        return

    if not tasks:
        print("❌ No tasks loaded. Check config.")
        return

    if not args.no_prefetch:
        has_refs = sum(1 for t in tasks if t.get("reference_files"))
        if has_refs:
            print(f"\n📦 Auto-prefetching reference files for {has_refs} tasks …")
            print(f"   (use --no-prefetch to skip, --prefetch-only to run separately)")
            prefetch_reference_files(tasks)

    rd = _results_dir(cfg)
    rd.mkdir(parents=True, exist_ok=True)

    # Save config for reproducibility
    run_cfg = {**cfg, "concurrency": concurrency}
    with open(rd / "config.json", "w") as f:
        json.dump(run_cfg, f, indent=2, ensure_ascii=False)

    tracker = TokenTracker(record_details=cfg.get("record_call_details", True))

    mode_str = f"concurrent ({concurrency} workers)" if concurrency > 1 else "serial"
    print(f"\n⚡ Execution mode: {mode_str}")

    # ── Phase 1 ──
    if not args.phase2_only:
        print("\n" + "🔵" * 30)
        print("  PHASE 1: Cold Start — Skills Accumulate")
        if concurrency > 1:
            print(f"  ⚡ Concurrent mode: {concurrency} workers")
            print("  ⚠️  Cross-task skill accumulation reduced (concurrent batches)")
        print("🔵" * 30)

        p1_results_file = rd / "phase1_results.jsonl"
        completed_p1 = _completed_task_ids(p1_results_file) if args.resume else set()

        if not args.resume:
            _wipe_skill_db()

        if completed_p1:
            print(f"📌 Resuming Phase 1: {len(completed_p1)} tasks already done")
        else:
            print(f"🚀 Running {len(tasks)} tasks...")

        await run_phase("phase1", tasks, cfg, tracker, completed_p1, concurrency)

        # Snapshot skills after Phase 1
        try:
            from openspace.skill_engine import SkillStore
            store = SkillStore()
            skills = _snapshot_skills(store)
            with open(rd / "skills_snapshot.json", "w") as f:
                json.dump(skills, f, indent=2, ensure_ascii=False, default=str)
            store.close()
            print(f"\n🧬 Phase 1 complete: {len(skills)} skills accumulated")
        except Exception as e:
            print(f"⚠️  Could not snapshot skills: {e}")

        # Backup the DB before Phase 2
        _backup_skill_db(rd / "phase1_skill_db.sqlite")

    # ── Phase 2 ──
    if not args.phase1_only:
        print("\n" + "🟢" * 30)
        print("  PHASE 2: Full Warm — All Phase 1 Skills Available")
        if concurrency > 1:
            print(f"  ⚡ Concurrent mode: {concurrency} workers")
        print("🟢" * 30)

        p2_results_file = rd / "phase2_results.jsonl"
        completed_p2 = _completed_task_ids(p2_results_file) if args.resume else set()

        if completed_p2:
            print(f"📌 Resuming Phase 2: {len(completed_p2)} tasks already done")
        else:
            print(f"🚀 Running {len(tasks)} tasks with accumulated skills...")

        await run_phase("phase2", tasks, cfg, tracker, completed_p2, concurrency)

        # ── Comparison ──
        print("\n📊 Building comparison...")
        build_comparison(cfg)
    else:
        print("\n📌 Phase 1 only mode — skipping Phase 2.")


def _check_environment(cfg: Dict) -> bool:
    """Pre-flight checks: API keys, dependencies, data availability."""
    print("🔍 Pre-flight checks...")
    ok = True

    # 1. Check LLM API key
    model = cfg.get("model", "")
    if "openrouter" in model:
        key = os.environ.get("OPENROUTER_API_KEY", "")
        if not key:
            print("  ❌ OPENROUTER_API_KEY not set")
            print("     export OPENROUTER_API_KEY='sk-or-...'")
            ok = False
        else:
            print(f"  ✅ OPENROUTER_API_KEY set ({key[:12]}...)")
    elif "openai" in model or "gpt" in model:
        key = os.environ.get("OPENAI_API_KEY", "")
        if not key:
            print("  ❌ OPENAI_API_KEY not set")
            ok = False
        else:
            print(f"  ✅ OPENAI_API_KEY set")
    elif "anthropic" in model or "claude" in model:
        key = os.environ.get("ANTHROPIC_API_KEY", "") or os.environ.get("OPENROUTER_API_KEY", "")
        if not key:
            print("  ❌ ANTHROPIC_API_KEY or OPENROUTER_API_KEY not set")
            ok = False
        else:
            print(f"  ✅ API key set for Anthropic model")
    else:
        print(f"  ⚠️  Model: {model} — make sure the corresponding API key is set")

    # 2. Check litellm
    try:
        import litellm
        ver = getattr(litellm, "__version__", getattr(litellm, "version", "unknown"))
        print(f"  ✅ litellm ({ver})")
    except ImportError:
        print("  ❌ litellm not installed — pip install litellm")
        ok = False

    # 3. Check openspace
    try:
        importlib.import_module("openspace")
        print(f"  ✅ openspace importable")
    except ImportError as e:
        print(f"  ❌ openspace not importable: {e}")
        print(f"     Run from OpenSpace directory or: pip install -e .")
        ok = False

    # 4. Check data availability (quick peek)
    root = Path(cfg.get("clawwork_root", ""))
    gdp = cfg.get("gdpval_path")
    has_data = False
    if gdp and Path(gdp).exists():
        has_data = True
        print(f"  ✅ GDPVal data at {gdp}")
    elif root.exists():
        for p in [
            root / "gdpval",
            root / "livebench" / "data" / "tasks" / "example_tasks.jsonl",
            root / "scripts" / "task_value_estimates" / "task_values.jsonl",
        ]:
            if p.exists():
                has_data = True
                print(f"  ✅ Task data found: {p}")
                break
    if not has_data:
        if importlib.util.find_spec("datasets") is not None:
            print(f"  ✅ HuggingFace datasets library installed — will auto-download")
            has_data = True
        else:
            print(f"  ⚠️  No local task data found & no HuggingFace datasets library")
            print(f"     Fix: pip install datasets  OR  set --clawwork-root")

    # 5. Check evaluation readiness
    if cfg.get("enable_evaluation", True):
        eval_key = os.environ.get("EVALUATION_API_KEY") or os.environ.get("OPENAI_API_KEY")
        meta_dir = root / "eval" / "meta_prompts" if root.exists() else None
        if eval_key:
            print(f"  ✅ Evaluation API key set")
        else:
            print(f"  ⚠️  No EVALUATION_API_KEY or OPENAI_API_KEY for evaluation")
            print(f"     Evaluation will fail. Use --no-eval to skip.")
        if meta_dir and meta_dir.exists():
            n_meta = len(list(meta_dir.glob("*.json")))
            print(f"  ✅ Evaluation meta-prompts: {n_meta} rubrics in {meta_dir}")
            print(f"     (If evaluation fails with 'No module named langchain_core', run: pip install -r benchmarks/gdpval/requirements-eval.txt)")
        elif meta_dir:
            print(f"  ⚠️  Meta-prompts dir not found: {meta_dir}")
            print(f"     Evaluation needs ClawWork/eval/meta_prompts/")
    else:
        print(f"  ℹ️  Evaluation disabled")

    print()
    return ok


def cli():
    parser = argparse.ArgumentParser(
        description="GDPVal Benchmark for OpenSpace skill-driven token savings"
    )
    parser.add_argument("--config", type=str, default=str(_DEFAULT_CONFIG),
                        help="Path to config JSON file (default: benchmarks/gdpval/config.json)")
    parser.add_argument("--max-tasks", type=int, default=None,
                        help="Max tasks to run (for testing)")
    parser.add_argument("--per-occupation", type=int, default=None,
                        help="Stratified sampling: pick N tasks per occupation "
                             "(e.g., --per-occupation 1 → 44 tasks covering all occupations)")
    parser.add_argument("--task-list", type=str, default=None,
                        help="Path to a task-list JSON file (e.g. tasks_50.json). "
                             "The file must have a 'task_ids' array. Overrides max-tasks / "
                             "per-occupation / sectors / occupations filters.")
    parser.add_argument("--model", type=str, default=None,
                        help="LLM model override")
    parser.add_argument("--run-name", type=str, default=None,
                        help="Run name (determines output directory)")
    parser.add_argument("--clawwork-root", type=str, default=None,
                        help="Path to ClawWork project (for loading tasks)")
    parser.add_argument("--concurrency", type=int, default=1,
                        help="Number of parallel OpenSpace workers per phase "
                             "(default: 1 = serial). Higher values reduce "
                             "cross-task skill accumulation within a phase.")
    parser.add_argument("--resume", action="store_true",
                        help="Resume from last checkpoint")
    parser.add_argument("--phase2-only", action="store_true",
                        help="Skip Phase 1, run Phase 2 only (requires Phase 1 DB)")
    parser.add_argument("--phase1-only", action="store_true",
                        help="Run Phase 1 only, skip Phase 2")
    parser.add_argument("--dry-run", action="store_true",
                        help="Only load tasks and check environment, don't execute")
    parser.add_argument("--no-eval", action="store_true",
                        help="Disable ClawWork-aligned evaluation after each task "
                             "(saves API calls to the evaluation model)")
    parser.add_argument("--prefetch-only", action="store_true",
                        help="Only pre-download all reference files to local cache, "
                             "then exit. Run this first to avoid SSL flakiness "
                             "during benchmark execution.")
    parser.add_argument("--no-prefetch", action="store_true",
                        help="Skip the automatic prefetch step (download on-the-fly instead)")
    parser.add_argument("--use-clawwork-productivity", action="store_true",
                        help="Enable ClawWork productivity tools (web_search, create_file, read, etc.) for fair comparison with ClawWork; requires livebench installed.")
    args = parser.parse_args()
    asyncio.run(main(args))


if __name__ == "__main__":
    cli()