Function send_command

core/daemon.py:657–701

Send a command to the daemon via socket.

(cmd: str, data: Dict = None, timeout: float = 60.0)

def send_command(cmd: str, data: Dict = None, timeout: float = 60.0) -> Dict:
    """Send a command to the daemon via socket."""
    if not SOCKET_FILE.exists():
        raise ConnectionError("Daemon socket not found. Is the daemon running?")

    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    sock.settimeout(timeout)

    try:
        sock.connect(str(SOCKET_FILE))

        # Send request
        request = {"cmd": cmd, "data": data or {}}
        sock.sendall(json.dumps(request).encode('utf-8'))

        # Receive response
        response_data = b""
        while True:
            chunk = sock.recv(65536)
            if not chunk:
                break
            response_data += chunk
        
        if not response_data:
            raise ConnectionError("No response from daemon. Connection closed.")

        response = json.loads(response_data.decode('utf-8'))
        
        # Check for error response
        if response.get("status") == "error":
            raise RuntimeError(response.get("message", "Unknown error"))
        
        return response

    except socket.timeout:
        raise ConnectionError("Connection timed out. Daemon may be busy.")
    except socket.error as e:
        raise ConnectionError(f"Socket error: {e}. Is the daemon running?")
    except json.JSONDecodeError as e:
        raise RuntimeError(f"Invalid response from daemon: {e}")
    finally:
        try:
            sock.close()
        except:
            pass


def daemon_status() -> Dict: