#include <errno.h>
#include <stdarg.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include "xce/errnox.h"  /* E_XCE_NOT_READY, E_XCE_STANDBY */

#include <openssl/ssl.h>

#include "mylock.h"
#include "netio.h"
#include "request.h"
#include "typemap.h"
#include "utils.h"
#include "xce/types.h"

#define REQUEST_FROM(req)                                                      \
    ((req)->from == NULL)                                                      \
        ? "Unknown"                                                            \
        : (((req)->from->name == NULL) ? "Unknown" : (req->from->name))

int xce_encode(xce_request_t* req, io_packet_t* pkt)
{
    struct capn          c;
    struct Request       rq;
    struct capn_segment* cs;
    Request_ptr          rp;
    int                  rc;

    TRACE("%s request op %d seq %ld from %s", __func__, req->opcode, req->seq,
          REQUEST_FROM(req));

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;

    encode_Request(cs, &rq, req);

    cs = capn_root(&c).seg;
    rp = new_Request(cs);
    write_Request(&rq, rp);

    capn_setp(capn_root(&c), 0, rp.p);

    pkt->sz = capn_size(&c);
    rc      = alloc_packet(pkt);
    if (rc < 0) {
        capn_free(&c);
        return -errno;
    }

    capn_write_mem(&c, (uint8_t*)pkt->data, pkt->sz, 0);
    capn_free(&c);

    return 1;
}

int xce_decode(io_packet_t* pkt, xce_request_t* req)
{
    struct capn    c;
    struct Request rq;
    Request_ptr    rp;

    capn_init_mem(&c, (const uint8_t*)pkt->data, pkt->sz, 0);
    rp.p = capn_getp(capn_root(&c), 0, 1);

    read_Request(&rq, rp);

    decode_Request(req, &rq);

    /* OPCode_upgradeConfig carries sha256 and bytes as List(UInt8) on the wire.
     * decode_UpgradeConfigRequest() intentionally leaves raw_sha256/raw_bytes as
     * NULL (the project's Data union cannot carry raw bytes via decode_Data_ptr).
     * Extract them here using UpgradeConfigRequest_get_* + capn_getv8 BEFORE
     * capn_free() destroys the capnp segment. */
    if (rq.opcode == OPCode_upgradeConfig
        && req->body_which == Request_body_upgradeconfigrequestmsg
        && req->body.upgradeconfigrequestmsg != NULL) {

        upgrade_config_req_t *ucr = req->body.upgradeconfigrequestmsg;
        UpgradeConfigRequest_ptr u_ptr = rq.body.upgradeconfigrequestmsg;

        /* sha256 */
        {
            Data_ptr    dp        = UpgradeConfigRequest_get_sha256(u_ptr);
            capn_list8  sha_list;
            sha_list.p = dp.p;
            if (sha_list.p.type != CAPN_NULL) {
                int slen = capn_len(sha_list);
                if (slen > 0) {
                    uint8_t *sbuf = (uint8_t *)malloc((size_t)slen);
                    if (sbuf && capn_getv8(sha_list, 0, sbuf, slen) == slen) {
                        ucr->raw_sha256 = sbuf;
                    } else {
                        free(sbuf);
                    }
                }
            }
        }

        /* bytes */
        {
            Data_ptr    dp         = UpgradeConfigRequest_get_bytes(u_ptr);
            capn_list8  bytes_list;
            bytes_list.p = dp.p;
            if (bytes_list.p.type != CAPN_NULL) {
                int blen = capn_len(bytes_list);
                if (blen > 0) {
                    uint8_t *bbuf = (uint8_t *)malloc((size_t)blen);
                    if (bbuf && capn_getv8(bytes_list, 0, bbuf, blen) == blen) {
                        ucr->raw_bytes     = bbuf;
                        ucr->raw_bytes_len = (size_t)blen;
                    } else {
                        free(bbuf);
                    }
                }
            }
        }
    }

    /* OPCode_replConfigPush: same List(UInt8) issue — pre-extract raw bytes. */
    if (rq.opcode == OPCode_replConfigPush
        && req->body_which == Request_body_replconfigpushrequestmsg
        && req->body.replconfigpushrequestmsg != NULL) {

        repl_config_push_req_t   *rcpr  = req->body.replconfigpushrequestmsg;
        ReplConfigPushRequest_ptr r_ptr = rq.body.replconfigpushrequestmsg;

        /* sha256 */
        {
            Data_ptr   dp = ReplConfigPushRequest_get_sha256(r_ptr);
            capn_list8 sha_list; sha_list.p = dp.p;
            if (sha_list.p.type != CAPN_NULL && capn_len(sha_list) > 0) {
                int slen = capn_len(sha_list);
                uint8_t *sbuf = (uint8_t *)malloc((size_t)slen);
                if (sbuf && capn_getv8(sha_list, 0, sbuf, slen) == slen)
                    rcpr->raw_sha256 = sbuf;
                else
                    free(sbuf);
            }
        }

        /* bytes */
        {
            Data_ptr   dp = ReplConfigPushRequest_get_bytes(r_ptr);
            capn_list8 bytes_list; bytes_list.p = dp.p;
            if (bytes_list.p.type != CAPN_NULL && capn_len(bytes_list) > 0) {
                int blen = capn_len(bytes_list);
                uint8_t *bbuf = (uint8_t *)malloc((size_t)blen);
                if (bbuf && capn_getv8(bytes_list, 0, bbuf, blen) == blen) {
                    rcpr->raw_bytes     = bbuf;
                    rcpr->raw_bytes_len = (size_t)blen;
                } else {
                    free(bbuf);
                }
            }
        }
    }

    TRACE("%s request op %d seq %ld from %s", __func__, req->opcode, req->seq,
          REQUEST_FROM(req));
    capn_free(&c);

    free_packet(pkt);
    FREEUP(pkt);

    return 0;
}

void init_reply_by_req(xce_reply_t* reply, xce_request_t* req)
{
    reply->seq    = req->seq;
    reply->opcode = req->opcode;
}

/*
 * xce_free_request: free an xce_request_t and ALL owned sub-fields.
 *
 * The generated free_Request() handles the capnp-decoded fields but does
 * not know about raw_sha256/raw_bytes on upgrade_config_req_t (those are
 * extra fields populated by xce_decode() before capn_free()).  This wrapper
 * frees those extra fields first, then calls free_Request().
 *
 * Use this everywhere instead of calling free_Request() directly.
 */
