azad.main Module

azad.main

Attributes

PID_FILE_DIR module-attribute

PID_FILE_DIR = 'logs'

PID_FILE_NAME module-attribute

PID_FILE_NAME = 'azad.pid'

PID_FILE_PATH module-attribute

PID_FILE_PATH = join(PID_FILE_DIR, PID_FILE_NAME)

shutdown_event module-attribute

shutdown_event = None

Classes

Functions

setup_main_logging

setup_main_logging(log_level: str = 'INFO') -> None

Set up logging configuration.

Source code in azad/main.py

def setup_main_logging(log_level: str = 'INFO') -> None:
    """Set up logging configuration."""
    setup_log_levels()
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG)
    root_logger = logging.getLogger()
    for handler in root_logger.handlers[:]:
        root_logger.removeHandler(handler)
    root_logger.addHandler(console_handler)
    root_logger.setLevel(logging.DEBUG)
    root_logger.debug(f"Setting up logging with level: {log_level}")
    os.makedirs(PID_FILE_DIR, exist_ok=True) # Ensure logs dir exists for PID file too
    try:
        level_name = log_level.upper()
        try:
            numeric_level = getattr(LogLevel, level_name).value
        except AttributeError:
            numeric_level = getattr(logging, level_name)
    except AttributeError:
        raise ValueError(f'Invalid log level: {log_level}')
    setup_logging(level=numeric_level)
    root_logger = logging.getLogger()
    root_logger.debug("Logging system initialized", extra={
        "event": "initialization",
        "level": level_name,
        "numeric_level": numeric_level
    })

write_pid_file

write_pid_file() -> None

Write the current process ID to the PID file.

Source code in azad/main.py

def write_pid_file() -> None:
    """Write the current process ID to the PID file."""
    pid = os.getpid()
    try:
        os.makedirs(PID_FILE_DIR, exist_ok=True)
        with open(PID_FILE_PATH, 'w') as f:
            f.write(str(pid))
        logging.getLogger(__name__).debug(f"PID {pid} written to {PID_FILE_PATH}", extra={"event": "pid_write"})
    except IOError as e:
        logging.getLogger(__name__).error(f"Failed to write PID file {PID_FILE_PATH}: {e}", extra={"event": "pid_error"})
        # Exit if we can't write the PID file, as 'stop' won't work
        sys.exit(f"Error: Could not write PID file to {PID_FILE_PATH}. Aborting.")

remove_pid_file

remove_pid_file() -> None

Remove the PID file if it exists.

Source code in azad/main.py

def remove_pid_file() -> None:
    """Remove the PID file if it exists."""
    try:
        if os.path.exists(PID_FILE_PATH):
            os.remove(PID_FILE_PATH)
            logging.getLogger(__name__).debug(f"PID file {PID_FILE_PATH} removed", extra={"event": "pid_remove"})
    except IOError as e:
        logging.getLogger(__name__).warning(f"Failed to remove PID file {PID_FILE_PATH}: {e}", extra={"event": "pid_error"})

read_pid_from_file

read_pid_from_file() -> Optional[int]

Read the PID from the PID file.

Source code in azad/main.py

def read_pid_from_file() -> Optional[int]:
    """Read the PID from the PID file."""
    try:
        if os.path.exists(PID_FILE_PATH):
            with open(PID_FILE_PATH, 'r') as f:
                pid_str = f.read().strip()
                if pid_str:
                    return int(pid_str)
    except (IOError, ValueError) as e:
        logging.getLogger(__name__).warning(f"Failed to read or parse PID file {PID_FILE_PATH}: {e}", extra={"event": "pid_error"})
    return None

create_argument_parser

create_argument_parser() -> ArgumentParser

Create and return the main argument parser with subcommands.

Source code in azad/main.py

