#include "controller_connector.h"
#include "netio.h"
#include "uv_ipc.h"
#include "xcecd_cgroups.h"
#include "xcecd_exdata.h"
#include "xcecd_worker_utils.h"
#include "xcelog.h"
#include <errno.h>
#include <fcntl.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#ifdef _WIN32
#include <windows.h>
#else
#include <pthread.h>
#include <unistd.h>
#endif

#ifdef XRUN
hostattrs_t host_attrs = { 0 };
#else
extern hostattrs_t host_attrs;
#endif

xcecli_ctx_t                xcecli_ctx     = { 0 };
xce_client_config_t         xce_client_cfg = { 0 };
controller_connector_ctx_t* cc_ctx         = NULL;

static int myreq_handler(xce_request_t*, void*);

/* make compiler happy */
typedef struct {
    task_func_t function;
    void*       arg;
} thread_wrapper_t;

static const cc_callbacks_t cc_callbacks
    = { .job_finished     = xcecd_job_finished,
        .job_stopped      = xcecd_job_stopped,
        .job_resumed      = xcecd_job_resumed,
        .job_start_error  = xcecd_job_start_error,
        .job_status_error = xcecd_job_status_error,
        .job_ckpted       = xcecd_job_ckpted,
        .job_remote_start = xcecd_job_remote_start,
        .reply_startjob   = xcecd_reply_startjob };

static void free_startinfo(job_startinfo_t* info)
{
    if (!info) {
        return;
    }
    FREEUP(info->cmdline);
    FREEUP(info->execuser);
    FREEUP(info->exdata);
    FREEUP(info->exechome);
    FREEUP(info->execcwd);
    FREEUP(info->q_pre_exec);
    FREEUP(info->q_post_exec);
    FREEUP(info->job_pre_exec);
    FREEUP(info->job_post_exec);
    FREEUP(info);
}

static void free_limits(res_limits_t* limits)
{
    if (!limits) {
        return;
    }
    FREEUP(limits->cpu_affinity);
    FREEUP(limits->io_device_path);
    FREEUP(limits->user_name);
    FREEUP(limits);
}

/**
 * @brief Initializes the host attributes with default values.
 */
const cc_callbacks_t* get_cc_callbacks(void) { return &cc_callbacks; }

static void* thread_start_wrapper(void* arg)
{
    thread_wrapper_t* wrapper = (thread_wrapper_t*)arg;
    wrapper->function(wrapper->arg);
    FREEUP(wrapper);
    return NULL;
}

/* my old implementation and currently is disabled */
int start_controller_connector_thread(task_func_t function, void* arg)
{
    XCECD_LOG(XCE_LOG_DEBUG, "Starting controller connector thread...");
#ifdef _WIN32
    HANDLE status_thread
        = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)function, arg, 0, NULL);
    if (status_thread == NULL) {
        return -1;
    }
    CloseHandle(status_thread);
#else
    pthread_t         status_thread;
    thread_wrapper_t* wrapper = calloc(1, sizeof(thread_wrapper_t));
    if (!wrapper) {
        return -1;
    }
    wrapper->function = function;
    wrapper->arg      = arg;
    if (pthread_create(&status_thread, NULL, thread_start_wrapper, wrapper)
        != 0) {
        FREEUP(wrapper);
        return -1;
    }
    pthread_detach(status_thread);
#endif

    XCECD_LOG(XCE_LOG_DEBUG, "Controller connector thread started.");
    return 0;
}

/**
 * @brief Initializes the client configuration for the controller connector.
 *
 * @param config Pointer to the xce_client_config_t structure to be initialized.
 */
static void init_client_config(xce_client_config_t* config)
{
    int  ret;
    char local_host[MAXHOSTNAMELEN] = { 0 };

    if (!config) {
        return;
    }

    ret = gethostname(local_host, sizeof(local_host));
    if (ret < 0) {
        XCECD_LOG(XCE_LOG_ERROR, "gethostname failed: %s", strerror(errno));
        strncpy(local_host, "localhost", sizeof(local_host) - 1);
    }

    config->nmasters        = 1;
    config->masters[0]      = cc_ctx->remote_host;
    config->master_ports[0] = cc_ctx->remote_port;
#ifdef VIRTUAL_CLUSTER
    config->from = strdup(cc_ctx->virtual_name);
#else
    config->from = strdup(local_host);
#endif
    config->key = strdup("keepSecret");

    XCECD_LOG(XCE_LOG_DEBUG, "Controller connector configuration initialized.");
    XCECD_LOG(XCE_LOG_DEBUG,
              "Join cluster configuration - Local host: %s, Remote host: %s, "
              "Remote port: %d, Key: %s",
              config->from, config->masters[0], config->master_ports[0],
              config->key);
}

static void handle_disconnect()
{
    XCECD_LOG(XCE_LOG_DEBUG, "Handling disconnect and shutdown XCECD...");

    // Shutdown worker and waiting for restart
    cc_ctx->running = false;

    XCECD_LOG(XCE_LOG_DEBUG, "Disconnect handled.");
}

/**
 * @brief Handles I/O events for the endpoint, focusing on error conditions
 *        and connection liveness checks using EPOLLIN/EPOLLOUT.
 *
 * @return int Always returns 0.
 */
int checking_connection()
{
    const int MAX_RETRY_COUNT = 5;
    const int RETRY_INTERVAL  = 5; // seconds
    int       retry_count     = 0;
    int       ret;

    while (retry_count < MAX_RETRY_COUNT) {
        ret = xcecli_hello(xcecli_ctx, "hello");
        if (ret == 0) {
            // Connection successful
            if (retry_count > 0) {
                XCECD_LOG(XCE_LOG_DEBUG,
                          "Connection to controller restored after %d retries",
                          retry_count);
            }
            return 0;
        }

        retry_count++;
        XCECD_LOG(XCE_LOG_ERROR,
                  "Connection to controller is broken. Retry %d/%d failed",
                  retry_count, MAX_RETRY_COUNT);

        if (0 == xcecd_join_cluster(&host_attrs)) {
            XCECD_LOG(XCE_LOG_INFO,
                      "Rejoined cluster successfully after retry %d",
                      retry_count);

            // api_ctx_t* api_ctx = (api_ctx_t*)xcecli_ctx;
            // cc_ctx->xce_ctx  = &(api_ctx->ctx);
            return 0;
        }

        if (retry_count < MAX_RETRY_COUNT) {
            XCECD_LOG(XCE_LOG_INFO, "Retrying connection in %d seconds...",
                      RETRY_INTERVAL);
            sleep(RETRY_INTERVAL);
        }
    }

    // All retries exhausted
    XCECD_LOG(XCE_LOG_ERROR,
              "Connection to controller failed after %d retries. Handling "
              "disconnect...",
              MAX_RETRY_COUNT);
    handle_disconnect();

    return 0;
}

