my-task/
├── instruction.md       # 任务指令（给 agent 看）
├── task.toml            # 配置 & 元数据
├── environment/         # 环境定义
│   └── Dockerfile
├── solution/            # （可选）Oracle 解决方案
│   └── solve.sh
└── tests/               # 验证脚本
    └── test.sh

schema_version = "1.4"

[task]
name = "org/name"
version = "1.0.0"
description = "..."
authors = [{ name = "...", email = "..." }]
keywords = ["programming"]

[metadata]
difficulty_explanation = "..."
category = "programming"

[verifier]
timeout_sec = 120.0
env = { API_KEY = "sk-test-123" }
user = "root"

[agent]
timeout_sec = 120.0
user = "agent"

[environment]
network_mode = "no-network"     # public | no-network | allowlist
build_timeout_sec = 600.0
docker_image = "some-org/some-name:some-tag"
os = "linux"                    # 或 "windows"
cpus = 1
memory_mb = 2048
storage_mb = 10240
gpus = 0
gpu_types = ["H100", "A100"]

[environment.tpu]               # 可选
type = "v6e"
topology = "2x4"

[[environment.mcp_servers]]
name = "mcp-server"
transport = "streamable-http"
url = "http://mcp-server:8000/mcp"

[environment.healthcheck]
command = "curl -f http://localhost:8080/health"
interval_sec = 5.0
timeout_sec = 30.0
retries = 3

# Installed Agent
from harbor.agents.installed.base import BaseInstalledAgent, with_prompt_template

class MyAgent(BaseInstalledAgent):
    async def install(self, environment):
        await self.exec_as_root(environment, "apt-get update && apt-get install -y curl")
        await self.exec_as_agent(environment, "pip install my-agent")

    @with_prompt_template
    async def run(self, instruction, environment, context):
        await self.exec_as_agent(environment, f"my-agent run {shlex.quote(instruction)}")

    def populate_context_post_run(self, context):
        pass

harbor run -d "dataset@version" --agent path.to.agent:MyAgent

# YAML 配置文件
harbor run -c "path/to/job.yaml"

# 或 adhoc 模式
harbor run -d "dataset@version" -m "model" -a "agent"

AgentConfig(
    name=AgentName.TERMINUS_2,
    model_name="openai/gpt-5",
    kwargs={
        "parser_name": "json",              # json | xml
        "temperature": 0.7,
        "max_turns": 100,
        "enable_summarize": True,            # 上下文摘要
        "proactive_summarization_threshold": 8000,
        "collect_rollout_details": False,    # RL 训练用
        "reasoning_effort": "medium",        # none..max
        "max_thinking_tokens": 2048,         # Anthropic extended thinking
        "session_id": "auto",               # 用于 prefix cache 路由
    }
)

Previous History → 1. Summary Subagent → 2. Question Subagent → 3. Answer Subagent → Compressed History

uv tool install harbor-rewardkit
# 或带图片/文档解析能力
uv tool install harbor-rewardkit[all]

# tests/test.sh
uvx --with harbor-rewardkit@0.1 rewardkit /tests

import rewardkit as rk

# 内置 criteria（20+）
rk.file_exists("output.txt")
rk.file_contains("output.txt", "hello")
rk.command_succeeds("python main.py")

# 自定义 criteria
from rewardkit import criterion

@criterion
def has_valid_output(workspace) -> bool:
    output = (workspace / "output.txt").read_text()
    return len(output.splitlines()) >= 10

# 带参数
@criterion(description="output has at least {n} lines")
def has_n_lines(workspace, n: int) -> bool:
    return len((workspace / "output.txt").read_text().splitlines()) >= n

rk.has_n_lines(10, weight=2.0)
rk.has_n_lines(50, weight=1.0)

# tests/quality.toml
[judge]
judge = "anthropic/claude-sonnet-4-6"
files = ["/app/main.py"]

[[criterion]]
description = "Is the code correct?"
type = "binary"

[[criterion]]
description = "How readable is the code?"
type = "likert"
points = 5

# tests/structure/reward.toml
weight = 2.0

[scoring]
aggregation = "all-pass"  # weighted-mean | all-pass | any-pass | threshold | required-pass
threshold = 0.5

SkyRL Trainer (PPO) ←→ HarborGenerator ←→ Harbor Trial
     ↑                      ↓
  Policy Update      Harbor Trial Lifecycle:
                     1. Start Environment (Docker/Daytona/Modal)
                     2. Run Agent (Terminus-2 → vLLM HTTP)
                     3. Run Verifier (test.sh → reward.txt)
                     4. Cleanup

# 安装
uv tool install harbor        # 推荐
pip install harbor

# 运行 eval
harbor run -d "terminal-bench@2.0" -a claude-code -m anthropic/claude-opus-4-1

