ab_gp_channel.c

Go to the documentation of this file.

 
#include "ab_gp_channel.h"
 
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
 
#include "rte/redundancy/checkpoint/rte_checkpoint.h"
#include "rte/redundancy/channel_service/rte_channel_service.h"
#include "rte/redundancy/channel_service/rte_channel_service_flow_osadapter.h"
#include "rte/posix_osadapter/rte_posix_osadapter_channel_service.h"
#include "rte/protocol_adapter_dds/rte_protocol_adapter_dds.h"
#include "rte_osadapter/flow/rte_osadapter_flow.h"
#include "rte/oal/log/rte_log.h"
#include "rte/oal/memory/rte_mem_util.h"
#include "rte/utils/safestate/rte_safestate.h"
 
#include "../../common/common_config.h"
#include "../../common/rbc_util_frame.h"
#include "ab_gp_channel_checkpoint.h"
#include "ab_gp_channel_crosscompare.h"
#include "ab_gp_channel_dual.h"
#include "ab_gp_channel_negotiate.h"
#include "ab_gp_standby_policy.h"
#include "ab_gp_channel_types.h"
#include "ab_gp_channel_util.h"
#include "ab_gp_channel_payload.h"
#include "ab_gp_km114_relay.h"
#include "../ga_interface/ga_interface.h"
#include "../main/ab_gp_ga_hooks.h" /* ab_gp_build_ctc_alarm() - see that file's own doc */
#include "../train/ab_gp_train.h"
#include "../ertms_procedures/ab_gp_ertms_engine.h"
 
static ab_gp_channel_context_t g_ab_ctx;
 
/* Shared by both config-driven resolution paths below (previously two verbatim-duplicated
 * copies of this exact if/else-if chain, one per path - a real complexity and drift source in
 * its own right). Only touches config->host; every other field is the caller's responsibility.
 * Precedence, unchanged from before this extraction: an explicit env var override, else the
 * config-supplied host, else a well-known "same host as this build's own default docker/host
 * name" fallback to loopback. */
static void apply_host_env_override(const char *name, const ab_gp_channel_context_t *ctx, rte_netlink_config_t *config)
{
    const char *env_host;
 
    if ((strcmp(name, "ab-peer") == 0) || (strcmp(name, "ab-crosscompare") == 0))
    {
        env_host = getenv(RTE_EXAMPLE_A_HOST_ENV);
        if ((env_host != NULL) && (env_host[0] != '\0'))
        {
            config->host = env_host;
        }
        else if ((config->host == NULL) || (config->host[0] == '\0') ||
                 (strcmp(config->host, "a-west") == 0) || (strcmp(config->host, "a-east") == 0))
        {
            config->host = "127.0.0.1";
        }
    }
    else if (strcmp(name, "ab-negotiate") == 0)
    {
        const char *env_var = ctx->is_role_a ? RTE_EXAMPLE_REMOTE_A_HOST_ENV
                                             : RTE_EXAMPLE_REMOTE_B_HOST_ENV;
        env_host = getenv(env_var);
        if ((env_host != NULL) && (env_host[0] != '\0'))
        {
            config->host = env_host;
        }
        else if ((config->host == NULL) || (config->host[0] == '\0') ||
                 (strcmp(config->host, "a-west") == 0) || (strcmp(config->host, "b-west") == 0))
        {
            config->host = "127.0.0.1";
        }
    }
    else
    {
        env_host = getenv(RTE_EXAMPLE_C_HOST_ENV);
        if ((env_host != NULL) && (env_host[0] != '\0'))
        {
            config->host = env_host;
        }
        else if ((config->host == NULL) || (config->host[0] == '\0') ||
                 (strcmp(config->host, "c-west") == 0) || (strcmp(config->host, "c-east") == 0))
        {
            config->host = "127.0.0.1";
        }
    }
}
 
/* Highest-priority resolution path: the runtime redundancy config (JSON, ctx->redundancy).
 * Returns false (config left untouched) when name isn't in that config, or when ctx itself is
 * NULL - unchanged from before this extraction, this is the ONLY one of the three paths that
 * tolerates a NULL ctx; the other two dereference it unconditionally, exactly as the original,
 * un-extracted function did (not "fixed" as a side effect of this refactor). */
static bool try_resolve_from_redundancy_config(const ab_gp_channel_context_t *ctx, const char *name,
                                                rte_netlink_config_t *config)
{
    const rte_channel_def_t *found_ch = NULL;
 
    if (ctx != NULL)
    {
        uint8_t ch_i;
        for (ch_i = 0U; ch_i < ctx->redundancy.channel_count; ch_i++)
        {
            if (strcmp(ctx->redundancy.channels[ch_i].name, name) == 0)
            {
                found_ch = &ctx->redundancy.channels[ch_i];
                break;
            }
        }
    }
    if (found_ch == NULL)
    {
        return false;
    }
 
    rte_mem_set(config, 0, sizeof(*config));
    config->role = (found_ch->role == RTE_CHANNEL_ROLE_CONNECT) ? RTE_NETLINK_ROLE_CONNECT : RTE_NETLINK_ROLE_LISTEN;
    config->host = (config->role == RTE_NETLINK_ROLE_CONNECT && found_ch->host[0] != '\0') ? found_ch->host : NULL;
    if (config->role == RTE_NETLINK_ROLE_CONNECT)
    {
        apply_host_env_override(name, ctx, config);
    }
    config->port = found_ch->port;
    config->message_size = (found_ch->message_size > 0U) ? found_ch->message_size : RBC_ENVELOPE_WIRE_SIZE;
    if ((strcmp(name, "ab-peer") == 0) || (strcmp(name, "ab-crosscompare") == 0))
    {
        /* The A/B frame size is a compile-time wire fact of this build, not a deployment setting. */
        config->message_size = PEER_WIRE_FRAME_SIZE;
    }
    config->connect_timeout_ms = (found_ch->connect_timeout_ms > 0U) ? found_ch->connect_timeout_ms
                                                                  : ctx->config.timers.reconnect_attempt_timeout_ms;
    return true;
}
 
/* Second-priority resolution path: the legacy etc/config/<role>_<site>.conf file. Dereferences
 * ctx->config unconditionally - unchanged from before this extraction (see the NULL-handling
 * note on try_resolve_from_redundancy_config above). */
static bool try_resolve_from_legacy_config(const ab_gp_channel_context_t *ctx, const char *name,
                                            rte_netlink_config_t *config)
{
    if (ab_gp_config_find_channel(&ctx->config, name, config) != RTE_STATUS_OK)
    {
        return false;
    }
    if (config->role == RTE_NETLINK_ROLE_CONNECT)
    {
        apply_host_env_override(name, ctx, config);
    }
    if ((strcmp(name, "ab-peer") == 0) || (strcmp(name, "ab-crosscompare") == 0))
    {
        config->message_size = PEER_WIRE_FRAME_SIZE;
    }
    else if (config->message_size == 0U)
    {
        config->message_size = RBC_ENVELOPE_WIRE_SIZE;
    }
    return true;
}
 
/* ADR-036: one CONNECT link per {kind} to C - train_id demux is done against the envelope, not
 * the link. Matches resolve_channel_hardcoded_fallback()'s own "ab-relay-*" name set. */
