feat: add empirical difficulty evaluator

README.md

@@ -102,6 +102,30 @@ uv run cvelab verify enterprise_3tier \
   --environment-only
 ```
 
+Run an independent empirical difficulty evaluation:
+
+```bash
+# Range: deploys an isolated copy once per model and writes only the report.
+uv run cvelab difficulty range data/scenarios/enterprise3-demo \
+  --output reports/difficulty-enterprise3-demo.json \
+  --api-key "$LLM_API_KEY" --base-url "$LLM_BASE_URL"
+
+# Atom: evaluate an already running target container. Add --reset-command
+# when the target can be restored between model runs.
+uv run cvelab difficulty atom data/atoms/CVE-2014-6271 \
+  --container cve-target --target-ip 172.18.0.2 \
+  --output reports/difficulty-CVE-2014-6271.json \
+  --api-key "$LLM_API_KEY" --base-url "$LLM_BASE_URL"
+```
+
+The evaluator runs the fixed Qwen model set
+(`qwen3.6-27b`, `qwen3.6-35b-a3b`, `qwen3.6-plus`, `qwen3.6-flash`) with a
+30-turn / 1800-second default budget. It records solution rate, turns, wall
+time, and tool calls in a separate JSON artifact. It never writes evaluation
+results into an Atom or Range. Atom evaluation is marked
+`state_isolated=false` unless `--reset-command` is supplied; a non-isolated
+run should be treated as exploratory rather than a clean comparison.
+
 Run focused tests before changing a subsystem:
 
 ```bash

src/clab_builder/atomizer/agent/researcher.py

@@ -84,9 +84,15 @@ def __post_init__(self):
 class SecurityResearcherAgent:
     """Agent 容器生命周期管理器"""
 
-    def __init__(self, agent_image: str = "clab-agent:latest", max_turns: int = 80):
+    def __init__(
+        self,
+        agent_image: str = "clab-agent:latest",
+        max_turns: int = 80,
+        agent_timeout: int = 900,
+    ):
         self.agent_image = agent_image
         self.max_turns = max_turns
+        self.agent_timeout = agent_timeout
         self.container_id: str | None = None
         self.container_name = f"agent-{uuid.uuid4().hex[:8]}"
         self.model: str = ""  # set by start(); read by run() to pick the harness
@@ -331,9 +337,9 @@ def read_stderr():
             reader = threading.Thread(target=read_stderr, daemon=True)
             reader.start()
 
-            # 墙钟上限：max_turns=80 正常约 12 分钟（170 次 API 调用 × ~4s）。
-            # 15 分钟足够跑满 80 turns 的慢任务，超时后 self.stop() 强制终止容器。
-            wall_timeout = 900
+            # 墙钟上限独立于 max_turns：达到上限后 stop() 会强制终止 Agent
+            # 容器，避免 API 卡住时评估永远不返回。
+            wall_timeout = self.agent_timeout
             proc.wait(timeout=wall_timeout)
             reader.join(timeout=5)
 

src/clab_builder/cli.py

@@ -2,14 +2,10 @@
 
 import click
 import os
-import yaml
 from pathlib import Path
 from dotenv import load_dotenv
-
-# 自动加载 .env
-load_dotenv()
-
 from clab_builder import __version__
+load_dotenv()
 
 
 @click.group()
@@ -19,6 +15,93 @@ def main():
     pass
 
 