/**
 * @brief Joins the cluster using the provided host attributes.
 *
 * @param attrs Pointer to the host attributes structure.
 * @return int 0 on success, -1 on failure.
 */
int xcecd_join_cluster(hostattrs_t* attrs)
{
    if (!attrs) {
        return -1;
    }

    int rc;
    int role = HostRole_worker;

    xcecli_ctx_init(&xcecli_ctx, &xce_client_cfg);
    xcecli_set_handler(xcecli_ctx, &myreq_handler, NULL);
    xcecli_start(xcecli_ctx);

    rc = xcecli_join(xcecli_ctx, role, attrs);

    if (rc < 0) {
        errno = -rc;
        XCECD_LOG(XCE_LOG_ERROR, "Call xcecli_join() failure: %s",
                  strerror(errno));
        return rc;
    }

    return 0;
}

#ifndef XRUN
static void dump_startjob_request(xce_job_exec_t* d)
{
    if (!d) {
        return;
    }

    XCECD_LOG(XCE_LOG_DEBUG,
              "Job ID: %lu, Command: %s, Exec User: %s, Exdata: %s", d->jobid,
              d->start_info->cmdline ? d->start_info->cmdline : "NULL",
              d->start_info->execuser ? d->start_info->execuser : "NULL",
              d->start_info->exdata ? d->start_info->exdata : "NULL");
}
#endif // XRUN

static char* create_mpi_cmdline(char* cmd, char* type, xce_launch_t* launchinfo)
{
    if (!cmd || !type || !launchinfo || launchinfo->nhosts <= 0) {
        XCECD_LOG(XCE_LOG_ERROR, "Invalid parameters for create_mpi_cmdline");
        return NULL;
    }

    char* mpi_executable = NULL;
    char* app_command    = NULL;
    char* token          = NULL;
    char* cmd_copy       = strdup(cmd);
    if (!cmd_copy) {
        XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for cmd_copy");
        return NULL;
    }

    // Parse the original command to extract mpi executable path and application
    // command
    token = strtok(cmd_copy, " ");
    if (!token) {
        XCECD_LOG(XCE_LOG_ERROR, "Failed to parse command: %s", cmd);
        FREEUP(cmd_copy);
        return NULL;
    }

    // Check if first token contains mpirun or mpiexec
    if (strstr(token, "mpirun") || strstr(token, "mpiexec")) {
        mpi_executable = strdup(token);
        // Get the rest of the command as application
        char* rest = cmd + strlen(token);
        while (*rest == ' ')
            rest++; // skip spaces
        app_command = strdup(rest);
    } else {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Command does not contain mpirun or mpiexec: %s", cmd);
        FREEUP(cmd_copy);
        return NULL;
    }

    FREEUP(cmd_copy);

    if (!mpi_executable || !app_command) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Failed to extract mpi executable or application command");
        FREEUP(mpi_executable);
        FREEUP(app_command);
        return NULL;
    }

    // Calculate the size needed for host list
    size_t hostlist_size = 0;
    for (int i = 0; i < launchinfo->nhosts; i++) {
        if (launchinfo->hosts[i] && launchinfo->hosts[i]->host) {
            // host:nrank + comma + null terminator
            hostlist_size += strlen(launchinfo->hosts[i]->host)
                           + 20; // 20 should be enough for :nrank
        }
    }

    if (hostlist_size == 0) {
        XCECD_LOG(XCE_LOG_ERROR, "No valid hosts found in launch info");
        FREEUP(mpi_executable);
        FREEUP(app_command);
        return NULL;
    }

    // Build host list string
    char* hostlist = calloc(hostlist_size, sizeof(char));
    if (!hostlist) {
        XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for hostlist");
        FREEUP(mpi_executable);
        FREEUP(app_command);
        return NULL;
    }

    for (int i = 0; i < launchinfo->nhosts; i++) {
        if (launchinfo->hosts[i] && launchinfo->hosts[i]->host) {
            if (strlen(hostlist) > 0) {
                strcat(hostlist, ",");
            }
            strcat(hostlist, launchinfo->hosts[i]->host);
            strcat(hostlist, ":");

            char nrank_str[16];
            snprintf(nrank_str, sizeof(nrank_str), "%d",
                     launchinfo->hosts[i]->nranks);
            strcat(hostlist, nrank_str);
        }
    }

    char  launcher[MAXFILENAMELEN] = { 0 };
    char* server_dir               = getenv("XCE_SERVER_DIR");
    if (server_dir) {
        snprintf(launcher, sizeof(launcher), "%s/xrun", server_dir);
    } else {
        snprintf(launcher, sizeof(launcher), "/opt/xce/bin/xrun");
    }

    if (access(launcher, X_OK) != 0) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "xrun launcher not found or not executable at %s", launcher);
        FREEUP(mpi_executable);
        FREEUP(app_command);
        FREEUP(hostlist);
        return NULL;
    }

    // Determine the host parameter based on MPI type
    const char* host_param     = NULL;
    const char* launcher_param = NULL;
    if (strcmp(type, "openmpi") == 0) {
        host_param     = "-host";
        launcher_param = "--mca plm_rsh_agent";
    } else if (strcmp(type, "mpich") == 0) {
        host_param     = "-hosts";
        launcher_param = "-launcher-exec";
    } else {
        XCECD_LOG(XCE_LOG_ERROR, "Unsupported MPI type: %s", type);
        FREEUP(mpi_executable);
        FREEUP(app_command);
        FREEUP(hostlist);
        return NULL;
    }

    // Calculate final command size
    size_t final_cmd_size
        = strlen(mpi_executable) + 1 + // executable + space
          strlen(host_param) + 1 +     // host parameter + space
          strlen(hostlist) + 1 +       // hostlist + space
          strlen(launcher_param) + 1 + // launcher parameter + space
          strlen(launcher) + 1 +       // launcher + space
          strlen(app_command) + 1;     // application + null terminator

    char* final_cmd = calloc(final_cmd_size, sizeof(char));
    if (!final_cmd) {
        XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for final command");
        FREEUP(mpi_executable);
        FREEUP(app_command);
        FREEUP(hostlist);
        return NULL;
    }

    // Build the final command
    snprintf(final_cmd, final_cmd_size, "%s %s %s %s %s %s", mpi_executable,
             host_param, hostlist, launcher_param, launcher, app_command);

    XCECD_LOG(XCE_LOG_DEBUG3, "Generated MPI command: %s", final_cmd);

    // Cleanup
    FREEUP(mpi_executable);
    FREEUP(app_command);
    FREEUP(hostlist);

    return final_cmd;
}