static bool relay_kind_from_name(const char *name, rte_example_relay_kind_t *out_kind)
{
    if (strcmp(name, "ab-relay-train") == 0)
    {
        *out_kind = RTE_EXAMPLE_RELAY_KIND_TRAIN;
        return true;
    }
    if (strcmp(name, "ab-relay-il") == 0)
    {
        *out_kind = RTE_EXAMPLE_RELAY_KIND_IL;
        return true;
    }
    if (strcmp(name, "ab-relay-ctc") == 0)
    {
        *out_kind = RTE_EXAMPLE_RELAY_KIND_CTC;
        return true;
    }
    if (strcmp(name, "ab-relay-util") == 0)
    {
        *out_kind = RTE_EXAMPLE_RELAY_KIND_UTIL;
        return true;
    }
    return false;
}
 
/* ADR-038: the UTIL link carries rbc_util_frame_t (256 B), not an rbc_envelope_t. Train is
 * widened to RBC_TRAIN_ENVELOPE_WIRE_SIZE (common_config.h) to carry RBC_MSG_ERTMS_ENVELOPE
 * frames alongside its existing flat-scheme kinds - see that constant's own doc. */
static uint32_t message_size_for_relay_kind(rte_example_relay_kind_t rk)
{
    if (rk == RTE_EXAMPLE_RELAY_KIND_UTIL)
    {
        return (uint32_t)RBC_UTIL_FRAME_SIZE;
    }
    if (rk == RTE_EXAMPLE_RELAY_KIND_TRAIN)
    {
        return (uint32_t)RBC_TRAIN_ENVELOPE_WIRE_SIZE;
    }
    return (uint32_t)RBC_ENVELOPE_WIRE_SIZE;
}
 
/* Last-resort resolution path: common_config.h's own hardcoded port/host formulas
 * (site_config_*), used when neither config-driven path above found this channel name.
 * Dereferences ctx unconditionally - see the NULL-handling note on
 * try_resolve_from_redundancy_config above. */
static rte_status_t resolve_channel_hardcoded_fallback(const ab_gp_channel_context_t *ctx, const char *name,
                                                         rte_netlink_config_t *config)
{
    rte_example_relay_kind_t rk;
 
    rte_mem_set(config, 0, sizeof(*config));
    config->connect_timeout_ms = ctx->config.timers.reconnect_attempt_timeout_ms;
 
    if (strcmp(name, "ab-peer") == 0)
    {
        config->role = ctx->is_role_a ? RTE_NETLINK_ROLE_LISTEN : RTE_NETLINK_ROLE_CONNECT;
        config->host = ctx->is_role_a ? NULL : site_config_host(RTE_EXAMPLE_A_HOST_ENV);
        config->port = site_config_ab_port(ctx->site);
        config->message_size = PEER_WIRE_FRAME_SIZE;
        return RTE_STATUS_OK;
    }
    if (strcmp(name, "ab-crosscompare") == 0)
    {
        config->role = ctx->is_role_a ? RTE_NETLINK_ROLE_LISTEN : RTE_NETLINK_ROLE_CONNECT;
        config->host = ctx->is_role_a ? NULL : site_config_host(RTE_EXAMPLE_A_HOST_ENV);
        config->port = site_config_ab_crosscompare_port(ctx->site);
        config->message_size = PEER_WIRE_FRAME_SIZE;
        return RTE_STATUS_OK;
    }
    if (relay_kind_from_name(name, &rk))
    {
        config->role = RTE_NETLINK_ROLE_CONNECT;
        config->host = site_config_host(RTE_EXAMPLE_C_HOST_ENV);
        config->port = ctx->is_role_a ? site_config_c_port_for_a(ctx->site, rk)
                                      : site_config_c_port_for_b(ctx->site, rk);
        config->message_size = message_size_for_relay_kind(rk);
        return RTE_STATUS_OK;
    }
    return RTE_STATUS_INVALID_PARAM;
}
 
/* TODO.md "Single source for GP ports and message sizes" (was ISSUES.md "etc/config/
 * {a,b}_{west,east}.conf hardcode ports independently of common_config.h", 2026-09-18):
 * ab-peer/ab-crosscompare used to be the two channel names with BOTH a config-file literal
 * port (JSON redundancy config, then the legacy .conf, in that priority order above) AND a
 * compiled common_config.h formula (resolve_channel_hardcoded_fallback()) able to disagree
 * silently - unlike every other channel name, which is deliberately resolved from
 * common_config.h every run for exactly this reason (see that function's own doc). A prior
 * pass (2026-09-23) added detection (a loud CONFIG_DRIFT warning) without changing which port
 * won, confirming the drift is real against the shipped default config
 * (etc/config/a_west.json's own ab-crosscompare literal, 15016, vs the formula, 15017).
 *
 * This now goes the rest of the way: the compiled formula is the single source of truth for
 * these two channels' PORT, same as every other channel already gets - a config file's own
 * port for ab-peer/ab-crosscompare is overridden, not merely flagged, closing the actual
 * footgun (changing common_config.h and rebuilding now always takes effect). The config file's
 * HOST for these two channels is untouched - that genuinely is deployment-specific (a real
 * site's A/B may be on different machines) and already has its own env-var override
 * (apply_host_env_override()); only the port number was ever the two-independent-sources-of-
 * truth problem. Still logs when an override actually happened, so a config file's leftover
 * literal port is visible, not silently ignored either. */
static void enforce_hardcoded_port(const ab_gp_channel_context_t *ctx, const char *name,
                                    rte_netlink_config_t *config)
{
    uint16_t formula_port;
 
    if (strcmp(name, "ab-peer") == 0)
    {
        formula_port = site_config_ab_port(ctx->site);
    }
    else if (strcmp(name, "ab-crosscompare") == 0)
    {
        formula_port = site_config_ab_crosscompare_port(ctx->site);
    }
    else
    {
        return;
    }
 
    if (formula_port != config->port)
    {
        char info[96];
 
        (void)snprintf(info, sizeof(info), "channel=%s config_port=%u formula_port=%u", name,
                       (unsigned int)config->port, (unsigned int)formula_port);
        rte_log_write_event(RTE_LOG_LEVEL_WARNING, site_config_name(ctx->site), 0U, ctx->role_tag, "-",
                              "CONFIG_DRIFT", "config-file port ignored - common_config.h formula wins", info);
        config->port = formula_port;
    }
}
 
static rte_status_t resolve_channel(const char *name, rte_netlink_config_t *config, void *context)
{
    const ab_gp_channel_context_t *ctx = (const ab_gp_channel_context_t *)context;
 
    if (try_resolve_from_redundancy_config(ctx, name, config))
    {
        if (ctx != NULL)
        {
            enforce_hardcoded_port(ctx, name, config);
        }
        return RTE_STATUS_OK;
    }
    if (try_resolve_from_legacy_config(ctx, name, config))
    {
        if (ctx != NULL)
        {
            enforce_hardcoded_port(ctx, name, config);
        }
        return RTE_STATUS_OK;
    }
    return resolve_channel_hardcoded_fallback(ctx, name, config);
}

static rte_status_t resolve_channel_flow(const char *name, rte_netlink_config_t *config, void *context)
{
    rte_status_t status = resolve_channel(name, config, context);
    if (status == RTE_STATUS_OK)
    {
        (void)rte_protocol_adapter_dds_add_route(name, config->host, config->port);
    }
    return status;
}