void xce_free_request(xce_request_t *req)
{
    if (req == NULL)
        return;

    /* Free raw_* fields that the generated free_*Request skips */
    if (req->body_which == Request_body_upgradeconfigrequestmsg
        && req->body.upgradeconfigrequestmsg != NULL) {
        upgrade_config_req_t *ucr = req->body.upgradeconfigrequestmsg;
        free(ucr->raw_sha256);
        ucr->raw_sha256    = NULL;
        free(ucr->raw_bytes);
        ucr->raw_bytes     = NULL;
        ucr->raw_bytes_len = 0;
    }

    if (req->body_which == Request_body_replconfigpushrequestmsg
        && req->body.replconfigpushrequestmsg != NULL) {
        repl_config_push_req_t *rcpr = req->body.replconfigpushrequestmsg;
        free(rcpr->raw_sha256);
        rcpr->raw_sha256    = NULL;
        free(rcpr->raw_bytes);
        rcpr->raw_bytes     = NULL;
        rcpr->raw_bytes_len = 0;
    }

    free_Request(req);
}

int xce_request(endpoint_t* ep, xce_request_t* req)
{
    int          rc;
    io_packet_t* pkt;

    pkt = (io_packet_t*)calloc(1, sizeof(*pkt));
    if (pkt == NULL) {
        return -ENOMEM;
    }

    req->seq  = endpoint_getseq(ep);
    req->from = ep->peer;

    rc = xce_encode(req, pkt);
    if (rc < 0) {
        free_packet(pkt);
        free(pkt);

        return rc;
    }

    endpoint_enq(ep, &(pkt->link));
    return 0;
}

int xce_reply(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_reply,
        .body_which = Request_body_replymsg,
    };
    /* clang-format on */
    int          rc;
    io_packet_t* pkt;

    TRACE("%s seq %lu opcode %d status %d data which %d", __func__, reply->seq,
          reply->opcode, reply->status, reply->data_which);

    pkt = (io_packet_t*)calloc(1, sizeof(*pkt));
    if (pkt == NULL) {
        return -ENOMEM;
    }

    request.from          = from;
    request.body.replymsg = reply;

    request.seq = endpoint_getseq(ep);

    rc = xce_encode(&request, pkt);
    if (rc < 0) {
        free_packet(pkt);
        free(pkt);

        return rc;
    }

    endpoint_enq(ep, &(pkt->link));
    return 0;
}

/*
 * xce_replconfigpush_req_raw: send OP_replConfigPush with raw byte arrays.
 *
 * Bypasses encode_ReplConfigPushRequest_ptr (which maps sha256/bytes through
 * xce_data_t — a cluster-data discriminated union that cannot carry raw bytes).
 * Instead, constructs the struct Request / struct ReplConfigPushRequest directly
 * using the capnp wire-format API (capn_new_list8 + capn_setv8), then calls
 * write_Request to serialise.
 */
int xce_replconfigpush_req_raw(endpoint_t *ep, xce_peerinfo_t *from,
                                uint64_t version_tag,
                                const uint8_t *sha256, size_t sha256_len,
                                const uint8_t *config_bytes, size_t config_len)
{
    struct capn               c;
    struct capn_segment      *cs;
    struct Request            rq;
    struct ReplConfigPushRequest push_req;
    ReplConfigPushRequest_ptr push_ptr;
    Request_ptr               rp;
    io_packet_t              *pkt;
    int                       rc;

    if (ep == NULL)
        return -EINVAL;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;

    memset(&rq,       0, sizeof(rq));
    memset(&push_req, 0, sizeof(push_req));

    push_req.versionTag = version_tag;

    /* sha256: write as a List(UInt8) — capn_list8 and capn_data share {capn_ptr p} */
    if (sha256 && sha256_len > 0) {
        capn_list8 sha_list = capn_new_list8(cs, (int)sha256_len);
        capn_setv8(sha_list, 0, sha256, (int)sha256_len);
        push_req.sha256.p = sha_list.p;
    }

    /* config bytes */
    if (config_bytes && config_len > 0) {
        capn_list8 bytes_list = capn_new_list8(cs, (int)config_len);
        capn_setv8(bytes_list, 0, config_bytes, (int)config_len);
        push_req.bytes.p = bytes_list.p;
    }

    push_ptr = new_ReplConfigPushRequest(cs);
    write_ReplConfigPushRequest(&push_req, push_ptr);

    rq.opcode     = OPCode_replConfigPush;
    rq.seq        = endpoint_getseq(ep);
    rq.body_which = Request_body_replconfigpushrequestmsg;
    rq.body.replconfigpushrequestmsg = push_ptr;
    encode_PeerInfo_ptr(cs, &rq.from, from);

    rp = new_Request(cs);
    write_Request(&rq, rp);
    capn_setp(capn_root(&c), 0, rp.p);

    pkt = (io_packet_t *)calloc(1, sizeof(*pkt));
    if (!pkt) {
        capn_free(&c);
        return -ENOMEM;
    }

    pkt->sz = capn_size(&c);
    rc = alloc_packet(pkt);
    if (rc < 0) {
        free(pkt);
        capn_free(&c);
        return -errno;
    }

    capn_write_mem(&c, (uint8_t *)pkt->data, pkt->sz, 0);
    capn_free(&c);

    endpoint_enq(ep, &(pkt->link));
    return 0;
}

/*
 * xce_replconfig_reply_raw: primary sends OP_replConfig reply with real
 * config bytes via an endpoint (uses iomgr queueing).
 *
 * Bypasses encode_ReplConfigReply_ptr (which routes sha256/bytes through
 * xce_data_t — a discriminated cluster-data union that cannot carry raw
 * bytes).  Constructs Request { opcode=OPCode_reply, body=replymsg }
 * with RequestReply { data=replconfigreplymsg } directly via the capnp
 * wire-format API (capn_new_list8 + capn_setv8), then enqueues for send.
 */
