| |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import os |
| import shlex |
| import subprocess |
| import sys |
| import time |
| from pathlib import Path |
| from typing import Any |
|
|
|
|
| DG_ROOT = Path(__file__).resolve().parents[1] |
| DEFAULT_RUN_ROOT = DG_ROOT / "runlogs" / "mini-swe-agent" |
| DEFAULT_CONFIG = DG_ROOT / "configs" / "client_profiles" / "mini-swe-agent.dg.yaml" |
| DEFAULT_REGISTRY = DG_ROOT / "configs" / "client_profiles" / "litellm-local-model-registry.json" |
| LOCAL_MINI = DG_ROOT / ".tools" / "external-agents" / "bin" / "mini" |
| MINI_GLOBAL_CONFIG_DIR = DG_ROOT / ".tools" / "external-agents" / "mini-swe-config" |
|
|
|
|
| def timestamp() -> str: |
| return time.strftime("%Y%m%d-%H%M%S", time.localtime()) |
|
|
|
|
| def tail(text: str, limit: int = 6000) -> str: |
| return text[-limit:] if len(text) > limit else text |
|
|
|
|
| def resolve_repo(path: str) -> Path: |
| repo = Path(path or ".").resolve() |
| if not repo.exists() or not repo.is_dir(): |
| raise SystemExit(f"repo does not exist: {repo}") |
| return repo |
|
|
|
|
| def resolve_binary() -> str: |
| if LOCAL_MINI.exists(): |
| return str(LOCAL_MINI) |
| for name in ("mini", "mini-swe-agent"): |
| resolved = shutil_which(name) |
| if resolved: |
| return resolved |
| return "" |
|
|
|
|
| def shutil_which(name: str) -> str: |
| for part in os.environ.get("PATH", "").split(os.pathsep): |
| path = Path(part) / name |
| if path.exists() and os.access(path, os.X_OK): |
| return str(path) |
| return "" |
|
|
|
|
| def write_text(path: Path, text: str) -> None: |
| path.parent.mkdir(parents=True, exist_ok=True) |
| path.write_text(text, encoding="utf-8") |
|
|
|
|
| def ensure_mini_global_config() -> Path: |
| env_path = MINI_GLOBAL_CONFIG_DIR / ".env" |
| if not env_path.exists(): |
| write_text( |
| env_path, |
| "\n".join( |
| [ |
| "MSWEA_CONFIGURED=true", |
| "MSWEA_MODEL_NAME=openai/diffusiongemma-local", |
| "OPENAI_API_KEY=dummy", |
| "OPENAI_BASE_URL=http://127.0.0.1:4100/v1", |
| ] |
| ) |
| + "\n", |
| ) |
| return env_path |
|
|
|
|
| def analyze_trajectory(path: Path, stdout: str) -> dict[str, Any]: |
| analysis: dict[str, Any] = {"failed": False, "reasons": [], "exit_status": ""} |
| if "RepeatedFormatError" in stdout: |
| analysis["failed"] = True |
| analysis["reasons"].append("RepeatedFormatError in stdout") |
| if "Session status: failed" in stdout or "DG task runner finished: failed" in stdout: |
| analysis["failed"] = True |
| analysis["reasons"].append("delegated DG session failed") |
|
|
| if not path.exists(): |
| return analysis |
|
|
| try: |
| data = json.loads(path.read_text(encoding="utf-8")) |
| except Exception as exc: |
| analysis["failed"] = True |
| analysis["reasons"].append(f"trajectory is not valid JSON: {exc}") |
| return analysis |
|
|
| messages = data.get("messages", []) if isinstance(data, dict) else data |
| if not isinstance(messages, list): |
| return analysis |
|
|
| for message in messages: |
| if not isinstance(message, dict): |
| continue |
| content = str(message.get("content", "")) |
| extra = message.get("extra", {}) if isinstance(message.get("extra"), dict) else {} |
| if message.get("role") == "exit": |
| exit_status = str(extra.get("exit_status") or content).strip() |
| analysis["exit_status"] = exit_status |
| if exit_status in {"RepeatedFormatError"}: |
| analysis["failed"] = True |
| analysis["reasons"].append(f"mini-swe exit status: {exit_status}") |
| if "Session status: failed" in content or "DG task runner finished: failed" in content: |
| analysis["failed"] = True |
| analysis["reasons"].append("delegated DG session failed in trajectory") |
|
|
| analysis["reasons"] = sorted(set(analysis["reasons"])) |
| return analysis |
|
|
|
|
| def build_command(args: argparse.Namespace, repo: Path, out_dir: Path) -> tuple[list[str], Path, Path, Path]: |
| binary = args.binary or resolve_binary() |
| if not binary: |
| binary = "mini" |
|
|
| config = Path(args.config).resolve() if args.config else DEFAULT_CONFIG |
| registry = Path(args.model_registry).resolve() if args.model_registry else DEFAULT_REGISTRY |
| trajectory = Path(args.output).resolve() if args.output else out_dir / "trajectory.json" |
|
|
| cmd = [binary, "-c", str(config), "-t", args.task, "-o", str(trajectory)] |
| if args.yolo: |
| cmd.append("--yolo") |
| if args.exit_immediately: |
| cmd.append("--exit-immediately") |
| if args.cost_limit is not None: |
| cmd.extend(["--cost-limit", str(args.cost_limit)]) |
| for item in args.extra_config: |
| cmd.extend(["-c", item]) |
| if args.extra_args: |
| cmd.extend(args.extra_args) |
|
|
| return cmd, config, registry, trajectory |
|
|
|
|
| def command_string(cmd: list[str]) -> str: |
| return " ".join(shlex.quote(part) for part in cmd) |
|
|
|
|
| def run(args: argparse.Namespace) -> int: |
| repo = resolve_repo(args.repo) |
| run_dir = Path(args.out_dir).resolve() if args.out_dir else DEFAULT_RUN_ROOT / f"{timestamp()}-{repo.name}" |
| stdout_path = run_dir / "stdout.log" |
| stderr_path = run_dir / "stderr.log" |
| report_path = Path(args.report).resolve() if args.report else run_dir / "report.json" |
| command_path = run_dir / "command.sh" |
| cmd, config, registry, trajectory = build_command(args, repo, run_dir) |
|
|
| report: dict[str, Any] = { |
| "status": "dry-run" if args.dry_run else "running", |
| "repo": str(repo), |
| "task": args.task, |
| "run_dir": str(run_dir), |
| "command": cmd, |
| "command_text": command_string(cmd), |
| "config": str(config), |
| "model_registry": str(registry), |
| "trajectory": str(trajectory), |
| "command_file": str(command_path), |
| "stdout": str(stdout_path), |
| "stderr": str(stderr_path), |
| "timeout": args.timeout, |
| "started_at": time.time(), |
| } |
|
|
| run_dir.mkdir(parents=True, exist_ok=True) |
| write_text(command_path, "#!/usr/bin/env bash\nset -euo pipefail\ncd " + shlex.quote(str(repo)) + "\n" + command_string(cmd) + "\n") |
| command_path.chmod(command_path.stat().st_mode | 0o755) |
|
|
| env = os.environ.copy() |
| ensure_mini_global_config() |
| env["MSWEA_GLOBAL_CONFIG_DIR"] = str(MINI_GLOBAL_CONFIG_DIR) |
| env["MSWEA_CONFIGURED"] = "true" |
| env["MSWEA_MODEL_NAME"] = "openai/diffusiongemma-local" |
| env["MSWEA_SILENT_STARTUP"] = "1" |
| env["OPENAI_API_KEY"] = "dummy" |
| env["OPENAI_BASE_URL"] = "http://127.0.0.1:4100/v1" |
| env["LITELLM_MODEL_REGISTRY_PATH"] = str(registry) |
| if str(LOCAL_MINI.parent) not in env.get("PATH", "").split(os.pathsep): |
| env["PATH"] = str(LOCAL_MINI.parent) + os.pathsep + env.get("PATH", "") |
|
|
| if args.dry_run: |
| report["status"] = "dry-run" |
| report["returncode"] = None |
| write_text(stdout_path, "") |
| write_text(stderr_path, "") |
| write_text(report_path, json.dumps(report, ensure_ascii=False, indent=2) + "\n") |
| print(json.dumps(report, ensure_ascii=False, indent=2) if args.json else str(report_path)) |
| return 0 |
|
|
| started = time.time() |
| try: |
| proc = subprocess.run( |
| cmd, |
| cwd=str(repo), |
| env=env, |
| text=True, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE, |
| timeout=args.timeout, |
| check=False, |
| ) |
| elapsed = time.time() - started |
| write_text(stdout_path, proc.stdout) |
| write_text(stderr_path, proc.stderr) |
| mini_swe_analysis = analyze_trajectory(trajectory, proc.stdout) |
| report.update( |
| { |
| "status": "failed" if mini_swe_analysis["failed"] else ("success" if proc.returncode == 0 else "failed"), |
| "returncode": proc.returncode, |
| "elapsed_sec": round(elapsed, 3), |
| "trajectory_exists": trajectory.exists(), |
| "trajectory_bytes": trajectory.stat().st_size if trajectory.exists() else 0, |
| "mini_swe_analysis": mini_swe_analysis, |
| "stdout_tail": tail(proc.stdout), |
| "stderr_tail": tail(proc.stderr), |
| } |
| ) |
| except subprocess.TimeoutExpired as exc: |
| stdout = exc.stdout or "" |
| stderr = exc.stderr or "" |
| if isinstance(stdout, bytes): |
| stdout = stdout.decode("utf-8", errors="replace") |
| if isinstance(stderr, bytes): |
| stderr = stderr.decode("utf-8", errors="replace") |
| write_text(stdout_path, stdout) |
| write_text(stderr_path, stderr) |
| report.update( |
| { |
| "status": "timeout", |
| "returncode": None, |
| "elapsed_sec": round(time.time() - started, 3), |
| "trajectory_exists": trajectory.exists(), |
| "trajectory_bytes": trajectory.stat().st_size if trajectory.exists() else 0, |
| "stdout_tail": tail(stdout), |
| "stderr_tail": tail(stderr), |
| } |
| ) |
|
|
| write_text(report_path, json.dumps(report, ensure_ascii=False, indent=2) + "\n") |
| if args.json: |
| print(json.dumps(report, ensure_ascii=False, indent=2)) |
| else: |
| print(report_path) |
| return 0 if report["status"] == "success" else (124 if report["status"] == "timeout" else 1) |
|
|
|
|
| def parse_args() -> argparse.Namespace: |
| parser = argparse.ArgumentParser(description="Artifacted runner for mini-swe-agent over the local DG LiteLLM profile.") |
| parser.add_argument("--repo", required=True) |
| parser.add_argument("--task", required=True) |
| parser.add_argument("--out-dir", default="") |
| parser.add_argument("--report", default="") |
| parser.add_argument("--output", default="", help="Trajectory JSON path") |
| parser.add_argument("--config", default="") |
| parser.add_argument("--model-registry", default="") |
| parser.add_argument("--binary", default="") |
| parser.add_argument("--timeout", type=int, default=420) |
| parser.add_argument("--cost-limit", type=float, default=0.01) |
| parser.add_argument("--yolo", action="store_true") |
| parser.add_argument("--exit-immediately", action="store_true") |
| parser.add_argument("--extra-config", action="append", default=[]) |
| parser.add_argument("--extra-args", nargs=argparse.REMAINDER, default=[]) |
| parser.add_argument("--dry-run", action="store_true") |
| parser.add_argument("--json", action="store_true") |
| return parser.parse_args() |
|
|
|
|
| def main() -> int: |
| return run(parse_args()) |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