static bool ab_gp_supports_redundancy(rte_redundancy_topology_t topology, uint32_t replica_count, void *context)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)context;
    bool supported = (replica_count == 2U) &&
                      ((topology == RTE_REDUNDANCY_TOPOLOGY_2OO2) ||
                       (topology == RTE_REDUNDANCY_TOPOLOGY_2OO2_REDUNDANT));
 
    if (!supported)
    {
        RTE_EXAMPLE_LOG(
            "[%s] redundancy config names topology=%s replicas=%u - not yet implemented "
            "(only 2oo2/2oo2_redundant with exactly 2 replicas run today)\n",
            ctx->role_tag, rte_redundancy_config_topology_to_string(topology), (unsigned int)replica_count);
    }
    return supported;
}

static rte_status_t channel_service_setup_with_retry(rte_channel_service_t *storage, const char *name)
{
    rte_timestamp_ms_t waited_ms = 0U;
    const unsigned int retry_delay_ms = 100U;
    rte_status_t status;
 
    RTE_EXAMPLE_LOG("[%s] GP: opening channel '%s' to RTE\n", g_ab_ctx.role_tag, name);
    for (;;)
    {
        status = rte_channel_service_setup(storage, name);
        if (status == RTE_STATUS_OK)
        {
            RTE_EXAMPLE_LOG("[%s] GP: channel '%s' to RTE opened successfully\n", g_ab_ctx.role_tag, name);
            return RTE_STATUS_OK;
        }
        if (waited_ms >= RTE_EXAMPLE_CONNECT_TIMEOUT_MS)
        {
            return status;
        }
        ab_gp_channel_sleep_ms(retry_delay_ms);
        waited_ms += retry_delay_ms;
    }
}
 
/* Runs on the OSAdapter's timer thread (task context - see ab_gp_cycle_ticks.h for that assumption). The tick is
 * counted under the tick count's mutex (ISS-051). A failed post (the lock not acquired within
 * AB_GP_CYCLE_TICKS_LOCK_TIMEOUT_MS, or an OSAdapter fault) drops this tick and is deliberately not logged: a log from
 * the timer thread would enter the JRU forwarder / UTIL send path. A single dropped tick is covered by the
 * cross-compare cycle skew tolerance; a persistent failure leaves the count at 0 and is caught by the no-tick watchdog
 * timeout of ab_gp_channel_wait_cycle_tick(). */
static void cycle_tick_callback(rte_timer_handle_t handle, void *user_ctx)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)user_ctx;
    (void)handle;
    (void)ab_gp_cycle_ticks_post(&ctx->cycle_ticks);
}
 
static void ab_gp_channel_safestate_handler(rte_safestate_level_t level, rte_safestate_reason_t reason,
                                          const char *file, int32_t line, const char *message)
{
    (void)level;
    RTE_EXAMPLE_LOG("[%s] SAFE-state handler: reason=%u at %s:%d%s%s\n", g_ab_ctx.role_tag,
                         (unsigned int)reason, (file != NULL) ? file : "?", (int)line,
                         (message != NULL) ? " - " : "", (message != NULL) ? message : "");
    rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(g_ab_ctx.site), g_ab_ctx.cyc_cycle, g_ab_ctx.role_tag,
                          "-", "SAFE_HALT", (message != NULL) ? message : "entering SAFE state", NULL);
    /* Before any shutdown() below (CTC relay socket still open either
     * way here - see ab_gp_channel_raise_ctc_alarm()'s own doc on why
     * ordering matters): a dispatcher watching CTC needs to see every
     * safestate-driven fault this channel reacts to, not just this
     * process's own local log. */
    ab_gp_channel_raise_ctc_alarm(&g_ab_ctx, RBC_ALARM_SEVERITY_ERROR, (uint32_t)reason,
                                   (message != NULL) ? message : "entering SAFE state");
 
    if (reason == RTE_SAFESTATE_REASON_CHECKPOINT_TIMEOUT)
    {
        RTE_EXAMPLE_LOG(
            "[%s] checkpoint timeout treated as peer-link-down (REQ-RBC, ADR-029 2.6) - rebooting, not halting\n",
            g_ab_ctx.role_tag);
        g_ab_ctx.own_faulted = true;
        (void)ab_gp_channel_shutdown(&g_ab_ctx);
        RTE_SAFESTATE(RTE_SAFESTATE_LEVEL_REBOOT, RTE_EXAMPLE_REASON_AB_PEER_LINK_DOWN);
    }
}

static void load_legacy_config(ab_gp_channel_context_t *ctx, rte_example_site_t site, bool is_role_a)
{
    const char *env_cfg = getenv("RTE_CONFIG_FILE");
    const char *site_str = (site == RTE_EXAMPLE_SITE_WEST) ? "west" : "east";
    const char *role_str = is_role_a ? "a" : "b";
 
    if (env_cfg != NULL)
    {
        (void)ab_gp_config_load_file(env_cfg, &ctx->config);
    }
    else
    {
        char default_cfg_path[128];
        (void)snprintf(default_cfg_path, sizeof(default_cfg_path), "etc/config/%s_%s.conf", role_str, site_str);
        if (ab_gp_config_load_file(default_cfg_path, &ctx->config) != RTE_STATUS_OK)
        {
            (void)snprintf(default_cfg_path, sizeof(default_cfg_path), "/etc/sapi/%s_%s.conf", role_str, site_str);
            (void)ab_gp_config_load_file(default_cfg_path, &ctx->config);
        }
    }
}

static rte_status_t load_redundancy_config(ab_gp_channel_context_t *ctx, rte_example_site_t site, bool is_role_a)
{
    const char *env_redundancy_cfg = getenv("RTE_REDUNDANCY_CONFIG_FILE");
    const char *site_str = (site == RTE_EXAMPLE_SITE_WEST) ? "west" : "east";
    const char *role_str = is_role_a ? "a" : "b";
    char json_cfg_path[128];
    const char *redundancy_cfg_path = env_redundancy_cfg;
    rte_status_t redundancy_status;
 
    if (redundancy_cfg_path == NULL)
    {
        (void)snprintf(json_cfg_path, sizeof(json_cfg_path), "etc/config/%s_%s.json", role_str, site_str);
        redundancy_cfg_path = json_cfg_path;
    }
 
    redundancy_status = rte_redundancy_config_load(redundancy_cfg_path, &ctx->redundancy);
    if ((redundancy_status != RTE_STATUS_OK) && (env_redundancy_cfg == NULL))
    {
        redundancy_status = rte_redundancy_config_load("etc/config/redundancy.json", &ctx->redundancy);
    }
 
    if (redundancy_status == RTE_STATUS_NOT_SUPPORTED)
    {
        /* ab_gp_supports_redundancy() already logged which
         * topology/replica_count it rejected and why. */
        return RTE_STATUS_NOT_SUPPORTED;
    }
    if (redundancy_status != RTE_STATUS_OK)
    {
        rte_mem_set(&ctx->redundancy, 0, sizeof(ctx->redundancy));
        ctx->redundancy.topology = RTE_REDUNDANCY_TOPOLOGY_2OO2;
        ctx->redundancy.replica_count = 2U;
        ctx->redundancy.quorum_size = 2U;
        ctx->redundancy.standby_mode = RTE_STANDBY_MODE_COLD;
    }
 
    RTE_EXAMPLE_LOG("[%s] redundancy topology=%s replicas=%u quorum=%u standby_mode=%s\n", ctx->role_tag,
                         rte_redundancy_config_topology_to_string(ctx->redundancy.topology),
                         (unsigned int)ctx->redundancy.replica_count, (unsigned int)ctx->redundancy.quorum_size,
                         rte_redundancy_config_standby_mode_to_string(ctx->redundancy.standby_mode));
    return RTE_STATUS_OK;
}

