259 lines
11 KiB
Python
Executable File
259 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Resolve an Orca project or terminal into tester's structured agent config."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
try:
|
||
import yaml
|
||
except ImportError: # pragma: no cover
|
||
print("需要 PyYAML: python3 -m pip install pyyaml", file=sys.stderr)
|
||
raise SystemExit(2)
|
||
|
||
|
||
SUPPORTED_CLI = {"codex", "cursor-agent", "grok", "omp", "pi"}
|
||
|
||
|
||
def run_orca(args: list[str], environment: str | None = None) -> dict[str, Any]:
|
||
command = ["orca", *args]
|
||
if environment and environment != "local":
|
||
command.extend(["--environment", environment])
|
||
command.append("--json")
|
||
proc = subprocess.run(command, capture_output=True, text=True, check=False)
|
||
raw = (proc.stdout or "").strip()
|
||
try:
|
||
data = json.loads(raw)
|
||
except json.JSONDecodeError as exc:
|
||
detail = (proc.stderr or raw or str(exc)).strip()
|
||
raise RuntimeError(f"orca 命令没有返回 JSON: {' '.join(command)}\n{detail}") from exc
|
||
if not data.get("ok"):
|
||
error = data.get("error") or {}
|
||
message = error.get("message") or error.get("code") or "unknown error"
|
||
raise RuntimeError(f"{' '.join(command)}: {message}")
|
||
result = data.get("result")
|
||
return result if isinstance(result, dict) else {}
|
||
|
||
|
||
def environments() -> list[str]:
|
||
result = run_orca(["environment", "list"])
|
||
names = ["local"]
|
||
for item in result.get("environments") or []:
|
||
if isinstance(item, dict) and item.get("name"):
|
||
names.append(str(item["name"]))
|
||
return names
|
||
|
||
|
||
def worktree_from_terminal(terminal: dict[str, Any], environment: str) -> dict[str, Any]:
|
||
worktree_id = str(terminal.get("worktreeId") or "")
|
||
if not worktree_id:
|
||
raise RuntimeError("terminal 没有 worktreeId")
|
||
result = run_orca(["worktree", "show", "--worktree", worktree_id], environment)
|
||
worktree = result.get("worktree")
|
||
if not isinstance(worktree, dict):
|
||
raise RuntimeError("找不到 terminal 对应的 worktree")
|
||
return worktree
|
||
|
||
|
||
def config_from_terminal(handle: str, requested_environment: str | None) -> dict[str, Any]:
|
||
candidates = [requested_environment] if requested_environment else environments()
|
||
errors: list[str] = []
|
||
for environment in candidates:
|
||
if not environment:
|
||
continue
|
||
try:
|
||
result = run_orca(["terminal", "show", "--terminal", handle], environment)
|
||
terminal = result.get("terminal")
|
||
if not isinstance(terminal, dict):
|
||
raise RuntimeError("返回缺少 terminal")
|
||
worktree = worktree_from_terminal(terminal, environment)
|
||
identity = str(terminal.get("agentIdentity") or "")
|
||
cli = identity if identity in SUPPORTED_CLI else "codex"
|
||
return {
|
||
"environment": "" if environment == "local" else environment,
|
||
"host": str(terminal.get("executionHostId") or worktree.get("hostId") or "local"),
|
||
"project": str(worktree.get("projectId") or worktree.get("repoId") or ""),
|
||
"path": str(terminal.get("worktreePath") or worktree.get("path") or ""),
|
||
"role": "test",
|
||
"cli": cli,
|
||
"tier": "standard",
|
||
"model": None,
|
||
"reasoningEffort": "medium",
|
||
"permissionMode": "workspace-write",
|
||
"_source": f"terminal:{handle}",
|
||
"_project_name": str(worktree.get("displayName") or ""),
|
||
"_terminal_agent_identity": identity,
|
||
}
|
||
except RuntimeError as exc:
|
||
errors.append(f"{environment}: {exc}")
|
||
raise RuntimeError("找不到可用 terminal: " + "；".join(errors))
|
||
|
||
|
||
def worktrees(environment: str) -> list[dict[str, Any]]:
|
||
result = run_orca(["worktree", "list", "--limit", "200"], environment)
|
||
return [item for item in result.get("worktrees") or [] if isinstance(item, dict)]
|
||
|
||
|
||
def projects(environment: str) -> list[dict[str, Any]]:
|
||
result = run_orca(["project", "list"], environment)
|
||
return [item for item in result.get("projects") or [] if isinstance(item, dict)]
|
||
|
||
|
||
def config_from_project(
|
||
selector: str,
|
||
requested_environment: str | None,
|
||
requested_path: str | None = None,
|
||
) -> dict[str, Any]:
|
||
environments_to_check = [requested_environment] if requested_environment else environments()
|
||
matches: list[tuple[str, dict[str, Any]]] = []
|
||
for environment in environments_to_check:
|
||
if not environment:
|
||
continue
|
||
project_ids: set[str] = set()
|
||
for project in projects(environment):
|
||
project_values = {
|
||
str(project.get("id") or ""),
|
||
str(project.get("displayName") or ""),
|
||
}
|
||
if selector in project_values:
|
||
project_ids.add(str(project.get("id") or ""))
|
||
project_ids.update(str(repo_id) for repo_id in project.get("sourceRepoIds") or [])
|
||
|
||
for item in worktrees(environment):
|
||
values = {
|
||
str(item.get("id") or ""),
|
||
str(item.get("projectId") or ""),
|
||
str(item.get("repoId") or ""),
|
||
str(item.get("displayName") or ""),
|
||
str(item.get("path") or ""),
|
||
Path(str(item.get("path") or "")).name,
|
||
}
|
||
direct_match = selector in values
|
||
project_match = bool(
|
||
project_ids
|
||
and project_ids.intersection(
|
||
{
|
||
str(item.get("projectId") or ""),
|
||
str(item.get("repoId") or ""),
|
||
}
|
||
)
|
||
)
|
||
path_match = not requested_path or str(item.get("path") or "") == requested_path
|
||
if path_match and (direct_match or project_match):
|
||
matches.append((environment, item))
|
||
if not matches:
|
||
raise RuntimeError(f"找不到 project id/名称: {selector}")
|
||
if len(matches) > 1:
|
||
lines = [
|
||
f"{env}: {item.get('projectId') or item.get('repoId')} "
|
||
f"{item.get('displayName') or ''} {item.get('path') or ''}"
|
||
for env, item in matches
|
||
]
|
||
raise RuntimeError("project 匹配多个 worktree，请补充 environment 或 path:\n" + "\n".join(lines))
|
||
environment, item = matches[0]
|
||
return {
|
||
"environment": "" if environment == "local" else environment,
|
||
"host": str(item.get("hostId") or "local"),
|
||
"project": str(item.get("projectId") or item.get("repoId") or item.get("id") or ""),
|
||
"path": str(item.get("path") or ""),
|
||
"role": "test",
|
||
"cli": "codex",
|
||
"tier": "standard",
|
||
"model": None,
|
||
"reasoningEffort": "medium",
|
||
"permissionMode": "workspace-write",
|
||
"_source": f"project:{selector}",
|
||
"_project_name": str(item.get("displayName") or ""),
|
||
}
|
||
|
||
|
||
def clean_config(config: dict[str, Any]) -> dict[str, Any]:
|
||
return {key: value for key, value in config.items() if not key.startswith("_")}
|
||
|
||
|
||
def write_local(root: Path, config: dict[str, Any]) -> tuple[Path, dict[str, Any]]:
|
||
path = root / ".pouch" / "tester" / "local.yaml"
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
data: dict[str, Any] = {}
|
||
if path.is_file():
|
||
loaded = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
||
if not isinstance(loaded, dict):
|
||
raise RuntimeError(f"{path} 必须是 mapping")
|
||
data = loaded
|
||
current_agent = data.get("agent")
|
||
if current_agent is None:
|
||
current_agent = {}
|
||
if not isinstance(current_agent, dict):
|
||
raise RuntimeError(f"{path} 的 agent 必须是 mapping")
|
||
|
||
resolved = clean_config(config)
|
||
# Target location always comes from the user's selector. Keep explicit local
|
||
# choices for model/policy fields, so resolving a terminal cannot erase them.
|
||
location_fields = {"environment", "host", "project", "path"}
|
||
terminal_cli = bool(config.get("_terminal_agent_identity"))
|
||
merged = dict(current_agent)
|
||
for key, value in resolved.items():
|
||
if key in location_fields or key not in merged or (key == "cli" and terminal_cli):
|
||
merged[key] = value
|
||
data["agent"] = merged
|
||
path.write_text(yaml.safe_dump(data, allow_unicode=True, sort_keys=False), encoding="utf-8")
|
||
return path, merged
|
||
|
||
|
||
def print_confirmation(config: dict[str, Any], written: bool) -> None:
|
||
model = config.get("model") or "CLI 默认模型"
|
||
print("## tester Agent 配置更新结果" if written else "## tester Agent 配置解析预览")
|
||
print()
|
||
print(f"来源: {config.get('_source', '配置文件')}")
|
||
print(f"Agent: {config.get('cli')}")
|
||
print(f"角色 / 档位: {config.get('role')} / {config.get('tier')}")
|
||
print(f"模型: {model}")
|
||
print(f"推理强度: {config.get('reasoningEffort') or 'n/a'}")
|
||
print(f"权限: {config.get('permissionMode') or 'n/a'}")
|
||
print(f"Orca 环境: {config.get('environment') or 'local'}")
|
||
print(f"执行节点: {config.get('host') or 'local'}")
|
||
print(f"项目: {config.get('project') or 'n/a'}（{config.get('_project_name') or 'n/a'}）")
|
||
print(f"目录: {config.get('path') or 'n/a'}")
|
||
print("Terminal: 每轮新建，结束后关闭，不复用输入的 terminal")
|
||
print()
|
||
if written:
|
||
print("以上是已写入 local.yaml.agent 的生效配置；如需调整，直接修改配置后重新运行 tester。")
|
||
else:
|
||
print("以上是解析结果预览；加上 --write 后会写入 local.yaml.agent。")
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
parser = argparse.ArgumentParser(description="解析 Orca project/terminal 为 tester Agent 配置")
|
||
parser.add_argument("--root", default=".", help="项目根")
|
||
source = parser.add_mutually_exclusive_group(required=True)
|
||
source.add_argument("--terminal", help="Orca terminal handle")
|
||
source.add_argument("--project", help="Orca project id、名称或路径")
|
||
parser.add_argument("--path", help="项目有多个 worktree 时按目标机器上的绝对路径消歧")
|
||
parser.add_argument("--environment", help="限定 Orca environment")
|
||
parser.add_argument("--write", action="store_true", help="写入 .pouch/tester/local.yaml")
|
||
args = parser.parse_args(argv)
|
||
try:
|
||
config = (
|
||
config_from_terminal(args.terminal, args.environment)
|
||
if args.terminal
|
||
else config_from_project(args.project, args.environment, args.path)
|
||
)
|
||
if args.write:
|
||
path, saved_agent = write_local(Path(args.root).resolve(), config)
|
||
config = {**config, **saved_agent}
|
||
print(f"已写入: {path}")
|
||
print_confirmation(config, written=args.write)
|
||
return 0
|
||
except RuntimeError as exc:
|
||
print(f"environment-fail: {exc}", file=sys.stderr)
|
||
return 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|