codex-app-server/examples/harness2_run.py

"""用真实 codex 任务跑一轮 Harness² 改进（接线层 CLI）。

这会创建真实任务、花真实模型配额——所以默认只打印计划，不建任务：
先跑一遍不带 --yes 看计划，确认项目、任务数与预算无误后，再带
--yes 重跑。第一次请务必用测试项目，不要直接指生产项目。

密钥不走命令行参数（ps 可见）：用环境变量 CODEX_API_KEY。

示例：
    export CODEX_API_KEY='...'
    python3 examples/harness2_run.py --base-url http://127.0.0.1:8765 \\
        --project-id <test-project> --prompt-file task.md \\
        --base-harness-dir ./my-base-harness --source-dir ./task-source \\
        --mode parallel --k 1 --budget 5 --output runs/h2-real-1
    # 看计划无误后，加 --yes 重跑（同一 run_tag 可断点续跑；想从头
    # 重跑就换 --run-tag 或不传，让它自动生成新的时间戳 tag）。

base-harness-dir：一个目录，里面就是 harness 树（AGENTS.md、
.agents/、scripts/、.codex/……），按原样读入；不存在或 AGENTS.md
为空时直接报错，不跑。
source-dir（可选）：原任务需要的源文件，会随 harness 一起 seed 进
每个探针任务的 worktree；与 harness 路径重名时 harness 优先。
"""

import argparse
import os
import sys
import tempfile
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from cloud.harness2 import read_tree  # noqa: E402
from cloud.harness2_tasks import (  # noqa: E402
    HttpTaskClient, preflight_checks, run_improvement_via_tasks,
    validate_run_tag,
)