static rte_status_t open_relay_channels(ab_gp_channel_context_t *ctx)
{
    uint8_t kind_idx;
    rte_status_t status;
 
    for (kind_idx = 0U; kind_idx < (uint8_t)RTE_EXAMPLE_RELAY_KIND_COUNT; kind_idx++)
    {
        rte_example_relay_kind_t kind = (rte_example_relay_kind_t)kind_idx;
        const char *name;
 
        if (kind == RTE_EXAMPLE_RELAY_KIND_TRAIN)
        {
            name = "ab-relay-train";
        }
        else if (kind == RTE_EXAMPLE_RELAY_KIND_IL)
        {
            name = "ab-relay-il";
        }
        else if (kind == RTE_EXAMPLE_RELAY_KIND_UTIL)
        {
            name = "ab-relay-util";
        }
        else
        {
            name = "ab-relay-ctc";
        }
        status = channel_service_setup_with_retry(&ctx->com.relay_channel[kind_idx], name);
        if (status != RTE_STATUS_OK)
        {
            RTE_EXAMPLE_LOG("[%s] relay channel setup failed for %s: %s\n", ctx->role_tag, name, rte_status_to_string(status));
            return status;
        }
    }
    return RTE_STATUS_OK;
}
 
rte_status_t ab_gp_channel_open_link(bool is_role_a, rte_example_site_t site, ab_gp_channel_handle_t *out_handle)
{
    ab_gp_channel_context_t *ctx = &g_ab_ctx;
    rte_status_t status;
 
    if (out_handle == NULL)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    *out_handle = NULL;
 
    rte_mem_set(ctx, 0, sizeof(*ctx));
    ctx->is_role_a = is_role_a;
    ctx->site = site;
    (void)snprintf(ctx->role_tag_buf, sizeof(ctx->role_tag_buf), "%s/%s", is_role_a ? "A" : "B",
                    site_config_name(site));
    ctx->role_tag = ctx->role_tag_buf;
    ctx->stop_requested = false;
    /* ctx->startup_timestamp_ms stays 0 here: A and B agree it over the peer link in
     * ab_gp_channel_negotiate_init() (ADR-040 step 5a), once the link below is open. */
 
    (void)ab_gp_config_init_defaults(site, is_role_a, &ctx->config);
    load_legacy_config(ctx, site, is_role_a);
 
    (void)rte_safestate_register_handler(RTE_SAFESTATE_LEVEL_SAFE, ab_gp_channel_safestate_handler);
 
    /* RCA/OCORA Phase 8 (see docs/rca/ and TODO.md): see load_redundancy_config()'s own doc. */
    (void)rte_redundancy_config_register_capability(ab_gp_supports_redundancy, ctx);
    if (load_redundancy_config(ctx, site, is_role_a) != RTE_STATUS_OK)
    {
        return RTE_STATUS_NOT_SUPPORTED;
    }
 
    /* RCA/OCORA Phase 4: full replacement of GP's A/B and site-negotiation
     * transport with rte_flow (routed through today's DDS-stub OSAdapter,
     * safeAPIFlowBackendDDS - see docs/rca/ and TODO.md). Was
     * rte_posix_osadapter_channel_service_register_resolver(resolve_channel, ctx)
     * (still used unchanged by role C, see gateway_c_common.c - out of
     * scope for this phase, see TODO.md's own scoping note). */
    status = rte_osadapter_flow_register(rte_protocol_adapter_dds_get_vtable());
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[%s] flow osadapter registration failed: %s\n", ctx->role_tag,
                             rte_status_to_string(status));
        return status;
    }
    status = rte_channel_service_flow_osadapter_register_resolver(resolve_channel_flow, ctx);
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[%s] resolver registration failed: %s\n", ctx->role_tag, rte_status_to_string(status));
        return status;
    }
 
    status = channel_service_setup_with_retry(&ctx->com.peer_channel, "ab-peer");
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[%s] peer channel setup failed: %s\n", ctx->role_tag, rte_status_to_string(status));
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), 0U, ctx->role_tag,
                              ctx->is_role_a ? "B" : "A", "LINK", "peer channel setup failed", NULL);
        return status;
    }
 
    status = channel_service_setup_with_retry(&ctx->com.crosscompare_channel, "ab-crosscompare");
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[%s] crosscompare channel setup failed: %s\n", ctx->role_tag, rte_status_to_string(status));
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), 0U, ctx->role_tag,
                              ctx->is_role_a ? "B" : "A", "LINK", "crosscompare channel setup failed", NULL);
        return status;
    }
 
    status = open_relay_channels(ctx);
    if (status != RTE_STATUS_OK)
    {
        return status;
    }
 
    *out_handle = (ab_gp_channel_handle_t)ctx;
    return RTE_STATUS_OK;
}
 
void ab_gp_channel_close_link(ab_gp_channel_handle_t handle)
{
    if (handle != NULL)
    {
        (void)ab_gp_channel_shutdown((void *)handle);
    }
}
 
rte_status_t ab_gp_channel_checkpoint_setup(ab_gp_channel_handle_t handle)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
    rte_status_t status;
 
    if (ctx == NULL)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    {
        rte_channel_config_t checkpoint_vc_cfg;
        rte_voter_config_t checkpoint_voter_cfg;
 
        ctx->checkpoint_channel_handle = ctx;
        /* Named for lookup (rte_voter_get_channel_by_name()) and logging -
         * "ChannelAtoB" from A's own perspective, "ChannelBtoA" from B's. */
        (void)snprintf(ctx->checkpoint_channel_name_buf, sizeof(ctx->checkpoint_channel_name_buf),
                        "Channel%sto%s", ctx->is_role_a ? "A" : "B", ctx->is_role_a ? "B" : "A");
        rte_mem_set(&checkpoint_vc_cfg, 0, sizeof(checkpoint_vc_cfg));
        checkpoint_vc_cfg.channel_handle = ctx->checkpoint_channel_handle;
        checkpoint_vc_cfg.send = ab_gp_channel_checkpoint_backend_send;
        checkpoint_vc_cfg.recv = ab_gp_channel_checkpoint_backend_recv;
        checkpoint_vc_cfg.name = ctx->checkpoint_channel_name_buf;
 
        rte_mem_set(&checkpoint_voter_cfg, 0, sizeof(checkpoint_voter_cfg));
        checkpoint_voter_cfg.voting_strategy = RTE_VOTING_NMR;
        checkpoint_voter_cfg.quorum_size = 1U;
        checkpoint_voter_cfg.channel_timeout_ms = ctx->config.timers.checkpoint_max_delay_ms;
 
        status = rte_voter_init(&ctx->checkpoint_voter, &checkpoint_voter_cfg);
        if (status == RTE_STATUS_OK)
        {
            status = rte_channel_init(&ctx->checkpoint_vc, &checkpoint_vc_cfg);
        }
        if (status == RTE_STATUS_OK)
        {
            status = rte_voter_register_channel(&ctx->checkpoint_voter, &ctx->checkpoint_vc);
        }
        if (status != RTE_STATUS_OK)
        {
            RTE_EXAMPLE_LOG("[%s] checkpoint channel/voter init failed (%s) - built-in checkpoint disabled this run\n",
                   ctx->role_tag, rte_status_to_string(status));
            rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), 0U, ctx->role_tag,
                                  ctx->is_role_a ? "B" : "A", "CHECKPOINT",
                                  "channel/voter init failed - built-in checkpoint disabled this run", NULL);
        }
        else
        {
            rte_timestamp_ms_t init_now_ms = 0U;
 
            ctx->checkpoint_cfg.voter = &ctx->checkpoint_voter;
            ctx->checkpoint_cfg.max_delay_ms = (rte_duration_ms_t)ctx->config.timers.checkpoint_relaxed_max_delay_ms;
            ctx->checkpoint_cfg.expected_node_count = 1U;
            ctx->checkpoint_cfg.watchdog = NULL; /* wired to ctx->watchdog in
                                                   * ab_gp_channel_watchdog_setup()
                                                   * once that watchdog exists -
                                                   * see its own comment */
            (void)rte_timer_now(&init_now_ms);
            ctx->checkpoint_relaxed_until_ms = init_now_ms + (rte_timestamp_ms_t)ctx->config.timers.checkpoint_relax_duration_ms;
        }
    }
    return status;
}
 