int xce_replconfig_reply_raw(endpoint_t *ep, xce_peerinfo_t *from,
                              uint64_t seq, uint64_t tag,
                              const uint8_t *sha256, size_t sha256_len,
                              const uint8_t *config, size_t config_len)
{
    struct capn                c;
    struct capn_segment       *cs;
    struct Request             rq;
    struct RequestReply        rr;
    struct ReplConfigReply     reply_struct;
    ReplConfigReply_ptr        reply_ptr;
    RequestReply_ptr           rr_ptr;
    Request_ptr                rp;
    io_packet_t               *pkt;
    int                        rc;

    if (ep == NULL)
        return -EINVAL;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;

    memset(&rq,           0, sizeof(rq));
    memset(&rr,           0, sizeof(rr));
    memset(&reply_struct, 0, sizeof(reply_struct));

    /* ReplConfigReply fields */
    reply_struct.upToDate   = 0;
    reply_struct.versionTag = tag;

    if (sha256 && sha256_len > 0) {
        capn_list8 sha_list = capn_new_list8(cs, (int)sha256_len);
        capn_setv8(sha_list, 0, sha256, (int)sha256_len);
        reply_struct.sha256.p = sha_list.p;
    }
    if (config && config_len > 0) {
        capn_list8 bytes_list = capn_new_list8(cs, (int)config_len);
        capn_setv8(bytes_list, 0, config, (int)config_len);
        reply_struct.bytes.p = bytes_list.p;
    }

    reply_ptr = new_ReplConfigReply(cs);
    write_ReplConfigReply(&reply_struct, reply_ptr);

    /* RequestReply envelope */
    rr.seq        = seq;
    rr.opcode     = OPCode_replConfig;   /* opcode being replied to */
    rr.status     = 0;
    rr.data_which = RequestReply_data_replconfigreplymsg;
    rr.data.replconfigreplymsg = reply_ptr;

    rr_ptr = new_RequestReply(cs);
    write_RequestReply(&rr, rr_ptr);

    /* Outer Request */
    rq.opcode     = OPCode_reply;
    rq.seq        = seq;
    rq.body_which = Request_body_replymsg;
    rq.body.replymsg = rr_ptr;
    encode_PeerInfo_ptr(cs, &rq.from, from);

    rp = new_Request(cs);
    write_Request(&rq, rp);
    capn_setp(capn_root(&c), 0, rp.p);

    pkt = (io_packet_t *)calloc(1, sizeof(*pkt));
    if (!pkt) {
        capn_free(&c);
        return -ENOMEM;
    }

    pkt->sz = capn_size(&c);
    rc = alloc_packet(pkt);
    if (rc < 0) {
        free(pkt);
        capn_free(&c);
        return -errno;
    }

    capn_write_mem(&c, (uint8_t *)pkt->data, pkt->sz, 0);
    capn_free(&c);

    endpoint_enq(ep, &(pkt->link));
    return 0;
}

/*
 * xce_replconfig_req_ssl: standby sends OP_replConfig over a raw SSL
 * socket (no iomgr).  Used during early-boot config fetch from primary
 * before the iomgr is fully wired.
 *
 * Wire framing matches netio.c: uint64_t size prefix (host byte order)
 * followed by `size` bytes of capnp data.
 */
int xce_replconfig_req_ssl(SSL *ssl, uint64_t version_tag)
{
    struct capn                  c;
    struct capn_segment         *cs;
    struct Request               rq;
    struct ReplConfigRequest     req_struct;
    ReplConfigRequest_ptr        req_ptr;
    Request_ptr                  rp;
    uint8_t                     *buf  = NULL;
    int64_t                      blen = 0;
    int                          rc   = 0;

    if (ssl == NULL)
        return -EINVAL;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;

    memset(&rq,         0, sizeof(rq));
    memset(&req_struct, 0, sizeof(req_struct));

    req_struct.versionTag = version_tag;

    req_ptr = new_ReplConfigRequest(cs);
    write_ReplConfigRequest(&req_struct, req_ptr);

    rq.opcode     = OPCode_replConfig;
    rq.seq        = 0;
    rq.body_which = Request_body_replconfigrequestmsg;
    rq.body.replconfigrequestmsg = req_ptr;
    /* from = NULL (no peerinfo for one-shot SSL fetch) */
    encode_PeerInfo_ptr(cs, &rq.from, NULL);

    rp = new_Request(cs);
    write_Request(&rq, rp);
    capn_setp(capn_root(&c), 0, rp.p);

    blen = capn_size(&c);
    if (blen <= 0) {
        capn_free(&c);
        return -EIO;
    }

    buf = (uint8_t *)malloc((size_t)blen);
    if (!buf) {
        capn_free(&c);
        return -ENOMEM;
    }

    if (capn_write_mem(&c, buf, (size_t)blen, 0) < 0) {
        free(buf);
        capn_free(&c);
        return -EIO;
    }
    capn_free(&c);

    /* Wire framing: 8-byte size header then payload (matches netio.c
     * io_fsm_perform: writes &pkt->sz raw, then pkt->data). */
    {
        uint64_t hdr   = (uint64_t)blen;
        int      nsent = 0;

        nsent = SSL_write(ssl, &hdr, (int)sizeof(hdr));
        if (nsent != (int)sizeof(hdr)) {
            rc = -EIO;
            goto out;
        }

        nsent = SSL_write(ssl, buf, (int)blen);
        if (nsent != (int)blen) {
            rc = -EIO;
            goto out;
        }
    }

out:
    free(buf);
    return rc;
}

/* Helper: read exactly `len` bytes from SSL, looping on partial reads. */
static int ssl_read_all(SSL *ssl, void *dst, size_t len)
{
    size_t off = 0;
    while (off < len) {
        int n = SSL_read(ssl, (char *)dst + off, (int)(len - off));
        if (n <= 0) {
            return -EIO;
        }
        off += (size_t)n;
    }
    return 0;
}

/*
 * xce_replconfig_reply_ssl_decode: standby reads the primary's
 * OP_replConfig reply from a raw SSL socket and extracts up_to_date,
 * version_tag, sha256 and bytes.
 *
 * Bypasses xce_decode (which would route sha256/bytes through
 * decode_Data_ptr → xce_data_t and lose the raw byte content).
 * Instead, reads the capnp message directly into capn, walks down
 * Request → RequestReply → ReplConfigReply, and pulls sha256/bytes
 * via capn_len + capn_getv8 on the raw capn_list8.
 *
 * Output buffers (*out_sha256 and *out_bytes) are malloc'd and must
 * be freed by the caller.  Both may be NULL on a fast path
 * (up_to_date == 1).  Caller is responsible for distinguishing.
 */