def create_argument_parser() -> argparse.ArgumentParser:
    """Create and return the main argument parser with subcommands."""
    parser = argparse.ArgumentParser(description='Azad AI Development Agent CLI.')
    subparsers = parser.add_subparsers(dest='command', help='Subcommand to run')
    parser.set_defaults(func=lambda args: parser.print_help())

    # --- Start Subcommand ---
    start_parser = subparsers.add_parser('start', help='Start the Azad agent daemon')
    start_parser.add_argument(
        '--log-level', type=str, choices=['DEBUG', 'INFO', 'PROD', 'WARNING', 'ERROR', 'CRITICAL'],
        default='INFO', help='Set logging level for the daemon (default: INFO)'
    )
    start_parser.add_argument(
        '--host', type=str, default='', help='Interface to listen on (default: all interfaces)'
    )
    print(f"Host: {start_parser._get_args()}")
    GlobalConfig.add_cli_arguments(start_parser)
    start_parser.set_defaults(func=run_start_command)

    # --- Introspect Subcommand ---
    if introspect_main:
        introspect_parser = subparsers.add_parser('introspect', help='Introspect server interface and generate client')
        introspect_parser.add_argument('--verbose', '-v', action='store_true', help='Enable verbose output')
        introspect_parser.add_argument('--output', '-o', default='azad-ts-client', help='Output directory')
        introspect_parser.add_argument('--http-output', default='azad/http_server/client', help='Output directory for HTTP client')
        introspect_parser.add_argument('--methods', '-m', help='Comma-separated list of methods to generate')
        introspect_parser.set_defaults(func=run_introspect_command)

    # --- Stop Subcommand ---
    if psutil: # Only add if psutil is available
        stop_parser = subparsers.add_parser('stop', help='Stop the running Azad agent daemon')
        stop_parser.set_defaults(func=run_stop_command)
    else:
         # Add a disabled stop command if psutil is missing
        stop_parser = subparsers.add_parser('stop', help='Stop the running Azad agent daemon (disabled: psutil not installed)')
        stop_parser.set_defaults(func=lambda args: print("Error: 'azad stop' requires the 'psutil' library. Please install it."))


    return parser

start_daemon async

start_daemon(args: Namespace, config: GlobalConfig) -> None

Asynchronous function to start the agent daemon.

Source code in azad/main.py

async def start_daemon(args: argparse.Namespace, config: GlobalConfig) -> None:
    """Asynchronous function to start the agent daemon."""
    global shutdown_event
    shutdown_event = asyncio.Event()

    # Write PID file *before* setting up signal handlers that might remove it
    write_pid_file()

    def signal_handler(signum, frame):
        logger = logging.getLogger(__name__)
        logger.debug(f"Signal {signum} received", extra={"event": "shutdown", "signal": signum})
        if shutdown_event:
            shutdown_event.set()
        # Ensure PID file is removed even on direct signal handling if possible
        # Note: This might not always run depending on the signal and OS.
        remove_pid_file()

    # Register signal handlers
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)

    logger = logging.getLogger(__name__)
    http_server_task = None
    try:
        logger.debug("Initializing agent for daemon mode...", extra={"event": "initialization", "component": "agent"})
        if settings.AWS_DB_URL and uvicorn and http_server_app and LOGGING_CONFIG:
            # Force a verbose access log format to see the raw request details
            LOGGING_CONFIG["formatters"]["access"]["fmt"] = '%(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s'

            http_config = uvicorn.Config(
                http_server_app,
                host="0.0.0.0",
                port=3333,
                log_config=LOGGING_CONFIG,
                h11_max_incomplete_event_size=64 * 1024 * 1024,  # Increase max size to 64MB
            )
            http_server = uvicorn.Server(http_config)
            http_server_task = asyncio.create_task(http_server.serve())
            logger.info("HTTP server for failed edits started on port 8088 with increased size limits and verbose logging.")

        await AzadAgent.enter_daemon_mode(config)
    except asyncio.CancelledError:
        logger.debug("Daemon shutdown initiated by task cancellation")
    except Exception as e:
        logger.error("Daemon application error", extra={"event": "error", "error": str(e), "traceback": True})
    finally:
        # Ensure PID file is removed on clean exit or cancellation
        if http_server_task:
            http_server_task.cancel()
        remove_pid_file()
        logger.debug("Daemon shutdown complete")

run_start_command

run_start_command(args: Namespace)

Handler for the 'start' subcommand.

Source code in azad/main.py

def run_start_command(args: argparse.Namespace):
    """Handler for the 'start' subcommand."""
    setup_main_logging(args.log_level)
    logger = logging.getLogger(__name__)

    # Check if already running
    pid = read_pid_from_file()
    if pid and psutil and psutil.pid_exists(pid):
         try:
             proc = psutil.Process(pid)
             # Basic check if it looks like our process (can be improved)
             if 'python' in proc.name().lower() and any('azad' in cmd for cmd in proc.cmdline()):
                 print(f"Error: Azad agent appears to be already running with PID {pid}.")
                 logger.warning(f"Start aborted: Process with PID {pid} already exists.", extra={"event": "start_aborted"})
                 sys.exit(1)
             else:
                 logger.warning(f"Stale PID file found ({PID_FILE_PATH}) for PID {pid}, but process doesn't look like Azad. Overwriting.", extra={"event": "pid_stale"})
                 remove_pid_file() # Remove stale file
         except (psutil.NoSuchProcess, psutil.AccessDenied):
             logger.warning(f"Stale PID file found ({PID_FILE_PATH}) for PID {pid}, but process not found or inaccessible. Overwriting.", extra={"event": "pid_stale"})
             remove_pid_file() # Remove stale file

    logger.info("Starting Azad Agent Daemon...")
    config = GlobalConfig.load(args.config)
    config.set_cli_args(args)

    try:
        asyncio.run(start_daemon(args=args, config=config))
    except KeyboardInterrupt:
        logger.info("Daemon stopped by user (KeyboardInterrupt).")