rte_status_t ab_gp_channel_dual_setup(ab_gp_channel_handle_t handle)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
 
    if (ctx == NULL)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    ctx->single_mode = false;
    ctx->was_online = false;
    ctx->last_peer_sample_cycle = 0U;
    return RTE_STATUS_OK;
}
 
const rte_appmanager_checkpoint_config_t *ab_gp_channel_checkpoint_config(ab_gp_channel_handle_t handle)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
 
    return (ctx != NULL) ? &ctx->checkpoint_cfg : NULL;
}
 
rte_status_t ab_gp_channel_crosscompare_setup(ab_gp_channel_handle_t handle)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
 
    if (ctx == NULL)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    return ab_gp_channel_crosscompare_init(ctx);
}
 
rte_status_t ab_gp_channel_init(void *context)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)context;
    uint32_t discarded_ticks = 0U;
    rte_status_t status;
 
    /* Ticks posted between the timer start and here do not count: the first cycle starts from an empty count. */
    status = ab_gp_cycle_ticks_take(&ctx->cycle_ticks, &discarded_ticks);
    if (status != RTE_STATUS_OK)
    {
        return status;
    }
    ctx->cycle = 1U;
    RTE_EXAMPLE_LOG("[%s] cyclic executive armed (period=%ums, starting cycle=%u)\n", ctx->role_tag,
                         (unsigned int)ctx->config.timers.cycle_period_ms, ctx->cycle);
    return RTE_STATUS_OK;
}
 
rte_status_t ab_gp_channel_train_codec_setup(ab_gp_channel_handle_t handle)
{
    const ab_gp_channel_context_t *ctx = (const ab_gp_channel_context_t *)handle;
    rte_status_t status;
 
    status = ab_gp_train_init();
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[%s] session codec registration (train) -> %s\n", ctx->role_tag,
                             rte_status_to_string(status));
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), 0U, ctx->role_tag, "-", "GP",
                              "train session codec registration failed", NULL);
    }
    return status;
}
 
rte_status_t ab_gp_channel_ga_setup(ab_gp_channel_handle_t handle)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
 
    ab_gp_ertms_engine_init(&ctx->db);
    /* Bare call now - GA calls back into GP's own functions (declared in
     * ga_interface.h) by ordinary symbol resolution against the real
     * librte_gp, not through a handed-off function-pointer struct.
     * See ga_interface.h's own header doc for the dummy-then-real
     * shared-library mechanism this replaced. */
    return ab_ga_ma_adaptation_init(ctx->sessions, &ctx->db, ctx->site, ctx->role_tag);
}
 
 
void ab_gp_channel_relay_queues_setup(ab_gp_channel_handle_t handle)
{
    (void)handle;
}
 
rte_status_t ab_gp_channel_negotiate_setup(ab_gp_channel_handle_t handle)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
    rte_status_t status;
 
    status = ab_gp_channel_negotiate_init(ctx);
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[%s] inter-site negotiation init failed (%s)\n", ctx->role_tag, rte_status_to_string(status));
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), 0U, ctx->role_tag, "-",
                              "NEGOTIATION", "init failed", NULL);
    }
    return status;
}
 
rte_status_t ab_gp_channel_watchdog_setup(ab_gp_channel_handle_t handle)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
    rte_watchdog_config_t wd_cfg;
    rte_status_t status;
 
    rte_mem_set(&wd_cfg, 0, sizeof(wd_cfg));
    wd_cfg.type = RTE_WATCHDOG_CHANNEL;
    wd_cfg.name = ctx->role_tag;
    wd_cfg.timeout_ms = (rte_duration_ms_t)ctx->config.timers.watchdog_timeout_ms;
    wd_cfg.action = RTE_WATCHDOG_ACTION_REBOOT;
    status = rte_watchdog_create(&ctx->watchdog, &wd_cfg);
    if (status == RTE_STATUS_OK)
    {
        (void)rte_watchdog_start(ctx->watchdog);
 
        /* Wire this same watchdog into the checkpoint config too (set up
         * earlier, in ab_gp_channel_checkpoint_setup() - this watchdog
         * did not exist yet at that point, hence the ordering here
         * instead of there). rte_channel_checkpoint() now kicks
         * config->watchdog once per internal retry round, not only on
         * success (see rte_checkpoint.h's own doc) - without this,
         * ctx->watchdog (kicked only once per full cycle, in
         * post_execute.c, AFTER a checkpoint call returns) fires and
         * reboots this process out from under a checkpoint call that is
         * still legitimately retrying within the relaxed startup budget
         * (RTE_EXAMPLE_AB_CHECKPOINT_RELAXED_MAX_DELAY_MS, far longer
         * than RTE_EXAMPLE_AB_WATCHDOG_TIMEOUT_MS) - confirmed live
         * via Docker: B/WEST logged "watchdog timeout - requesting
         * reboot" mid-retry, well before its own checkpoint call could
         * either succeed or genuinely exhaust its budget. */
        ctx->checkpoint_cfg.watchdog = ctx->watchdog;
    }
    else
    {
        RTE_EXAMPLE_LOG("[%s] watchdog_create failed (%s) - continuing without it\n", ctx->role_tag,
               rte_status_to_string(status));
        ctx->watchdog = NULL;
    }
    return status;
}
 
rte_status_t ab_gp_channel_cycle_timer_setup(ab_gp_channel_handle_t handle)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
    rte_timer_config_t timer_cfg;
    rte_status_t status;
 
    /* The tick count's mutex exists before the timer that posts to it (a mutex is a setup-phase resource). The status
     * text is printed by the caller's fatal path (main.c step 12). */
    status = ab_gp_cycle_ticks_setup(&ctx->cycle_ticks);
    if (status != RTE_STATUS_OK)
    {
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), 0U, ctx->role_tag, "-", "TIMER",
                              "cycle tick mutex_create failed", NULL);
        return status;
    }
 
    rte_mem_set(&timer_cfg, 0, sizeof(timer_cfg));
    timer_cfg.mode = RTE_TIMER_MODE_PERIODIC;
    timer_cfg.period_ms = (rte_duration_ms_t)ctx->config.timers.cycle_period_ms;
    timer_cfg.callback = cycle_tick_callback;
    timer_cfg.user_ctx = ctx;
    status = rte_timer_create(&ctx->cycle_timer_storage, &timer_cfg, &ctx->cycle_timer_handle);
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[%s] cycle timer_create failed (%s)\n", ctx->role_tag, rte_status_to_string(status));
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), 0U, ctx->role_tag, "-", "TIMER",
                              "cycle timer_create failed", NULL);
        return status;
    }
    status = rte_timer_start(ctx->cycle_timer_handle);
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[%s] cycle timer_start failed (%s)\n", ctx->role_tag, rte_status_to_string(status));
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), 0U, ctx->role_tag, "-", "TIMER",
                              "cycle timer_start failed", NULL);
    }
    return status;
}
 