int xce_replconfig_reply_ssl_decode(SSL *ssl,
                                     int *out_up_to_date,
                                     uint64_t *out_tag,
                                     uint8_t **out_sha256,
                                     uint8_t **out_bytes,
                                     size_t   *out_len)
{
    struct capn               c;
    Request_ptr               rp;
    struct Request            rq;
    RequestReply_ptr          rr_ptr;
    struct RequestReply       rr;
    ReplConfigReply_ptr       reply_ptr;
    struct ReplConfigReply    reply_struct;
    capn_list8                sha_list;
    capn_list8                bytes_list;
    uint64_t                  hdr  = 0;
    uint8_t                  *buf  = NULL;
    int                       rc   = 0;

    if (ssl == NULL || out_up_to_date == NULL || out_tag == NULL
        || out_sha256 == NULL || out_bytes == NULL || out_len == NULL) {
        return -EINVAL;
    }

    *out_up_to_date = 0;
    *out_tag        = 0;
    *out_sha256     = NULL;
    *out_bytes      = NULL;
    *out_len        = 0;

    rc = ssl_read_all(ssl, &hdr, sizeof(hdr));
    if (rc < 0) {
        return rc;
    }
    if (hdr == 0 || hdr > (uint64_t)(64 * 1024 * 1024)) {
        /* 64 MiB sanity ceiling */
        return -EMSGSIZE;
    }

    buf = (uint8_t *)malloc((size_t)hdr);
    if (!buf) {
        return -ENOMEM;
    }

    rc = ssl_read_all(ssl, buf, (size_t)hdr);
    if (rc < 0) {
        free(buf);
        return rc;
    }

    if (capn_init_mem(&c, buf, (size_t)hdr, 0) < 0) {
        free(buf);
        return -EPROTO;
    }

    rp.p = capn_getp(capn_root(&c), 0, 1);
    read_Request(&rq, rp);

    if (rq.body_which != Request_body_replymsg) {
        rc = -EPROTO;
        goto cleanup;
    }

    rr_ptr = rq.body.replymsg;
    read_RequestReply(&rr, rr_ptr);

    if (rr.data_which != RequestReply_data_replconfigreplymsg) {
        rc = -EPROTO;
        goto cleanup;
    }

    reply_ptr = rr.data.replconfigreplymsg;
    read_ReplConfigReply(&reply_struct, reply_ptr);

    *out_up_to_date = (int)reply_struct.upToDate;
    *out_tag        = reply_struct.versionTag;

    /* Pull sha256 and bytes via raw capn_list8 access (NOT decode_Data
     * which would corrupt via xce_data_t).  Data_ptr and capn_list8
     * share {capn_ptr p} layout. */
    sha_list.p   = reply_struct.sha256.p;
    bytes_list.p = reply_struct.bytes.p;

    if (sha_list.p.type != CAPN_NULL && capn_len(sha_list) > 0) {
        int slen = capn_len(sha_list);
        uint8_t *sbuf = (uint8_t *)malloc((size_t)slen);
        if (!sbuf) {
            rc = -ENOMEM;
            goto cleanup;
        }
        if (capn_getv8(sha_list, 0, sbuf, slen) != slen) {
            free(sbuf);
            rc = -EPROTO;
            goto cleanup;
        }
        *out_sha256 = sbuf;
    }

    if (bytes_list.p.type != CAPN_NULL && capn_len(bytes_list) > 0) {
        int blen = capn_len(bytes_list);
        uint8_t *bbuf = (uint8_t *)malloc((size_t)blen);
        if (!bbuf) {
            free(*out_sha256);
            *out_sha256 = NULL;
            rc = -ENOMEM;
            goto cleanup;
        }
        if (capn_getv8(bytes_list, 0, bbuf, blen) != blen) {
            free(bbuf);
            free(*out_sha256);
            *out_sha256 = NULL;
            rc = -EPROTO;
            goto cleanup;
        }
        *out_bytes = bbuf;
        *out_len   = (size_t)blen;
    }

cleanup:
    capn_free(&c);
    free(buf);
    return rc;
}

/*
 * xce_upgrade_config_req_extract: extract sha256 + bytes from a decoded
 * OP_upgradeConfig request.
 *
 * xce_decode() now extracts the raw List(UInt8) bytes from the capnp
 * segment before capn_free(), storing malloc'd copies in
 * req->body.upgradeconfigrequestmsg->raw_sha256 and ->raw_bytes.
 * This function returns shallow copies (pointers into the struct); the
 * caller must NOT free them — they are owned by the xce_request_t and
 * released by free_Request().
 *
 * Returns 0 on success, -EINVAL on bad args or missing fields.
 */
int xce_upgrade_config_req_extract(const xce_request_t *req,
                                    uint8_t **out_bytes, size_t *out_len,
                                    uint8_t **out_sha256)
{
    if (req == NULL || out_bytes == NULL || out_len == NULL
        || out_sha256 == NULL) {
        return -EINVAL;
    }
    *out_bytes  = NULL;
    *out_len    = 0;
    *out_sha256 = NULL;

    if (req->body_which != Request_body_upgradeconfigrequestmsg
        || req->body.upgradeconfigrequestmsg == NULL) {
        return -EINVAL;
    }

    const upgrade_config_req_t *ucr = req->body.upgradeconfigrequestmsg;
    if (ucr->raw_sha256 == NULL || ucr->raw_bytes == NULL
        || ucr->raw_bytes_len == 0) {
        return -EINVAL;
    }

    *out_sha256 = ucr->raw_sha256;
    *out_bytes  = ucr->raw_bytes;
    *out_len    = ucr->raw_bytes_len;
    return 0;
}

/*
 * xce_upgrade_config_req_extract_raw: parse OP_upgradeConfig request
 * bytes from a raw on-wire packet and extract sha256 + config bytes.
 *
 * Use this from handle_upgrade_config() once a path is wired to capture
 * the io_packet_t->data buffer before xce_decode() tears it down (see
 * the xce_upgrade_config_req_extract() comment for why the decoded
 * form is unusable).
 *
 * Output buffers (*out_bytes, *out_sha256) are malloc'd and owned by
 * the caller.
 */
