mi222eh
112 lines
3.4 KiB
Python
112 lines
3.4 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.gitea_tools import GiteaTools
|
|
from gitea.config import AGENT_MODEL_ID, AGENT_MAX_RETRIES
|
|
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()
|
|
tools: GiteaTools = GiteaTools(client)
|
|
model_name: str = AGENT_MODEL_ID
|
|
|
|
# Initialize orchestrator
|
|
orchestrator: AgentOrchestrator = AgentOrchestrator(client, 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())
|