knowledge-agent/agent_process.py

import os
import sys
import json
import time
import signal
import subprocess
from typing import Any, Dict, Optional

PID_FILE = os.path.join(os.path.dirname(__file__), 'agent_scheduler.pid')


def _write_pid_file(pid: int) -> None:
    with open(PID_FILE, 'w') as f:
        f.write(str(pid))


def _read_pid_file() -> Optional[int]:
    if not os.path.exists(PID_FILE):
        return None
    try:
        with open(PID_FILE, 'r') as f:
            content = f.read().strip()
            return int(content) if content else None
    except Exception:
        return None


def _is_process_running(pid: int) -> bool:
    try:
        os.kill(pid, 0)
        return True
    except Exception:
        return False


def start_daemon(host: str, port: int) -> Dict[str, Any]:
    old_pid = _read_pid_file()
    if old_pid and _is_process_running(old_pid):
        return {"status": "already_running", "pid": old_pid}

    python_exec = sys.executable
    script_path = os.path.join(os.path.dirname(__file__), 'agent.py')
    args = [python_exec, script_path, "--serve", "--host", host, "--port", str(port)]

    with open(os.devnull, 'wb') as devnull:
        proc = subprocess.Popen(
            args,
            stdout=devnull,
            stderr=devnull,
            stdin=devnull,
            close_fds=True,
            preexec_fn=os.setsid if hasattr(os, 'setsid') else None,
        )

    _write_pid_file(proc.pid)
    time.sleep(0.5)
    running = _is_process_running(proc.pid)
    return {"status": "started" if running else "failed", "pid": proc.pid}


def stop_daemon(timeout: float = 5.0) -> Dict[str, Any]:
    pid = _read_pid_file()
    if not pid:
        return {"status": "not_running"}

    if not _is_process_running(pid):
        try:
            os.remove(PID_FILE)
        except Exception:
            pass
        return {"status": "not_running"}

    try:
        os.kill(pid, signal.SIGTERM)
    except Exception as e:
        return {"status": "error", "error": str(e)}

    start_time = time.time()
    while time.time() - start_time < timeout:
        if not _is_process_running(pid):
            break
        time.sleep(0.2)

    if _is_process_running(pid):
        try:
            os.kill(pid, signal.SIGKILL)
        except Exception as e:
            return {"status": "error", "error": str(e)}

    try:
        os.remove(PID_FILE)
    except Exception:
        pass

    return {"status": "stopped"}


def status_daemon() -> Dict[str, Any]:
    pid = _read_pid_file()
    if pid and _is_process_running(pid):
        return {"status": "running", "pid": pid}
    return {"status": "not_running"}