Phase 3: convert to_binary to async enif_send (Option 3a)

Converts the last worker-bound NIF to the async model. `to_binary`
was the hardest case because it returns a resource binary that
aliases MLX-owned memory via `fine::make_resource_binary` — Spike B
validated that building the resource binary on the worker-allocated
msg_env correctly pins the Tensor resource across `enif_send`, so
no memcpy fallback is needed.

The worker thread now runs `mx::contiguous + mx::eval` and builds
the resource binary directly into msg_env. The binary carries one
resource ref into the caller's heap fragment, keeping the MLX
buffer alive for the lifetime of the binary on the receiver.

Drops the `ERL_NIF_DIRTY_JOB_CPU_BOUND` flag — `to_binary` was the
last dirty-flagged NIF in the project. With Phase 3 merged, no
NIF blocks any BEAM scheduler (regular or dirty) on MLX work.

Full suite: 425 tests, 0 failures.
Full heavy suite (including qwen3_full, vit_full, whisper_full,
distilbert_full, training_full, qwen3_quant_full, fast_kernels_full,
conformance): 453 tests, 0 failures.

Author: ausimian

c_src/emily_nif.cpp

@@ -69,29 +69,40 @@ fine::ResourcePtr<Tensor> from_binary(
 }
 FINE_NIF(from_binary, 0);
 
-// to_binary/2 — materialize the array and return its bytes as a BEAM
-// resource binary aliasing MLX storage (no memcpy). The binary pins a
-// Tensor resource → mx::array → MLX buffer; the buffer survives until
-// the BEAM binary is GC'd.
-fine::Term to_binary(ErlNifEnv *env, fine::ResourcePtr<WorkerThread> w, fine::ResourcePtr<Tensor> tensor) {
-  auto materialized = w->run_sync([&](mx::Stream &s) {
-    auto c = mx::contiguous(tensor->array, false, s);
-    mx::eval(c);
-    return c;
-  });
-
-  if (!materialized.flags().row_contiguous) {
-    throw std::runtime_error(
-        "to_binary: array is not row-contiguous after mx::contiguous");
-  }
+// to_binary_nif/2 — materialize the array on the worker thread and
+// return its bytes as a BEAM resource binary aliasing MLX storage
+// (no memcpy). Async: the NIF returns a ref immediately; the worker
+// posts `{ref, {:ok, bin}}` back to the caller once `mx::contiguous +
+// mx::eval` completes. The Elixir wrapper `Emily.Native.to_binary/2`
+// awaits via `Emily.Native.Async.call/1`.
+//
+// Pinning: `fine::make_resource_binary` called on the worker-allocated
+// msg_env bumps the Tensor resource's refcount; the binary carries
+// that ref into the receiving process, where it stays alive until
+// the binary is GC'd. Validated by Spike B.
+fine::Term to_binary_nif(ErlNifEnv *env,
+                         fine::ResourcePtr<WorkerThread> w,
+                         fine::ResourcePtr<Tensor> tensor) {
+  return emily::async_reply(
+      env, w,
+      [tensor = std::move(tensor)](mx::Stream &s, ErlNifEnv *msg_env) {
+        auto materialized = mx::contiguous(tensor->array, false, s);
+        mx::eval(materialized);
+
+        if (!materialized.flags().row_contiguous) {
+          throw std::runtime_error(
+              "to_binary: array is not row-contiguous after mx::contiguous");
+        }
 
-  auto nbytes = materialized.nbytes();
-  auto data = reinterpret_cast<const char *>(materialized.data<void>());
+        auto nbytes = materialized.nbytes();
+        auto data = reinterpret_cast<const char *>(materialized.data<void>());
 
-  auto pin = wrap(std::move(materialized));
-  return fine::make_resource_binary(env, std::move(pin), data, nbytes);
+        auto pin = wrap(std::move(materialized));
+        return fine::Term(
+            fine::make_resource_binary(msg_env, std::move(pin), data, nbytes));
+      });
 }
-FINE_NIF(to_binary, ERL_NIF_DIRTY_JOB_CPU_BOUND);
+FINE_NIF(to_binary_nif, 0);
 
 // shape/1 — return the array's shape as a list of ints.
 std::vector<int64_t> shape(ErlNifEnv *, fine::ResourcePtr<Tensor> tensor) {

lib/emily/native.ex

@@ -40,8 +40,12 @@ defmodule Emily.Native do
   @spec from_binary(binary(), [non_neg_integer()], dtype()) :: tensor()
   def from_binary(_data, _shape, _dtype), do: nif()
 
+  @doc false
+  @spec to_binary_nif(worker(), tensor()) :: reference()
+  def to_binary_nif(_w, _tensor), do: nif()
+
   @spec to_binary(worker(), tensor()) :: binary()
-  def to_binary(_w, _tensor), do: nif()
+  def to_binary(w, tensor), do: Async.call(to_binary_nif(w, tensor))
 
   @spec shape(tensor()) :: [non_neg_integer()]
   def shape(_tensor), do: nif()