int xce_upgrade_config_req_extract_raw(const uint8_t *pkt_data, size_t pkt_sz,
                                       uint8_t **out_bytes, size_t *out_len,
                                       uint8_t **out_sha256)
{
    struct capn                     c;
    Request_ptr                     rp;
    struct Request                  rq;
    UpgradeConfigRequest_ptr        u_ptr;
    struct UpgradeConfigRequest     u_struct;
    capn_list8                      sha_list;
    capn_list8                      bytes_list;
    int                             rc = 0;

    if (pkt_data == NULL || pkt_sz == 0 || out_bytes == NULL
        || out_len == NULL || out_sha256 == NULL) {
        return -EINVAL;
    }

    *out_bytes  = NULL;
    *out_len    = 0;
    *out_sha256 = NULL;

    if (capn_init_mem(&c, pkt_data, pkt_sz, 0) < 0) {
        return -EPROTO;
    }

    rp.p = capn_getp(capn_root(&c), 0, 1);
    read_Request(&rq, rp);

    if (rq.body_which != Request_body_upgradeconfigrequestmsg) {
        rc = -EPROTO;
        goto cleanup;
    }

    u_ptr = rq.body.upgradeconfigrequestmsg;
    read_UpgradeConfigRequest(&u_struct, u_ptr);

    sha_list.p   = u_struct.sha256.p;
    bytes_list.p = u_struct.bytes.p;

    if (sha_list.p.type == CAPN_NULL || capn_len(sha_list) <= 0
        || bytes_list.p.type == CAPN_NULL || capn_len(bytes_list) <= 0) {
        rc = -EPROTO;
        goto cleanup;
    }

    {
        int slen = capn_len(sha_list);
        int blen = capn_len(bytes_list);
        uint8_t *sbuf;
        uint8_t *bbuf;

        sbuf = (uint8_t *)malloc((size_t)slen);
        if (!sbuf) {
            rc = -ENOMEM;
            goto cleanup;
        }
        if (capn_getv8(sha_list, 0, sbuf, slen) != slen) {
            free(sbuf);
            rc = -EPROTO;
            goto cleanup;
        }

        bbuf = (uint8_t *)malloc((size_t)blen);
        if (!bbuf) {
            free(sbuf);
            rc = -ENOMEM;
            goto cleanup;
        }
        if (capn_getv8(bytes_list, 0, bbuf, blen) != blen) {
            free(sbuf);
            free(bbuf);
            rc = -EPROTO;
            goto cleanup;
        }

        *out_sha256 = sbuf;
        *out_bytes  = bbuf;
        *out_len    = (size_t)blen;
    }

cleanup:
    capn_free(&c);
    return rc;
}

int xce_nop_req(endpoint_t* ep)
{
    xce_request_t request
        = { .opcode = OPCode_nop, .body_which = Request_body_nomsg };

    if (ep == NULL) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

int xce_hello_req(endpoint_t* ep, char* msg)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode     = OPCode_hello,
        .body_which = Request_body_hellomsg,
        .body       = {
                  .hellomsg = msg
	}
    };
    /* clang-format on */

    if ((ep == NULL) || (msg == NULL)) {
        return -EINVAL;
    }
    return xce_request(ep, &request);
}

int xce_join_req(endpoint_t* ep, int role, hostattrs_t* attrs)
{
    /* clang-format off */
    xce_joindata_t joinmsg = {
        .role = role
    };
    xce_request_t request = {
        .opcode = OPCode_join,
        .body_which = Request_body_joinmsg,
        .body = {
            .joinmsg = &joinmsg 
        }
    };

    if ((role != HostRole_client) &&
        (role != HostRole_worker) &&
        (role != HostRole_apiserver) &&
        (role != HostRole_scheduler)) {
        return -EINVAL;
    }
    /* clang-format on */

    if (role == HostRole_worker) {
        if (attrs == NULL) {
            return -EINVAL;
        }

        joinmsg.data_which = JoinData_data_attrs;
        joinmsg.data.attrs = attrs;
    }
    return xce_request(ep, &request);
}

int xce_shutdown_req(endpoint_t* ep)
{
    int           rc;
    dataq_t*      outq;
    xce_request_t request
        = { .opcode = OPCode_shutdown, .body_which = Request_body_nomsg };

    if (ep == NULL) {
        return -EINVAL;
    }

    rc = xce_request(ep, &request);
    if (rc < 0) {
        return rc;
    }

    outq = &(ep->outq);

    while (!DATAQ_EMPTY(outq)) {
        sleep(1);
    }

    return 0;
}

int xce_pushconfig_req(endpoint_t* ep, xce_peerinfo_t* from,
                       xce_configdata_t* config)
{
    int rc = 0;
    /* clang-format off */
    xce_request_t request
        = { .opcode = OPCode_push, .body_which = Request_body_configpushmsg };
    /* clang-format on */

    if ((ep == NULL) || (config == NULL)) {
        return -EINVAL;
    }

    request.body.configpushmsg = config;
    rc                         = xce_request(ep, &request);

    return rc;
}

static int xce_push_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data)
{
    int           rc = 0;
    xce_request_t request
        = { .opcode = OPCode_push, .body_which = Request_body_pushmsg };

    if ((ep == NULL) || (from == NULL) || (data == NULL)) {
        return -EINVAL;
    }

    request.body.pushmsg = data;

    rc = xce_request(ep, &request);
    return rc;
}