+@main.group("difficulty")
+def difficulty():
+    # 这个命令组只产生外部评估报告，不回写 Atom 或 Range。
+    """Empirically evaluate Atom or Range difficulty without changing it."""
+    pass
+
+
+def _difficulty_common_options(command):
+    # 四个模型使用同一套 endpoint/key 和相同预算，保证比较条件一致。
+    command = click.option("--output", "-o", required=True, type=click.Path())(command)
+    command = click.option("--api-key", envvar="LLM_API_KEY", required=True)(command)
+    command = click.option("--base-url", envvar="LLM_BASE_URL", default="")(command)
+    command = click.option("--max-turns", type=int, default=30, show_default=True)(command)
+    command = click.option("--timeout", type=int, default=1800, show_default=True)(command)
+    command = click.option(
+        "--models", default=",".join(
+            ("qwen3.6-27b", "qwen3.6-35b-a3b", "qwen3.6-plus", "qwen3.6-flash")
+        ),
+        show_default=True,
+    )(command)
+    return command
+
+
+@difficulty.command("range")
+@click.argument("scenario_dir", type=click.Path(exists=True, file_okay=False))
+@click.option("--agent-context", type=click.Choice(["guided", "no_guide", "no_hint", "l0", "l1", "l2"]),
+              default="guided", show_default=True)
+@click.option("--keep-run-artifacts", is_flag=True)
+@_difficulty_common_options
+def difficulty_range(scenario_dir, agent_context, keep_run_artifacts, output, api_key,
+                     base_url, max_turns, timeout, models):
+    """Evaluate a generated Range scenario."""
+    from clab_builder.evaluation.difficulty import build_report, write_report
+    from clab_builder.evaluation.range_evaluator import evaluate_range
+
+    # 默认四个 Qwen 模型；显式覆盖时仍强制要求四个，避免不完整比较。
+    model_list = tuple(item.strip() for item in models.split(",") if item.strip())
+    if len(model_list) != 4:
+        raise click.ClickException("--models must contain exactly four models")
+    runs, valid, isolated = evaluate_range(
+        scenario_dir, models=model_list, api_key=api_key, base_url=base_url,
+        max_turns=max_turns, timeout=timeout, agent_context=agent_context,
+        keep_artifacts=keep_run_artifacts,
+    )
+    report = build_report(
+        kind="range", subject=scenario_dir, runs=runs,
+        environment_valid=valid, state_isolated=isolated,
+        config={"models": model_list, "max_turns": max_turns, "timeout_s": timeout,
+                "agent_context": agent_context, "runner": "openai_newapi"},
+    )
+    write_report(output, report)
+    click.echo(f"Difficulty report: {output}")
+
+
+@difficulty.command("atom")
+@click.argument("atom_dir", type=click.Path(exists=True, file_okay=False))
+@click.option("--container", required=True, help="Running target container name or ID")
+@click.option("--target-ip", required=True)
+@click.option("--reset-command", default="", help="Optional command run before each model")
+@_difficulty_common_options
+def difficulty_atom(atom_dir, container, target_ip, reset_command, output, api_key,
+                    base_url, max_turns, timeout, models):
+    """Evaluate a generated Atom against an already running target."""
+    import subprocess
+    from clab_builder.evaluation.atom_evaluator import evaluate_atom
+    from clab_builder.evaluation.difficulty import build_report, write_report
+
+    model_list = tuple(item.strip() for item in models.split(",") if item.strip())
+    if len(model_list) != 4:
+        raise click.ClickException("--models must contain exactly four models")
+    # reset-command 由用户提供，典型用途是重启/重建 Atom 目标容器。
+    reset = (lambda: subprocess.run(reset_command, shell=True, check=True, timeout=300)) \
+        if reset_command else None
+    runs, valid, isolated = evaluate_atom(
+        atom_dir, container=container, target_ip=target_ip, models=model_list,
+        api_key=api_key, base_url=base_url, max_turns=max_turns, timeout=timeout, reset=reset,
+    )
+    report = build_report(
+        kind="atom", subject=atom_dir, runs=runs, environment_valid=valid,
+        state_isolated=isolated,
+        config={"models": model_list, "max_turns": max_turns, "timeout_s": timeout,
+                "runner": "openai_newapi"},
+    )
+    write_report(output, report)
+    click.echo(f"Difficulty report: {output}")
+
+
 # ── Generate (only generate files, no deploy) ───────────────────────────
 
 @main.command("generate")

src/clab_builder/evaluation/__init__.py

@@ -0,0 +1,17 @@
+"""Independent empirical difficulty evaluation for Atom and Range artifacts."""
+
+from .difficulty import (
+    DEFAULT_MODELS,
+    EvaluationRun,
+    aggregate_runs,
+    classify_difficulty,
+    write_report,
+)
+
+__all__ = [
+    "DEFAULT_MODELS",
+    "EvaluationRun",
+    "aggregate_runs",
+    "classify_difficulty",
+    "write_report",
+]