rte_status_t ab_gp_channel_wait_cycle_tick(ab_gp_channel_handle_t handle)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
    unsigned int waited_ms;
    uint32_t pending = 0U;
    rte_status_t status;
 
    if (ctx == NULL)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    /* One take per 10 ms pass: the pending ticks are read and cleared in one critical section (ISS-051), so the value
     * that leaves the loop is the only read of the count. */
    waited_ms = 0U;
    status = ab_gp_cycle_ticks_take(&ctx->cycle_ticks, &pending);
    while ((status == RTE_STATUS_OK) && (pending == 0U))
    {
        ab_gp_channel_sleep_ms(10U);
        waited_ms += 10U;
        /* Fast internal cycle: keep the negotiation link serviced (warm/hot standby acks) between RBC cycles. */
        ab_gp_channel_negotiate_service_link(ctx);
        if (waited_ms >= ctx->config.timers.watchdog_timeout_ms)
        {
            return RTE_STATUS_TIMEOUT;
        }
        status = ab_gp_cycle_ticks_take(&ctx->cycle_ticks, &pending);
    }
    if (status != RTE_STATUS_OK)
    {
        return status;
    }
    {
        /* The tick callback counts every timer expiry; after a stall (this process was starved, or a cycle overran)
         * several ticks are pending. Consuming them all as ONE cycle made this channel's cycle number lag behind its
         * sibling's after every stall (found live: ~700 cycles apart after 40 minutes, the A/B cross-compare then
         * skipped every cycle). The missed ticks are counted, so the number keeps following real time. Bounded, so a
         * long suspension cannot make the number jump by more than 20 cycles at once. */
        const uint32_t missed = (pending > 1U) ? (pending - 1U) : 0U;
 
        ctx->cycle += (missed > 20U) ? 20U : missed;
    }
 
    if (ctx->checkpoint_relaxed_until_ms != 0U)
    {
        rte_timestamp_ms_t now_ms = 0U;
 
        (void)rte_timer_now(&now_ms);
        if (now_ms >= ctx->checkpoint_relaxed_until_ms)
        {
            ctx->checkpoint_cfg.max_delay_ms = (rte_duration_ms_t)ctx->config.timers.checkpoint_max_delay_ms;
            ctx->checkpoint_relaxed_until_ms = 0U;
        }
    }
 
    {
        /* Stall detector: the rule is ab_gp_stall_lost_contact() (pure, unit-tested); both times are the full 64-bit
         * monotonic time (ISS-052). */
        rte_timestamp_ms_t wall_now = 0U;
        rte_timestamp_ms_t stalled_ms = 0U;
 
        (void)rte_timer_now(&wall_now);
        if (ab_gp_stall_lost_contact(ctx->last_cycle_wall_ms, wall_now, ctx->is_online, &stalled_ms))
        {
            rte_site_role_note_lost_contact(&ctx->site_role);
            RTE_EXAMPLE_LOG("[%s] stalled for %llu ms while ONLINE - the counterpart may have taken over, yielding if it is ONLINE\n",
                            ctx->role_tag, (unsigned long long)stalled_ms);
        }
        ctx->last_cycle_wall_ms = wall_now;
    }
 
    ctx->cyc_cycle = ctx->cycle;
    ctx->cycle++;
 
    return RTE_STATUS_OK;
}
 
/* ISS-020: every UTIL send goes through here (per-cycle status publish, JRU log forwarder; both on the cycle thread).
 * Before the link's first successful open, each send is an open attempt that blocks until the status service answers
 * the handshake or the timeout runs out, so: at most one attempt per RTE_EXAMPLE_UTIL_REOPEN_BACKOFF_MS, each bounded
 * to RTE_EXAMPLE_UTIL_OPEN_ATTEMPT_TIMEOUT_MS; a refused send returns RTE_STATUS_TIMEOUT and the frame is dropped
 * (best-effort, as the callers already document). Once open, the gate always allows and the send is the one it
 * always was. */
static rte_status_t send_util_gated(ab_gp_channel_context_t *ctx, const uint8_t *bytes, size_t size)
{
    ab_gp_util_link_gate_t *gate = &ctx->com.util_gate;
    rte_timestamp_ms_t now_ms = 0U;
    rte_status_t status;
 
    (void)rte_timer_now(&now_ms);
    if (!ab_gp_util_link_gate_may_attempt(gate, now_ms))
    {
        status = RTE_STATUS_TIMEOUT;
    }
    else if (ab_gp_util_link_gate_is_open(gate))
    {
        status = rte_channel_service_send(&ctx->com.relay_channel[RTE_EXAMPLE_RELAY_KIND_UTIL], bytes, size,
                                          RTE_EXAMPLE_LINK_TIMEOUT_MS);
    }
    else
    {
        ab_gp_util_link_gate_on_attempt(gate, now_ms);
        status = rte_channel_service_send(&ctx->com.relay_channel[RTE_EXAMPLE_RELAY_KIND_UTIL], bytes, size,
                                          RTE_EXAMPLE_UTIL_OPEN_ATTEMPT_TIMEOUT_MS);
        (void)rte_timer_now(&now_ms);
        ab_gp_util_link_gate_on_result(gate, status, now_ms);
    }
    return status;
}
 