int xce_push_data_req(endpoint_t* ep, xce_peerinfo_t* from, xce_data_t* data,
                      xce_check_fn_t checkf, void* checkarg, int pktlimit)
{
    int        nrows = 0;
    int        pos   = 0;
    int        rc    = 0;
    xce_data_t data1;
    int        checkstatus = 0;

    if ((ep == NULL) || (from == NULL) || (data == NULL)) {
        return -EINVAL;
    }

    data1.table = data->table;
    switch (data->table) {
    case Data_data_schedjobs:
    case Data_data_jobs:
        nrows           = data->data.jobs->n_total;
        data1.data.jobs = (xce_jobs_data_t*)calloc(1, sizeof(xce_jobs_data_t));
        if (data1.data.jobs == NULL) {
            rc = -ENOMEM;
        }
        break;
    case Data_data_schedworkers:
    case Data_data_workers:
        nrows = data->data.workers->n_total;
        data1.data.workers
            = (xce_workers_data_t*)calloc(1, sizeof(xce_workers_data_t));
        if (data1.data.workers == NULL) {
            rc = -ENOMEM;
        }
        break;
    case Data_data_resources:
        break;
    case Data_data_resvals:
        break;
    default:
        rc = -EINVAL;
        break;
    }

    if (rc < 0) {
        return rc;
    }

    if (pktlimit <= 0) {
        pktlimit = nrows;
    }

    do {
        int nc = nrows - pos;

        if (nc > pktlimit) {
            nc = pktlimit;
        }

        if (checkf != NULL) {
            checkstatus = (*checkf)(checkarg);
            if (checkstatus < 0) {
                nc = 0;
            }
        }

        switch (data->table) {
        case Data_data_schedjobs:
        case Data_data_jobs:
            data1.data.jobs->n_total = data->data.jobs->n_total;
            data1.data.jobs->pos     = pos;
            data1.data.jobs->count   = nc;
            if (nc == 0) {
                data1.data.jobs->jobs = NULL;
            } else {
                data1.data.jobs->jobs = &(data->data.jobs->jobs[pos]);
            }
            break;
        case Data_data_schedworkers:
        case Data_data_workers:
            data1.data.workers->n_total = data->data.workers->n_total;
            data1.data.workers->pos     = pos;
            data1.data.workers->count   = nc;
            if (nc == 0) {
                data1.data.workers->workers = NULL;
            } else {
                data1.data.workers->workers
                    = &(data->data.workers->workers[pos]);
            }
            break;
        case Data_data_resources:
            break;
        case Data_data_resvals:
            break;
        default:
            break;
        }

        rc = xce_push_req(ep, from, &data1);

        if (rc < 0) {
            break;
        }

        pos += nc;

    } while ((pos < nrows) && (checkstatus >= 0));

    switch (data->table) {
    case Data_data_schedjobs:
    case Data_data_jobs:
        FREEUP(data1.data.jobs);
        break;
    case Data_data_schedworkers:
    case Data_data_workers:
        FREEUP(data1.data.workers);
        break;
    default:
        break;
    }

    return rc;
}

int xce_get_req(endpoint_t* ep, xce_get_t* get)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_get,
        .body_which = Request_body_getmsg,
        .body = {
            .getmsg = get
        }
    };
    /* clang-format on */

    return xce_request(ep, &request);
}

int xce_reply_get(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply,
                  xce_check_fn_t checkf, void* checkarg, int pktlimit)
{
    int rc    = 0;
    int nrows = 0;
    int pos   = 0;
    /* clang-format off */
    xce_data_t getdata = {
        .table = reply->data.getdata->table
    };
    xce_reply_t reply1 = {
        .seq        = reply->seq,
        .opcode     = reply->opcode,
        .status     = reply->status,
        .data_which = reply->data_which,
        .data       = {
            .getdata = &getdata
        }
    };
    /* clang-format on */
    int         checkstatus = 0;
    xce_data_t* data        = reply->data.getdata;

    if ((ep == NULL) || (from == NULL) || (reply == NULL)) {
        return -EINVAL;
    }

    switch (getdata.table) {
    case Data_data_schedjobs:
    case Data_data_jobs:
        nrows = data->data.jobs->n_total;
        getdata.data.jobs
            = (xce_jobs_data_t*)calloc(1, sizeof(xce_jobs_data_t));
        break;
    case Data_data_schedworkers:
    case Data_data_workers:
        nrows = data->data.workers->n_total;
        getdata.data.workers
            = (xce_workers_data_t*)calloc(1, sizeof(xce_workers_data_t));
        break;
    case Data_data_resources:
        nrows = data->data.resources->n_total;
        getdata.data.resources
            = (xce_resource_data_t*)calloc(1, sizeof(xce_resource_data_t));
        break;
    case Data_data_resvals:
        nrows = data->data.resvals->n_total;
        getdata.data.resvals
            = (xce_resval_data_t*)calloc(1, sizeof(xce_resval_data_t));
        break;
    default:
        rc = -EINVAL;
        break;
    }

    if (rc < 0) {
        /* FIXME we need to reply no data */
        return rc;
    }

    if (pktlimit <= 0) {
        pktlimit = nrows;
    }

    do {
        int nc = nrows - pos;

        if (nc > pktlimit) {
            nc = pktlimit;
        }

        if (checkf != NULL) {
            checkstatus = (*checkf)(checkarg);

            if (checkstatus < 0) {
                nc = 0;
            }
        }

        switch (getdata.table) {
        case Data_data_schedjobs:
        case Data_data_jobs:
            getdata.data.jobs->n_total = data->data.jobs->n_total;
            getdata.data.jobs->pos     = pos;
            getdata.data.jobs->count   = nc;
            if (nc == 0) {
                getdata.data.jobs->jobs = NULL;
            } else {
                getdata.data.jobs->jobs = &(data->data.jobs->jobs[pos]);
            }
            break;
        case Data_data_schedworkers:
        case Data_data_workers:
            getdata.data.workers->n_total = data->data.workers->n_total;
            getdata.data.workers->pos     = pos;
            getdata.data.workers->count   = nc;
            if (nc == 0) {
                getdata.data.workers->workers = NULL;
            } else {
                getdata.data.workers->workers
                    = &(data->data.workers->workers[pos]);
            }
            break;
        case Data_data_resources:
            getdata.data.resources->n_total = data->data.resources->n_total;
            getdata.data.resources->pos     = pos;
            getdata.data.workers->count     = nc;
            if (nc == 0) {
                getdata.data.resources->resources = NULL;
            } else {
                getdata.data.resources->resources
                    = &(data->data.resources->resources[pos]);
            }
            break;
        case Data_data_resvals:
            getdata.data.resvals->n_total = data->data.resvals->n_total;
            getdata.data.resvals->pos     = pos;
            getdata.data.resvals->count   = nc;
            if (nc == 0) {
                getdata.data.resvals->vals = NULL;
            } else {
                getdata.data.resvals->vals = &(data->data.resvals->vals[pos]);
            }
            break;
        default:
            /* should never goes here */
            break;
        }

        rc = xce_reply(ep, from, &reply1);
        if (rc < 0) {
            break;
        }

        pos += nc;
    } while ((pos < nrows) && (checkstatus >= 0));

    switch (getdata.table) {
    case Data_data_schedjobs:
    case Data_data_jobs:
        FREEUP(getdata.data.jobs);
        break;
    case Data_data_schedworkers:
    case Data_data_workers:
        FREEUP(getdata.data.workers);
        break;
    case Data_data_resources:
        FREEUP(getdata.data.resources);
        break;
    case Data_data_resvals:
        FREEUP(getdata.data.resvals);
        break;
    default:
        break;
    }

    return rc;
}

