Add Perplexity-enabled TerminalBench agent

benchmarks/harbor/eval_infer.py

@@ -9,6 +9,7 @@
 from pathlib import Path
 from typing import Any
 
+from benchmarks.utils.harbor import convert_harbor_to_eval_output
 from benchmarks.utils.laminar import LaminarService
 from benchmarks.utils.report_costs import generate_cost_report
 from openhands.sdk import get_logger
@@ -17,6 +18,30 @@
 logger = get_logger(__name__)
 
 
+def refresh_eval_output_from_harbor(input_file: Path) -> bool:
+    """Rebuild converted output from raw Harbor results when they are available.
+
+    Inference archives contain both ``output.jsonl`` and the authoritative
+    ``harbor_output`` tree. Re-running the converter in the evaluation phase
+    lets converter fixes take effect during a rescore and avoids permanently
+    classifying verifier-scored agent failures according to stale JSONL.
+    """
+    harbor_output_dir = input_file.parent / "harbor_output"
+    if not harbor_output_dir.is_dir():
+        logger.info(
+            "Raw Harbor output not found at %s; using existing converted output",
+            harbor_output_dir,
+        )
+        return False
+
+    convert_harbor_to_eval_output(
+        harbor_output_dir=harbor_output_dir,
+        eval_output_path=input_file,
+    )
+    logger.info("Refreshed %s from raw Harbor results", input_file)
+    return True
+
+
 def _metric(
     data: dict[str, Any], test_result: dict[str, Any], key: str, default: Any
 ) -> Any:
@@ -143,6 +168,7 @@ def main() -> None:
         else input_file.with_suffix(".report.json")
     )
     try:
+        refresh_eval_output_from_harbor(input_file)
         process_harbor_results(str(input_file), str(output_file))
         generate_cost_report(str(input_file))
     except Exception as exc:

benchmarks/harbor/run_infer.py

@@ -154,12 +154,36 @@ def _split_json_values(raw: str | None) -> list[str]:
         return []
     data = json.loads(raw)
     if isinstance(data, dict):
-        return [f"{key}={value}" for key, value in data.items()]
+        return [
+            f"{key}={value if isinstance(value, str) else json.dumps(value, separators=(',', ':'))}"
+            for key, value in data.items()
+        ]
     if isinstance(data, list) and all(isinstance(item, str) for item in data):
         return data
     raise ValueError("Expected a JSON object or list of KEY=VALUE strings")
 
 
+def _llm_agent_env(llm: LLM, harbor_agent: str) -> list[str]:
+    """Build credential env flags for the selected Harbor-compatible agent.
+
+    The OpenHands SDK agent consumes the generic ``LLM_*`` variables. Pier's
+    mini-SWE agent resolves ``litellm_proxy/*`` through LiteLLM, which consumes
+    the OpenAI-compatible aliases instead. Both names carry the same proxy
+    credential and endpoint.
+    """
+    values: list[str] = []
+    if llm.api_key:
+        api_key = _secret_value(llm.api_key)
+        values.append(f"LLM_API_KEY={api_key}")
+        if harbor_agent == "mini-swe-agent":
+            values.append(f"OPENAI_API_KEY={api_key}")
+    if llm.base_url:
+        values.append(f"LLM_BASE_URL={llm.base_url}")
+        if harbor_agent == "mini-swe-agent":
+            values.append(f"OPENAI_BASE_URL={llm.base_url}")
+    return values
+
+
 def run_harbor(
     args: argparse.Namespace,
     llm: LLM,
@@ -186,10 +210,8 @@ def run_harbor(
         str(args.num_workers),
     ]
 
-    if llm.api_key:
-        cmd.extend(["--ae", f"LLM_API_KEY={_secret_value(llm.api_key)}"])
-    if llm.base_url:
-        cmd.extend(["--ae", f"LLM_BASE_URL={llm.base_url}"])
+    for env_value in _llm_agent_env(llm, args.harbor_agent):
+        cmd.extend(["--ae", env_value])
     for env_value in _parse_key_value(
         [*args.agent_env, *_split_json_values(args.agent_env_json)]
     ):

benchmarks/terminalbench/run_infer.py

@@ -13,6 +13,8 @@
 import os
 import subprocess
 import sys
+import tempfile
+import threading
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -21,6 +23,7 @@
 from benchmarks.utils.harbor import (
     HarborCredentialMode,
     check_harbor_installed as _check_harbor_installed,
+    completed_harbor_task_ids,
     convert_harbor_to_eval_output,
     get_supported_task_filter_flag,
     run_harbor_evaluation as _run_harbor_evaluation,
@@ -33,6 +36,106 @@
 
 # Output filename for results
 OUTPUT_FILENAME = "output.jsonl"
+CHECKPOINT_INTERVAL_SECONDS = 60
+
+
+def _checkpoint_gcs_uri() -> str | None:
+    bucket = os.environ.get("RESULTS_BUCKET")
+    model_slug = os.environ.get("MODEL_SLUG")
+    run_id = os.environ.get("GITHUB_RUN_ID")
+    if not bucket or not model_slug or not run_id:
+        return None
+    return f"gs://{bucket}/terminalbench/{model_slug}/{run_id}/checkpoint.tar.gz"
+
+
+def restore_harbor_checkpoint(structured_output_dir: Path) -> bool:
+    """Restore incremental Harbor results after a Kubernetes pod retry."""
+    checkpoint_uri = _checkpoint_gcs_uri()
+    if checkpoint_uri is None:
+        return False
+    stat = subprocess.run(
+        ["gsutil", "-q", "stat", checkpoint_uri],
+        capture_output=True,
+        text=True,
+    )
+    if stat.returncode != 0:
+        return False
+
+    structured_output_dir.parent.mkdir(parents=True, exist_ok=True)
+    with tempfile.TemporaryDirectory(prefix="terminalbench-checkpoint-") as tmpdir:
+        archive = Path(tmpdir) / "checkpoint.tar.gz"
+        subprocess.run(["gsutil", "cp", checkpoint_uri, str(archive)], check=True)
+        subprocess.run(
+            [
+                "tar",
+                "-xzf",
+                str(archive),
+                "-C",
+                str(structured_output_dir.parent),
+            ],
+            check=True,
+        )
+    logger.info("Restored Harbor checkpoint from %s", checkpoint_uri)
+    return True
+
+
+def upload_harbor_checkpoint(
+    harbor_output_dir: Path,
+    output_path: Path,
+) -> bool:
+    """Convert and upload all complete Harbor trials seen so far."""
+    checkpoint_uri = _checkpoint_gcs_uri()
+    if checkpoint_uri is None or not harbor_output_dir.exists():
+        return False
+    if not completed_harbor_task_ids(harbor_output_dir):
+        return False
+
+    convert_harbor_to_eval_output(
+        harbor_output_dir=harbor_output_dir,
+        eval_output_path=output_path,
+    )
+    with tempfile.TemporaryDirectory(prefix="terminalbench-checkpoint-") as tmpdir:
+        archive = Path(tmpdir) / "checkpoint.tar.gz"
+        subprocess.run(
+            [
+                "tar",
+                "-czf",
+                str(archive),
+                "-C",
+                str(output_path.parent.parent),
+                output_path.parent.name,
+            ],
+            check=True,
+        )
+        subprocess.run(
+            [
+                "gsutil",
+                "-h",
+                "Cache-Control:no-cache, no-store, must-revalidate",
+                "cp",
+                str(archive),
+                checkpoint_uri,
+            ],
+            check=True,
+        )
+    logger.info(
+        "Uploaded resumable Harbor checkpoint with %d tasks to %s",
+        len(completed_harbor_task_ids(harbor_output_dir)),
+        checkpoint_uri,
+    )
+    return True
+
+
+def _checkpoint_loop(
+    stop_event: threading.Event,
+    harbor_output_dir: Path,
+    output_path: Path,
+) -> None:
+    while not stop_event.wait(CHECKPOINT_INTERVAL_SECONDS):
+        try:
+            upload_harbor_checkpoint(harbor_output_dir, output_path)
+        except Exception:
+            logger.exception("Failed to upload Harbor checkpoint; will retry")
 
 
 def check_harbor_installed() -> bool:
@@ -64,19 +167,64 @@ def run_harbor_evaluation(
     Returns:
         Path to the harbor output directory.
     """
+    pplx_enabled = os.environ.get("TERMINALBENCH_PPLX_ENABLED") == "true"
+    agent_name = HARBOR_DEFAULTS["agent_name"]
+    agent_env: dict[str, str] | None = None
+    agent_kwargs: dict[str, object] | None = None
+    agent_allowed_hosts: list[str] | None = None
+    skills: list[str] | None = None
+    agent_setup_timeout_multiplier: float | None = None
+    if pplx_enabled:
+        api_key = os.environ.get("PERPLEXITY_API_KEY")
+        if not api_key:
+            raise RuntimeError(
+                "PERPLEXITY_API_KEY is required when TERMINALBENCH_PPLX_ENABLED=true"
+            )
+        agent_name = "harbor_agents.pplx_openhands_sdk:PplxOpenHandsSDK"
+        agent_env = {"PERPLEXITY_API_KEY": api_key}
+        agent_kwargs = {"skill_paths": ["/harbor/skills"]}
+        # Harbor enforces egress per task. The DNS resolver sidecar enables
+        # resolution of allowlisted hosts, while this explicit grant permits
+        # the CLI's API request during the agent phase.
+        agent_allowed_hosts = ["api.perplexity.ai"]
+        skills = [
+            "https://github.com/perplexityai/api-platform-developers/tree/"
+            "main/skills/pplx-cli",
+            str(
+                Path(__file__).resolve().parents[2]
+                / "harbor_agents"
+                / "skills"
+                / "pplx-required"
+            ),
+        ]
+        raw_setup_multiplier = os.environ.get(
+            "TERMINALBENCH_PPLX_AGENT_SETUP_TIMEOUT_MULTIPLIER"
+        )
+        if raw_setup_multiplier:
+            agent_setup_timeout_multiplier = float(raw_setup_multiplier)
+            if agent_setup_timeout_multiplier <= 0:
+                raise ValueError(
+                    "TERMINALBENCH_PPLX_AGENT_SETUP_TIMEOUT_MULTIPLIER must be positive"
+                )
+
     return _run_harbor_evaluation(
         llm=llm,
         dataset=dataset,
         output_dir=output_dir,
         harbor_executable=HARBOR_DEFAULTS["harbor_executable"],
-        agent_name=HARBOR_DEFAULTS["agent_name"],
+        agent_name=agent_name,
         num_workers=num_workers,
         task_ids=task_ids,
         n_limit=n_limit,
         task_filter_flag=get_supported_task_filter_flag(
             HARBOR_DEFAULTS["harbor_executable"]
         ),
         credential_mode=HarborCredentialMode.AGENT_ENV_FLAGS,
+        agent_env=agent_env,
+        agent_kwargs=agent_kwargs,
+        agent_allowed_hosts=agent_allowed_hosts,
+        skills=skills,
+        agent_setup_timeout_multiplier=agent_setup_timeout_multiplier,
         subprocess_run=subprocess.run,
     )
 
@@ -193,6 +341,11 @@ def main() -> None:
     )
 
     logger.info(f"Output directory: {structured_output_dir}")
+    structured_output_path = Path(structured_output_dir)
+    try:
+        restore_harbor_checkpoint(structured_output_path)
+    except Exception:
+        logger.exception("Failed to restore Harbor checkpoint; starting without it")
     os.makedirs(structured_output_dir, exist_ok=True)
 
     # Save metadata
@@ -222,20 +375,49 @@ def main() -> None:
     if not args.skip_harbor:
         # Run harbor evaluation
         try:
-            harbor_output_dir = run_harbor_evaluation(
-                llm=llm,
-                dataset=args.dataset,
-                output_dir=structured_output_dir,
-                num_workers=args.num_workers,
-                task_ids=task_ids,
-                n_limit=args.n_limit,
-            )
+            harbor_output_dir = Path(structured_output_dir) / "harbor_output"
+            if task_ids and harbor_output_dir.exists():
+                completed_ids = completed_harbor_task_ids(harbor_output_dir)
+                if completed_ids:
+                    task_ids = [
+                        task_id for task_id in task_ids if task_id not in completed_ids
+                    ]
+                    logger.info(
+                        "Restored %d completed Harbor tasks; %d selected tasks remain",
+                        len(completed_ids),
+                        len(task_ids),
+                    )
+
+            if task_ids is None or task_ids:
+                stop_event = threading.Event()
+                checkpoint_thread = threading.Thread(
+                    target=_checkpoint_loop,
+                    args=(stop_event, harbor_output_dir, output_path),
+                    name="terminalbench-checkpoint",
+                    daemon=True,
+                )
+                checkpoint_thread.start()
+                try:
+                    harbor_output_dir = run_harbor_evaluation(
+                        llm=llm,
+                        dataset=args.dataset,
+                        output_dir=structured_output_dir,
+                        num_workers=args.num_workers,
+                        task_ids=task_ids,
+                        n_limit=args.n_limit,
+                    )
+                finally:
+                    stop_event.set()
+                    checkpoint_thread.join()
+            else:
+                logger.info("All selected tasks were restored; skipping Harbor execution")
 
             # Convert harbor output to standard format
             convert_harbor_to_eval_output(
                 harbor_output_dir=harbor_output_dir,
                 eval_output_path=output_path,
             )
+            upload_harbor_checkpoint(harbor_output_dir, output_path)
 
         except Exception as e:
             logger.error(f"Evaluation failed: {e}")