run_introspect_command

run_introspect_command(args: Namespace)

Handler for the 'introspect' subcommand.

Source code in azad/main.py

def run_introspect_command(args: argparse.Namespace):
    """Handler for the 'introspect' subcommand."""
    if not introspect_main:
         print("Error: Introspect command is not available due to import issues.")
         return
    print("Running introspection...")
    introspect_main(args=args)
    print("Introspection complete.")

run_stop_command

run_stop_command(args: Namespace)

Handler for the 'stop' subcommand.

Source code in azad/main.py

def run_stop_command(args: argparse.Namespace):
    """Handler for the 'stop' subcommand."""
    if not psutil: # Should not happen if parser setup is correct, but double-check
        print("Error: 'azad stop' requires the 'psutil' library.")
        return

    # Setup basic logging for the stop command itself if not already configured
    if not logging.getLogger().hasHandlers():
         logging.basicConfig(level=logging.INFO, format='%(levelname)-8s [%(name)s] %(message)s')
    logger = logging.getLogger(__name__)
    pid = read_pid_from_file()

    if not pid:
        print("Azad agent does not appear to be running (PID file not found).")
        logger.info("Stop command: PID file not found.", extra={"event": "stop_noop"})
        return

    try:
        if psutil.pid_exists(pid):
            process = psutil.Process(pid)
            print(f"Attempting to stop Azad agent (PID: {pid})...")
            logger.info(f"Sending SIGTERM to process {pid}", extra={"event": "stop_signal"})
            process.terminate() # Send SIGTERM for graceful shutdown

            # Optional: Wait a bit and check if it stopped, then send SIGKILL if needed
            try:
                process.wait(timeout=5) # Wait up to 5 seconds
                print("Azad agent stopped successfully.")
                logger.info(f"Process {pid} terminated successfully.", extra={"event": "stop_success"})
            except psutil.TimeoutExpired:
                print(f"Azad agent (PID: {pid}) did not stop gracefully after 5s. Forcing stop...")
                logger.warning(f"Process {pid} did not terminate gracefully. Sending SIGKILL.", extra={"event": "stop_force"})
                process.kill() # Force kill
                print("Azad agent force-stopped.")
                logger.info(f"Process {pid} killed.", extra={"event": "stop_killed"})

            # Clean up PID file even if we killed it
            remove_pid_file()

        else:
            print(f"Azad agent process with PID {pid} not found (PID file might be stale).")
            logger.warning(f"Stop command: Process {pid} not found (stale PID file?).", extra={"event": "stop_stale"})
            remove_pid_file() # Remove stale file

    except psutil.NoSuchProcess:
        print(f"Azad agent process with PID {pid} not found (PID file might be stale).")
        logger.warning(f"Stop command: Process {pid} not found (NoSuchProcess).", extra={"event": "stop_stale"})
        remove_pid_file() # Remove stale file
    except psutil.AccessDenied:
        print(f"Error: Permission denied to stop process {pid}.")
        logger.error(f"Permission denied to signal process {pid}.", extra={"event": "stop_error"})
    except Exception as e:
        print(f"An unexpected error occurred while trying to stop the agent: {e}")
        logger.error(f"Unexpected error during stop command for PID {pid}: {e}", extra={"event": "stop_error"})

run_main_sync

run_main_sync()

Synchronous wrapper that parses args and calls the appropriate subcommand handler.

Source code in azad/main.py

def run_main_sync():
    """Synchronous wrapper that parses args and calls the appropriate subcommand handler."""
    # Basic logging setup before parsing args, in case parsing fails
    logging.basicConfig(level=logging.WARNING, format='%(levelname)-8s [%(name)s] %(message)s')
    parser = create_argument_parser()
    args = parser.parse_args()
    args.func(args)