src/clab_builder/evaluation/atom_evaluator.py

@@ -0,0 +1,103 @@
+"""Atom adapter for an already running single-CVE environment."""
+
+from __future__ import annotations
+
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import Callable
+
+import yaml
+
+from clab_builder.atomizer.agent.researcher import CVEInput, SecurityResearcherAgent
+
+from .difficulty import EvaluationRun, session_metrics, timed_run
+
+
+def evaluate_atom(
+    atom_dir: str,
+    *,
+    container: str,
+    target_ip: str,
+    models: tuple[str, ...],
+    api_key: str,
+    base_url: str,
+    max_turns: int,
+    timeout: int,
+    reset: Callable[[], None] | None = None,
+) -> tuple[list[EvaluationRun], bool, bool]:
+    """让四个模型连接到同一个已启动的 Atom 目标容器。
+
+    A reset callback is required for strict isolation. Without one we still
+    collect measurements, but mark the report as state-isolated=false.
+
+    Atom 当前没有像 Range 那样的统一 deploy/destroy wrapper，因此 CLI 要求
+    调用者提供目标容器；`reset` 用来在模型之间恢复目标状态。
+    """
+    root = Path(atom_dir).resolve()
+    atom = yaml.safe_load((root / "atom.yaml").read_text(encoding="utf-8")) or {}
+    cve_id = str(atom.get("cve_id") or root.name)
+    runtime = atom.get("runtime_spec") or {}
+    ports = [int(p) for p in runtime.get("ports", [])]
+    description = str(atom.get("description") or atom.get("vulnerability_type") or cve_id)
+    writeup = ""
+    for candidate in (root / "README.md", root / "exploit_guide.yaml"):
+        if candidate.is_file():
+            writeup = candidate.read_text(encoding="utf-8", errors="replace")
+            break
+
+    runs: list[EvaluationRun] = []
+    # Agent 容器需要加入目标所在的 Docker network，而不是加入目标容器本身。
+    inspect = subprocess.run(
+        ["docker", "inspect", "--format", "{{json .NetworkSettings.Networks}}", container],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if inspect.returncode != 0:
+        raise RuntimeError(f"cannot inspect target container: {inspect.stderr.strip()}")
+    import json
+    networks = json.loads(inspect.stdout or "{}")
+    network_name = next(iter(networks), "")
+    if not network_name:
+        raise RuntimeError("target container has no Docker network")
+    workspace_root = Path(tempfile.mkdtemp(prefix="cvelab-difficulty-atom-"))
+    try:
+        for index, model in enumerate(models, 1):
+            # 没有 reset 时仍允许做探索性实验，但报告会明确标记隔离不完整。
+            if reset:
+                reset()
+            workspace = workspace_root / f"run-{index}"
+            workspace.mkdir()
+            agent = SecurityResearcherAgent(max_turns=max_turns, agent_timeout=timeout)
+            agent.start(network_name, str(workspace), api_key, base_url, model)
+            try:
+                output, elapsed = timed_run(
+                    agent.run,
+                    CVEInput(
+                        cve_id=cve_id,
+                        description=description,
+                        target_ip=target_ip,
+                        target_ports=ports,
+                        writeup=writeup,
+                    ),
+                    str(workspace),
+                )
+                metrics = session_metrics(workspace / "session.json")
+                runs.append(EvaluationRun(
+                    model=model,
+                    success=bool(output.success),
+                    turns=metrics["turns"],
+                    tool_calls=metrics["tool_calls"],
+                    wall_time_s=elapsed,
+                    termination_reason="completed" if output.success else "agent_failed",
+                    verifier={"environment_valid": True, "agent_success": bool(output.success)},
+                ))
+            except Exception as exc:  # noqa: BLE001
+                runs.append(EvaluationRun(model=model, error=str(exc)))
+            finally:
+                agent.stop()
+    finally:
+        shutil.rmtree(workspace_root, ignore_errors=True)
+    return runs, True, reset is not None