coding-agent-gitea/main.py

import asyncio
import os
import time
import logging
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 = logging.FileHandler(LOG_FILE)
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())