/**
 * @brief Runs a job task on remote hosts.
 *
 * @param jobid The ID of the job to run.
 * @param startinfo Pointer to the job start information.
 * @param launchinfo Pointer to the job launch information.
 * @return int 0 on success, -1 on failure.
 */
static int job_start_remote_task(uint64_t jobid, job_startinfo_t* startinfo,
                                 xce_launch_t* launchinfo)
{
    xce_taskhost_t hosts[launchinfo->nhosts - 1];
    xce_taskrun_t  taskrun;
    int            i;
    int            rc;

    if (!startinfo || !launchinfo) {
        XCECD_LOG(XCE_LOG_ERROR, "Invalid parameters for job_run_task_remote");
        return -1;
    }

    if ((strcmp(startinfo->launch_type, "openmpi") == 0)
        || (strcmp(startinfo->launch_type, "mpich") == 0)) {
        XCECD_LOG(XCE_LOG_DEBUG3,
                  "MPI launch detected, using MPI task run method.");
        return 0;
    }

    for (i = 1; i < launchinfo->nhosts; i++) {
        hosts[i - 1].exechome    = startinfo->exechome;
        hosts[i - 1].execcwd     = startinfo->execcwd;
        hosts[i - 1].cmdline     = startinfo->cmdline;
        hosts[i - 1].exdata      = startinfo->exdata;
        hosts[i - 1].execuser    = startinfo->execuser;
        hosts[i - 1].launch_type = startinfo->launch_type;

        hosts[i - 1].host   = launchinfo->hosts[i]->host;
        hosts[i - 1].ntasks = launchinfo->hosts[i]->nranks;

        XCECD_LOG(XCE_LOG_DEBUG3, "Remote task setup for host %s with %d ranks",
                  hosts[i - 1].host, hosts[i - 1].ntasks);
    }

    taskrun.taskhosts = (xce_taskhost_t**)calloc(launchinfo->nhosts - 1,
                                                 sizeof(xce_taskhost_t*));
    taskrun.clientidx = -1;

    if (taskrun.taskhosts != NULL) {
        for (i = 1; i < launchinfo->nhosts; i++) {
            taskrun.taskhosts[i - 1] = &hosts[i - 1];
        }

        taskrun.nhosts = launchinfo->nhosts - 1;
        taskrun.jobid  = jobid;

        XCECD_LOG(XCE_LOG_DEBUG, "Launching %d remote tasks for job %lu",
                  taskrun.nhosts, jobid);

        xcecli_unblock_tasks(xcecli_ctx);
        rc = xcecli_runtask(xcecli_ctx, &taskrun);
        if (rc < 0) {
            XCECD_LOG(XCE_LOG_ERROR, "Job %lu run remote task failed.", jobid);
        }
    } else {
        rc = -ENOMEM;
    }

    FREEUP(taskrun.taskhosts);

    return rc;
}

/**
 * @brief Starts a job task on the local host.
 *
 * @param request Pointer to the xce_request_t structure containing the request
 * data.
 * @return int 0 on success, -1 on failure.
 */
