fix(grpc): drain log stream gracefully on success path in solve_lp/mip

stop_log_streaming() calls TryCancel() which races against the server's
final log drain: poll_for_completion() returns as soon as the job is
marked complete, then TryCancel() kills the stream before the server has
finished sending remaining lines (e.g. "Best objective …").

Add drain_log_streaming(): polls log_stream_done_ (set when the thread
exits naturally after receiving the job_complete sentinel) for up to 5s,
then falls back to stop_log_streaming(). Use it on the success path in
solve_lp and solve_mip so all final log lines are received before the
stream is torn down.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Signed-off-by: Ramakrishna Prabhu <ramakrishnap@nvidia.com>

Author: ramakrishnap-nv

cpp/src/grpc/client/grpc_client.cpp

@@ -281,12 +281,14 @@ void grpc_client_t::start_log_streaming(const std::string& job_id)
   }
 
   stop_logs_.store(false);
+  log_stream_done_.store(false);
   log_thread_ = std::make_unique<std::thread>([this, job_id]() {
     stream_logs(job_id, 0, [this](const std::string& line, bool /*job_complete*/) {
       if (stop_logs_.load()) return false;
       if (config_.log_callback) { config_.log_callback(line); }
       return true;
     });
+    log_stream_done_.store(true);
   });
 }
 
@@ -308,6 +310,21 @@ void grpc_client_t::stop_log_streaming()
   if (t && t->joinable()) { t->join(); }
 }
 
+void grpc_client_t::drain_log_streaming()
+{
+  // On the success path the server drains all pending log lines and then sends
+  // a job_complete sentinel; the streaming thread exits naturally when that
+  // sentinel arrives.  Poll for natural completion (up to kDrainTimeout) so
+  // callers receive the final log lines (e.g. "Best objective …") before we
+  // force-cancel.  Fall back to stop_log_streaming() if the deadline passes.
+  static constexpr auto kDrainTimeout = std::chrono::seconds(5);
+  auto deadline                       = std::chrono::steady_clock::now() + kDrainTimeout;
+  while (!log_stream_done_.load() && std::chrono::steady_clock::now() < deadline) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+  }
+  stop_log_streaming();
+}
+
 // =============================================================================
 // Proto → Client Enum Conversion
 // =============================================================================
@@ -848,7 +865,13 @@ remote_lp_result_t<i_t, f_t> grpc_client_t::solve_lp(
 
   start_log_streaming(sub.job_id);
   auto poll = poll_for_completion(sub.job_id);
-  stop_log_streaming();
+  // On success, wait for the server to send the job_complete sentinel so all
+  // final log lines are received before the stream is torn down.
+  if (poll.completed) {
+    drain_log_streaming();
+  } else {
+    stop_log_streaming();
+  }
 
   if (!poll.completed) { return {.error_message = poll.error_message}; }
 
@@ -875,7 +898,11 @@ remote_mip_result_t<i_t, f_t> grpc_client_t::solve_mip(
 
   start_log_streaming(sub.job_id);
   auto poll = poll_for_completion(sub.job_id);
-  stop_log_streaming();
+  if (poll.completed) {
+    drain_log_streaming();
+  } else {
+    stop_log_streaming();
+  }
 
   if (!poll.completed) { return {.error_message = poll.error_message}; }
 

cpp/src/grpc/client/grpc_client.hpp

@@ -408,6 +408,10 @@ class grpc_client_t {
   // Activated when config_.stream_logs is true and config_.log_callback is set.
   void start_log_streaming(const std::string& job_id);
   void stop_log_streaming();
+  // Graceful variant: waits up to kLogDrainTimeout for the server to send the
+  // job_complete sentinel before force-cancelling.  Use on the success path so
+  // final log lines (e.g. "Best objective …") are not dropped.
+  void drain_log_streaming();
 
   // Shared polling loop used by solve_lp and solve_mip.
   struct poll_result_t {
@@ -419,6 +423,7 @@ class grpc_client_t {
 
   std::unique_ptr<std::thread> log_thread_;
   std::atomic<bool> stop_logs_{false};
+  std::atomic<bool> log_stream_done_{false};
   mutable std::mutex log_context_mutex_;
   // Points to the grpc::ClientContext* of the in-flight StreamLogs RPC (if
   // any).  Typed as void* to avoid exposing grpc headers in the public API.