fix(470): auto-recovery probe + force T3 rebuild on RPC peer reconnection

ggml/include/ggml-rpc.h

@@ -84,6 +84,16 @@ GGML_BACKEND_API void ggml_backend_rpc_clear_local_tensors(void);
 // must rebind (or fall back to solo). Cheap — reads one atomic.
 GGML_BACKEND_API uint32_t ggml_backend_rpc_get_registry_epoch(void);
 
+// #470 Option B: check if the peer for a given device reconnected since the
+// last call. Returns true once per reconnection event (flag is cleared on read).
+// The engine calls this after graph_compute fails to decide whether to trigger
+// a T3 rebuild (re-provision) or treat it as a transient error.
+GGML_BACKEND_API bool ggml_backend_rpc_check_peer_reconnection(uint32_t device_idx);
+
+// #470: check if ANY RPC peer has reconnected (used by PREFILL handler
+// to detect peer restarts before graph_compute runs)
+GGML_BACKEND_API bool ggml_backend_rpc_check_any_peer_reconnection();
+
 // #368: fetch the peer's current registry epoch over RPC (uses
 // RPC_CMD_RESOLVE_TENSOR on a sentinel name and reads back the epoch).
 // Returns 0 on any failure (peer unreachable, RPC error, peer doesn't

ggml/src/ggml-rpc/ggml-rpc.cpp

@@ -280,6 +280,10 @@ struct ggml_backend_rpc_device_context {
     // identity at compute time (lock-free; get_socket never touches the
     // registry mutex, so no lock-order inversion with ggml_backend_rpc_add_server).
     std::weak_ptr<socket_t> last_sock;
+    // #470 Option B: set when a peer reconnection is detected (last_sock changed).
+    // The engine checks this after graph_compute and triggers a T3 rebuild to
+    // re-provision model layers on the fresh peer. Cleared by check function.
+    std::atomic<bool> peer_reconnected{false};
 };
 
 // Forward declaration — defined after ggml_backend_rpc_reg_context (needs the