uint32_t ab_gp_channel_poll_relays(ab_gp_channel_handle_t handle, ab_gp_com_envelope_cb cb, void *user_ctx)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
    /* Sized for the widest relay kind (Train, RBC_TRAIN_ENVELOPE_WIRE_SIZE)
     * and shared across IL/CTC too - a genuinely-96-byte IL/CTC datagram
     * read into this bigger buffer just leaves its own tail untouched,
     * harmless since those domains' own decode never reads past their
     * own known 96-byte shape. */
    uint8_t env_bytes[RBC_TRAIN_ENVELOPE_WIRE_SIZE];
    uint32_t drained_count = 0U;
    uint8_t kind_idx;
 
    if (ctx == NULL)
    {
        return 0U;
    }
 
    /* ADR-036: one link per {kind}; the callback's index is always 0 -
     * the receiving domain code routes by the envelope's own train_id.
     * ADR-038: the UTIL link carries rbc_util_frame_t, not rbc_envelope_t. A/B mostly only
     * PUBLISH on it (post_execute.c) - but Gateway now also relays RBC_UTIL_TOPIC_KM114_FILE
     * chunks down this same link (Platform_Protocol_KM114/CLAUDE.md's "Consumption design"),
     * so it must be drained too - handled separately below, not through @p cb (util frames are
     * not rbc_envelope_t and would misparse). */
    for (kind_idx = 0U; kind_idx < (uint8_t)RTE_EXAMPLE_RELAY_KIND_COUNT; kind_idx++)
    {
        rte_example_relay_kind_t kind = (rte_example_relay_kind_t)kind_idx;
 
        if (kind == RTE_EXAMPLE_RELAY_KIND_UTIL)
        {
            uint8_t          util_bytes[RBC_UTIL_FRAME_SIZE];
            rbc_util_frame_t util_frame;
 
            /* ISS-020: no read before the link's first successful open. A read on the unopened link triggers the
             * lazy open, and its timeout 0 means "the channel's connect_timeout_ms" (1000 ms for ab-relay-util), so
             * this poll blocked the cycle thread ~1.3 s every cycle while the status service was absent (measured
             * 1298 ms: the 1000 ms open plus a nested 300 ms open from the JRU forwarder republishing the open's
             * own INFO line). The open is driven by the gated UTIL sends (ab_gp_channel_send_relay()); the Gateway
             * cannot send on this link before it has heard from A/B anyway, so nothing is lost by not reading. */
            if (!ab_gp_util_link_gate_is_open(&ctx->com.util_gate))
            {
                continue;
            }
            while (rte_channel_service_read(&ctx->com.relay_channel[kind_idx], util_bytes,
                                             sizeof(util_bytes), 0U) == RTE_STATUS_OK)
            {
                if (rbc_util_frame_decode(util_bytes, &util_frame) &&
                    (util_frame.msg_type == RBC_UTIL_MSG_PUBLISH) &&
                    (util_frame.topic_id == (uint16_t)RBC_UTIL_TOPIC_KM114_FILE))
                {
                    ab_gp_km114_relay_on_frame(&util_frame, ctx->role_tag, ctx->site, (ab_gp_channel_handle_t)ctx);
                }
                else if (rbc_util_frame_decode(util_bytes, &util_frame) &&
                         (util_frame.msg_type == RBC_UTIL_MSG_PUBLISH) &&
                         (util_frame.topic_id == (uint16_t)RBC_UTIL_TOPIC_TAKEOVER_CONFIRM))
                {
                    ab_gp_channel_negotiate_on_takeover_confirm(ctx, util_frame.payload_z);
                }
                /* Any other util topic/msg_type arriving here has no receiver today - dropped,
                 * same as before this Gateway->A/B direction existed. */
            }
            continue;
        }
 
        /* Zeroed before every read (not just once) so a genuinely-smaller
         * frame (e.g. a 96-byte flat-scheme kind, or IL/CTC's own always-96
         * traffic) leaves this buffer's own tail deterministic zero rather
         * than the previous iteration's bytes - callers key off each
         * kind's own known shape, but must never see stale data past it. */
        size_t read_size = (kind == RTE_EXAMPLE_RELAY_KIND_TRAIN) ? RBC_TRAIN_ENVELOPE_WIRE_SIZE
                                                                      : RBC_ENVELOPE_WIRE_SIZE;
        rte_mem_set(env_bytes, 0, sizeof(env_bytes));
        while (rte_channel_service_read(&ctx->com.relay_channel[kind_idx],
                                         env_bytes, read_size, 0U) == RTE_STATUS_OK)
        {
            drained_count++;
            if (cb != NULL)
            {
                cb(user_ctx, kind, 0U, env_bytes);
            }
            rte_mem_set(env_bytes, 0, sizeof(env_bytes));
        }
    }
 
    return drained_count;
}
 
void ab_gp_channel_poll_peer(ab_gp_channel_handle_t handle, ab_gp_com_peer_frame_cb cb, void *user_ctx)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
    uint8_t in_bytes[PEER_WIRE_FRAME_SIZE];
 
    if (ctx == NULL)
    {
        return;
    }
 
    while (rte_channel_service_read(&ctx->com.peer_channel, in_bytes, sizeof(in_bytes), 0U) == RTE_STATUS_OK)
    {
        if (cb != NULL)
        {
            cb(user_ctx, in_bytes[0], &in_bytes[1], sizeof(in_bytes) - 1U);
        }
    }
}
 
void ab_gp_channel_poll_crosscompare(ab_gp_channel_handle_t handle, ab_gp_com_peer_frame_cb cb, void *user_ctx)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
    uint8_t in_bytes[PEER_WIRE_FRAME_SIZE];
 
    if (ctx == NULL)
    {
        return;
    }
 
    while (rte_channel_service_read(&ctx->com.crosscompare_channel, in_bytes, sizeof(in_bytes), 0U) == RTE_STATUS_OK)
    {
        if (cb != NULL)
        {
            cb(user_ctx, in_bytes[0], &in_bytes[1], sizeof(in_bytes) - 1U);
        }
    }
}
 
rte_status_t ab_gp_channel_send_crosscompare(ab_gp_channel_handle_t handle, uint8_t kind, const uint8_t *payload, size_t size)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
    uint8_t out_bytes[PEER_WIRE_FRAME_SIZE];
 
    if ((ctx == NULL) || (payload == NULL) || (size > (sizeof(out_bytes) - 1U)))
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    rte_mem_set(out_bytes, 0, sizeof(out_bytes));
    out_bytes[0] = kind;
    rte_mem_copy(&out_bytes[1], payload, size);
    return rte_channel_service_send(&ctx->com.crosscompare_channel, out_bytes, sizeof(out_bytes),
                                      RTE_EXAMPLE_LINK_TIMEOUT_MS);
}
 
rte_status_t ab_gp_channel_send_relay(ab_gp_channel_handle_t handle, rte_example_relay_kind_t kind,
                                     uint8_t index, const uint8_t *bytes, size_t size)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
 
    (void)index; /* ADR-036: one link per kind - envelope carries train_id */
 
    if (ctx == NULL || bytes == NULL || size == 0U || kind >= RTE_EXAMPLE_RELAY_KIND_COUNT)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    /* Historical note: the Train relay channel's own transport
     * (gateway_c_train_handler.c, C-side) used to forward every
     * A/B->TrainSim datagram at a FIXED RBC_TRAIN_ENVELOPE_WIRE_SIZE
     * width, because rte_channel_service_read() had no "actual bytes
     * received" out-parameter for it to forward only what really
     * arrived. That gap is closed (rte_channel_service_read_ex(),
     * Platform_RTE) and gateway_c_train_handler.c now uses it both
     * directions, so a genuinely shorter datagram on this channel would
     * reach the far end at its real length today, not silently regrown to
     * full width. The zero-pad below is kept anyway, as a belt-and-
     * suspenders match for TrainRBCSim's own train_sim.py (_pad(), same
     * reasoning) and, more importantly, for RBC_Test_CommServer's TCP hop
     * further downstream, which still requires fixed rbc.message_size=512
     * framing (TCP has no datagram boundaries of its own to preserve) -
     * removing this padding is only safe once that hop is also made
     * length-aware, which it is not yet (see docs/architecture/
     * SAFE_COMM_INTEGRATION.md). Zero-pad every Train-kind send up to the
     * channel's own full width here, at this one choke point, so "every
     * datagram on this channel is exactly RBC_TRAIN_ENVELOPE_WIRE_SIZE
     * bytes" still holds for BOTH directions. */
    if ((kind == RTE_EXAMPLE_RELAY_KIND_TRAIN) && (size < (size_t)RBC_TRAIN_ENVELOPE_WIRE_SIZE))
    {
        uint8_t padded[RBC_TRAIN_ENVELOPE_WIRE_SIZE];
 
        rte_mem_set(padded, 0, sizeof(padded));
        rte_mem_copy(padded, bytes, size);
        return rte_channel_service_send(&ctx->com.relay_channel[kind], padded, sizeof(padded),
                                          RTE_EXAMPLE_LINK_TIMEOUT_MS);
    }
 
    if (kind == RTE_EXAMPLE_RELAY_KIND_UTIL)
    {
        return send_util_gated(ctx, bytes, size);
    }
 
    return rte_channel_service_send(&ctx->com.relay_channel[kind], bytes, size, RTE_EXAMPLE_LINK_TIMEOUT_MS);
}
 