static int job_start_local_task(xce_request_t* request)
{
#ifndef XRUN
    xce_job_exec_t* d = request->body.jobstartmsg;

    dump_startjob_request(d);

    fork_task_t* job_ctx = calloc(1, sizeof(fork_task_t));
    if (job_ctx) {
        job_ctx->startinfo = calloc(1, sizeof(job_startinfo_t));
        if (!job_ctx->startinfo) {
            XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for startinfo");
            xcecd_job_start_error(d->jobid);
            FREEUP(job_ctx);
            return -1;
        }

        if (d->start_info->execuser && d->start_info->execuser[0] != '\0') {
            job_ctx->startinfo->execuser
                = strdup((char*)d->start_info->execuser);
        } else {
            job_ctx->startinfo->execuser = strdup("nobody");
            XCECD_LOG(XCE_LOG_WARN, "No execuser provided in startjob request, "
                                    "defaulting to 'nobody'");
        }

        if (!job_ctx->startinfo->execuser) {
            XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for execuser");
            xcecd_job_start_error(d->jobid);
            free_startinfo(job_ctx->startinfo);
            FREEUP(job_ctx);
            return -1;
        }

#if 0
        char data[1024] = {0};
        snprintf(data, sizeof(data), 
                 "{\"sub_env_vars\": {\"XCE_TEST1\": \"abab\", \"XCE_TEST2\": \"cdcd\"}, \"transfer_file\": [{\"op\": \">\", \"local_file\": \"/tmp/file1.txt\", \"remote_file\": \"\"}, {\"op\": \">\", \"local_file\": \"/tmp/file2.txt\", \"remote_file\": \"\"}], \"interactive\": 1}");
        job_ctx->exdata = strdup(data);
#else

        if (!d->start_info->exdata || d->start_info->exdata[0] == '\0') {
            job_ctx->startinfo->exdata = NULL;
            XCECD_LOG(XCE_LOG_DEBUG3, "No exdata provided in startjob request");
        } else {
            job_ctx->startinfo->exdata = strdup((char*)d->start_info->exdata);
            XCECD_LOG(XCE_LOG_DEBUG3, "Received exdata: %s",
                      job_ctx->startinfo->exdata);

            if (!job_ctx->startinfo->exdata) {
                XCECD_LOG(XCE_LOG_ERROR,
                          "Failed to allocate memory for exdata");
                xcecd_job_start_error(d->jobid);
                free_startinfo(job_ctx->startinfo);
                FREEUP(job_ctx);
                return -1;
            }
        }
#endif

        if (!d->start_info->exechome || d->start_info->exechome[0] == '\0') {
            job_ctx->startinfo->exechome = NULL;
            XCECD_LOG(XCE_LOG_DEBUG3,
                      "No exechome provided in startjob request");
        } else {
            job_ctx->startinfo->exechome
                = strdup((char*)d->start_info->exechome);
            if (!job_ctx->startinfo->exechome) {
                XCECD_LOG(XCE_LOG_ERROR,
                          "Failed to allocate memory for exechome");
                xcecd_job_start_error(d->jobid);
                free_startinfo(job_ctx->startinfo);
                FREEUP(job_ctx);
                return -1;
            }
        }

        if (!d->start_info->execcwd || d->start_info->execcwd[0] == '\0') {
            job_ctx->startinfo->execcwd = NULL;
            XCECD_LOG(XCE_LOG_DEBUG3,
                      "No execcwd provided in startjob request");
        } else {
            job_ctx->startinfo->execcwd = strdup((char*)d->start_info->execcwd);
            if (!job_ctx->startinfo->execcwd) {
                XCECD_LOG(XCE_LOG_ERROR,
                          "Failed to allocate memory for execcwd");
                xcecd_job_start_error(d->jobid);
                free_startinfo(job_ctx->startinfo);
                FREEUP(job_ctx);
                return -1;
            }
        }

        if (!d->start_info->q_pre_exec
            || d->start_info->q_pre_exec[0] == '\0') {
            job_ctx->startinfo->q_pre_exec = NULL;
            XCECD_LOG(XCE_LOG_DEBUG3,
                      "No queue level pre-exec provided in startjob request");
        } else {
            job_ctx->startinfo->q_pre_exec
                = strdup((char*)d->start_info->q_pre_exec);
            if (!job_ctx->startinfo->q_pre_exec) {
                XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for queue "
                                         "level pre-exec command.");
                xcecd_job_start_error(d->jobid);
                free_startinfo(job_ctx->startinfo);
                FREEUP(job_ctx);
                return -1;
            }
        }

        if (!d->start_info->q_post_exec
            || d->start_info->q_post_exec[0] == '\0') {
            job_ctx->startinfo->q_post_exec = NULL;
            XCECD_LOG(XCE_LOG_DEBUG3,
                      "No queue level post-exec provided in startjob request");
        } else {
            job_ctx->startinfo->q_post_exec
                = strdup((char*)d->start_info->q_post_exec);
            if (!job_ctx->startinfo->q_post_exec) {
                XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for queue "
                                         "level post-exec command.");
                xcecd_job_start_error(d->jobid);
                free_startinfo(job_ctx->startinfo);
                FREEUP(job_ctx);
                return -1;
            }
        }

        if (!d->start_info->job_pre_exec
            || d->start_info->job_pre_exec[0] == '\0') {
            job_ctx->startinfo->job_pre_exec = NULL;
            XCECD_LOG(XCE_LOG_DEBUG3,
                      "No job level pre-exec provided in startjob request");
        } else {
            job_ctx->startinfo->job_pre_exec
                = strdup((char*)d->start_info->job_pre_exec);
            if (!job_ctx->startinfo->job_pre_exec) {
                XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for job "
                                         "level pre-exec command.");
                xcecd_job_start_error(d->jobid);
                free_startinfo(job_ctx->startinfo);
                FREEUP(job_ctx);
                return -1;
            }
        }

        if (!d->start_info->job_post_exec
            || d->start_info->job_post_exec[0] == '\0') {
            job_ctx->startinfo->job_post_exec = NULL;
            XCECD_LOG(XCE_LOG_DEBUG3,
                      "No job level post-exec provided in startjob request");
        } else {
            job_ctx->startinfo->job_post_exec
                = strdup((char*)d->start_info->job_post_exec);
            if (!job_ctx->startinfo->job_post_exec) {
                XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for job "
                                         "level post-exec command.");
                xcecd_job_start_error(d->jobid);
                free_startinfo(job_ctx->startinfo);
                FREEUP(job_ctx);
                return -1;
            }
        }

#ifndef DBUS_MANAGER
        job_ctx->manager = (void*)cc_ctx->process_manager;
#endif

        if (d->start_info->launch_type) {
            job_ctx->startinfo->launch_type
                = strdup((char*)d->start_info->launch_type);
        } else {
            job_ctx->startinfo->launch_type = NULL;
        }

        job_ctx->func      = xcecd_process_command;
        job_ctx->xce_jobid = d->jobid;
        job_ctx->ntasks    = d->launchinfo->hosts[0]->nranks;
        job_ctx->clientidx = -1;

        /* Create MPI command line */
        if (strcmp(job_ctx->startinfo->launch_type, "openmpi") == 0
            || strcmp(job_ctx->startinfo->launch_type, "mpich") == 0) {
            job_ctx->startinfo->cmdline
                = create_mpi_cmdline(d->start_info->cmdline,
                                     d->start_info->launch_type, d->launchinfo);
            if (job_ctx->startinfo->cmdline == NULL) {
                XCECD_LOG(XCE_LOG_ERROR, "Failed to create MPI command line");
                xcecd_job_start_error(d->jobid);
                free_startinfo(job_ctx->startinfo);
                FREEUP(job_ctx);
                return -1;
            }

            /* In MPI case, we set ntasks to 1 on all nodes as we have passed
             * the host options command line to all of them.
             */
            job_ctx->ntasks = 1;
        } else {
            job_ctx->startinfo->cmdline = strdup((char*)d->start_info->cmdline);
            if (!job_ctx->startinfo->cmdline) {
                XCECD_LOG(XCE_LOG_ERROR,
                          "Failed to allocate memory for command");
                xcecd_job_start_error(d->jobid);
                free_startinfo(job_ctx->startinfo);
                FREEUP(job_ctx);
                return -1;
            }
        }

        XCECD_LOG(XCE_LOG_DEBUG, "Job %lu ntasks is: %d", d->jobid,
                  job_ctx->ntasks);

        /* Job limits */
        job_ctx->limits = (res_limits_t*)calloc(1, sizeof(res_limits_t));
        if (!job_ctx->limits) {
            XCECD_LOG(XCE_LOG_ERROR,
                      "Failed to allocate memory for job limits");
            xcecd_job_start_error(d->jobid);
            free_startinfo(job_ctx->startinfo);
            FREEUP(job_ctx);
            return -1;
        }

        /* Even exdata is empty, we still need to initialize limits as for
         * runtime user */
        if (!parse_limits(job_ctx->startinfo->exdata, job_ctx->limits)) {
            XCECD_LOG(XCE_LOG_ERROR, "Failed to parse job limits from exdata");
            xcecd_job_start_error(d->jobid);
            free_startinfo(job_ctx->startinfo);
            free_limits(job_ctx->limits);
            FREEUP(job_ctx);
            return -1;
        }

        if (job_ctx->startinfo->execuser != NULL) {
            job_ctx->limits->user_name = strdup(job_ctx->startinfo->execuser);
        } else {
            job_ctx->limits->user_name = strdup("nobody");
            XCECD_LOG(XCE_LOG_WARN,
                      "No execuser provided, defaulting to 'nobody'");
        }

        job_ctx->limits->cpu_affinity   = NULL;
        job_ctx->limits->io_device_path = NULL;

        if (submit_fork_task(cc_ctx->thread_pool, job_ctx) != 0) {
            XCECD_LOG(XCE_LOG_ERROR, "Failed to add task to thread pool");

            xcecd_job_start_error(d->jobid);
            free_startinfo(job_ctx->startinfo);
            free_limits(job_ctx->limits);
            FREEUP(job_ctx);
            return -1;
        } else {
            XCECD_LOG(XCE_LOG_DEBUG, "Task added to thread pool queue");
        }
    } else {
        XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for task context");
        xcecd_job_start_error(d->jobid);
        return -1;
    }
