meeks
140 lines
4.3 KiB
Python
140 lines
4.3 KiB
Python
import asyncio
|
|
import os
|
|
import time
|
|
import logging
|
|
from logging.handlers import RotatingFileHandler
|
|
from pathlib import Path
|
|
from dotenv import load_dotenv
|
|
|
|
from gitea.client import GiteaClient
|
|
from gitea.tools.issue_tools import IssueTools
|
|
from gitea.tools.pr_tools import PRTools
|
|
from gitea.tools.file_tools import FileTools
|
|
from gitea.tools.git_tools import GitTools
|
|
from gitea.config import AGENT_MODEL_ID, AGENT_MAX_RETRIES, GITEA_REPOS_ROOT
|
|
from core.orchestrator import AgentOrchestrator
|
|
|
|
import json
|
|
|
|
# Setup logging to logs folder
|
|
LOG_DIR: Path = Path(__file__).parent.parent.parent / "logs"
|
|
LOG_DIR.mkdir(exist_ok=True)
|
|
LOG_FILE: Path = LOG_DIR / "agent.log"
|
|
|
|
|
|
class JSONFormatter(logging.Formatter):
|
|
"""Formats log records as JSON objects for structured logging."""
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
log_data = {
|
|
"timestamp": self.formatTime(record, self.datefmt),
|
|
"level": record.levelname,
|
|
"logger": record.name,
|
|
"message": record.getMessage(),
|
|
}
|
|
if record.exc_info:
|
|
log_data["exception"] = self.formatException(record.exc_info)
|
|
return json.dumps(log_data)
|
|
|
|
|
|
file_handler = RotatingFileHandler(LOG_FILE, maxBytes=5 * 1024 * 1024, backupCount=5)
|
|
file_handler.setFormatter(JSONFormatter())
|
|
|
|
stream_handler = logging.StreamHandler()
|
|
stream_handler.setFormatter(
|
|
logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
|
|
)
|
|
|
|
logging.basicConfig(level=logging.INFO, handlers=[file_handler, stream_handler])
|
|
logger: logging.Logger = logging.getLogger("coding-agent")
|
|
|
|
|
|
async def main() -> None:
|
|
load_dotenv()
|
|
|
|
# Ensure we're in the repo directory
|
|
repo_root: Path = Path(__file__).parent.parent.parent
|
|
if Path.cwd() != repo_root:
|
|
os.chdir(repo_root)
|
|
logger.info(f"Changed directory to {repo_root}")
|
|
|
|
# Initialize Gitea components
|
|
client: GiteaClient = GiteaClient()
|
|
try:
|
|
user = client.repos.get_authenticated_user()
|
|
if not user or not user.login:
|
|
raise RuntimeError("No authenticated user found.")
|
|
logger.info(f"Authenticated as user: {user.login}")
|
|
except Exception as e:
|
|
logger.critical(
|
|
f"Critical initialization error: No authenticated user found. {e}"
|
|
)
|
|
raise SystemExit(1)
|
|
|
|
issue_tools: IssueTools = IssueTools(client)
|
|
pr_tools: PRTools = PRTools(client)
|
|
file_tools: FileTools = FileTools(client, GITEA_REPOS_ROOT)
|
|
git_tools: GitTools = GitTools(client)
|
|
model_name: str = AGENT_MODEL_ID
|
|
|
|
# Initialize orchestrator
|
|
orchestrator: AgentOrchestrator = AgentOrchestrator(
|
|
client,
|
|
issue_tools,
|
|
pr_tools,
|
|
file_tools,
|
|
git_tools,
|
|
model_name,
|
|
AGENT_MAX_RETRIES,
|
|
)
|
|
|
|
logger.info("--- Autonomous Coding Agent Active ---")
|
|
logger.info(f"Model: {model_name}")
|
|
logger.info(f"Max retries: {AGENT_MAX_RETRIES}")
|
|
logger.info(f"Checking Gitea at {client.base_url}")
|
|
logger.info("Press Ctrl+C to stop.")
|
|
|
|
consecutive_errors: int = 0
|
|
max_consecutive_errors: int = 5
|
|
|
|
while True:
|
|
try:
|
|
logger.info("=== Checking for pending tasks ===")
|
|
|
|
# Use orchestrator to poll and dispatch
|
|
await orchestrator.poll_and_dispatch()
|
|
|
|
# Reset error counter on successful run
|
|
consecutive_errors = 0
|
|
|
|
logger.info("=== Waiting 60 seconds before next check ===")
|
|
await asyncio.sleep(60)
|
|
|
|
except KeyboardInterrupt:
|
|
logger.info("Agent stopped.")
|
|
break
|
|
except Exception as e:
|
|
consecutive_errors += 1
|
|
logger.error(
|
|
f"Error in main loop (attempt {consecutive_errors}/{max_consecutive_errors}): {e}"
|
|
)
|
|
|
|
# If too many consecutive errors, wait longer
|
|
if consecutive_errors >= max_consecutive_errors:
|
|
logger.error(
|
|
f"Too many consecutive errors ({max_consecutive_errors}). Waiting 5 minutes before retry."
|
|
)
|
|
await asyncio.sleep(300)
|
|
consecutive_errors = 0
|
|
else:
|
|
await asyncio.sleep(60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|
|
|
|
|
|
def start_agent() -> None:
|
|
"""Entry point for uv run."""
|
|
asyncio.run(main())
|