def _flatten(root: Path) -> dict[str, bytes]:
    out: dict[str, bytes] = {}
    if not root or not root.is_dir():
        return out
    for p in sorted(root.rglob("*")):
        if p.is_file() and not p.is_symlink() and "__pycache__" not in p.parts:
            out[str(p.relative_to(root))] = p.read_bytes()
    return out


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--base-url", required=True, help="控制面地址，如 http://127.0.0.1:8765")
    ap.add_argument("--project-id", required=True)
    ap.add_argument("--prompt", default="")
    ap.add_argument("--prompt-file", default="")
    ap.add_argument("--task-id", default="h2-task-1")
    ap.add_argument("--domain", default="general")
    ap.add_argument("--base-harness-dir", required=True)
    ap.add_argument("--source-dir", default="")
    ap.add_argument("--mode", default="parallel", choices=["parallel", "sequential"])
    ap.add_argument("--k", type=int, default=1)
    ap.add_argument("--budget", type=float, default=None)
    ap.add_argument("--max-sessions", type=int, default=None,
                    help="会话数上限（探针 + 阶段尝试；unpriced 模型上美元门不触发时的兜底）")
    ap.add_argument("--check", action="store_true",
                    help="只做建任务前预检（本地目录/提示/tag + 控制面 health 探针），不建任务")
    ap.add_argument("--run-tag", default="")
    ap.add_argument("--output", default="")
    ap.add_argument("--library-root", default="")
    ap.add_argument("--model", default=None)
    ap.add_argument("--model-probe", default=None)
    ap.add_argument("--model-editor", default=None)
    ap.add_argument("--model-selector", default=None)
    ap.add_argument("--model-reflector", default=None)
    ap.add_argument("--effort", default=None)
    ap.add_argument("--approval", default=None)
    ap.add_argument("--environment-id", default=None)
    ap.add_argument("--executor", default=None)
    ap.add_argument("--no-time-limit", action="store_true")
    ap.add_argument("--timeout", type=int, default=3600)
    ap.add_argument("--yes", action="store_true", help="确认创建真实任务并消费配额")
    args = ap.parse_args()

    if args.check:
        prompt = args.prompt
        if args.prompt_file:
            try:
                prompt = Path(args.prompt_file).read_text(encoding="utf-8")
            except OSError as exc:
                print(f"FAIL: prompt-file 读不到：{exc}", file=sys.stderr)
                return 2
        errors, warnings = preflight_checks(
            base_harness_dir=args.base_harness_dir,
            source_dir=args.source_dir or None, prompt=prompt,
            run_tag=args.run_tag, base_url=args.base_url)
        for e in errors:
            print(f"FAIL: {e}", file=sys.stderr)
        for w in warnings:
            print(f"WARN: {w}")
        if not os.environ.get("CODEX_API_KEY"):
            print("WARN: 未设置 CODEX_API_KEY（建任务时需要）")
        if errors:
            return 2
        print("preflight: 通过" + ("（有告警）" if warnings else ""))
        return 0

    try:
        validate_run_tag(args.run_tag)
    except ValueError as exc:
        print(f"run-tag 非法：{exc}", file=sys.stderr)
        return 2
    api_key = os.environ.get("CODEX_API_KEY")
    if not api_key:
        print("缺少 CODEX_API_KEY 环境变量（密钥不接受命令行参数）", file=sys.stderr)
        return 2
    prompt = args.prompt
    if args.prompt_file:
        prompt = Path(args.prompt_file).read_text(encoding="utf-8")
    if not prompt.strip():
        print("需要 --prompt 或 --prompt-file", file=sys.stderr)
        return 2
    base = read_tree(args.base_harness_dir)
    if not base.get("AGENTS.md", "").strip():
        print(f"base-harness-dir 里没有可用的 AGENTS.md：{args.base_harness_dir}",
              file=sys.stderr)
        return 2
    run_tag = args.run_tag or time.strftime("h2-%Y%m%d-%H%M%S")
    # 任务数（k=1 parallel）：1 base + 2 候选 + 1 终版探针 + 3 编辑
    # + 1 反思 = 8（k=1 时选择 trivial，不建选择任务）；sequential
    # 同为 8；k 越大线性涨，上限 k=3 时 parallel 为 1 + 6 + 3 探针
    # + 9 编辑 + 1 选择 + 1 反思 = 21。
    n_probes = 1 + 3 * args.k  # 两种模式同款：1 base + 每轮 2 候选 + 1 终版
    n_edits = 3 * args.k
    n_select = 0 if args.k == 1 else 1  # k=1 trivial，不建选择任务
    plan = (f"计划：项目 {args.project_id}，任务 {args.task_id}，"
            f"mode={args.mode} k={args.k}，run_tag={run_tag}，"
            f"约 {n_probes} 个探针 + {n_edits} 个编辑 + {n_select} 选择 + 1 反思 = "
            f"{n_probes + n_edits + n_select + 1} 个真实任务，"
            f"budget_usd={args.budget}，max_sessions={args.max_sessions}，"
            f"base 文件 {len(base)} 个。")
    print(plan)
    if not args.yes:
        print("未带 --yes：只打印计划，不创建任何任务。确认无误后加 --yes 重跑。")
        return 0
    work_dir = Path(args.output) if args.output else Path(tempfile.mkdtemp(prefix="h2run-"))
    client = HttpTaskClient(args.base_url, api_key)
    result = run_improvement_via_tasks(
        client, project_id=args.project_id, task_id=args.task_id, prompt=prompt,
        base_harness=base, run_tag=run_tag, work_dir=work_dir, domain=args.domain,
        source_files=_flatten(Path(args.source_dir)) if args.source_dir else None,
        k=args.k, mode=args.mode,
        library_root=Path(args.library_root) if args.library_root else None,
        budget_usd=args.budget, max_sessions=args.max_sessions,
        model=args.model,
        role_models={"probe": args.model_probe, "editor": args.model_editor,
                     "selector": args.model_selector,
                     "reflector": args.model_reflector},
        effort=args.effort,
        approval=args.approval, environment_id=args.environment_id,
        executor=args.executor, no_time_limit=args.no_time_limit,
        timeout_seconds=args.timeout)
    d = result.to_dict()
    chosen = result.mode_result.rollouts.get(result.selection.chosen_label)
    print(f"ok={result.ok} chosen={result.selection.chosen_label} "
          f"fallback={result.selection.fallback} "
          f"fell_back={result.mode_result.fell_back or 'none'} "
          f"budget_exceeded={result.budget_exceeded}")
    print(f"deliverable={(chosen.deliverable_text if chosen else '')[:500]}")
    print(f"total_cost_usd={result.mode_result.total_cost_usd} "
          f"library_entry={result.library_entry}")
    print(f"work_dir={work_dir}")
    print(f"full_result={d}")
    return 0 if result.ok else 1


if __name__ == "__main__":
    raise SystemExit(main())