#endif
    return 0;
}

static int handle_startjob(xce_request_t* request)
{
    int rc = 0;

    XCECD_LOG(XCE_LOG_DEBUG, "Received startjob request from controller");

    if (request->body.jobstartmsg == NULL
        || request->body.jobstartmsg->start_info == NULL
        || request->body.jobstartmsg->launchinfo == NULL) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Received startjob request with invalid parameters");
        return -1;
    }

    if (request->body.jobstartmsg->jobid <= 0) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Received startjob request with invalid job ID");
        return -1;
    }

    uint64_t jobid = request->body.jobstartmsg->jobid;

    if (request->body.jobstartmsg->start_info->cmdline == NULL) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Received startjob request with empty command");
        xcecd_job_start_error(jobid);
        return -1;
    }

    if (request->body.jobstartmsg->start_info->execuser == NULL) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Received startjob request with empty execuser");
        xcecd_job_start_error(jobid);
        return -1;
    }

    job_startinfo_t* startinfo  = request->body.jobstartmsg->start_info;
    xce_launch_t*    launchinfo = request->body.jobstartmsg->launchinfo;

    if (launchinfo->nhosts > 1) {
        rc = job_start_remote_task(jobid, startinfo, launchinfo);
        if (rc < 0) {
            XCECD_LOG(XCE_LOG_ERROR, "Job %lu run remote task failed, rc %d.",
                      jobid, rc);
            xcecd_task_fail(jobid, request->body.runtaskmsg->clientidx, -1, rc);
            return rc;
        }
    }

    rc = job_start_local_task(request);

    return rc;
}

/**
 * @brief Handles the runtasks request from the controller.
 *
 * @param request Pointer to the xce_request_t structure containing the request
 * data.
 * @return int 0 on success, -1 on failure.
 */