# 本地 task
harbor run -p "path/to/task" -a oracle

# 云沙箱（水平扩展）
harbor run -d "terminal-bench@2.0" -a claude-code -m anthropic/claude-opus-4-1 \
  --env daytona -n 100

# 查看结果（Web UI）
harbor view

# 列出可用数据集
harbor datasets list

# 列出可用 agents
harbor run --help | grep -A 20 "--agent"

# 创建新 task
harbor init --task "org/name"

# 恢复中断的 job
harbor jobs resume --job-path ~/.harbor/jobs/<job-name>

# AI 分析失败原因
harbor jobs summarize ~/.harbor/jobs/<job-name> --model sonnet

# 导出训练数据
harbor run ... --export-traces --export-sharegpt

# Job 配置文件
harbor run -c "path/to/job.yaml"

# 1. 安装（推荐 uv）
uv tool install harbor

# 2. 验证安装
harbor --version

# 3. 跑一个本地简单 task（Oracle agent 验证任务可解）
harbor run -p examples/tasks/hello-world -a oracle

# 4. 用真实 agent 跑
harbor run -p examples/tasks/hello-world -a claude-code -m anthropic/claude-sonnet-4-5

# 5. 查看 Web UI 结果
harbor view

harbor task init ssh-key-pair
cd ssh-key-pair

ssh-key-pair/
├── instruction.md
├── task.toml
├── environment/
│   └── Dockerfile
├── solution/
│   └── solve.sh
└── tests/
    ├── test.sh
    └── test_outputs.py

<!-- instruction.md -->
Generate an SSH key pair using the ssh-keygen command.
Save the private key to /app/id_rsa and the public key to /app/id_rsa.pub.
Use the following options:
- Key type: RSA
- Key size: 4096 bits
- No passphrase (empty passphrase)
- Comment: "harbor-test-key"

# task.toml
[metadata]
difficulty_explanation = "Simple SSH key generation command"
category = "system-administration"
tags = ["ssh", "cryptography", "linux"]

[verifier]
timeout_sec = 120.0

[agent]
timeout_sec = 120.0

[environment]
build_timeout_sec = 600.0

# 启动交互式容器测试
harbor task start-env -p ssh-key-pair -e docker -a -i

#!/bin/bash
# solution/solve.sh
ssh-keygen -t rsa -b 4096 -f /app/id_rsa -N "" -C "harbor-test-key"

#!/bin/bash
# tests/test.sh
cd /app
pip install pytest -q
python -m pytest /tests/test_outputs.py -v

# tests/test_outputs.py
import subprocess

def test_private_key_exists():
    result = subprocess.run(["ssh-keygen", "-l", "-f", "/app/id_rsa"],
                          capture_output=True, text=True)
    assert result.returncode == 0
    assert "4096" in result.stdout  # 4096-bit key

def test_public_key_exists():
    with open("/app/id_rsa.pub") as f:
        content = f.read()
    assert "harbor-test-key" in content

harbor run -p ssh-key-pair -a oracle
# 成功应显示 reward = 1

harbor run -p ssh-key-pair -a terminus-2 -m anthropic/claude-haiku-4-5

harbor view
# 在 Web UI 中浏览：agent 每步操作、verifier 输出、最终 reward

harbor run -p harbor_cookbook/recipes/<name> -a claude-code -m anthropic/claude-opus-4-6

steps/
├── scaffold/          # Step 1: 创建 /app/greet.sh
│   ├── instruction.md
│   ├── tests/test.sh
│   └── solution/solve.sh
├── implement/         # Step 2: 读取 config.txt，打印 "Hello, <name>!"
│   ├── instruction.md
│   ├── tests/test.sh
│   ├── workdir/
│   │   ├── config.txt       # per-step 文件上传
│   │   └── setup.sh         # per-step 预处理脚本（运行后自删除）
│   └── solution/solve.sh
└── document/          # Step 3: 写 README.md
    ├── instruction.md
    ├── tests/test.sh
    └── solution/solve.sh

[[steps]]
name = "scaffold"
min_reward = 1.0          # 低于 1.0 立即终止后续步骤（没有脚手架就没法继续）

[steps.agent]
timeout_sec = 60.0

[steps.verifier]
timeout_sec = 30.0

[[steps]]
name = "implement"
min_reward = 0.5          # 允许部分完成

[steps.agent]
timeout_sec = 120.0

[steps.verifier]
timeout_sec = 30.0
env = { EXPECTED_NAME = "World", EXPECTED_GREETING = "Hello, World!" }

[[steps]]
name = "document"

[steps.agent]
timeout_sec = 60.0

# Oracle 验证
harbor run -p harbor_cookbook/recipes/multi-step --agent oracle

# 真实 agent
harbor run -p harbor_cookbook/recipes/multi-step --agent claude-code --model anthropic/claude-sonnet-4-6

services:
  main:
    depends_on:
      mcp-server:
        condition: service_healthy

  mcp-server:
    build:
      context: ./mcp-server
    expose:
      - "8000"
    healthcheck:
      test: ["CMD", "python", "-c", "import socket; ..."]
      interval: 2s
      timeout: 5s
      retries: 15

[[environment.mcp_servers]]
name = "mcp-server"
transport = "streamable-http"   # sse | streamable-http | stdio
url = "http://mcp-server:8000/mcp"

[[environment.mcp_servers]]
name = "user-server"
transport = "streamable-http"
url = "http://user-server:8000/mcp"

# openai_agent.py
import os, json
from openai import AsyncOpenAI
from harbor.agents.base import BaseAgent
from harbor.environments.base import BaseEnvironment
from harbor.models.agent.context import AgentContext

class OpenAIAgent(BaseAgent):
    def __init__(self, max_iterations: int = 10, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._max_iterations = max_iterations
        self._client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

    @staticmethod
    def name() -> str:
        return "openai-agent"

    def version(self) -> str | None:
        return "1.0.0"

    async def setup(self, environment: BaseEnvironment) -> None:
        pass  # 无需安装

    async def run(self, instruction, environment, context):
        messages = [
            {"role": "system", "content": "You are a helpful assistant that solves tasks."},
            {"role": "user", "content": instruction}
        ]

        total_input_tokens = 0
        total_output_tokens = 0
        iterations = 0

        for i in range(self._max_iterations):
            iterations += 1
            response = await self._client.chat.completions.create(
                model=self.model_name or "gpt-4",
                messages=messages
            )

            if response.usage:
                total_input_tokens += response.usage.prompt_tokens
                total_output_tokens += response.usage.completion_tokens

            message = response.choices[0].message
            messages.append(message.model_dump())

            if message.tool_calls:
                for tool_call in message.tool_calls:
                    if tool_call.function.name == "execute_bash":
                        args = json.loads(tool_call.function.arguments)
                        result = await environment.exec(args["command"])
                        messages.append({
                            "role": "tool",
                            "tool_call_id": tool_call.id,
                            "content": result.stdout or result.stderr or ""
                        })

            if message.content and "task complete" in message.content.lower():
                break

        context.commands_executed = iterations
        context.n_input_tokens = total_input_tokens
        context.n_output_tokens = total_output_tokens
        context.exit_code = 0

export OPENAI_API_KEY=<YOUR-KEY>
harbor run --tasks ./my-task \
  --agent openai_agent:OpenAIAgent \
  --agent-import-path openai_agent.py \
  --model gpt-4 \
  --ak max_iterations:int=20

# 下载 Terminal-Bench 2.0 任务
uvx harbor datasets download terminal-bench@2.0

# 启动 RL 训练
uv run python tinker_cookbook/recipes/harbor_rl/scripts/train_terminal_bench.py \
    model_name=moonshotai/Kimi-K2.6 \
    max_tokens=8192 \
    group_size=4 \
    groups_per_batch=8 \
    learning_rate=1e-5 \
    lora_rank=32 \
    wandb_project=cookbook_harbor_rl

python tinker_cookbook/recipes/terminal_rl/train.py \
  --task-dir harbor_envs/devops_task \
  --env docker \
  --model-name Qwen/Qwen3-30B-A3B-Instruct-2507 \
  --renderer-name qwen3 \
  --max-tokens 4096 \
  --temperature 0.7 \
  --groups-per-batch 4 \
  --group-size 8 \
  --num-batches 25 \
  --learning-rate 4e-5 \
  --lora-rank 32

# 使用 SkyRL + Daytona 云沙箱做大规模 RL
# 详见 12.7 节完整示例

# 在 MedAgentBench 上优化 Codex agent 的 prompt
uv run harbor_cookbook/gepa/optimize.py

tests/
├── test.sh                 # 入口脚本
├── check.py                # 编程式 criteria
├── quality.toml            # LLM Judge criteria
└── structure/
    ├── check.py            # 子目录 criteria
    └── reward.toml         # 聚合配置

#!/bin/bash
uvx --with harbor-rewardkit@0.1 rewardkit /tests

import rewardkit as rk
from rewardkit import criterion
from pathlib import Path

# 内置 criteria
rk.file_exists("/app/main.py")
rk.file_contains("/app/main.py", "def solve")
rk.command_succeeds("python /app/main.py --test")

# 自定义 criteria
@criterion
def no_hardcoded_secrets(workspace: Path) -> bool:
    code = (workspace / "main.py").read_text()
    return "sk-" not in code and "password" not in code.lower()

# 带参数 + 权重
@criterion(description="at least {n} test cases")
def has_n_test_cases(workspace: Path, n: int) -> bool:
    code = (workspace / "main.py").read_text()
    return code.count("def test_") >= n

rk.has_n_test_cases(3, weight=1.0)
rk.has_n_test_cases(10, weight=0.5)

[judge]
judge = "anthropic/claude-sonnet-4-6"
files = ["/app/main.py"]

[[criterion]]
description = "Does the code handle edge cases correctly?"
type = "binary"

[[criterion]]
description = "How well-structured is the code?"
type = "likert"
points = 5

[[criterion]]
description = "Is the error handling adequate?"
type = "binary"

weight = 2.0   # 此 group 在维度中占比

[scoring]
aggregation = "all-pass"  # 任一检查失败 → 整组 0 分

// /logs/verifier/reward.json
{ "reward": 0.75 }

export ANTHROPIC_API_KEY=<KEY>
export DAYTONA_API_KEY=<KEY>

# 100 并发跑 Terminal-Bench 2.0
harbor run \
  --dataset terminal-bench@2.0 \
  --agent claude-code \
  --model anthropic/claude-opus-4-1 \
  --n-concurrent 100 \
  --env daytona

export AWS_ACCESS_KEY_ID=<KEY>
export AWS_REGION=us-east-1

harbor run \
  --dataset terminal-bench@2.0 \
  --agent claude-code \
  --model anthropic/claude-opus-4-1 \
  --n-concurrent 32 \
  --env ec2 \
  --ae AWS_ACCESS_KEY_ID=$AWS_ACCESS_KEY_ID \
  --ae AWS_REGION=us-east-1

# Job 配置文件方式：同时跑多个 agent + model 组合
harbor run -c job.yaml

# job.yaml 示例
environment:
  type: daytona

agents:
  - name: claude-code
    model: anthropic/claude-opus-4-1
  - name: openhands
    model: anthropic/claude-sonnet-4
  - name: terminus-2
    model: openai/gpt-5

datasets:
  - terminal-bench@2.0

orchestrator:
  n_concurrent_trials: 32

from harbor.models.trial.config import JobConfig, AgentConfig, EnvironmentConfig
from harbor.models.agent_name import AgentName
from harbor.models.environment_type import EnvironmentType
from harbor.job import Job
from pathlib import Path

class HarborRolloutInterface:
    async def run(self, task_ids: list[str]) -> list[dict]:
        # 构造 JobConfig
        task_configs = [...]  # 从 task_ids 构造 TaskConfig 列表

        job = Job(
            config=JobConfig(
                jobs_dir=Path("jobs"),
                environment=EnvironmentConfig(
                    type=EnvironmentType.DAYTONA,
                ),
                agents=[
                    AgentConfig(
                        name=AgentName.TERMINUS_2,
                        model_name="hosted_vllm/my-model",
                        kwargs={
                            "base_url": "https://vllm-server:8000",
                            "collect_rollout_details": True,  # 收集 token_ids + logprobs
                        }
                    )
                ],
                orchestrator=OrchestratorConfig(
                    n_concurrent_trials=32,
                ),
                tasks=task_configs,
            ),
        )

        result = await job.run()

        rollouts = []
        for trial_result in result.trial_results:
            # 提取 reward
            reward = (
                trial_result.verifier_result.rewards.get("reward", 0)
                if trial_result.verifier_result and trial_result.verifier_result.rewards
                else 0
            )

            # 提取 token_ids 和 mask_ids（用于 RL 训练）
            if (trial_result.agent_result
                and trial_result.agent_result.metadata
                and "token_ids" in trial_result.agent_result.metadata
                and "mask_ids" in trial_result.agent_result.metadata):
                token_ids = trial_result.agent_result.metadata["token_ids"]
                mask_ids = trial_result.agent_result.metadata["mask_ids"]
            else:
                raise ValueError(f"Missing token_ids or mask_ids for trial {trial_result.trial_name}")

            rollouts.append({
                "reward": reward,
                "token_ids": token_ids,
                "mask_ids": mask_ids,
            })

        return rollouts

harbor view

# 用 Claude 分析失败原因
harbor jobs summarize ~/.harbor/jobs/<job-name> --model sonnet

# 导出轨迹用于 SFT
harbor run ... --export-traces

# 导出 ShareGPT 格式对话数据
harbor run ... --export-sharegpt

# 中断后恢复未完成的 trial
harbor jobs resume --job-path ~/.harbor/jobs/<job-name>

# 创建 task 的交互式引导
npx skills add harbor-framework/harbor --skill create-task

# 设计 verifier 的交互式引导
npx skills add harbor-framework/harbor --skill rewardkit