int extract_data(dataset_t* dataset, xce_data_t* getdata)
{
    int ntotal;
    int rc    = 0;
    int pos   = 0;
    int count = 0;

    TRACE("%s ...", __func__);

    switch (getdata->table) {
    case Data_data_schedjobs:
    case Data_data_jobs:
        if (getdata->data.jobs->count <= 0) {
            return 1;
        }

        ntotal = getdata->data.jobs->n_total;
        pos    = getdata->data.jobs->pos;
        count  = getdata->data.jobs->count;

        if (dataset->rows == NULL) {
            dataset->rows  = (xce_job_t**)calloc(ntotal, sizeof(xce_job_t*));
            dataset->nrows = 0;
        }

        memcpy((void*)(((char*)dataset->rows) + sizeof(xce_job_t*) * pos),
               getdata->data.jobs->jobs, count * sizeof(xce_job_t*));
        memset(getdata->data.jobs->jobs, 0x0, count * sizeof(xce_job_t*));

        break;
    case Data_data_schedworkers:
    case Data_data_workers:
        if (getdata->data.workers->count <= 0) {
            return 1;
        }

        ntotal = getdata->data.workers->n_total;
        if (dataset->rows == NULL) {
            dataset->rows
                = (xce_worker_t**)calloc(ntotal, sizeof(xce_worker_t*));
            dataset->nrows = 0;
        }
        pos   = getdata->data.workers->pos;
        count = getdata->data.workers->count;

        memcpy((void*)(((char*)dataset->rows) + sizeof(xce_worker_t*) * pos),
               getdata->data.workers->workers, count * sizeof(xce_worker_t*));
        memset(getdata->data.workers->workers, 0x0,
               count * sizeof(xce_worker_t*));

        break;
    case Data_data_resources:
        if (getdata->data.resources->count <= 0) {
            return 1;
        }

        ntotal = getdata->data.resources->n_total;
        if (dataset->rows == NULL) {
            dataset->rows
                = (xce_resource_t**)calloc(ntotal, sizeof(xce_resource_t*));
            dataset->nrows = 0;
        }
        pos   = getdata->data.resources->pos;
        count = getdata->data.resources->count;

        memcpy((void*)(((char*)dataset->rows) + sizeof(xce_resource_t*) * pos),
               getdata->data.resources->resources,
               count * sizeof(xce_resource_t*));

        memset(getdata->data.resources->resources, 0x0,
               count * sizeof(xce_resource_t*));
        break;
    case Data_data_resvals:
        if (getdata->data.resvals->count <= 0) {
            return 1;
        }

        ntotal = getdata->data.resvals->n_total;
        if (dataset->rows == NULL) {
            dataset->rows
                = (xce_resval_t**)calloc(ntotal, sizeof(xce_resval_t*));
            dataset->nrows = 0;
        }

        pos   = getdata->data.resvals->pos;
        count = getdata->data.resvals->count;

        memcpy((void*)(((char*)dataset->rows) + sizeof(xce_resval_t*) * pos),
               getdata->data.resvals->vals, count * sizeof(xce_resval_t*));

        memset(getdata->data.resvals->vals, 0x0, count * sizeof(xce_resval_t*));
        break;
    default:
        rc = -EINVAL;
        break;
    }

    if (rc < 0) {
        return rc;
    }

    dataset->nrows += count;
    if (dataset->nrows < ntotal) {
        return 0;
    }

    return 1;
}

int xce_jobdispatch_req(endpoint_t* ep, xce_job_dispatch_t* info)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_jobdispatch,
        .body_which = Request_body_dispatchmsg,
        .body = {
            .dispatchmsg = info 
        }
    };
    /* clang-format on */

    /* clang-format off */
    if ((ep == NULL) ||
        (info == NULL)) {
        return -EINVAL;
    }
    /* clang-format on */

    return xce_request(ep, &request);
}

int xce_notify_req(endpoint_t* ep, xce_notify_data_t* data)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_notify,
        .body_which = Request_body_notifymsg,
        .body = {
            .notifymsg = data
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (data == NULL)) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

int xce_jobsubmit_req(endpoint_t* ep, xce_job_submit_t* submit)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_jobsub,
        .body_which = Request_body_jobsubmsg,
        .body = {
            .jobsubmsg = submit
        }
    };
    /*clang-format on*/
    
    /* clang-format off */
    if ((ep == NULL) ||
        (submit == NULL)) {
        return -EINVAL;
    }
    /* clang-format on */

    return xce_request(ep, &request);
}

int xce_startjob_req(endpoint_t* ep, xce_job_exec_t* execinfo)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_startjob,
        .body_which = Request_body_jobstartmsg,
        .body = {
            .jobstartmsg = execinfo
        }
    };
    /* clang-format on */

    return xce_request(ep, &request);
}

int xce_clusterconf_req(endpoint_t* ep)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_clusterconf,
        .body_which = Request_body_nomsg
    };
    /* clang-format on */

    return xce_request(ep, &request);
}

int xce_update_req(endpoint_t* ep, xce_update_data_t* data)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_update,
        .body_which = Request_body_updatemsg,
        .body = {
            .updatemsg = data
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (data == NULL)) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

int xce_jobctrl_req(endpoint_t* ep, xce_jobctrl_t* ctrinfo)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_jobcontrol,
        .body_which = Request_body_jobctrlmsg,
        .body = {
            .jobctrlmsg = ctrinfo
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (ctrinfo == NULL)) {
        return -EINVAL;
    }

    if (ctrinfo->jobid == 0) {
        return -EINVAL;
    }

    switch (ctrinfo->action) {
    case JobControl_u_siginfo:
        break;
    case JobControl_u_ckptoption:
        if (ctrinfo->u.ckptoption->method == NULL) {
            ctrinfo->u.ckptoption->method = "default";
        }
        break;
    default:
        return -EINVAL;
        break;
    }
    return xce_request(ep, &request);
}