static int handle_runtasks(xce_request_t* request)
{
#ifndef XRUN
    XCECD_LOG(XCE_LOG_DEBUG, "Received runtasks request from controller");

    xce_taskrun_t* runmsg = request->body.runtaskmsg;

    if (!runmsg) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Received runtasks request with invalid message");
        return -1;
    }

    if (runmsg->jobid <= 0) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Received runtasks request with invalid job ID");
        return -1;
    }

    if (runmsg->nhosts != 1) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Received runtasks request with invalid number of hosts: %d",
                  runmsg->nhosts);
        xcecd_task_fail(runmsg->jobid, runmsg->clientidx, -1, -EINVAL);
        return -1;
    }

    if (runmsg->taskhosts[0]->host == NULL) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Received runtasks request with empty host name");
        return -1;
    }

    if (strcmp(xce_client_cfg.from, runmsg->taskhosts[0]->host) != 0) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Received runtasks request for a different host: %s",
                  runmsg->taskhosts[0]->host);
        xcecd_task_fail(runmsg->jobid, runmsg->clientidx, -1, -EINVAL);
        return -1;
    }

    XCECD_LOG(XCE_LOG_DEBUG3, "run task: jobid = %lu", runmsg->jobid);
    XCECD_LOG(XCE_LOG_DEBUG3, "run task: user = %s",
              runmsg->taskhosts[0]->execuser ? runmsg->taskhosts[0]->execuser
                                             : "NULL");
    XCECD_LOG(XCE_LOG_DEBUG3, "run task: cmdline = %s",
              runmsg->taskhosts[0]->cmdline ? runmsg->taskhosts[0]->cmdline
                                            : "NULL");
    XCECD_LOG(XCE_LOG_DEBUG3, "run task: exdata = %s",
              runmsg->taskhosts[0]->exdata ? runmsg->taskhosts[0]->exdata
                                           : "NULL");
    XCECD_LOG(XCE_LOG_DEBUG3, "run task: ntasks = %d",
              runmsg->taskhosts[0]->ntasks);
    XCECD_LOG(XCE_LOG_DEBUG3, "run task: exechome = %s",
              runmsg->taskhosts[0]->exechome ? runmsg->taskhosts[0]->exechome
                                             : "NULL");
    XCECD_LOG(XCE_LOG_DEBUG3, "run task: execcwd = %s",
              runmsg->taskhosts[0]->execcwd ? runmsg->taskhosts[0]->execcwd
                                            : "NULL");

    fork_task_t* task_ctx = calloc(1, sizeof(fork_task_t));
    if (task_ctx) {
        task_ctx->startinfo = calloc(1, sizeof(job_startinfo_t));
        if (!task_ctx->startinfo) {
            XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for startinfo");
            FREEUP(task_ctx);
            return -1;
        }

        task_ctx->startinfo->cmdline
            = strdup((char*)runmsg->taskhosts[0]->cmdline);
        if (!task_ctx->startinfo->cmdline) {
            XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for command");
            free_startinfo(task_ctx->startinfo);
            FREEUP(task_ctx);
            xcecd_task_fail(runmsg->jobid, runmsg->clientidx, -1, -ENOMEM);
            return -1;
        }

        task_ctx->startinfo->execuser
            = strdup((char*)runmsg->taskhosts[0]->execuser);
        if (!task_ctx->startinfo->execuser) {
            XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for execuser");
            free_startinfo(task_ctx->startinfo);
            FREEUP(task_ctx);
            xcecd_task_fail(runmsg->jobid, runmsg->clientidx, -1, -ENOMEM);
            return -1;
        }

        task_ctx->ntasks = runmsg->taskhosts[0]->ntasks;
        if (task_ctx->ntasks <= 0) {
            XCECD_LOG(
                XCE_LOG_ERROR,
                "Received runtasks request with invalid number of tasks: %d",
                task_ctx->ntasks);
            free_startinfo(task_ctx->startinfo);
            FREEUP(task_ctx);
            xcecd_task_fail(runmsg->jobid, runmsg->clientidx, -1, -EINVAL);
            return -1;
        }

        if (!runmsg->taskhosts[0]->exdata
            || runmsg->taskhosts[0]->exdata[0] == '\0') {
            task_ctx->startinfo->exdata = NULL;
            XCECD_LOG(XCE_LOG_DEBUG3, "No exdata provided in runtasks request");
        } else {
            task_ctx->startinfo->exdata
                = strdup((char*)runmsg->taskhosts[0]->exdata);
            XCECD_LOG(XCE_LOG_DEBUG3, "Received exdata: %s",
                      task_ctx->startinfo->exdata);

            if (!task_ctx->startinfo->exdata) {
                XCECD_LOG(XCE_LOG_ERROR,
                          "Failed to allocate memory for exdata");
                free_startinfo(task_ctx->startinfo);
                FREEUP(task_ctx);
                xcecd_task_fail(runmsg->jobid, runmsg->clientidx, -1, -ENOMEM);
                return -1;
            }
        }

        if (!runmsg->taskhosts[0]->exechome
            || runmsg->taskhosts[0]->exechome[0] == '\0') {
            task_ctx->startinfo->exechome = NULL;
            XCECD_LOG(XCE_LOG_DEBUG3,
                      "No exechome provided in runtasks request");
        } else {
            task_ctx->startinfo->exechome
                = strdup((char*)runmsg->taskhosts[0]->exechome);
            if (!task_ctx->startinfo->exechome) {
                XCECD_LOG(XCE_LOG_ERROR,
                          "Failed to allocate memory for exechome");
                free_startinfo(task_ctx->startinfo);
                FREEUP(task_ctx);
                xcecd_task_fail(runmsg->jobid, runmsg->clientidx, -1, -ENOMEM);
                return -1;
            }
        }

        if (!runmsg->taskhosts[0]->execcwd
            || runmsg->taskhosts[0]->execcwd[0] == '\0') {
            task_ctx->startinfo->execcwd = NULL;
            XCECD_LOG(XCE_LOG_DEBUG3,
                      "No execcwd provided in runtasks request");
        } else {
            task_ctx->startinfo->execcwd
                = strdup((char*)runmsg->taskhosts[0]->execcwd);
            if (!task_ctx->startinfo->execcwd) {
                XCECD_LOG(XCE_LOG_ERROR,
                          "Failed to allocate memory for execcwd");
                free_startinfo(task_ctx->startinfo);
                FREEUP(task_ctx);
                xcecd_task_fail(runmsg->jobid, runmsg->clientidx, -1, -ENOMEM);
                return -1;
            }
        }

        /* currently we do not consider pre/post exec for tasks */
#if 0
        if (!runmsg->taskhosts[0]->q_pre_exec ||
            runmsg->taskhosts[0]->q_pre_exec[0] == '\0') {
                task_ctx->startinfo->q_pre_exec = NULL;
                XCECD_LOG(XCE_LOG_DEBUG3,
                          "No queue level pre-exec provided in runtasks request");
        } else {
            task_ctx->startinfo->q_pre_exec = strdup((char*)runmsg->taskhosts[0]->q_pre_exec);
            if (!task_ctx->startinfo->q_pre_exec) {
                XCECD_LOG(XCE_LOG_ERROR,
                          "Failed to allocate memory for queue level pre-exec command.");
                free_startinfo(task_ctx->startinfo);
                FREEUP(task_ctx);
                xcecd_task_fail(runmsg->jobid,
                                runmsg->clientidx,
                                -1, -ENOMEM);
                return -1;
            }
        }

        if (!runmsg->taskhosts[0]->q_post_exec ||
            runmsg->taskhosts[0]->q_post_exec[0] == '\0') {
                task_ctx->startinfo->q_post_exec = NULL;
                XCECD_LOG(XCE_LOG_DEBUG3,
                          "No queue level post-exec provided in runtasks request");
        } else {
            task_ctx->startinfo->q_post_exec = strdup((char*)runmsg->taskhosts[0]->q_post_exec);
            if (!task_ctx->startinfo->q_post_exec) {
                XCECD_LOG(XCE_LOG_ERROR,
                          "Failed to allocate memory for queue level post-exec command.");
                free_startinfo(task_ctx->startinfo);
                FREEUP(task_ctx);
                xcecd_task_fail(runmsg->jobid,
                                runmsg->clientidx,
                                -1, -ENOMEM);
                return -1;
            }
        }

        if (!runmsg->taskhosts[0]->job_pre_exec ||
            runmsg->taskhosts[0]->job_pre_exec[0] == '\0') {
                task_ctx->startinfo->job_pre_exec = NULL;
                XCECD_LOG(XCE_LOG_DEBUG3,
                          "No job level pre-exec provided in runtasks request");
        } else {
            task_ctx->startinfo->job_pre_exec = strdup((char*)runmsg->taskhosts[0]->job_pre_exec);
            if (!task_ctx->startinfo->job_pre_exec) {
                XCECD_LOG(XCE_LOG_ERROR,
                          "Failed to allocate memory for job level pre-exec command.");
                free_startinfo(task_ctx->startinfo);
                FREEUP(task_ctx);
                xcecd_task_fail(runmsg->jobid,
                                runmsg->clientidx,
                                -1, -ENOMEM);
                return -1;
            }
        }

        if (!runmsg->taskhosts[0]->job_post_exec ||
            runmsg->taskhosts[0]->job_post_exec[0] == '\0') {
                task_ctx->startinfo->job_post_exec = NULL;
                XCECD_LOG(XCE_LOG_DEBUG3,
                          "No job level post-exec provided in runtasks request");
        } else {
            task_ctx->startinfo->job_post_exec = strdup((char*)runmsg->taskhosts[0]->job_post_exec);
            if (!task_ctx->startinfo->job_post_exec) {
                XCECD_LOG(XCE_LOG_ERROR,
                          "Failed to allocate memory for job level post-exec command.");
                free_startinfo(task_ctx->startinfo);
                FREEUP(task_ctx);
                xcecd_task_fail(runmsg->jobid,
                                runmsg->clientidx,
                                -1, -ENOMEM);
                return -1;
            }
        }
#endif

#ifndef DBUS_MANAGER
        task_ctx->manager = (void*)cc_ctx->process_manager;
#endif

        task_ctx->func      = xcecd_process_command;
        task_ctx->xce_jobid = runmsg->jobid;
        task_ctx->clientidx = runmsg->clientidx;

        if (submit_fork_task(cc_ctx->thread_pool, task_ctx) != 0) {
            XCECD_LOG(XCE_LOG_ERROR, "Failed to add task to thread pool");
            free_startinfo(task_ctx->startinfo);
            FREEUP(task_ctx);
            xcecd_task_fail(runmsg->jobid, runmsg->clientidx, -1, -ENOMEM);
            return -1;
        } else {
            XCECD_LOG(XCE_LOG_DEBUG, "Task added to thread pool queue");
        }

    } else {
        XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for task context");
        xcecd_task_fail(runmsg->jobid, runmsg->clientidx, -1, -ENOMEM);
        return -1;
    }
