| |
| """BeamArena CLI — Autonomous Coding & Evaluation Engine. |
| |
| Run autonomous repairs, ingest 1M repositories, and evaluate benchmarks locally. |
| """ |
|
|
| import argparse |
| import json |
| import os |
| import sys |
| from core.client import BeamClient |
| from core.sandbox import Sandbox |
| from core.agent import AutonomousCodingAgent |
| from core.ingest import RepoIngestor |
| from scenarios import ALL_SCENARIOS |
|
|
|
|
| def cmd_solve(args): |
| client = BeamClient() |
| agent = AutonomousCodingAgent(client=client, max_turns=args.max_turns) |
|
|
| def on_start(turn_idx): |
| print(f"\n>>> TURN {turn_idx} STARTING...") |
|
|
| def on_complete(turn): |
| print(f">>> TURN {turn.turn_index} FINISHED. Tests Passed: {turn.passed}") |
| print(f" Reasoning Tokens: {turn.usage.reasoning_tokens} | Total: {turn.usage.total_tokens}") |
| for log in turn.edits_applied: |
| print(f" • {log}") |
| for cr in turn.command_results: |
| status = "PASS" if cr.exit_code == 0 else f"FAIL (code {cr.exit_code})" |
| print(f" $ {cr.command} -> {status}") |
|
|
| agent.on_turn_start = on_start |
| agent.on_turn_complete = on_complete |
|
|
| if args.task: |
| |
| print("=" * 60) |
| print("⚡ [Beam-SWE] Autonomous Local Codebase Repair") |
| print(f"📋 Task: {args.task}") |
| print(f"🧪 Test Command: {args.test_cmd}") |
| print("=" * 60) |
|
|
| ingestor = RepoIngestor() |
| packed = ingestor.ingest_directory(".") |
| file_map = {p: packed.raw_files[p] for p in packed.files if p in packed.raw_files} |
|
|
| with Sandbox() as sb: |
| sb.load_files(file_map) |
| trace = agent.solve( |
| task_description=args.task, |
| sandbox=sb, |
| test_command=args.test_cmd, |
| scenario_id="local_custom", |
| ) |
|
|
| print("\n" + "=" * 60) |
| print(f"RESULT: {'✅ SOLVED (TESTS GREEN)' if trace.resolved else '❌ FAILED'}") |
| print(f"Turns: {len(trace.turns)} | Time: {trace.wall_clock_seconds}s") |
| print(f"Total Tokens: {trace.total_tokens} (Reasoning: {trace.total_reasoning_tokens})") |
| print("=" * 60) |
| return |
|
|
| |
| scen_id = args.scenario or "01_fastapi_auth" |
| scen = ALL_SCENARIOS.get(scen_id) |
| if not scen: |
| print(f"Error: Unknown scenario '{scen_id}'. Available: {list(ALL_SCENARIOS.keys())}") |
| sys.exit(1) |
|
|
| print("=" * 60) |
| print(f"⚡ [BeamArena] Solving Scenario: {scen.title}") |
| print(f"📁 Workspace: {len(scen.files)} files | Verification: {scen.test_command}") |
| print("=" * 60) |
|
|
| with Sandbox() as sb: |
| sb.load_files(scen.files) |
| trace = agent.solve( |
| task_description=scen.description, |
| sandbox=sb, |
| test_command=scen.test_command, |
| scenario_id=scen.id |
| ) |
|
|
| print("\n" + "=" * 60) |
| print(f"RESULT: {'✅ SOLVED (TESTS GREEN)' if trace.resolved else '❌ FAILED'}") |
| print(f"Turns: {len(trace.turns)} | Time: {trace.wall_clock_seconds}s") |
| print(f"Total Tokens: {trace.total_tokens} (Reasoning: {trace.total_reasoning_tokens})") |
| print("=" * 60) |
|
|
| if not trace.resolved: |
| sys.exit(1) |
|
|
|
|
| def cmd_ingest(args): |
| target_dir = os.path.abspath(args.path) |
| if not os.path.exists(target_dir): |
| print(f"Error: Path does not exist: {target_dir}") |
| sys.exit(1) |
|
|
| print("=" * 60) |
| print(f"🌐 [BeamArena] Ingesting Codebase (1M Context Mode): {target_dir}") |
| print("=" * 60) |
|
|
| ingestor = RepoIngestor() |
| packed = ingestor.ingest_directory(target_dir) |
|
|
| print(f"Scanned Files: {len(packed.files)}") |
| print(f"Total Size: {round(packed.total_bytes / 1024, 2)} KB") |
| print(f"Estimated Tokens: ~{packed.estimated_tokens:,} tokens (Capacity: 1,048,576)") |
| print("\n--- File Tree Sample ---") |
| tree_preview = "\n".join(packed.file_tree.splitlines()[:25]) |
| print(tree_preview) |
| if len(packed.file_tree.splitlines()) > 25: |
| print("... (truncated)") |
|
|
|
|
| def cmd_list(args): |
| print("Available Scenarios in BeamArena:") |
| for s_id, s in ALL_SCENARIOS.items(): |
| print(f" • {s_id:25} [{s.difficulty:6}] {s.title}") |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description="BeamArena Autonomous CLI") |
| subparsers = parser.add_subparsers(dest="command") |
|
|
| |
| solve_parser = subparsers.add_parser("solve", help="Run autonomous self-healing on a scenario or local codebase") |
| solve_parser.add_argument("--scenario", "-s", help="Benchmark scenario ID to solve") |
| solve_parser.add_argument("--task", help="Custom task description to solve in current directory") |
| solve_parser.add_argument("--test-cmd", default="pytest", help="Verification command to execute (default: pytest)") |
| solve_parser.add_argument("--max-turns", "-t", type=int, default=3, help="Max repair turns") |
|
|
| |
| ingest_parser = subparsers.add_parser("ingest", help="Ingest a local repository into 1M context") |
| ingest_parser.add_argument("path", help="Local directory path to ingest") |
|
|
| |
| subparsers.add_parser("list", help="List all benchmark scenarios") |
|
|
| args = parser.parse_args() |
| if args.command == "solve": |
| cmd_solve(args) |
| elif args.command == "ingest": |
| cmd_ingest(args) |
| elif args.command == "list": |
| cmd_list(args) |
| else: |
| parser.print_help() |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