@@ -289,6 +293,21 @@ struct ggml_backend_rpc_device_context {
 // full GRAPH_COMPUTE (re-serialized with current data pointers).
 static void ggml_backend_rpc_invalidate_recompute(const std::string & endpoint);
 
+// #470 Option B: helper to set peer_reconnected flag on the device that owns
+// a buffer, given the buffer's buffer_type. Used by fail-soft paths in buffer
+// functions when the peer restarts (RPC fails, stale pointers, etc.).
+static void rpc_set_reconnect_flag(ggml_backend_buffer_t buffer, const char * func) {
+    ggml_backend_dev_t dev = ggml_backend_buft_get_device(buffer->buft);
+    if (dev) {
+        ggml_backend_rpc_device_context * dev_ctx =
+            (ggml_backend_rpc_device_context *)dev->context;
+        if (!dev_ctx->peer_reconnected.exchange(true, std::memory_order_acq_rel)) {
+            GGML_LOG_WARN("[%s] peer reconnection detected — "
+                          "engine will trigger T3 rebuild\n", func);
+        }
+    }
+}
+
 struct ggml_backend_rpc_buffer_type_context {
     std::string endpoint;
     uint32_t    device;
@@ -523,16 +542,17 @@ static void ggml_backend_rpc_buffer_free_buffer(ggml_backend_buffer_t buffer) {
     ggml_backend_rpc_buffer_context * ctx = (ggml_backend_rpc_buffer_context *)buffer->context;
     // #470: re-resolve — the free must ride the CURRENT connection, not the
     // socket captured at alloc time (dead after a head teardown/re-attach).
-    // No error channel: a genuinely unreachable peer aborts loudly rather
-    // than silently leaking the peer allocation (correctness-preserving).
     auto sock = get_socket(ctx->endpoint);
     if (sock == nullptr) {
-        GGML_ABORT("[%s] cannot reach peer %s (reconnect failed) — refusing to silently skip FREE_BUFFER\n",
-                   __func__, ctx->endpoint.c_str());
+        // #470 Option B: fail-soft
+        rpc_set_reconnect_flag(buffer, __func__);
+        return;
     }
     rpc_msg_free_buffer_req request = {ctx->remote_ptr};
     bool status = send_rpc_cmd(sock, RPC_CMD_FREE_BUFFER, &request, sizeof(request), nullptr, 0);
-    RPC_STATUS_ASSERT(status);
+    if (!status) {
+        rpc_set_reconnect_flag(buffer, __func__);
+    }
     // #470: freeing the buffer frees the peer memory the server's stored graph
     // points into. Reset the recompute fast-path so the next compute with a
     // recycled graph uid is a full GRAPH_COMPUTE (fresh data pointers), never
@@ -552,13 +572,17 @@ static void * ggml_backend_rpc_buffer_get_base(ggml_backend_buffer_t buffer) {
     // happy path (mutex + map lookup, no network).
     auto sock = get_socket(ctx->endpoint);
     if (sock == nullptr) {
-        GGML_ABORT("[%s] cannot reach peer %s (reconnect failed) — refusing to return a stale base pointer\n",
-                   __func__, ctx->endpoint.c_str());
+        // #470 Option B: fail-soft
+        rpc_set_reconnect_flag(buffer, __func__);
+        return nullptr;
     }
     rpc_msg_buffer_get_base_req request = {ctx->remote_ptr};
     rpc_msg_buffer_get_base_rsp response;
     bool status = send_rpc_cmd(sock, RPC_CMD_BUFFER_GET_BASE, &request, sizeof(request), &response, sizeof(response));
-    RPC_STATUS_ASSERT(status);
+    if (!status) {
+        rpc_set_reconnect_flag(buffer, __func__);
+        return nullptr;
+    }
     ctx->base_ptr = reinterpret_cast<void *>(response.base_ptr);
     return ctx->base_ptr;
 }
@@ -623,15 +647,14 @@ static enum ggml_status ggml_backend_rpc_buffer_init_tensor(ggml_backend_buffer_
         request.tensor = serialize_tensor(tensor);
 
         if (sock == nullptr) {
-            // No error channel: abort loudly rather than silently skipping
-            // the server-side padding init (correctness-preserving). In
-            // practice get_socket's reconnect fixes the dead fd, so this
-            // only fires when the peer is genuinely unreachable.
-            GGML_ABORT("[%s] cannot reach peer %s (reconnect failed) — refusing to silently skip INIT_TENSOR\n",
-                       __func__, ctx->endpoint.c_str());
+            // #470 Option B: fail-soft
+            rpc_set_reconnect_flag(buffer, __func__);
+            return GGML_STATUS_SUCCESS;
         }
         bool status = send_rpc_cmd(sock, RPC_CMD_INIT_TENSOR, &request, sizeof(request), nullptr, 0);
-        RPC_STATUS_ASSERT(status);
+        if (!status) {
+            rpc_set_reconnect_flag(buffer, __func__);
+        }
     }
     return GGML_STATUS_SUCCESS;
 }
@@ -644,8 +667,9 @@ static void ggml_backend_rpc_buffer_set_tensor(ggml_backend_buffer_t buffer, ggm
     // reconnect fixes the dead fd in the COMBINED teardown/re-attach case.
     auto sock = get_socket(ctx->endpoint);
     if (sock == nullptr) {
-        GGML_ABORT("[%s] cannot reach peer %s (reconnect failed) — refusing to silently skip SET_TENSOR\n",
-                   __func__, ctx->endpoint.c_str());
+        // #470 Option B: fail-soft
+        rpc_set_reconnect_flag(buffer, __func__);
+        return;
     }
     rpc_tensor rpc_tensor = serialize_tensor(tensor);
     if (size > HASH_THRESHOLD) {
@@ -655,7 +679,10 @@ static void ggml_backend_rpc_buffer_set_tensor(ggml_backend_buffer_t buffer, ggm
         request.hash = fnv_hash((const uint8_t*)data, size);
         rpc_msg_set_tensor_hash_rsp response;
         bool status = send_rpc_cmd(sock, RPC_CMD_SET_TENSOR_HASH, &request, sizeof(request), &response, sizeof(response));
-        RPC_STATUS_ASSERT(status);
+        if (!status) {
+            rpc_set_reconnect_flag(buffer, __func__);
+            return;
+        }
         if (response.result) {
             // the server has the same data, no need to send it
             return;
@@ -668,7 +695,9 @@ static void ggml_backend_rpc_buffer_set_tensor(ggml_backend_buffer_t buffer, ggm
     memcpy(input.data() + sizeof(rpc_tensor), &offset, sizeof(offset));
     memcpy(input.data() + sizeof(rpc_tensor) + sizeof(offset), data, size);
     bool status = send_rpc_cmd(sock, RPC_CMD_SET_TENSOR, input.data(), input.size());
-    RPC_STATUS_ASSERT(status);
+    if (!status) {
+        rpc_set_reconnect_flag(buffer, __func__);
+    }
 }
 
 static void ggml_backend_rpc_buffer_get_tensor(ggml_backend_buffer_t buffer, const ggml_tensor * tensor, void * data, size_t offset, size_t size) {
@@ -677,15 +706,19 @@ static void ggml_backend_rpc_buffer_get_tensor(ggml_backend_buffer_t buffer, con
     // the read; the reconnect rides the current connection instead.
     auto sock = get_socket(ctx->endpoint);
     if (sock == nullptr) {
-        GGML_ABORT("[%s] cannot reach peer %s (reconnect failed) — refusing to return garbage for GET_TENSOR\n",
-                   __func__, ctx->endpoint.c_str());
+        // #470 Option B: fail-soft
+        rpc_set_reconnect_flag(buffer, __func__);
+        return;
     }
     rpc_msg_get_tensor_req request;
     request.tensor = serialize_tensor(tensor);
     request.offset = offset;
     request.size = size;
     bool status = send_rpc_cmd(sock, RPC_CMD_GET_TENSOR, &request, sizeof(request), data, size);
-    RPC_STATUS_ASSERT(status);
+    if (!status) {
+        // #470 Option B: fail-soft instead of RPC_STATUS_ASSERT crash
+        rpc_set_reconnect_flag(buffer, __func__);
+    }
 }
 
 static bool ggml_backend_rpc_buffer_cpy_tensor(ggml_backend_buffer_t buffer, const ggml_tensor * src, ggml_tensor * dst) {
@@ -733,12 +766,15 @@ static void ggml_backend_rpc_buffer_clear(ggml_backend_buffer_t buffer, uint8_t
     // never on a dead cached sock.
     auto sock = get_socket(ctx->endpoint);
     if (sock == nullptr) {
-        GGML_ABORT("[%s] cannot reach peer %s (reconnect failed) — refusing to silently skip BUFFER_CLEAR\n",
-                   __func__, ctx->endpoint.c_str());
+        // #470 Option B: fail-soft
+        rpc_set_reconnect_flag(buffer, __func__);
+        return;
     }
     rpc_msg_buffer_clear_req request = {ctx->remote_ptr, value};
     bool status = send_rpc_cmd(sock, RPC_CMD_BUFFER_CLEAR, &request, sizeof(request), nullptr, 0);
-    RPC_STATUS_ASSERT(status);
+    if (!status) {
+        rpc_set_reconnect_flag(buffer, __func__);
+    }
 }
 
 static ggml_backend_buffer_i ggml_backend_rpc_buffer_interface = {
@@ -938,9 +974,27 @@ static enum ggml_status ggml_backend_rpc_graph_compute(ggml_backend_t backend, g
     // fresh instance with an empty stored-graph cache — a GRAPH_RECOMPUTE
     // would be refused and the connection would churn. Detect it here (lock
     // free, by socket identity) and fall back to a full GRAPH_COMPUTE.
-    if (rpc_dev_ctx->last_sock.lock() != sock) {
+    auto prev_sock = rpc_dev_ctx->last_sock.lock();
+    if (prev_sock != sock) {
         rpc_dev_ctx->last_graph_uid = 0;
         rpc_dev_ctx->last_sock      = sock;
+        // #470 Option B: a GENUINE peer reconnection means we previously held a
+        // live socket (prev_sock != nullptr) that now differs from the new one.
+        // On a fresh context (engine start, or right after a T3 rebuild which
+        // recreates the device context with an empty last_sock) prev_sock is
+        // nullptr — that is the FIRST connect, NOT a reconnection, and must
+        // proceed normally. Without this guard every engine start (and every
+        // post-rebuild compute) would spuriously return FAILED and loop on
+        // T3 rebuilds, so the engine could never serve.
+        if (prev_sock != nullptr) {
+            rpc_dev_ctx->peer_reconnected.store(true, std::memory_order_release);
+            GGML_LOG_WARN("[%s] peer %s reconnected — stale buffers, "
+                          "engine should trigger re-provision\n",
+                          __func__, rpc_ctx->endpoint.c_str());
+            // Return FAILED so the engine knows the peer needs re-provision.
+            // The caller (engine) will detect this and trigger T3 rebuild.
+            return GGML_STATUS_FAILED;
+        }
     }
     bool reuse = cgraph->uid != 0 && rpc_dev_ctx->last_graph_uid == cgraph->uid;
     if (reuse) {
@@ -1086,6 +1140,58 @@ uint32_t ggml_backend_rpc_get_registry_epoch(void) {
     return g_hydra_registry_epoch.load(std::memory_order_acquire);
 }
 
+// #470 Option B: check if the peer for a given device reconnected since the
+// last call. Returns true once per reconnection event (flag is cleared on read).
+// The engine calls this after graph_compute fails to decide whether to trigger
+// a T3 rebuild (re-provision) or treat it as a transient error.
+bool ggml_backend_rpc_check_peer_reconnection(uint32_t device_idx) {
+    // Get the RPC backend registry (index 0 is the first registered backend)
+    ggml_backend_reg_t reg = ggml_backend_reg_get(0);
+    if (!reg) {
+        return false;
+    }
+    // Check if the device index is valid
+    size_t dev_count = ggml_backend_reg_dev_count(reg);
+    if (device_idx >= dev_count) {
+        return false;
+    }
+    ggml_backend_dev_t dev = ggml_backend_reg_dev_get(reg, device_idx);
+    if (!dev) {
+        return false;
+    }
+    // Check if this is an RPC device by name
+    const char * name = ggml_backend_dev_name(dev);
+    if (!name || strncmp(name, "RPC", 3) != 0) {
+        return false;
+    }
+    // The device context is stored in dev->context — but we need to cast it
+    // to our specific context type. Since we know this is an RPC device
+    // (checked by name), we can safely cast.
+    ggml_backend_rpc_device_context * ctx = (ggml_backend_rpc_device_context *)dev->context;
+    return ctx->peer_reconnected.exchange(false, std::memory_order_acquire);
+}
+
+// #470: Check if ANY RPC peer has reconnected. Called from the PREFILL
+// handler to detect peer restarts before graph_compute runs.
+bool ggml_backend_rpc_check_any_peer_reconnection() {
+    ggml_backend_reg_t reg = ggml_backend_reg_get(0);
+    if (!reg) {
+        return false;
+    }
+    size_t dev_count = ggml_backend_reg_dev_count(reg);
+    for (size_t i = 0; i < dev_count; i++) {
+        ggml_backend_dev_t dev = ggml_backend_reg_dev_get(reg, i);
+        if (!dev) continue;
+        const char * name = ggml_backend_dev_name(dev);
+        if (!name || strncmp(name, "RPC", 3) != 0) continue;
+        ggml_backend_rpc_device_context * ctx = (ggml_backend_rpc_device_context *)dev->context;
+        if (ctx && ctx->peer_reconnected.exchange(false, std::memory_order_acquire)) {
+            return true;
+        }
+    }
+    return false;
+}
+
 void ggml_backend_rpc_register_local_tensor(const char * name, struct ggml_tensor * tensor) {
     if (!tensor || !tensor->buffer || !tensor->data) {
         return;
@@ -2605,6 +2711,12 @@ static void * ggml_backend_rpc_get_proc_address(ggml_backend_reg_t reg, const ch
     if (std::strcmp(name, "ggml_backend_rpc_get_registry_epoch") == 0) {
         return (void *)ggml_backend_rpc_get_registry_epoch;
     }
+    if (std::strcmp(name, "ggml_backend_rpc_check_peer_reconnection") == 0) {
+        return (void *)ggml_backend_rpc_check_peer_reconnection;
+    }
+    if (std::strcmp(name, "ggml_backend_rpc_check_any_peer_reconnection") == 0) {
+        return (void *)ggml_backend_rpc_check_any_peer_reconnection;
+    }
     if (std::strcmp(name, "ggml_backend_rpc_get_remote_registry_epoch") == 0) {
         return (void *)ggml_backend_rpc_get_remote_registry_epoch;
     }

src/llama-context.cpp

@@ -1466,6 +1466,25 @@ llm_graph_result * llama_context::process_ubatch(const llama_ubatch & ubatch, ll
     const auto status = graph_compute(res->get_gf(), ubatch.n_tokens > 1);
     if (status != GGML_STATUS_SUCCESS) {
         LLAMA_LOG_ERROR("%s: failed to compute graph, compute status: %d\n", __func__, status);
+        // #470 Option B: after graph_compute failure, check if any RPC peer
+        // reconnected. If so, set the flag so the server can trigger a T3 rebuild.
+        ggml_backend_reg_t reg = ggml_backend_reg_get(0);
+        if (reg) {
+            for (uint32_t i = 0; i < ggml_backend_reg_dev_count(reg); i++) {
+                ggml_backend_dev_t dev = ggml_backend_reg_dev_get(reg, i);
+                if (!dev) continue;
+                const char * dev_name = ggml_backend_dev_name(dev);
+                if (dev_name && strncmp(dev_name, "RPC", 3) == 0) {
+                    auto check_fn = (bool(*)(uint32_t)) ggml_backend_reg_get_proc_address(
+                        reg, "ggml_backend_rpc_check_peer_reconnection");
+                    if (check_fn && check_fn(i)) {
+                        LLAMA_LOG_WARN("%s: peer reconnection detected on device %u (%s) — "
+                                       "flagging T3 rebuild\n", __func__, i, dev_name);
+                        peer_reconnection_pending = true;
+                    }
+                }
+            }
+        }
         ret = status;
         return nullptr;
     }
@@ -2515,6 +2534,26 @@ ggml_status llama_context::graph_compute(
     auto status = ggml_backend_sched_graph_compute_async(sched.get(), gf);
     if (status != GGML_STATUS_SUCCESS) {
         LLAMA_LOG_ERROR("%s: ggml_backend_sched_graph_compute_async failed with error %d\n", __func__, status);
+        // #470 Option B: check if any RPC peer reconnected — if so, set the
+        // peer_reconnection_pending flag so the server triggers T3 rebuild.
+        ggml_backend_reg_t reg = ggml_backend_reg_get(0);
+        if (reg) {
+            for (uint32_t i = 0; i < ggml_backend_reg_dev_count(reg); i++) {
+                ggml_backend_dev_t dev = ggml_backend_reg_dev_get(reg, i);
+                if (!dev) continue;
+                const char * dev_name = ggml_backend_dev_name(dev);
+                if (dev_name && strncmp(dev_name, "RPC", 3) == 0) {
+                    auto check_fn = (bool(*)(uint32_t)) ggml_backend_reg_get_proc_address(
+                        reg, "ggml_backend_rpc_check_peer_reconnection");
+                    if (check_fn && check_fn(i)) {
+                        LLAMA_LOG_WARN("%s: peer reconnection detected on device %u (%s) — "
+                                       "setting peer_reconnection_pending for T3 rebuild\n",
+                                       __func__, i, dev_name);
+                        peer_reconnection_pending = true;
+                    }
+                }
+            }
+        }
     } else {
         // Force completion before unlocking so a concurrent RPC-driven
         // compute on the shared backend can't start while this async

src/llama-context.h

@@ -450,4 +450,10 @@ struct llama_context {
     std::string hydra_pending_config_json;
     std::string hydra_pending_config_tier;
     time_t      hydra_pending_config_set_at = 0;
+
+public:
+    // #470 Option B: flag set by graph_compute when a peer reconnection is
+    // detected. The server checks this in update_slots() and triggers a
+    // T3 rebuild to re-provision model layers on the fresh peer.
+    bool peer_reconnection_pending = false;
 };