#endif // XRUN
    return 0;
}

/**
 * @brief Handles the runtasks request from the controller.
 *
 * @param request Pointer to the xce_request_t structure containing the request
 * data.
 * @return int 0 on success, -1 on failure.
 */
static int handle_jobcontrol(xce_request_t* request)
{
#ifndef XRUN
    XCECD_LOG(XCE_LOG_DEBUG, "Received job control request from controller");
    xce_jobctrl_t* d = request->body.jobctrlmsg;

    if (d->jobid <= 0) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Received job control request with invalid job ID");
        return -1;
    }

    if (d->action == JobControl_u_siginfo) {
        XCECD_LOG(
            XCE_LOG_DEBUG,
            "Received job control request to send SIGINFO for job ID: %lu",
            d->jobid);

        job_control_task_t* task_ctx = calloc(1, sizeof(job_control_task_t));
        if (!task_ctx) {
            XCECD_LOG(XCE_LOG_ERROR,
                      "Failed to allocate memory for job control task context");
            return -1;
        }

        task_ctx->jobid  = d->jobid;
        task_ctx->signal = d->u.siginfo->signo;
        task_ctx->force  = d->u.siginfo->force;

        xcecd_worker_t* worker = find_worker_by_jobid(
            (xcecd_worker_t*)*cc_ctx->worker_list, d->jobid);

        if (worker) {
            if (worker->job_status == JOB_PENDING
                || worker->job_status == JOB_STARTING) {
                XCECD_LOG(XCE_LOG_ERROR,
                          "Job is pending or starting, controller should not "
                          "pass the signal to %lu.",
                          d->jobid);

                // really should not happen, but if it does, we handle it
                // gracefully
                xcecd_job_status_error(d->jobid);
                FREEUP(task_ctx);
                return 0;
            }
        } else {
            XCECD_LOG(XCE_LOG_ERROR, "Failed to find worker for job ID: %lu",
                      d->jobid);
            FREEUP(task_ctx);
            return -1;
        }

#if !defined(VIRTUAL_CLUSTER)
        if (!ipc_master_send_job_signal(task_ctx->jobid, task_ctx->signal,
                                        task_ctx->force)) {
            XCECD_LOG(XCE_LOG_ERROR,
                      "Failed to submit job control task for job ID: %lu",
                      d->jobid);
            FREEUP(task_ctx);
            return -1;
        } else {
            XCECD_LOG(XCE_LOG_DEBUG,
                      "Job control task submitted for job ID: %lu", d->jobid);
        }
        FREEUP(task_ctx);
#else
        // For virtual cluster, we handle job control tasks differently
        XCECD_LOG(XCE_LOG_DEBUG,
                  "Virtual cluster mode, handling job control task directly");
        FREEUP(task_ctx);
        XCECD_LOG(XCE_LOG_DEBUG,
                  "Job control task handled directly for job ID: %lu",
                  d->jobid);

        worker->killed_signal = d->u.siginfo->signo; // Mark job as killed
#endif
    } else if (d->action == JobControl_u_ckptoption) {
#if !defined(VIRTUAL_CLUSTER)
        XCECD_LOG(XCE_LOG_DEBUG,
                  "Received job control request to set checkpoint options for "
                  "job ID: %lu",
                  d->jobid);

        if (!ipc_master_send_job_ckpt(d->jobid, d->u.ckptoption->kill,
                                      d->u.ckptoption->method)) {
            XCECD_LOG(XCE_LOG_ERROR,
                      "Failed to send checkpoint options for job ID: %lu",
                      d->jobid);
            return -1;
        } else {
            XCECD_LOG(XCE_LOG_DEBUG, "Checkpoint options sent for job ID: %lu",
                      d->jobid);
        }
#else
        XCECD_LOG(XCE_LOG_WARN, "Received job control request in virtual "
                                "cluster or xrun mode, which is not supported");
#endif
    } else {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Received unknown job control action: %d for job ID: %lu",
                  d->action, d->jobid);
        return -1;
    }
#endif // XRUN
    return 0;
}

/**
 * @brief Handles incoming requests from the controller.
 *
 * @param ctx Pointer to the I/O context structure.
 * @return int 0 on success, -1 on failure.
 */
static int myreq_handler(xce_request_t* request, void* args)
{
    int ret = 0;

    XCECD_LOG(XCE_LOG_DEBUG, "Calling request handler...");

    switch (request->opcode) {
    case OPCode_startjob:
        ret = handle_startjob(request);
        break;
    case OPCode_runtasks:
        ret = handle_runtasks(request);
        break;
    case OPCode_jobcontrol:
        ret = handle_jobcontrol(request);
        break;
    default:
        break;
    }

    return ret;
}

/**
 * @brief Sends a start job reply to the controller.
 *
 * @param job Pointer to the job structure.
 * @return bool true on success, false on failure.
 */
bool xcecd_reply_startjob(xcecd_job_t* job)
{
    /* clang-format off */
    //xce_ctx_t* ctx          = cc_ctx->xce_ctx;
    unsigned int pids[] = {job->job_pid};
    job_pidinfo_t pidinfo = {
        .jobid = job->job_id,
        .npids = 1,
        .pids = &pids[0],
        .npgids = 0,
        .pgids = NULL,
        .exdata = NULL,
    };

    if (job->exdata[0] != '\0') {
        pidinfo.exdata = strdup(job->exdata);
    }

    xcecli_notify_jobstart(xcecli_ctx, &pidinfo);

    FREEUP(pidinfo.exdata);

    return true;
}

/**
 * @brief Starts the controller connector by joining the cluster and setting up
 *        the endpoint handler.
 *
 * @return bool true on success, false on failure.
 */
bool controller_connector_start(void)
{
    int        rc;
    int        retry_limit = 0;
    static int error_count = 0;

    init_client_config(&xce_client_cfg);

    do {
        rc = xcecd_join_cluster(&host_attrs);
        if (rc < 0) {
            XCECD_LOG(XCE_LOG_DEBUG,
                      "Failed to join cluster, retrying in %d seconds",
                      cc_ctx->join_interval);
            error_count++;
            if ((retry_limit > 0) && (error_count >= retry_limit)) {
                XCECD_LOG(XCE_LOG_ERROR,
                          "Failed to join cluster after %d retries",
                          retry_limit);
                return false;
            }
        }
        sleep(cc_ctx->join_interval);
    } while (rc < 0 && cc_ctx->running);

    // api_ctx_t* api_ctx = (api_ctx_t*)xcecli_ctx;
    // cc_ctx->xce_ctx  = &(api_ctx->ctx);

    return true;
}

