404 lines
12 KiB
Python
404 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
OpenHands Build Agent
|
|
|
|
Executes build/test tasks using OpenHands SDK.
|
|
Designed for GitHub Actions integration with n8n orchestration.
|
|
|
|
Usage:
|
|
python agent_script.py
|
|
|
|
Environment Variables:
|
|
TASK: Task description to execute
|
|
WORKSPACE_PATH: Path to project workspace
|
|
RETRY_COUNT: Current retry attempt number
|
|
LLM_API_KEY: API key for LLM (required)
|
|
LLM_MODEL: Model to use (default: anthropic/claude-sonnet-4-5-20250929)
|
|
PROJECT_TYPE: Type of project (nodejs, python, etc.)
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import sys
|
|
import subprocess
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Dict, Any, List
|
|
|
|
from openhands.sdk import LLM, Conversation, get_logger
|
|
from openhands.tools.preset.default import get_default_agent
|
|
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
def detect_project_type(workspace_path: str) -> str:
|
|
"""Detect project type from workspace."""
|
|
workspace = Path(workspace_path)
|
|
|
|
if (workspace / "package.json").exists():
|
|
return "nodejs"
|
|
elif (workspace / "requirements.txt").exists() or (workspace / "pyproject.toml").exists():
|
|
return "python"
|
|
elif (workspace / "Cargo.toml").exists():
|
|
return "rust"
|
|
elif (workspace / "go.mod").exists():
|
|
return "go"
|
|
else:
|
|
return "unknown"
|
|
|
|
|
|
def load_previous_errors(workspace_path: str) -> Dict[str, Any]:
|
|
"""Load errors from previous build attempts."""
|
|
error_file = Path(workspace_path) / "build-errors.json"
|
|
|
|
if not error_file.exists():
|
|
return {}
|
|
|
|
try:
|
|
with open(error_file) as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
logger.warning(f"Could not load previous errors: {e}")
|
|
return {}
|
|
|
|
|
|
def create_build_task(
|
|
workspace_path: str,
|
|
task: str,
|
|
retry_count: int,
|
|
previous_errors: Dict[str, Any],
|
|
project_type: str
|
|
) -> str:
|
|
"""Create enhanced build task with context and previous errors."""
|
|
|
|
# Build enhanced prompt
|
|
build_prompt = f"""
|
|
You are an expert build engineer. Execute the following task for a {project_type} project.
|
|
|
|
PROJECT: {workspace_path}
|
|
TASK: {task}
|
|
|
|
CURRENT ATTEMPT: {retry_count + 1}
|
|
|
|
EXECUTION STEPS:
|
|
1. Analyze the project structure and detect build requirements
|
|
2. Install dependencies (npm install / pip install / etc.)
|
|
3. Run the build process
|
|
4. Execute tests if available
|
|
5. Generate a comprehensive report
|
|
|
|
IMPORTANT REQUIREMENTS:
|
|
- Capture ALL errors (stdout, stderr, exit codes)
|
|
- Save errors to: build-errors.json
|
|
- Save detailed report to: build-report.json
|
|
- Report success/failure clearly
|
|
- Fix common issues automatically if possible
|
|
|
|
OUTPUT FORMAT (JSON only):
|
|
{{
|
|
"success": true/false,
|
|
"exit_code": 0/1,
|
|
"errors": [
|
|
{{
|
|
"type": "dependency/build/test/runtime",
|
|
"message": "error description",
|
|
"file": "file path if applicable",
|
|
"line": line number if applicable,
|
|
"fix_suggestion": "suggested solution"
|
|
}}
|
|
],
|
|
"build_time": "execution time in seconds",
|
|
"artifacts": ["list of generated files"],
|
|
"warnings": ["list of warnings"],
|
|
"test_results": {{
|
|
"passed": number,
|
|
"failed": number,
|
|
"total": number
|
|
}},
|
|
"recommendations": ["list of improvement suggestions"]
|
|
}}
|
|
|
|
SPECIFIC FOCUS AREAS:
|
|
- Package installation issues (npm install, pip install)
|
|
- Build script failures (npm run build, python setup.py build)
|
|
- Test failures and coverage
|
|
- TypeScript/JavaScript errors
|
|
- Python import/module errors
|
|
- Configuration problems (missing files, wrong paths)
|
|
- Dependency version conflicts
|
|
|
|
Be thorough and provide actionable error messages.
|
|
"""
|
|
|
|
# Add previous errors if this is a retry
|
|
if previous_errors and retry_count > 0:
|
|
build_prompt += f"\n\n" + "="*80 + "\n"
|
|
build_prompt += f"PREVIOUS BUILD ERRORS (attempt #{retry_count}):\n"
|
|
build_prompt += "="*80 + "\n\n"
|
|
build_prompt += json.dumps(previous_errors, indent=2)
|
|
build_prompt += "\n\n" + "="*80 + "\n"
|
|
build_prompt += "IMPORTANT: Fix these specific issues first.\n"
|
|
build_prompt += "Provide clear error analysis and solutions.\n"
|
|
build_prompt += "="*80 + "\n"
|
|
|
|
return build_prompt
|
|
|
|
|
|
def save_build_report(workspace_path: str, result: Dict[str, Any]) -> Path:
|
|
"""Save build report to file."""
|
|
report = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"workspace": workspace_path,
|
|
"result": result,
|
|
}
|
|
|
|
report_path = Path(workspace_path) / "build-report.json"
|
|
with open(report_path, "w") as f:
|
|
json.dump(report, f, indent=2)
|
|
|
|
logger.info(f"Build report saved to: {report_path}")
|
|
return report_path
|
|
|
|
|
|
def save_build_errors(workspace_path: str, errors: List[Dict[str, Any]]) -> Path:
|
|
"""Save build errors to file."""
|
|
error_report = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"errors": errors,
|
|
}
|
|
|
|
error_path = Path(workspace_path) / "build-errors.json"
|
|
with open(error_path, "w") as f:
|
|
json.dump(error_report, f, indent=2)
|
|
|
|
logger.info(f"Build errors saved to: {error_path}")
|
|
return error_path
|
|
|
|
|
|
def run_basic_build_commands(workspace_path: str, project_type: str) -> Dict[str, Any]:
|
|
"""Run basic build commands to get initial feedback."""
|
|
workspace = Path(workspace_path)
|
|
result = {
|
|
"commands_run": [],
|
|
"errors": [],
|
|
"warnings": []
|
|
}
|
|
|
|
try:
|
|
if project_type == "nodejs":
|
|
# Check if package.json exists
|
|
if (workspace / "package.json").exists():
|
|
# Try npm install
|
|
cmd_result = subprocess.run(
|
|
["npm", "ci", "--silent"],
|
|
cwd=workspace,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=300
|
|
)
|
|
result["commands_run"].append("npm ci")
|
|
|
|
if cmd_result.returncode != 0:
|
|
result["errors"].append({
|
|
"type": "dependency",
|
|
"message": cmd_result.stderr or cmd_result.stdout,
|
|
"fix_suggestion": "Check package.json dependencies and versions"
|
|
})
|
|
|
|
# Try npm run build if available
|
|
cmd_result = subprocess.run(
|
|
["npm", "run", "build"],
|
|
cwd=workspace,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=300
|
|
)
|
|
result["commands_run"].append("npm run build")
|
|
|
|
if cmd_result.returncode != 0:
|
|
result["errors"].append({
|
|
"type": "build",
|
|
"message": cmd_result.stderr or cmd_result.stdout,
|
|
"fix_suggestion": "Check build script in package.json"
|
|
})
|
|
|
|
elif project_type == "python":
|
|
if (workspace / "requirements.txt").exists():
|
|
cmd_result = subprocess.run(
|
|
["pip", "install", "-r", "requirements.txt"],
|
|
cwd=workspace,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=300
|
|
)
|
|
result["commands_run"].append("pip install -r requirements.txt")
|
|
|
|
if cmd_result.returncode != 0:
|
|
result["errors"].append({
|
|
"type": "dependency",
|
|
"message": cmd_result.stderr or cmd_result.stdout,
|
|
"fix_suggestion": "Check requirements.txt format and packages"
|
|
})
|
|
|
|
except subprocess.TimeoutExpired:
|
|
result["errors"].append({
|
|
"type": "timeout",
|
|
"message": "Build command timed out",
|
|
"fix_suggestion": "Check for infinite loops or network issues"
|
|
})
|
|
except Exception as e:
|
|
result["errors"].append({
|
|
"type": "runtime",
|
|
"message": str(e),
|
|
"fix_suggestion": "Check project configuration"
|
|
})
|
|
|
|
return result
|
|
|
|
|
|
def main():
|
|
"""Execute build task with OpenHands SDK."""
|
|
start_time = datetime.now()
|
|
|
|
# Get configuration from environment
|
|
task = os.getenv("TASK", "Build and test the project")
|
|
workspace_path = os.getenv("WORKSPACE_PATH", os.getcwd())
|
|
retry_count = int(os.getenv("RETRY_COUNT", "0"))
|
|
api_key = os.getenv("LLM_API_KEY")
|
|
model = os.getenv("LLM_MODEL", "anthropic/claude-sonnet-4-5-20250929")
|
|
project_type = os.getenv("PROJECT_TYPE", detect_project_type(workspace_path))
|
|
|
|
# Validate inputs
|
|
if not api_key:
|
|
logger.error("LLM_API_KEY environment variable is required")
|
|
sys.exit(1)
|
|
|
|
if not os.path.exists(workspace_path):
|
|
logger.error(f"Workspace path does not exist: {workspace_path}")
|
|
sys.exit(1)
|
|
|
|
logger.info("="*80)
|
|
logger.info(f"OpenHands Build Starting")
|
|
logger.info("="*80)
|
|
logger.info(f"Attempt: #{retry_count + 1}")
|
|
logger.info(f"Workspace: {workspace_path}")
|
|
logger.info(f"Project Type: {project_type}")
|
|
logger.info(f"Task: {task[:100]}...")
|
|
logger.info("="*80)
|
|
|
|
try:
|
|
# Load previous errors
|
|
previous_errors = load_previous_errors(workspace_path)
|
|
|
|
# Run basic build commands first for initial feedback
|
|
logger.info("Running initial build commands...")
|
|
basic_result = run_basic_build_commands(workspace_path, project_type)
|
|
|
|
if basic_result["errors"]:
|
|
logger.warning(f"Initial build errors detected: {len(basic_result['errors'])}")
|
|
save_build_errors(workspace_path, basic_result["errors"])
|
|
|
|
# Configure LLM
|
|
logger.info("Initializing OpenHands SDK...")
|
|
llm = LLM(
|
|
model=model,
|
|
api_key=api_key,
|
|
usage_id="openhands-build",
|
|
drop_params=True,
|
|
)
|
|
|
|
# Create agent with all tools
|
|
logger.info("Creating OpenHands agent...")
|
|
agent = get_default_agent(
|
|
llm=llm,
|
|
cli_mode=True,
|
|
)
|
|
|
|
# Create conversation
|
|
conversation = Conversation(
|
|
agent=agent,
|
|
workspace=workspace_path,
|
|
)
|
|
|
|
# Build task with context
|
|
build_task = create_build_task(
|
|
workspace_path,
|
|
task,
|
|
retry_count,
|
|
previous_errors,
|
|
project_type
|
|
)
|
|
|
|
logger.info("Sending task to OpenHands agent...")
|
|
logger.info("="*80)
|
|
|
|
# Execute task
|
|
conversation.send_message(build_task)
|
|
conversation.run()
|
|
|
|
logger.info("="*80)
|
|
logger.info("OpenHands agent completed")
|
|
logger.info("="*80)
|
|
|
|
# Load and parse build report
|
|
report_path = Path(workspace_path) / "build-report.json"
|
|
error_path = Path(workspace_path) / "build-errors.json"
|
|
|
|
# Calculate build time
|
|
build_time = (datetime.now() - start_time).total_seconds()
|
|
|
|
if report_path.exists():
|
|
with open(report_path) as f:
|
|
result = json.load(f)
|
|
|
|
# Add build time to result
|
|
result["result"]["build_time_seconds"] = build_time
|
|
|
|
# Save updated report
|
|
save_build_report(workspace_path, result)
|
|
|
|
# Print result for GitHub Actions
|
|
print(json.dumps(result["result"], indent=2))
|
|
|
|
# Exit with appropriate code
|
|
exit_code = 0 if result["result"].get("success", False) else 1
|
|
logger.info(f"Build {'successful' if exit_code == 0 else 'failed'}")
|
|
logger.info(f"Build time: {build_time:.2f} seconds")
|
|
sys.exit(exit_code)
|
|
else:
|
|
logger.error("Build report not found")
|
|
if error_path.exists():
|
|
with open(error_path) as f:
|
|
error_data = json.load(f)
|
|
print(json.dumps(error_data, indent=2))
|
|
|
|
sys.exit(1)
|
|
|
|
except KeyboardInterrupt:
|
|
logger.error("Build interrupted by user")
|
|
sys.exit(130)
|
|
except Exception as e:
|
|
logger.error(f"Build failed with exception: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
# Save error report
|
|
error_report = {
|
|
"success": False,
|
|
"error": str(e),
|
|
"error_type": type(e).__name__,
|
|
"timestamp": datetime.now().isoformat(),
|
|
"build_time_seconds": (datetime.now() - start_time).total_seconds(),
|
|
}
|
|
error_path = Path(workspace_path) / "build-errors.json"
|
|
with open(error_path, "w") as f:
|
|
json.dump(error_report, f, indent=2)
|
|
|
|
print(json.dumps(error_report, indent=2))
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|