rte_status_t ab_gp_channel_send_peer(ab_gp_channel_handle_t handle, const uint8_t *bytes, size_t size)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
 
    if (ctx == NULL || bytes == NULL || size == 0U)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    return rte_channel_service_send(&ctx->com.peer_channel, bytes, size, RTE_EXAMPLE_LINK_TIMEOUT_MS);
}
 
ab_gp_saf_train_t *ab_gp_channel_saf_train(ab_gp_channel_handle_t handle)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
 
    return (ctx == NULL) ? NULL : &ctx->saf_train;
}
 
rte_status_t ab_gp_channel_saf_train_setup(ab_gp_channel_handle_t handle)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
    const char *value = getenv("RTE_SAFECOMM_TRAIN_RBC_ID");
    char *end = NULL;
    unsigned long id;
    char info[32];
 
    if (ctx == NULL)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    if ((value == NULL) || (value[0] == '\0'))
    {
        ab_gp_saf_train_init(&ctx->saf_train, false, ctx->is_role_a, 0U);
        return RTE_STATUS_OK;
    }
    id = strtoul(value, &end, 0);
    if ((end == value) || (*end != '\0') || (id == 0UL) || (id > 0xFFFFFFUL))
    {
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), 0U, ctx->role_tag, "-", "SAF",
                              "RTE_SAFECOMM_TRAIN_RBC_ID is not a valid ETCS identity - refusing to start", NULL);
        return RTE_STATUS_INVALID_PARAM;
    }
    ab_gp_saf_train_init(&ctx->saf_train, true, ctx->is_role_a, (uint32_t)id);
    {
        const char *keys_path = getenv("RTE_SAFECOMM_TRAIN_KEYS_FILE");
        uint32_t loaded = 0U;
        char kinfo[32];
 
        if ((keys_path != NULL) && (keys_path[0] != '\0'))
        {
            if (!ab_gp_saf_keys_load_file(keys_path, &loaded))
            {
                rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), 0U, ctx->role_tag, "-", "SAF",
                                      "RTE_SAFECOMM_TRAIN_KEYS_FILE unreadable or malformed - refusing to start",
                                      NULL);
                return RTE_STATUS_INVALID_PARAM;
            }
            (void)snprintf(kinfo, sizeof(kinfo), "keys=%u", (unsigned int)loaded);
            rte_log_write_event(RTE_LOG_LEVEL_INFO, site_config_name(ctx->site), 0U, ctx->role_tag, "-", "SAF",
                                  "train KMACs loaded from RTE_SAFECOMM_TRAIN_KEYS_FILE", kinfo);
        }
    }
    (void)snprintf(info, sizeof(info), "rbc_etcs_id=%lu", id);
    rte_log_write_event(RTE_LOG_LEVEL_INFO, site_config_name(ctx->site), 0U, ctx->role_tag, "-", "SAF",
                          "train link protected (Subset-037 SaF terminated in this channel)", info);
    return RTE_STATUS_OK;
}
 
void ab_gp_channel_raise_ctc_alarm(ab_gp_channel_handle_t handle, rbc_alarm_severity_t severity, uint32_t code,
                                    const char *log_message)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
    rte_log_level_t log_level;
    uint8_t bytes[RBC_ENVELOPE_WIRE_SIZE];
    rte_timestamp_ms_t now_ms = 0U;
    rte_status_t status;
 
    if (ctx == NULL)
    {
        return;
    }
 
    switch (severity)
    {
        case RBC_ALARM_SEVERITY_WARNING:
            log_level = RTE_LOG_LEVEL_WARNING;
            break;
        case RBC_ALARM_SEVERITY_ERROR:
            log_level = RTE_LOG_LEVEL_ERROR;
            break;
        case RBC_ALARM_SEVERITY_INFO:
        default:
            log_level = RTE_LOG_LEVEL_INFO;
            break;
    }
    /* Guaranteed side of this function's contract - see its own header
     * doc: this local log line happens regardless of whether the CTC
     * send below succeeds. */
    rte_log_write_event(log_level, site_config_name(ctx->site), ctx->cyc_cycle, ctx->role_tag, "CTC", "ALARM",
                          (log_message != NULL) ? log_message : "-", NULL);
 
    (void)rte_timer_now(&now_ms);
    ab_gp_build_ctc_alarm(bytes, severity, code, ctx->cyc_cycle, (uint32_t)now_ms);
    /* Direct ab_gp_channel_send_relay(), not ab_gp_channel_stage_relay():
     * every current caller is reacting to an already-decided fault right
     * before shutdown/reboot, so there is no next checkpoint round to
     * gate this on - see this function's own header doc on the ordering
     * requirement (must run before ab_gp_channel_shutdown()). */
    status = ab_gp_channel_send_relay(handle, RTE_EXAMPLE_RELAY_KIND_CTC, 0U, bytes, RBC_ENVELOPE_WIRE_SIZE);
    if (status != RTE_STATUS_OK)
    {
        /* Best-effort, per this function's own doc: a failed CTC send
         * must never block or fail the caller's own (already-decided)
         * fault reaction - the local log line above is the guaranteed
         * part of this contract, this is only a diagnostic best-effort
         * addition on top of it. */
        rte_log_write_event(RTE_LOG_LEVEL_WARNING, site_config_name(ctx->site), ctx->cyc_cycle, ctx->role_tag,
                              "CTC", "ALARM_SEND_FAILED", rte_status_to_string(status), NULL);
    }
}
 
rte_status_t ab_gp_channel_shutdown(void *context)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)context;
    uint8_t kind_idx;
 
    if (ctx == NULL)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    ctx->stop_requested = true;
 
    (void)rte_voter_destroy(&ctx->checkpoint_voter);
    (void)rte_channel_destroy(&ctx->checkpoint_vc);
    ctx->checkpoint_cfg.voter = NULL;
 
    if (ctx->cycle_timer_handle != NULL)
    {
        (void)rte_timer_stop(ctx->cycle_timer_handle);
        (void)rte_timer_destroy(ctx->cycle_timer_handle);
        ctx->cycle_timer_handle = NULL;
    }
    /* After the timer is destroyed (its thread joined), so nothing posts any more; also when the timer was never
     * created but the mutex was. Not set up (a shutdown before the cycle timer setup) is refused by the call and
     * ignored here. */
    (void)ab_gp_cycle_ticks_teardown(&ctx->cycle_ticks);
    if (ctx->watchdog != NULL)
    {
        (void)rte_watchdog_stop(ctx->watchdog);
        (void)rte_watchdog_destroy(ctx->watchdog);
        ctx->watchdog = NULL;
    }
 
    (void)rte_channel_service_close(&ctx->com.peer_channel);
    (void)rte_channel_service_close(&ctx->com.crosscompare_channel);
 
    for (kind_idx = 0U; kind_idx < (uint8_t)RTE_EXAMPLE_RELAY_KIND_COUNT; kind_idx++)
    {
        (void)rte_channel_service_close(&ctx->com.relay_channel[kind_idx]);
    }
 
    ab_gp_channel_negotiate_shutdown(ctx);
    RTE_EXAMPLE_LOG("[%s] done\n", ctx->role_tag);
    rte_log_write_event(RTE_LOG_LEVEL_INFO, site_config_name(ctx->site), 0U, ctx->role_tag, "-", "LIFECYCLE",
                          "channel done", NULL);
    return RTE_STATUS_OK;
}