int xce_queuectrl_req(endpoint_t* ep, xce_queuectrl_t* ctrlinfo)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_queuecontrol,
        .body_which = Request_body_queuectrlmsg,
        .body = {
            .queuectrlmsg = ctrlinfo
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (ctrlinfo == NULL)) {
        return -EINVAL;
    }

    if ((ctrlinfo->queue == NULL) || (strlen(ctrlinfo->queue) == 0)) {
        return -EINVAL;
    }

    if (ctrlinfo->action == QUEUE_DO_NONE) {
        return 0;
    }

    if ((ctrlinfo->action < 0) || (ctrlinfo->action > QUEUE_ACTIVE)) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

int xce_hostctrl_req(endpoint_t* ep, xce_hostctrl_t* ctrlinfo)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_hostcontrol,
        .body_which = Request_body_hostctrlmsg,
        .body = {
            .hostctrlmsg = ctrlinfo
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (ctrlinfo == NULL)) {
        return -EINVAL;
    }

    if ((ctrlinfo->hostname == NULL) || (strlen(ctrlinfo->hostname) == 0)) {
        return -EINVAL;
    }

    if (ctrlinfo->action == HOST_DO_NONE) {
        return 0;
    }

    if (ctrlinfo->action != HOST_OPEN && ctrlinfo->action != HOST_CLOSE) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

int xce_runtask_req(endpoint_t* ep, xce_taskrun_t* runinfo)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_runtasks,
        .body_which = Request_body_runtaskmsg,
        .body = {
            .runtaskmsg = runinfo
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (runinfo == NULL)) {
        return -EINVAL;
    }

    if (runinfo->nhosts <= 0) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

int xce_set_pend_reason_req(endpoint_t* ep, xce_preason_set_t* reasons)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_setpreason,
        .body_which = Request_body_preasonmsg,
        .body = {
            .preasonmsg = reasons
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (reasons == NULL)) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

int xce_clear_reason_req(endpoint_t* ep, xce_clear_reason_t* reason)
{
    /* clang-format off */
    xce_request_t request = {
        .opcode = OPCode_clrpreason,
        .body_which = Request_body_clrpreasonmsg,
        .body = {
            .clrpreasonmsg = reason
        }
    };
    /* clang-format on */

    if ((ep == NULL) || (reason == NULL)) {
        return -EINVAL;
    }

    return xce_request(ep, &request);
}

/* ── Task 10.2: Client-Side Transparent Retry (Req 7.1–7.6) ────────────────
 *
 * xce_retry_send_hook: test-injectable send+receive round-trip.
 * NULL in production; set by unit tests to return controlled replies.
 */
int (*xce_retry_send_hook)(endpoint_t *ep, xce_request_t *req,
                            xce_reply_t *reply_out) = NULL;

const xce_retry_config_t XCE_RETRY_DEFAULTS = {
    .max_retries        = 20,
    .initial_retry_ms   = 50,
    .max_retry_ms       = 5000,
    .backoff_multiplier = 2.0,
};

/* is_non_retriable: errors that should surface immediately without retry */
static int is_non_retriable(int status)
{
    return (status == -EPERM   || status == -EACCES ||
            status == -EINVAL  || status == -ENOENT);
}

/* ms_sleep_jittered: sleep delay_ms + random jitter via rand_r (Req 7.6) */
static void ms_sleep_jittered(int delay_ms, unsigned int *seed)
{
    if (delay_ms <= 0) return;
    int jitter_ms = (int)((rand_r(seed) % (delay_ms / 4 + 1)));
    long total_ns = ((long)delay_ms + jitter_ms) * 1000000L;
    struct timespec ts = { .tv_sec = total_ns / 1000000000L,
                           .tv_nsec = total_ns % 1000000000L };
    nanosleep(&ts, NULL);
}

int xce_request_with_retry(endpoint_t *ep,
                            xce_request_t *req,
                            const xce_retry_config_t *cfg)
{
    if (cfg == NULL)
        cfg = &XCE_RETRY_DEFAULTS;

    unsigned int rand_seed = (unsigned int)(uintptr_t)ep ^ (unsigned int)time(NULL);
    int delay_ms = cfg->initial_retry_ms;
    int last_rc  = -ETIMEDOUT;

    for (int attempt = 0; attempt <= cfg->max_retries; attempt++) {
        xce_reply_t reply;
        memset(&reply, 0, sizeof(reply));

        int rc;
        if (xce_retry_send_hook) {
            rc = xce_retry_send_hook(ep, req, &reply);
        } else {
            /* Production path: fire-and-forget send; reply handling is caller's
             * responsibility via the event loop. Return send error directly. */
            rc = xce_request(ep, req);
            if (rc != 0)
                return rc;
            return 0;  /* async: caller reads reply via event loop */
        }

        /* Transport-level failure (ECONNREFUSED, ETIMEDOUT) → retry (Req 7.3) */
        if (rc == -ECONNREFUSED || rc == -ETIMEDOUT) {
            last_rc = rc;
            if (attempt < cfg->max_retries)
                ms_sleep_jittered(delay_ms, &rand_seed);
            goto next_backoff;
        }

        /* Any other transport error → surface immediately */
        if (rc != 0)
            return rc;

        /* Application-layer status from reply */
        int status = (int)reply.status;

        if (status == 0)
            return 0;  /* success */

        last_rc = status;

        /* Non-retriable errors: surface immediately (Req 7.4) */
        if (is_non_retriable(status))
            return status;

        /* XCE_ERR_STANDBY with redirect → reconnect (Req 7.2).
         * In hook mode, the hook already updated ep/redirect state.
         * Just retry; redirect handling is at the transport layer. */
        if (status == -E_XCE_STANDBY && reply.redirect_host != NULL
            && reply.redirect_port != 0) {
            /* no extra sleep for redirect — just retry immediately */
            continue;
        }

        /* XCE_ERR_NOT_READY or XCE_ERR_STANDBY without redirect → backoff */
        if (attempt < cfg->max_retries)
            ms_sleep_jittered(delay_ms, &rand_seed);

next_backoff:
        {
            int next = (int)((double)delay_ms * cfg->backoff_multiplier);
            delay_ms = next > cfg->max_retry_ms ? cfg->max_retry_ms : next;
        }
    }

    return last_rc;
}