Implement distributed reply channels for remote await (#169)

* Implement distributed reply channels for remote await

Add a request-response protocol over the mesh so that
`await actor.method()` on a remote actor returns the actual
reply instead of falling back to fire-and-forget.

Wire protocol:
- Extend HewWireEnvelope with request_id (field 5, varint) and
  source_node_id (field 6, varint). Fields are only encoded when
  non-zero, preserving backward compatibility with existing nodes.

Reply routing table (hew_node.rs):
- Process-global HashMap<u64, Arc<PendingReply>> keyed by request_id.
- Atomic u64 counter for unique request ID generation.
- Condvar-based blocking with timeout for the calling thread.

Outbound remote ask (hew_node_api_ask):
- If local PID: delegates to hew_actor_ask_by_id (new).
- If remote PID: registers pending reply, sends envelope with
  request_id and source_node_id, blocks on condvar (5s timeout),
  returns reply data or zeroed fallback buffer.

Inbound ask handling (node_inbound_router):
- Detects request_id > 0 with source_node_id > 0.
- Spawns a thread to perform a local blocking ask via
  hew_actor_ask_by_id, then sends the reply envelope back.

Reply receipt (reader_loop in connection.rs):
- Detects reply envelopes (request_id > 0, source_node_id == 0).
- Deposits payload in the reply routing table, signalling the
  condvar to wake the blocked caller.

Codegen (codegen.cpp ActorAskOpLowering):
- Remote i64 targets now call hew_node_api_ask instead of
  hew_actor_send_by_id, enabling actual return values.

New runtime helpers:
- hew_actor_ask_by_id: blocking ask by actor ID (not pointer).
- hew_reply_wait_with_size: returns both value and size.
- InboundRouter extended to 6 params (+ request_id, source_node_id).

Tests:
- Wire envelope roundtrip with request_id/source_node_id.
- Backward compatibility (fields default to zero).
- Reply routing table: register, complete, remove, concurrent wake.

Closes #165

* Fix unclosed mod tests delimiter from merge conflict

The merge of main into issue-165-remote-await lost the closing brace
for the mod tests block, causing the Node API wrapper functions and
reply table code to be parsed inside the test module. This produced
an 'unclosed delimiter' error on all CI platforms.

Author: slepp

hew-codegen/src/codegen.cpp

@@ -969,28 +969,57 @@ struct ActorAskOpLowering : public mlir::OpConversionPattern<hew::ActorAskOp> {
     auto clonedArgs = deepCopyOwnedArgs(rewriter, loc, module, op.getArgs(), adaptor.getArgs());
     auto [dataPtr, dataSize] = emitPackArgs(rewriter, loc, clonedArgs);
 
-    // Remote actors (i64 PID from Node::lookup): use fire-and-forget send
-    // via hew_actor_send_by_id. Full remote await (request-response) requires
-    // a distributed reply channel protocol — planned for a future release.
+    // Remote actors (i64 PID from Node::lookup): distributed ask via
+    // hew_node_api_ask, which sends the message with a request_id over
+    // the mesh and blocks until the reply arrives.
     if (targetVal.getType() == i64Type) {
-      auto sendFuncType =
-          rewriter.getFunctionType({i64Type, i32Type, ptrType, sizeType}, {i32Type});
-      getOrInsertFuncDecl(module, rewriter, "hew_actor_send_by_id", sendFuncType);
-      mlir::func::CallOp::create(rewriter, loc, "hew_actor_send_by_id", mlir::TypeRange{i32Type},
-                                 mlir::ValueRange{targetVal, msgTypeVal, dataPtr, dataSize});
+      auto askFuncType = rewriter.getFunctionType({i64Type, i32Type, ptrType, sizeType}, {ptrType});
+      getOrInsertFuncDecl(module, rewriter, "hew_node_api_ask", askFuncType);
+      auto call =
+          mlir::func::CallOp::create(rewriter, loc, "hew_node_api_ask", mlir::TypeRange{ptrType},
+                                     mlir::ValueRange{targetVal, msgTypeVal, dataPtr, dataSize});
+      auto replyPtr = call.getResult(0);
 
-      // Return a zero/null result since we can't await remote actors yet.
       auto resultType = op.getResult().getType();
-      mlir::Value zeroResult;
+
+      // Void-return handler: free the reply buffer and erase the op.
+      // free(null) is a safe no-op if the ask failed.
       if (llvm::isa<mlir::NoneType>(resultType)) {
+        auto freeFuncType = rewriter.getFunctionType({ptrType}, {});
+        getOrInsertFuncDecl(module, rewriter, "free", freeFuncType);
+        mlir::func::CallOp::create(rewriter, loc, "free", mlir::TypeRange{},
+                                   mlir::ValueRange{replyPtr});
         rewriter.eraseOp(op);
         return mlir::success();
-      } else if (resultType == ptrType || llvm::isa<mlir::LLVM::LLVMPointerType>(resultType)) {
-        zeroResult = mlir::LLVM::ZeroOp::create(rewriter, loc, ptrType).getResult();
+      }
+
+      // Non-void: load the result from the reply pointer.
+      // hew_node_api_ask guarantees a non-null return for non-void asks
+      // (returns a zeroed allocation on timeout/failure).
+      mlir::Value resultVal;
+      if (resultType == ptrType || llvm::isa<mlir::LLVM::LLVMPointerType>(resultType)) {
+        auto loaded = mlir::LLVM::LoadOp::create(rewriter, loc, ptrType, replyPtr);
+        resultVal = loaded.getResult();
+      } else if (resultType == i32Type || resultType == rewriter.getI64Type() ||
+                 llvm::isa<mlir::IntegerType>(resultType) ||
+                 llvm::isa<mlir::FloatType>(resultType) ||
+                 llvm::isa<mlir::LLVM::LLVMStructType>(resultType)) {
+        auto loaded = mlir::LLVM::LoadOp::create(rewriter, loc, resultType, replyPtr);
+        resultVal = loaded.getResult();
       } else {
-        zeroResult = mlir::arith::ConstantIntOp::create(rewriter, loc, resultType, 0);
+        auto loaded = mlir::LLVM::LoadOp::create(rewriter, loc, ptrType, replyPtr);
+        resultVal = mlir::UnrealizedConversionCastOp::create(rewriter, loc, resultType,
+                                                             mlir::ValueRange{loaded.getResult()})
+                        .getResult(0);
       }
-      rewriter.replaceOp(op, zeroResult);
+
+      // Free the malloc'd reply buffer.
+      auto freeFuncType = rewriter.getFunctionType({ptrType}, {});
+      getOrInsertFuncDecl(module, rewriter, "free", freeFuncType);
+      mlir::func::CallOp::create(rewriter, loc, "free", mlir::TypeRange{},
+                                 mlir::ValueRange{replyPtr});
+
+      rewriter.replaceOp(op, resultVal);
       return mlir::success();
     }
 

hew-runtime/src/actor.rs

@@ -1372,6 +1372,118 @@ pub unsafe extern "C" fn hew_actor_ask_with_channel(
     send_result
 }
 
+/// Perform a blocking ask against an actor identified by PID.
+///
+/// Looks up the actor in `LIVE_ACTORS`, packs a reply channel into the
+/// message, and waits for the reply. Returns the reply pointer and writes
+/// the reply size to `*out_size`.
+///
+/// Returns null if the actor is not found locally or the send fails.
+///
+/// # Safety
+///
+/// - `data` must point to at least `size` readable bytes, or be null when
+///   `size` is 0.
+/// - `out_size` must be a valid, non-null writable pointer.
+#[cfg(not(target_arch = "wasm32"))]
+pub(crate) unsafe fn hew_actor_ask_by_id(
+    actor_id: u64,
+    msg_type: i32,
+    data: *mut c_void,
+    size: usize,
+) -> *mut c_void {
+    let ptr_size = std::mem::size_of::<*mut c_void>();
+    let Some(total) = size.checked_add(ptr_size) else {
+        return std::ptr::null_mut();
+    };
+
+    let ch = reply_channel::hew_reply_channel_new();
+
+    // Pack: [original_data | reply_channel_ptr]
+    // SAFETY: malloc for packed buffer.
+    let packed = unsafe { libc::malloc(total) };
+    if packed.is_null() {
+        // SAFETY: ch was created by hew_reply_channel_new.
+        unsafe { reply_channel::hew_reply_channel_free(ch) };
+        return ptr::null_mut();
+    }
+    // SAFETY: copying data into packed buffer; write_unaligned handles the
+    // potentially-misaligned channel slot.
+    #[expect(
+        clippy::cast_ptr_alignment,
+        reason = "write_unaligned on the next line handles misalignment"
+    )]
+    unsafe {
+        if size > 0 && !data.is_null() {
+            ptr::copy_nonoverlapping(data.cast::<u8>(), packed.cast::<u8>(), size);
+        }
+        let ch_slot = packed.cast::<u8>().add(size).cast::<*mut c_void>();
+        ptr::write_unaligned(ch_slot, ch.cast());
+    }
+
+    // SAFETY: the actor now holds the sender-side reference until it replies.
+    unsafe { reply_channel::hew_reply_channel_retain(ch) };
+
+    // Look up actor and send packed message.
+    let sent = {
+        let guard = match LIVE_ACTORS.lock() {
+            Ok(g) => g,
+            Err(e) => e.into_inner(),
+        };
+        guard.as_ref().is_some_and(|set| {
+            set.iter().any(|actor_ptr| {
+                let actor = actor_ptr.0;
+                if actor.is_null() {
+                    return false;
+                }
+                // SAFETY: LIVE_ACTORS pointers are valid while locked.
+                let matches = unsafe { (*actor).id == actor_id };
+                if matches {
+                    // SAFETY: actor and packed data are valid.
+                    let rc = unsafe { actor_send_result_internal(actor, msg_type, packed, total) };
+                    rc == HewError::Ok as i32
+                } else {
+                    false
+                }
+            })
+        })
+    };
+
+    // SAFETY: packed was malloc'd above.
+    unsafe { libc::free(packed) };
+
+    if !sent {
+        // SAFETY: release the sender-side reference retained for the failed send.
+        unsafe { reply_channel::hew_reply_channel_free(ch) };
+        // SAFETY: ch was created by hew_reply_channel_new.
+        unsafe { reply_channel::hew_reply_channel_free(ch) };
+        return std::ptr::null_mut();
+    }
+
+    let mut reply_size: usize = 0;
+    // SAFETY: ch is valid and single-reader; reply_size is a valid stack pointer.
+    let result = unsafe { reply_channel::hew_reply_wait_with_size(ch, &raw mut reply_size) };
+
+    // Store the reply size in a thread-local so the caller can retrieve it.
+    LAST_REPLY_SIZE.set(reply_size);
+
+    // SAFETY: ch was created by hew_reply_channel_new.
+    unsafe { reply_channel::hew_reply_channel_free(ch) };
+
+    result
+}
+
+// Thread-local storage for the reply size from the last `hew_actor_ask_by_id`.
+std::thread_local! {
+    static LAST_REPLY_SIZE: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
+}
+
+/// Retrieve the size of the reply data from the most recent
+/// `hew_actor_ask_by_id` call on the current thread.
+pub(crate) unsafe fn hew_reply_data_size(_ptr: *mut c_void) -> usize {
+    LAST_REPLY_SIZE.get()
+}
+
 // ── Trap / Error ────────────────────────────────────────────────────────
 
 /// Trap (panic) an actor: store an error code, close the mailbox, and