int xcecd_notify_job_fail(uint64_t jobid,
                          int      stage,
                          int      code)
{
    job_failinfo_t failinfo = {
        .jobid = jobid,
        .stage = stage,
        .code  = code,
        .detail  = "no detail information"
    };

    XCECD_LOG(XCE_LOG_DEBUG, "Job failed and calling xcecli_notify_jobfail...");
    return xcecli_notify_jobfail(xcecli_ctx, &failinfo);
}

/**
 * @brief Marks a job as finished and sends the exit code to the controller.
 *
 * @param job Pointer to the job structure.
 * @return int 0 on success, -1 on failure.
 */
int xcecd_job_finished(xcecd_job_t* job)
{
    if (job->exit_code != 0) {
        return xcecd_notify_job_fail(job->job_id, 
                                     job->stage,
                                     job->exit_code);
    }

    xce_jobfinish_t myjob = {
        .jobid     = job->job_id,
        .exit_code = 0
    };

    XCECD_LOG(XCE_LOG_DEBUG, "calling xcecli_jobfinish...");
    return xcecli_notify_jobfinish(xcecli_ctx, &myjob);
}

/**
 * @brief Marks a job as stopped and sends a notification to the controller.
 *
 * @param jobid The ID of the job to be marked as stopped.
 * @return int 0 on success, -1 on failure.
 */
int xcecd_job_stopped(uint64_t jobid)
{
    XCECD_LOG(XCE_LOG_DEBUG,
              "Job stopped and calling xcecli_notify_jobstop...");

    return xcecli_notify_jobstop(xcecli_ctx, jobid);
}

/**
 * @brief Marks a job as resumed and sends a notification to the controller.
 *
 * @param jobid The ID of the job to be marked as resumed.
 * @return int 0 on success, -1 on failure.
 */
int xcecd_job_resumed(uint64_t jobid)
{
    XCECD_LOG(XCE_LOG_DEBUG,
              "Job resumed and calling xcecli_notify_jobresume...");

    return xcecli_notify_jobresume(xcecli_ctx, jobid);
}

/**
 * @brief Reply job start error information to controller
 *        for some cases, we need to requeue the job
 *        and let controller to retry
 * @return int 0 on success, -1 on failure.
 */
int xcecd_job_start_error(uint64_t jobid)
{
    return xcecd_notify_job_fail(jobid, JOB_EXEC_STAGE_MAIN, -1);
}

/**
 * @brief Reply job status error information to controller
 *        for some cases, we need to requeue the job
 *        and let controller to retry
 * @return int 0 on success, -1 on failure.
 */
int xcecd_job_status_error(uint64_t jobid)
{
    xce_jobfinish_t myjob = {
        .jobid     = jobid,
        .exit_code = -1
    };

    XCECD_LOG(XCE_LOG_DEBUG, "Job status error and calling xcecli_jobfinish...");
    return xcecli_jobfinish(xcecli_ctx, &myjob);
}

/**
 * @brief Reply job checkpoint result to controller
 * @return int 0 on success, -1 on failure
 */
int xcecd_job_ckpted(uint64_t jobid, int status, const char* info)
{
    if (0 == status) {
        XCECD_LOG(XCE_LOG_DEBUG, "Job checkpoint success and calling xcecli_notify_jobchkpt...");
    } else {
        XCECD_LOG(XCE_LOG_ERROR, "Job checkpoint failed and calling xcecli_notify_jobchkpt...");
    }

    return xcecli_notify_jobchkpt(xcecli_ctx, jobid, (char*)info);
}

/**
 * @brief Starts a job on a remote host.
 *
 * @param hostname The name of the remote host.
 * @param command The command to be executed on the remote host.
 * @return int 0 on success, -1 on failure.
 */
int xcecd_job_remote_start(uint64_t jobid, 
                           const char* hostname, 
                           const char* command)
{
    int rc = 0;

    if (!cc_ctx || !cc_ctx->worker_list) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Controller connector context is not initialized");
        return -1;
    }

    if (!hostname || !command) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Invalid parameters for remote job start");
        return -1;
    }

    if (jobid <= 0) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Invalid job ID for remote job start");
        return -1;
    }
    
    xcecd_worker_t* worker = find_worker_by_jobid((xcecd_worker_t *)*cc_ctx->worker_list, 
                                                  jobid);
    if (!worker) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Worker not found for job ID %lu", jobid);
        return -1;
    }

    xce_taskhost_t host[] = {
        {.exechome    = worker->exechome,
        .execcwd     = worker->execcwd,
        .cmdline     = strdup(command),
        .exdata      = worker->job_exdata,
        .execuser    = worker->execuser,
        .launch_type = worker->launch_type,
        .host        = strdup(hostname),
        .ntasks      = 1},
        {0}
    };

    xce_taskrun_t  taskrun = {
        .jobid = jobid,
        .clientidx = -1,
        .taskhosts = (xce_taskhost_t**)calloc(1, sizeof(xce_taskhost_t*)),
        .nhosts = 1
    };

    taskrun.taskhosts[0] = &host[0];

    rc = xcecli_runtask(xcecli_ctx, &taskrun);
    if (rc < 0) {
        XCECD_LOG(XCE_LOG_ERROR,
                  "Failed to start job %lu on remote host %s, rc: %d",
                  jobid, hostname, rc);
        xcecd_task_fail(jobid, -1, -1, rc);
        return rc;
    }

    XCECD_LOG(XCE_LOG_DEBUG,
              "Remote task of job %lu will run on %s",
              jobid, hostname);
    return 0;
}

int xcecd_task_fail(uint64_t jobid, int clientidx,
                    int taskidx, int code) 
{
    xce_task_failinfo_t failinfo = {
        .jobid = jobid,
        .host = xce_client_cfg.from,
        .clientidx = clientidx,
        .taskidx = taskidx,
        .code = code,
        .detail = NULL
    };

    return xcecli_notify_taskfail(xcecli_ctx, &failinfo);
}

int xcecd_jobpost_done(uint64_t jobid)
{
    XCECD_LOG(XCE_LOG_DEBUG,
              "Job post done and calling xcecli_notify_jobpostdone...");
    return xcecli_notify_jobpostdone(xcecli_ctx, jobid);
}