fix(sdks): provider reconnect across csharp/cpp/go/python/java demos

All six language demos stayed offline after the agent restarted
(incident 2026-08-16 05:33 / 07:38 UTC); only the JS demo recovered.

Root causes per SDK (all on the SDK->Agent local link):

- csharp: on remote close the read loop failed pending requests with
  SetCanceled, which the heartbeat loop misread as graceful shutdown and
  exited, so ReconnectAsync never ran. Fail pendings with an exception
  instead, only break on real cancellation, and trigger reconnect when
  the transport is already disconnected.
- cpp: heartbeat failure only logged and broke out of the loop; the
  reconnect_thread_ state existed but was never started from this path.
  Run a blocking reconnect loop on the heartbeat thread (self-join-safe
  via detached thread + thread-id guard).
- go: TCPClient.Call has no deadline, so a heartbeat on a half-dead
  connection blocked forever in the pending channel. Add a 30s timeout
  on heartbeat calls so the failure counter can trigger the existing
  reconnect path.
- python: _recover_connection held _state_lock across blocking dial/
  register calls, and _send_heartbeat/_send_drain_complete performed
  network I/O under the same lock; a blocked write stalled every state
  user (observed as a permanent futex wait on the heartbeat thread,
  tid 7, via /proc wchan). Snapshot state under the lock, do network
  I/O outside it.
- java: the demo image ships only slf4j-api, so the NOP logger silently
  swallowed every log line and the container looked dead. Add
  slf4j-simple to the demo runtime classpath.

Author: cuihairu

sdks/cpp/src/croupier_client.cpp

@@ -22,6 +22,7 @@
 #include <sstream>
 #include <stdexcept>
 #include <thread>
+#include <optional>
 #include <unordered_map>
 
 // Logging macros with configuration support
@@ -329,6 +330,7 @@ class CroupierClient::Impl {
     std::string session_id_;
     std::thread heartbeat_thread_;
     std::atomic<bool> should_stop_heartbeat_{false};
+    std::optional<std::thread::id> heartbeat_thread_id_;
     std::string last_error_;
 
     // Reconnection state
@@ -612,7 +614,11 @@ class CroupierClient::Impl {
     void startHeartbeatLoop() {
         stopHeartbeatLoop();
         should_stop_heartbeat_ = false;
+        // detach: the heartbeat thread may run the reconnect loop, and
+        // Connect() inside it calls startHeartbeatLoop() again — a join there
+        // would self-deadlock. Ownership is tracked via heartbeat_thread_id_.
         heartbeat_thread_ = std::thread([this]() {
+            heartbeat_thread_id_ = std::this_thread::get_id();
             const auto interval = std::max(1, config_.heartbeat_interval);
             while (!should_stop_heartbeat_) {
                 for (int elapsed = 0; elapsed < interval * 10 && !should_stop_heartbeat_; ++elapsed) {
@@ -628,10 +634,51 @@ class CroupierClient::Impl {
                     last_error_ = e.what();
                     connected_ = false;
                     SDK_LOG_WARN("Heartbeat failed: " + last_error_);
+                    if (!should_stop_heartbeat_ && running_) {
+                        reconnectLoop();
+                    }
                     break;
                 }
             }
         });
+        heartbeat_thread_.detach();
+    }
+
+    void stopHeartbeatLoop() {
+        should_stop_heartbeat_ = true;
+        // Cannot join from the heartbeat thread itself (reconnect path calls
+        // Connect() -> startHeartbeatLoop() -> stopHeartbeatLoop()).
+        if (heartbeat_thread_.joinable()) {
+            if (heartbeat_thread_id_.has_value() &&
+                *heartbeat_thread_id_ == std::this_thread::get_id()) {
+                return;  // self-stop: thread will exit via should_stop_heartbeat_
+            }
+            heartbeat_thread_.join();
+        }
+    }
+
+    // Blocking reconnect loop used after a heartbeat/connection failure.
+    // Retries with a fixed interval until Stop() is called or the
+    // connection is re-established. Runs on the heartbeat thread.
+    void reconnectLoop() {
+        std::lock_guard<std::mutex> lock(reconnect_mutex_);
+        if (is_reconnecting_.exchange(true)) {
+            return;  // another thread is already reconnecting
+        }
+        int attempt = 0;
+        while (!should_stop_heartbeat_ && running_) {
+            ++attempt;
+            SDK_LOG_INFO("Reconnecting to agent (attempt " + std::to_string(attempt) + ")...");
+            if (Connect()) {
+                SDK_LOG_INFO("Reconnected and re-registered after " + std::to_string(attempt) + " attempt(s)");
+                is_reconnecting_ = false;
+                return;
+            }
+            for (int waited = 0; waited < 5 * 10 && !should_stop_heartbeat_ && running_; ++waited) {
+                std::this_thread::sleep_for(std::chrono::milliseconds(100));
+            }
+        }
+        is_reconnecting_ = false;
     }
 
     void stopHeartbeatLoop() {

sdks/csharp/src/Croupier.Sdk/CroupierClient.cs

@@ -719,9 +719,17 @@ private async Task HeartbeatLoopAsync(CancellationToken cancellationToken)
             try
             {
                 await Task.Delay(interval, cancellationToken);
+
+                // The read loop may have marked the transport disconnected
+                // without this loop observing a failed heartbeat yet.
+                if (_transport is not { IsConnected: true })
+                {
+                    throw new InvalidOperationException("Not connected to Agent");
+                }
+
                 await SendHeartbeatAsync(cancellationToken);
             }
-            catch (OperationCanceledException)
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
             {
                 break;
             }

sdks/csharp/src/Croupier.Sdk/Transport/TCPTransport.cs

@@ -362,10 +362,12 @@ private async Task ReadLoop(CancellationToken cancellationToken)
             }
         }
 
-        // Fail all pending requests
+        // Fail all pending requests with an error (NOT cancellation): callers
+        // like the heartbeat loop treat OperationCanceledException as normal
+        // shutdown and would silently exit instead of reconnecting.
         foreach (var (reqId, tcs) in _pending)
         {
-            tcs.SetCanceled(cancellationToken);
+            tcs.TrySetException(new InvalidOperationException("connection closed"));
         }
         _pending.Clear();
     }

sdks/go/pkg/croupier/tcp_manager.go

@@ -412,7 +412,9 @@ func (m *TCPManager) sendHeartbeat(ctx context.Context) error {
 		return fmt.Errorf("marshal heartbeat: %w", err)
 	}
 
-	_, _, err = client.Call(ctx, protocol.MsgProviderHeartbeatRequest, reqBody)
+	hbCtx, hbCancel := context.WithTimeout(ctx, 30*time.Second)
+	defer hbCancel()
+	_, _, err = client.Call(hbCtx, protocol.MsgProviderHeartbeatRequest, reqBody)
 	return err
 }
 