hew-runtime/src/connection.rs

@@ -194,7 +194,11 @@ struct ReconnectPlan {
 }
 
 /// Inbound message routing callback.
-type InboundRouter = unsafe extern "C" fn(u64, i32, *mut u8, usize);
+///
+/// Parameters: `(target_actor_id, msg_type, data, size, request_id, source_node_id)`.
+/// `request_id` > 0 with `source_node_id` > 0 means this is an ask that expects
+/// a reply. `request_id` == 0 is fire-and-forget.
+type InboundRouter = unsafe extern "C" fn(u64, i32, *mut u8, usize, u64, u16);
 
 // SAFETY: HewConnMgr is only accessed through C ABI functions that
 // serialize access via the internal Mutex. The transport pointer is
@@ -673,6 +677,8 @@ unsafe fn hew_conn_encode_envelope(
         msg_type,
         payload_size: payload_len as u32,
         payload,
+        request_id: 0,
+        source_node_id: 0,
     };
     // SAFETY: zeroed is valid for HewWireBuf.
     let mut wire_buf: HewWireBuf = unsafe { std::mem::zeroed() };
@@ -892,12 +898,30 @@ fn reader_loop(
                 let mut envelope: HewWireEnvelope = std::mem::zeroed();
                 let rc = hew_wire_decode_envelope(&raw mut wire_buf, &raw mut envelope);
                 if rc == 0 {
-                    router_fn(
-                        envelope.target_actor_id,
-                        envelope.msg_type,
-                        envelope.payload,
-                        envelope.payload_size as usize,
-                    );
+                    // Reply envelopes (request_id > 0, source_node_id == 0) are
+                    // deposited directly into the reply routing table, bypassing
+                    // the normal inbound router.
+                    if envelope.request_id > 0 && envelope.source_node_id == 0 {
+                        let reply_payload =
+                            if envelope.payload_size > 0 && !envelope.payload.is_null() {
+                                std::slice::from_raw_parts(
+                                    envelope.payload,
+                                    envelope.payload_size as usize,
+                                )
+                            } else {
+                                &[]
+                            };
+                        crate::hew_node::complete_remote_reply(envelope.request_id, reply_payload);
+                    } else {
+                        router_fn(
+                            envelope.target_actor_id,
+                            envelope.msg_type,
+                            envelope.payload,
+                            envelope.payload_size as usize,
+                            envelope.request_id,
+                            envelope.source_node_id,
+                        );
+                    }
                 }
             }
         }
@@ -1157,6 +1181,7 @@ pub unsafe extern "C" fn hew_connmgr_add(mgr: *mut HewConnMgr, conn_id: c_int) -
         #[cfg(feature = "quic")]
         {
             // QUIC provides TLS 1.3 encryption — skip Noise when using QUIC transport.
+            // SAFETY: mgr.transport is valid while the connection manager is alive.
             unsafe { crate::quic_transport::hew_transport_is_quic(mgr.transport) }
         }
         #[cfg(not(feature = "quic"))]