sdks/java/examples/Dockerfile

@@ -22,6 +22,11 @@ COPY --from=build /out/croupier-java-sdk.jar /app/lib/croupier-java-sdk.jar
 COPY --from=build /out/classes /app/classes
 COPY --from=build /out/lib /app/lib
 
+# The demo only ships slf4j-api; without a binding every log line is silently
+# swallowed (NOP logger), making the container look dead. Add slf4j-simple so
+# demo logs (connect/register/reconnect) are visible on stdout.
+ADD --chmod=644 https://repo1.maven.org/maven2/org/slf4j/slf4j-simple/2.0.16/slf4j-simple-2.0.16.jar /app/lib/slf4j-simple.jar
+
 # The game-demo build occasionally bundles the SDK's own classes (io/github/.../wire)
 # into target/classes. Since the classpath puts /app/classes before the SDK jar,
 # those shadow copies would override the jar and pin stale wire encoding. Strip

sdks/python/croupier/__init__.py

@@ -646,7 +646,7 @@ def _send_drain_complete(self) -> None:
                 return
             transport = self._transport
 
-        # ProviderDrainCompleteRequest body is empty (no proto definition needed).
+        # Network call outside the state lock.
         transport.call(protocol.MSG_PROVIDER_DRAIN_COMPLETE_REQUEST, b"")
         LOG.info("DrainComplete sent to Agent")
 
@@ -673,23 +673,32 @@ def _heartbeat_loop(self) -> None:
                 self._recover_connection()
 
     def _send_heartbeat(self) -> None:
+        # Snapshot state under the lock, then perform the network call
+        # WITHOUT holding _state_lock: a blocked socket write must never
+        # stall every other state user (observed as a permanent futex wait
+        # on the heartbeat thread after the agent restarts).
         with self._state_lock:
             if not self._transport or not self._session_id:
                 raise RuntimeError("Client is not registered")
             transport = self._transport
-            request = provider_pb2.ProviderHeartbeatRequest(
-                service_id=self._config.service_id,
-                session_id=self._session_id,
-            )
-            req_data = request.SerializeToString()
+            session_id = self._session_id
+            service_id = self._config.service_id
+
+        request = provider_pb2.ProviderHeartbeatRequest(
+            service_id=service_id,
+            session_id=session_id,
+        )
+        req_data = request.SerializeToString()
 
         transport.call(protocol.MSG_PROVIDER_HEARTBEAT_REQUEST, req_data)
 
     def _recover_connection(self) -> None:
         while not self._heartbeat_stop.is_set():
             try:
+                # Do NOT hold _state_lock across the (blocking) dial/register
+                # network calls; only take it to publish the new state.
+                self._connect_and_register()
                 with self._state_lock:
-                    self._connect_and_register()
                     self._connected = True
                 LOG.info("Reconnected and re-registered service %s", self._config.service_id)
                 return