test(support): Introduce threaded Tox runner and simulation barrier

- Add `MpscQueue` for thread-safe task scheduling.
- Add `ToxRunner` to execute Tox instances in dedicated threads.
- Update `Simulation` to coordinate time steps across multiple runners using a synchronization barrier.
- Refactor `FakeMemory` and `FakeClock` to be thread-safe.
- Update `tox_network` helpers and tests to utilize the threaded runner infrastructure.

other/docker/modules/check

@@ -20,6 +20,7 @@ STD_MODULE = """module std [system] {
   textual header "/usr/include/c++/14.2.0/chrono"
   textual header "/usr/include/c++/14.2.0/climits"
   textual header "/usr/include/c++/14.2.0/compare"
+  textual header "/usr/include/c++/14.2.0/condition_variable"
   textual header "/usr/include/c++/14.2.0/cstddef"
   textual header "/usr/include/c++/14.2.0/cstdint"
   textual header "/usr/include/c++/14.2.0/cstdio"
@@ -27,6 +28,7 @@ STD_MODULE = """module std [system] {
   textual header "/usr/include/c++/14.2.0/cstring"
   textual header "/usr/include/c++/14.2.0/deque"
   textual header "/usr/include/c++/14.2.0/functional"
+  textual header "/usr/include/c++/14.2.0/future"
   textual header "/usr/include/c++/14.2.0/iomanip"
   textual header "/usr/include/c++/14.2.0/iosfwd"
   textual header "/usr/include/c++/14.2.0/iostream"

testing/support/BUILD.bazel

@@ -18,6 +18,7 @@ cc_library(
         "src/simulated_environment.cc",
         "src/simulation.cc",
         "src/tox_network.cc",
+        "src/tox_runner.cc",
     ],
     hdrs = [
         "doubles/fake_clock.hh",
@@ -31,11 +32,13 @@ cc_library(
         "public/fuzz_data.hh",
         "public/fuzz_helpers.hh",
         "public/memory.hh",
+        "public/mpsc_queue.hh",
         "public/network.hh",
         "public/random.hh",
         "public/simulated_environment.hh",
         "public/simulation.hh",
         "public/tox_network.hh",
+        "public/tox_runner.hh",
     ],
     copts = select({
         "//tools/config:windows": ["/wd4200"],  # Zero-sized array in struct/union
@@ -46,6 +49,7 @@ cc_library(
         "//c-toxcore/toxcore:mem",
         "//c-toxcore/toxcore:network",
         "//c-toxcore/toxcore:tox",
+        "//c-toxcore/toxcore:tox_events",
         "//c-toxcore/toxcore:tox_memory",
         "//c-toxcore/toxcore:tox_options",
         "//c-toxcore/toxcore:tox_random",

testing/support/CMakeLists.txt

@@ -18,6 +18,7 @@ set(support_SOURCES
   src/simulated_environment.cc
   src/simulation.cc
   src/tox_network.cc
+  src/tox_runner.cc
   doubles/fake_clock.hh
   doubles/fake_memory.hh
   doubles/fake_network_stack.hh
@@ -29,11 +30,13 @@ set(support_SOURCES
   public/fuzz_data.hh
   public/fuzz_helpers.hh
   public/memory.hh
+  public/mpsc_queue.hh
   public/network.hh
   public/random.hh
   public/simulated_environment.hh
   public/simulation.hh
   public/tox_network.hh
+  public/tox_runner.hh
 )
 
 add_library(support STATIC ${support_SOURCES})

testing/support/doubles/fake_clock.hh

@@ -1,6 +1,8 @@
 #ifndef C_TOXCORE_TESTING_SUPPORT_DOUBLES_FAKE_CLOCK_H
 #define C_TOXCORE_TESTING_SUPPORT_DOUBLES_FAKE_CLOCK_H
 
+#include <atomic>
+
 #include "../public/clock.hh"
 
 namespace tox::test {
@@ -15,7 +17,7 @@ public:
     void advance(uint64_t ms);
 
 private:
-    uint64_t now_ms_;
+    std::atomic<uint64_t> now_ms_;
 };
 
 }  // namespace tox::test

testing/support/doubles/fake_memory.hh

@@ -1,6 +1,7 @@
 #ifndef C_TOXCORE_TESTING_SUPPORT_DOUBLES_FAKE_MEMORY_H
 #define C_TOXCORE_TESTING_SUPPORT_DOUBLES_FAKE_MEMORY_H
 
+#include <atomic>
 #include <functional>
 
 #include "../public/memory.hh"
@@ -31,7 +32,13 @@ public:
     // Get the C-compatible struct
     struct Tox_Memory get_c_memory();
 
+    size_t current_allocation() const;
+    size_t max_allocation() const;
+
 private:
+    void on_allocation(size_t size);
+    void on_deallocation(size_t size);
+
     struct Header {
         size_t size;
         size_t magic;
@@ -39,8 +46,8 @@ private:
     static constexpr size_t kMagic = 0xDEADC0DE;
     static constexpr size_t kFreeMagic = 0xBAADF00D;
 
-    size_t current_allocation_ = 0;
-    size_t max_allocation_ = 0;
+    std::atomic<size_t> current_allocation_{0};
+    std::atomic<size_t> max_allocation_{0};
 
     FailureInjector failure_injector_;
     Observer observer_;

testing/support/public/mpsc_queue.hh

@@ -0,0 +1,83 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later
+ * Copyright © 2026 The TokTok team.
+ */
+#ifndef C_TOXCORE_TESTING_SUPPORT_MPSC_QUEUE_H
+#define C_TOXCORE_TESTING_SUPPORT_MPSC_QUEUE_H
+
+#include <condition_variable>
+#include <deque>
+#include <mutex>
+
+namespace tox::test {
+
+/**
+ * @brief Multiple Producer, Single Consumer Queue.
+ *
+ * This queue implementation provides thread-safe access for multiple producers
+ * pushing items and a single consumer popping items. It uses a `std::mutex`
+ * and `std::condition_variable` for synchronization.
+ *
+ * @tparam T The type of elements stored in the queue.
+ */
+template <typename T>
+class MpscQueue {
+public:
+    MpscQueue() = default;
+    ~MpscQueue() = default;
+
+    // Disable copy/move to prevent accidental sharing/slicing issues
+    MpscQueue(const MpscQueue &) = delete;
+    MpscQueue &operator=(const MpscQueue &) = delete;
+
+    /**
+     * @brief Pushes a value onto the queue.
+     * Thread-safe (Multiple Producers).
+     */
+    void push(T value)
+    {
+        {
+            std::lock_guard<std::mutex> lock(mutex_);
+            queue_.push_back(std::move(value));
+        }
+        cv_.notify_one();
+    }
+
+    /**
+     * @brief Pops a value from the queue, blocking if empty.
+     * Thread-safe (Single Consumer).
+     */
+    T pop()
+    {
+        std::unique_lock<std::mutex> lock(mutex_);
+        cv_.wait(lock, [this] { return !queue_.empty(); });
+        T value = std::move(queue_.front());
+        queue_.pop_front();
+        return value;
+    }
+
+    /**
+     * @brief Tries to pop a value from the queue without blocking.
+     * Thread-safe (Single Consumer).
+     *
+     * @param out Reference to store the popped value.
+     * @return true if a value was popped, false if the queue was empty.
+     */
+    bool try_pop(T &out)
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        if (queue_.empty())
+            return false;
+        out = std::move(queue_.front());
+        queue_.pop_front();
+        return true;
+    }
+
+private:
+    std::deque<T> queue_;
+    std::mutex mutex_;
+    std::condition_variable cv_;
+};
+
+}  // namespace tox::test
+
+#endif  // C_TOXCORE_TESTING_SUPPORT_MPSC_QUEUE_H

testing/support/public/simulation.hh

@@ -1,6 +1,7 @@
 #ifndef C_TOXCORE_TESTING_SUPPORT_SIMULATION_H
 #define C_TOXCORE_TESTING_SUPPORT_SIMULATION_H
 
+#include <condition_variable>
 #include <functional>
 #include <memory>
 #include <vector>
@@ -42,6 +43,57 @@ public:
     void advance_time(uint64_t ms);
     void run_until(std::function<bool()> condition, uint64_t timeout_ms = 5000);
 
+    // Synchronization Barrier
+    // These methods coordinate the lock-step execution of multiple Tox runners.
+
+    /**
+     * @brief Registers a new runner with the simulation barrier.
+     * @return The current generation ID of the simulation.
+     */
+    uint64_t register_runner();
+
+    /**
+     * @brief Unregisters a runner from the simulation barrier.
+     *
+     * This ensures the simulation does not block waiting for a terminated runner.
+     */
+    void unregister_runner();
+
+    using TickListenerId = int;
+
+    /**
+     * @brief Registers a callback to be invoked when a new simulation tick starts.
+     *
+     * @param listener The function to call with the new generation ID.
+     * @return An ID handle for unregistering the listener.
+     */
+    TickListenerId register_tick_listener(std::function<void(uint64_t)> listener);
+
+    /**
+     * @brief Unregisters a tick listener.
+     */
+    void unregister_tick_listener(TickListenerId id);
+
+    /**
+     * @brief Blocks until the simulation advances to the next tick.
+     *
+     * Called by runner threads to wait for the global clock to advance.
+     *
+     * @param last_gen The generation ID of the last processed tick.
+     * @param stop_token Atomic flag to signal termination while waiting.
+     * @param timeout_ms Maximum time to wait for the tick.
+     * @return The new generation ID, or `last_gen` on timeout/stop.
+     */
+    uint64_t wait_for_tick(
+        uint64_t last_gen, const std::atomic<bool> &stop_token, uint64_t timeout_ms = 10);
+
+    /**
+     * @brief Signals that a runner has completed its work for the current tick.
+     *
+     * @param next_delay_ms The requested delay until the next tick (from `tox_iteration_interval`).
+     */
+    void tick_complete(uint32_t next_delay_ms = 50);
+
     // Global Access
     FakeClock &clock() { return *clock_; }
     NetworkUniverse &net() { return *net_; }
@@ -53,6 +105,21 @@ private:
     std::unique_ptr<FakeClock> clock_;
     std::unique_ptr<NetworkUniverse> net_;
     uint32_t node_count_ = 0;
+
+    // Barrier State
+    std::mutex barrier_mutex_;
+    std::condition_variable barrier_cv_;
+    uint64_t current_generation_ = 0;
+    int registered_runners_ = 0;
+    std::atomic<int> active_runners_{0};
+    std::atomic<uint32_t> next_step_min_{50};
+
+    struct TickListener {
+        TickListenerId id;
+        std::function<void(uint64_t)> callback;
+    };
+    std::vector<TickListener> tick_listeners_;
+    TickListenerId next_listener_id_ = 0;
 };
 
 /**
@@ -90,6 +157,8 @@ public:
     struct Tox_Random get_c_random() { return random_->get_c_random(); }
     struct Tox_Memory get_c_memory() { return memory_->get_c_memory(); }
 
+    Simulation &simulation() { return sim_; }
+
     // For fuzzing compatibility (exposes first bound UDP socket as "endpoint")
     FakeUdpSocket *get_primary_socket();
 

testing/support/public/tox_network.hh

@@ -8,18 +8,19 @@
 #include <vector>
 
 #include "simulation.hh"
+#include "tox_runner.hh"
 
 namespace tox::test {
 
 struct ConnectedFriend {
     std::unique_ptr<SimulatedNode> node;
-    SimulatedNode::ToxPtr tox;
+    std::unique_ptr<ToxRunner> runner;
     uint32_t friend_number;
 
-    ConnectedFriend(std::unique_ptr<SimulatedNode> node_in, SimulatedNode::ToxPtr tox_in,
+    ConnectedFriend(std::unique_ptr<SimulatedNode> node_in, std::unique_ptr<ToxRunner> runner_in,
         uint32_t friend_number_in)
         : node(std::move(node_in))
-        , tox(std::move(tox_in))
+        , runner(std::move(runner_in))
         , friend_number(friend_number_in)
     {
     }
@@ -44,7 +45,8 @@ struct ConnectedFriend {
  * @return A vector of ConnectedFriend structures, each representing a friend.
  */
 std::vector<ConnectedFriend> setup_connected_friends(Simulation &sim, Tox *main_tox,
-    SimulatedNode &main_node, int num_friends, const Tox_Options *options = nullptr);
+    SimulatedNode &main_node, int num_friends, const Tox_Options *options = nullptr,
+    bool verbose = false);
 
 /**
  * @brief Connects two existing Tox instances as friends.

testing/support/public/tox_runner.hh

@@ -0,0 +1,116 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later
+ * Copyright © 2026 The TokTok team.
+ */
+
+#ifndef C_TOXCORE_TESTING_SUPPORT_TOX_RUNNER_H
+#define C_TOXCORE_TESTING_SUPPORT_TOX_RUNNER_H
+
+#include <atomic>
+#include <functional>
+#include <future>
+#include <thread>
+#include <type_traits>
+#include <vector>
+
+#include "../../../toxcore/tox_events.h"
+#include "mpsc_queue.hh"
+#include "simulation.hh"
+
+namespace tox::test {
+
+class ToxRunner {
+public:
+    explicit ToxRunner(SimulatedNode &node, const Tox_Options *options = nullptr);
+    ~ToxRunner();
+
+    ToxRunner(const ToxRunner &) = delete;
+    ToxRunner &operator=(const ToxRunner &) = delete;
+
+    struct ToxEventsDeleter {
+        void operator()(Tox_Events *e) const { tox_events_free(e); }
+    };
+    using ToxEventsPtr = std::unique_ptr<Tox_Events, ToxEventsDeleter>;
+
+    /**
+     * @brief Schedules a task for execution on the runner's thread.
+     *
+     * This method is thread-safe and non-blocking. The task is queued and will
+     * be executed during the runner's event loop cycle.
+     *
+     * @param task The function to execute, taking a raw Tox pointer.
+     */
+    void execute(std::function<void(Tox *)> task);
+
+    /**
+     * @brief Executes a task on the runner's thread and waits for the result.
+     *
+     * This method blocks the calling thread until the task has been executed
+     * by the runner. It automatically handles return value propagation and
+     * exception safety (though exceptions are not currently propagated).
+     *
+     * @tparam Func The type of the callable object.
+     * @param func The callable to execute, taking a raw Tox pointer.
+     * @return The result of the callable execution.
+     */
+    template <typename Func>
+    auto invoke(Func &&func) -> std::invoke_result_t<Func, Tox *>
+    {
+        using R = std::invoke_result_t<Func, Tox *>;
+        auto promise = std::make_shared<std::promise<R>>();
+        auto future = promise->get_future();
+
+        execute([p = promise, f = std::forward<Func>(func)](Tox *tox) {
+            if constexpr (std::is_void_v<R>) {
+                f(tox);
+                p->set_value();
+            } else {
+                p->set_value(f(tox));
+            }
+        });
+
+        return future.get();
+    }
+
+    /**
+     * @brief Retrieves all accumulated Tox event batches.
+     *
+     * Returns a vector of unique pointers to Tox_Events structures that have
+     * been collected by the runner since the last call. Ownership is transferred
+     * to the caller. This method is thread-safe.
+     *
+     * @return A vector of Tox_Events pointers.
+     */
+    std::vector<ToxEventsPtr> poll_events();
+
+    /**
+     * @brief Accesses the underlying Tox instance directly.
+     *
+     * @warning Thread-Safety Violation: This method provides unsafe access to the
+     * Tox instance. It should ONLY be used when the runner thread is known to be
+     * idle (e.g., before the loop starts) or for accessing constant/read-only properties.
+     * For all other operations, use `execute` or `invoke`.
+     */
+    Tox *unsafe_tox() { return tox_.get(); }
+
+private:
+    void loop();
+
+    SimulatedNode::ToxPtr tox_;
+    std::thread thread_;
+
+    struct Message {
+        enum Type { Task, Tick, Stop } type;
+        std::function<void(Tox *)> task;
+        uint64_t generation = 0;
+    };
+
+    MpscQueue<Message> queue_;
+    MpscQueue<ToxEventsPtr> events_queue_;
+
+    Simulation::TickListenerId tick_listener_id_ = -1;
+    SimulatedNode &node_;
+};
+
+}  // namespace tox::test
+
+#endif  // C_TOXCORE_TESTING_SUPPORT_TOX_RUNNER_H

testing/support/src/fake_memory.cc

@@ -45,15 +45,9 @@ void *FakeMemory::malloc(size_t size)
     header->size = size;
     header->magic = kMagic;
 
-    current_allocation_ += size;
-    if (current_allocation_ > max_allocation_) {
-        max_allocation_ = current_allocation_;
-    }
+    on_allocation(size);
 
-    void *res = header + 1;
-    // std::cerr << "[FakeMemory] malloc(" << size << ") -> " << res << " (header=" << header << ")"
-    // << std::endl;
-    return res;
+    return header + 1;
 }
 
 void *FakeMemory::realloc(void *ptr, size_t size)
@@ -82,7 +76,6 @@ void *FakeMemory::realloc(void *ptr, size_t size)
     }
 
     if (fail) {
-        // If realloc fails, original block is left untouched.
         return nullptr;
     }
 
@@ -92,19 +85,14 @@ void *FakeMemory::realloc(void *ptr, size_t size)
         return nullptr;
     }
 
-    Header *header = static_cast<Header *>(new_ptr);
-    current_allocation_ -= old_size;
-    current_allocation_ += size;
-    if (current_allocation_ > max_allocation_) {
-        max_allocation_ = current_allocation_;
-    }
+    on_deallocation(old_size);
+    on_allocation(size);
 
+    Header *header = static_cast<Header *>(new_ptr);
     header->size = size;
     header->magic = kMagic;
-    void *res = header + 1;
-    // std::cerr << "[FakeMemory] realloc(" << ptr << ", " << size << ") -> " << res << " (header="
-    // << header << ")" << std::endl;
-    return res;
+
+    return header + 1;
 }
 
 void FakeMemory::free(void *ptr)
@@ -127,7 +115,7 @@ void FakeMemory::free(void *ptr)
     }
 
     size_t size = header->size;
-    current_allocation_ -= size;
+    on_deallocation(size);
     header->magic = kFreeMagic;  // Mark as free
     std::free(header);
 }
@@ -141,4 +129,17 @@ void FakeMemory::set_observer(Observer observer) { observer_ = std::move(observe
 
 struct Tox_Memory FakeMemory::get_c_memory() { return Tox_Memory{&kFakeMemoryVtable, this}; }
 
+size_t FakeMemory::current_allocation() const { return current_allocation_.load(); }
+
+size_t FakeMemory::max_allocation() const { return max_allocation_.load(); }
+
+void FakeMemory::on_allocation(size_t size)
+{
+    size_t current = current_allocation_.fetch_add(size) + size;
+    size_t max = max_allocation_.load(std::memory_order_relaxed);
+    while (current > max && !max_allocation_.compare_exchange_weak(max, current)) { }
+}
+
+void FakeMemory::on_deallocation(size_t size) { current_allocation_.fetch_sub(size); }
+
 }  // namespace tox::test

testing/support/src/simulation.cc

@@ -1,7 +1,9 @@
 #include "../public/simulation.hh"
 
 #include <cassert>
+#include <chrono>
 #include <iostream>
+#include <thread>
 
 namespace tox::test {
 
@@ -24,11 +26,123 @@ void Simulation::advance_time(uint64_t ms)
 void Simulation::run_until(std::function<bool()> condition, uint64_t timeout_ms)
 {
     uint64_t start_time = clock_->current_time_ms();
-    while (!condition()) {
+
+    // Initial check
+    if (condition())
+        return;
+
+    while (true) {
         if (clock_->current_time_ms() - start_time > timeout_ms) {
             break;
         }
-        advance_time(10);  // 10ms ticks
+
+        // 1. Advance Global Time
+        // Determine the time step based on the minimum requested delay from all runners
+        // during the previous tick. We default to 50ms if no specific request was made.
+        // The `exchange` operation resets the minimum accumulator for the current tick.
+        uint32_t step = next_step_min_.exchange(50);
+        advance_time(step);
+
+        // 2. Start Barrier (Signal Runners)
+        // Notify all registered runners that time has advanced and they should proceed
+        // with their next iteration.
+        {
+            std::lock_guard<std::mutex> lock(barrier_mutex_);
+            current_generation_++;
+            // Initialize the countdown of active runners for this tick.
+            active_runners_.store(registered_runners_);
+
+            for (const auto &l : tick_listeners_) {
+                l.callback(current_generation_);
+            }
+        }
+        barrier_cv_.notify_all();
+
+        // 3. End Barrier (Wait for Completion)
+        // Block until all active runners have reported completion via `tick_complete()`.
+        {
+            std::unique_lock<std::mutex> lock(barrier_mutex_);
+            // We use a lambda predicate to handle spurious wakeups.
+            // The wait finishes when `active_runners_` reaches zero.
+            barrier_cv_.wait(lock, [this] { return active_runners_.load() == 0; });
+        }
+
+        // 4. Check condition
+        if (condition())
+            return;
+    }
+}
+
+Simulation::TickListenerId Simulation::register_tick_listener(
+    std::function<void(uint64_t)> listener)
+{
+    std::lock_guard<std::mutex> lock(barrier_mutex_);
+    TickListenerId id = next_listener_id_++;
+    tick_listeners_.push_back({id, std::move(listener)});
+    return id;
+}
+
+void Simulation::unregister_tick_listener(TickListenerId id)
+{
+    std::lock_guard<std::mutex> lock(barrier_mutex_);
+    for (auto it = tick_listeners_.begin(); it != tick_listeners_.end(); ++it) {
+        if (it->id == id) {
+            tick_listeners_.erase(it);
+            break;
+        }
+    }
+}
+
+uint64_t Simulation::register_runner()
+{
+    std::lock_guard<std::mutex> lock(barrier_mutex_);
+    registered_runners_++;
+    return current_generation_;
+}
+
+void Simulation::unregister_runner()
+{
+    std::lock_guard<std::mutex> lock(barrier_mutex_);
+    registered_runners_--;
+
+    // If we are currently running a tick (active_runners > 0), we need to decrement it
+    // because this runner will not be calling tick_complete()
+    if (active_runners_.load() > 0) {
+        if (active_runners_.fetch_sub(1) == 1) {
+            barrier_cv_.notify_all();
+        }
+    }
+}
+
+uint64_t Simulation::wait_for_tick(
+    uint64_t last_gen, const std::atomic<bool> &stop_token, uint64_t timeout_ms)
+{
+    std::unique_lock<std::mutex> lock(barrier_mutex_);
+    // Wait until generation increases (new tick started) OR we are stopped OR timeout
+    bool result = barrier_cv_.wait_for(lock, std::chrono::milliseconds(timeout_ms),
+        [&] { return current_generation_ > last_gen || stop_token; });
+
+    if (stop_token)
+        return last_gen;
+    if (!result)
+        return last_gen;  // Timeout
+    return current_generation_;
+}
+
+void Simulation::tick_complete(uint32_t next_delay_ms)
+{
+    // Atomic min reduction
+    uint32_t current = next_step_min_.load(std::memory_order_relaxed);
+    while (
+        next_delay_ms < current && !next_step_min_.compare_exchange_weak(current, next_delay_ms)) {
+        // If exchange failed, current was updated to actual value, so loop checks again
+    }
+
+    // We don't need the mutex to decrement the atomic
+    if (active_runners_.fetch_sub(1) == 1) {
+        // Last runner to finish: notify main thread
+        std::lock_guard<std::mutex> lock(barrier_mutex_);
+        barrier_cv_.notify_all();
     }
 }
 

testing/support/src/tox_network.cc

@@ -5,17 +5,21 @@
 #include "../public/tox_network.hh"
 
 #include <cstring>
+#include <future>
 #include <iostream>
+#include <vector>
 
 #include "../../../toxcore/network.h"
 #include "../../../toxcore/tox.h"
+#include "../../../toxcore/tox_events.h"
+#include "../public/tox_runner.hh"
 
 namespace tox::test {
 
 ConnectedFriend::~ConnectedFriend() = default;
 
 std::vector<ConnectedFriend> setup_connected_friends(Simulation &sim, Tox *main_tox,
-    SimulatedNode &main_node, int num_friends, const Tox_Options *options)
+    SimulatedNode &main_node, int num_friends, const Tox_Options *options, bool verbose)
 {
     std::vector<ConnectedFriend> friends;
     friends.reserve(num_friends);
@@ -42,32 +46,30 @@ std::vector<ConnectedFriend> setup_connected_friends(Simulation &sim, Tox *main_
 
     for (int i = 0; i < num_friends; ++i) {
         auto node = sim.create_node();
-        auto tox = node->create_tox(options);
-        if (!tox) {
-            return {};
-        }
+        auto runner = std::make_unique<ToxRunner>(*node, options);
 
         uint8_t friend_pk[TOX_PUBLIC_KEY_SIZE];
-        tox_self_get_public_key(tox.get(), friend_pk);
+        runner->invoke([&](Tox *tox) { tox_self_get_public_key(tox, friend_pk); });
 
         Tox_Err_Friend_Add err;
         uint32_t fn = tox_friend_add_norequest(main_tox, friend_pk, &err);
         if (fn == UINT32_MAX || err != TOX_ERR_FRIEND_ADD_OK) {
             return {};
         }
-        if (tox_friend_add_norequest(tox.get(), main_pk, &err) == UINT32_MAX
-            || err != TOX_ERR_FRIEND_ADD_OK) {
-            return {};
-        }
 
-        // Bootstrap to the main node AND the PREVIOUS node in the chain
-        tox_bootstrap(tox.get(), main_ip_str, main_port, main_dht_id, nullptr);
-        if (i > 0) {
-            tox_bootstrap(tox.get(), prev_ip_str, prev_port, prev_dht_id, nullptr);
-        }
+        // Execute add friend and bootstrap on runner
+        runner->execute([=](Tox *tox) {
+            tox_friend_add_norequest(tox, main_pk, nullptr);
+            tox_bootstrap(tox, main_ip_str, main_port, main_dht_id, nullptr);
+            if (i > 0) {
+                tox_bootstrap(tox, prev_ip_str, prev_port, prev_dht_id, nullptr);
+            }
+        });
+
+        // Retrieve previous node's DHT ID and update IP for the next iteration.
+        // We use invoke to safely fetch data from the runner thread.
+        runner->invoke([&](Tox *tox) { tox_self_get_dht_id(tox, prev_dht_id); });
 
-        // Update prev for next node
-        tox_self_get_dht_id(tox.get(), prev_dht_id);
         ip_parse_addr(&node->ip, prev_ip_str, sizeof(prev_ip_str));
 
         FakeUdpSocket *node_socket = node->get_primary_socket();
@@ -76,67 +78,72 @@ std::vector<ConnectedFriend> setup_connected_friends(Simulation &sim, Tox *main_
         }
         prev_port = node_socket->local_port();
 
-        friends.push_back({std::move(node), std::move(tox), fn});
+        friends.push_back({std::move(node), std::move(runner), fn});
 
-        // Run simulation to let DHT stabilize
-        sim.run_until(
-            [&]() {
-                tox_iterate(main_tox, nullptr);
-                for (auto &f : friends) {
-                    tox_iterate(f.tox.get(), nullptr);
-                }
-                return false;
-            },
-            200);
+        // Run the simulation periodically to allow the DHT to stabilize incrementally
+        // as we add nodes, rather than waiting until the end.
+        if (friends.size() % 10 == 0) {
+            sim.run_until([&]() { return false; }, 20);
+        }
     }
 
-    // Optional: Bootstrap main_tox to the last node to complete the circle
     if (!friends.empty()) {
         tox_bootstrap(main_tox, prev_ip_str, prev_port, prev_dht_id, nullptr);
     }
 
-    // Run simulation until all are connected
+    std::vector<bool> friends_connected(friends.size(), false);
+
     sim.run_until(
         [&]() {
-            bool all_connected = true;
-            int connected_count = 0;
             tox_iterate(main_tox, nullptr);
-            for (auto &f : friends) {
-                tox_iterate(f.tox.get(), nullptr);
-                if (tox_friend_get_connection_status(main_tox, f.friend_number, nullptr)
-                        != TOX_CONNECTION_NONE
-                    && tox_friend_get_connection_status(f.tox.get(), 0, nullptr)
-                        != TOX_CONNECTION_NONE) {
-                    connected_count++;
-                } else {
-                    all_connected = false;
-                }
-            }
-            static uint64_t last_print = 0;
-            if (sim.clock().current_time_ms() - last_print > 1000) {
-                std::cerr << "[setup_connected_friends] Friends connected: " << connected_count
-                          << "/" << friends.size() << " (time: " << sim.clock().current_time_ms()
-                          << "ms)" << std::endl;
 
-                if (connected_count < static_cast<int>(friends.size())
-                    && sim.clock().current_time_ms() > 10000) {
-                    for (size_t i = 0; i < friends.size(); ++i) {
-                        auto s1 = tox_friend_get_connection_status(
-                            main_tox, friends[i].friend_number, nullptr);
-                        auto s2
-                            = tox_friend_get_connection_status(friends[i].tox.get(), 0, nullptr);
-                        if (s1 == TOX_CONNECTION_NONE || s2 == TOX_CONNECTION_NONE) {
-                            std::cerr << "  Friend " << i << " not connected (Main->F: " << s1
-                                      << ", F->Main: " << s2 << ")" << std::endl;
+            // Check connection status
+            int connected_count = 0;
+
+            for (size_t i = 0; i < friends.size(); ++i) {
+                // Check if main sees friend
+                bool main_sees_friend
+                    = tox_friend_get_connection_status(main_tox, friends[i].friend_number, nullptr)
+                    != TOX_CONNECTION_NONE;
+
+                // Check if friend sees main by polling events from the runner
+                auto batches = friends[i].runner->poll_events();
+                for (const auto &batch : batches) {
+                    size_t size = tox_events_get_size(batch.get());
+                    for (size_t k = 0; k < size; ++k) {
+                        const Tox_Event *e = tox_events_get(batch.get(), k);
+                        if (tox_event_get_type(e) == TOX_EVENT_FRIEND_CONNECTION_STATUS) {
+                            auto *ev = tox_event_get_friend_connection_status(e);
+                            if (tox_event_friend_connection_status_get_connection_status(ev)
+                                != TOX_CONNECTION_NONE) {
+                                friends_connected[i] = true;
+                            } else {
+                                friends_connected[i] = false;
+                            }
                         }
                     }
                 }
 
+                if (main_sees_friend && friends_connected[i]) {
+                    connected_count++;
+                }
+            }
+
+            if (connected_count == static_cast<int>(friends.size())) {
+                return true;
+            }
+
+            static uint64_t last_print = 0;
+            if (verbose && sim.clock().current_time_ms() - last_print > 1000) {
+                std::cerr << "[setup_connected_friends] Friends connected: " << connected_count
+                          << "/" << friends.size() << " (time: " << sim.clock().current_time_ms()
+                          << "ms)" << std::endl;
                 last_print = sim.clock().current_time_ms();
             }
-            return all_connected;
+
+            return false;
         },
-        300000);  // 5 minutes simulation time for 100 nodes to converge
+        300000);
 
     return friends;
 }
@@ -144,6 +151,10 @@ std::vector<ConnectedFriend> setup_connected_friends(Simulation &sim, Tox *main_
 bool connect_friends(
     Simulation &sim, SimulatedNode &node1, Tox *tox1, SimulatedNode &node2, Tox *tox2)
 {
+    // This helper function assumes the Tox instances are running in the current thread
+    // (e.g., standard unit test) or that the caller is handling thread safety if they
+    // are part of a runner. It uses direct tox_iterate calls.
+
     uint8_t pk1[TOX_PUBLIC_KEY_SIZE];
     uint8_t pk2[TOX_PUBLIC_KEY_SIZE];
     tox_self_get_public_key(tox1, pk1);
@@ -217,52 +228,26 @@ uint32_t setup_connected_group(
         &err_new);
 
     if (main_state.group_number == UINT32_MAX || err_new != TOX_ERR_GROUP_NEW_OK) {
-        std::cerr << "tox_group_new failed with error: " << err_new << std::endl;
         return UINT32_MAX;
     }
 
-    std::vector<std::unique_ptr<NodeGroupState>> friend_states;
-    friend_states.reserve(friends.size());
+    // Friend states tracked via events
+    std::vector<NodeGroupState> friend_states(friends.size());
 
-    for (size_t i = 0; i < friends.size(); ++i) {
-        auto state = std::make_unique<NodeGroupState>();
-        tox_callback_group_peer_join(
-            friends[i].tox.get(), [](Tox *, uint32_t, uint32_t, void *user_data) {
-                static_cast<NodeGroupState *>(user_data)->peer_count++;
-            });
+    // Main tox sends invites; friends accept via events polled from their runners.
 
-        tox_callback_group_invite(friends[i].tox.get(),
-            [](Tox *tox, uint32_t friend_number, const uint8_t *invite_data,
-                size_t invite_data_length, const uint8_t *, size_t, void *user_data) {
-                NodeGroupState *ng_state = static_cast<NodeGroupState *>(user_data);
-                Tox_Err_Group_Invite_Accept err_accept;
-                ng_state->group_number
-                    = tox_group_invite_accept(tox, friend_number, invite_data, invite_data_length,
-                        reinterpret_cast<const uint8_t *>("peer"), 4, nullptr, 0, &err_accept);
-                if (ng_state->group_number == UINT32_MAX
-                    || err_accept != TOX_ERR_GROUP_INVITE_ACCEPT_OK) {
-                    ng_state->group_number = UINT32_MAX;
-                }
-            });
-
-        friend_states.push_back(std::move(state));
-    }
-
-    // Run until all have joined and see everyone
     bool success = false;
-    uint64_t last_print = 0;
     size_t invites_sent = 0;
 
     sim.run_until(
         [&]() {
             tox_iterate(main_tox, &main_state);
 
-            // Throttle invites: keep max 5 pending
+            // Throttle invites
             size_t accepted_count = 0;
-            for (size_t k = 0; k < invites_sent; ++k) {
-                if (friend_states[k]->group_number != UINT32_MAX) {
+            for (const auto &fs : friend_states) {
+                if (fs.group_number != UINT32_MAX)
                     accepted_count++;
-                }
             }
 
             while (invites_sent < friends.size() && (invites_sent - accepted_count) < 5) {
@@ -271,58 +256,58 @@ uint32_t setup_connected_group(
                         friends[invites_sent].friend_number, &err_invite)) {
                     invites_sent++;
                 } else {
-                    if (err_invite != TOX_ERR_GROUP_INVITE_FRIEND_FAIL_SEND) {
-                        std::cerr << "Invite failed for friend " << invites_sent << ": "
-                                  << err_invite << std::endl;
-                    }
-                    break;  // Stop trying to send for this tick if we failed
+                    break;
                 }
             }
 
-            bool all_see_all = true;
-
-            if (main_state.peer_count < friends.size()) {
-                all_see_all = false;
-            }
-
+            // Process friend events
             for (size_t i = 0; i < friends.size(); ++i) {
-                tox_iterate(friends[i].tox.get(), friend_states[i].get());
-                if (friend_states[i]->group_number == UINT32_MAX
-                    || friend_states[i]->peer_count < friends.size()) {
-                    all_see_all = false;
-                }
-            }
+                auto batches = friends[i].runner->poll_events();
+                for (const auto &batch : batches) {
+                    size_t size = tox_events_get_size(batch.get());
+                    for (size_t k = 0; k < size; ++k) {
+                        const Tox_Event *e = tox_events_get(batch.get(), k);
+                        Tox_Event_Type type = tox_event_get_type(e);
 
-            if ((sim.clock().current_time_ms() - last_print) % 5000 == 0) {
-                int joined = 0;
-                int fully_connected = 0;
-                if (main_state.group_number != UINT32_MAX)
-                    joined++;
-                if (main_state.peer_count >= friends.size())
-                    fully_connected++;
+                        if (type == TOX_EVENT_GROUP_INVITE) {
+                            auto *ev = tox_event_get_group_invite(e);
+                            uint32_t friend_number = tox_event_group_invite_get_friend_number(ev);
+                            const uint8_t *data = tox_event_group_invite_get_invite_data(ev);
+                            size_t len = tox_event_group_invite_get_invite_data_length(ev);
 
-                for (const auto &fs : friend_states) {
-                    if (fs->group_number != UINT32_MAX) {
-                        joined++;
-                        if (fs->peer_count >= friends.size())
-                            fully_connected++;
+                            // Accept invite on runner thread.
+                            // We must copy data because the event structure will be freed.
+                            std::vector<uint8_t> invite_data(data, data + len);
+
+                            friends[i].runner->execute([=](Tox *tox) {
+                                Tox_Err_Group_Invite_Accept err;
+                                tox_group_invite_accept(tox, friend_number, invite_data.data(),
+                                    invite_data.size(), reinterpret_cast<const uint8_t *>("peer"),
+                                    4, nullptr, 0, &err);
+                            });
+                        } else if (type == TOX_EVENT_GROUP_PEER_JOIN) {
+                            friend_states[i].peer_count++;
+                        } else if (type == TOX_EVENT_GROUP_SELF_JOIN) {
+                            auto *ev = tox_event_get_group_self_join(e);
+                            friend_states[i].group_number
+                                = tox_event_group_self_join_get_group_number(ev);
+                        }
                     }
                 }
-                std::cerr << "[setup_connected_group] Main peer count: " << main_state.peer_count
-                          << "/" << friends.size() << ", Nodes joined: " << joined << "/"
-                          << (friends.size() + 1) << ", fully connected: " << fully_connected << "/"
-                          << (friends.size() + 1) << " (time: " << sim.clock().current_time_ms()
-                          << "ms)" << std::endl;
-                last_print = sim.clock().current_time_ms();
             }
 
-            if (all_see_all) {
-                success = true;
-                return true;
+            if (main_state.peer_count < friends.size())
+                return false;
+
+            for (const auto &fs : friend_states) {
+                if (fs.group_number == UINT32_MAX || fs.peer_count < friends.size())
+                    return false;
             }
-            return false;
+
+            success = true;
+            return true;
         },
-        300000);  // 5 minutes
+        300000);
 
     return success ? main_state.group_number : UINT32_MAX;
 }

testing/support/src/tox_runner.cc

@@ -0,0 +1,92 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later
+ * Copyright © 2026 The TokTok team.
+ */
+
+#include "../public/tox_runner.hh"
+
+namespace tox::test {
+
+ToxRunner::ToxRunner(SimulatedNode &node, const Tox_Options *options)
+    : tox_(node.create_tox(options))
+    , node_(node)
+{
+    tox_events_init(tox_.get());
+    node_.simulation().register_runner();
+
+    tick_listener_id_ = node_.simulation().register_tick_listener([this](uint64_t gen) {
+        Message msg;
+        msg.type = Message::Tick;
+        msg.generation = gen;
+        queue_.push(std::move(msg));
+    });
+
+    thread_ = std::thread([this] { loop(); });
+}
+
+ToxRunner::~ToxRunner()
+{
+    // Unregister first to prevent new ticks and update simulation counters
+    node_.simulation().unregister_tick_listener(tick_listener_id_);
+    node_.simulation().unregister_runner();
+
+    Message msg;
+    msg.type = Message::Stop;
+    queue_.push(std::move(msg));
+
+    if (thread_.joinable()) {
+        thread_.join();
+    }
+}
+
+void ToxRunner::execute(std::function<void(Tox *)> task)
+{
+    Message msg;
+    msg.type = Message::Task;
+    msg.task = std::move(task);
+    queue_.push(std::move(msg));
+}
+
+std::vector<ToxRunner::ToxEventsPtr> ToxRunner::poll_events()
+{
+    std::vector<ToxEventsPtr> ret;
+    ToxEventsPtr ptr;
+    while (events_queue_.try_pop(ptr)) {
+        ret.push_back(std::move(ptr));
+        ptr = nullptr;  // Reset ptr to avoid use-after-move warning, although try_pop overwrites
+                        // it.
+    }
+    return ret;
+}
+
+void ToxRunner::loop()
+{
+    while (true) {
+        Message msg = queue_.pop();  // Blocking wait
+
+        switch (msg.type) {
+        case Message::Stop:
+            return;
+
+        case Message::Task:
+            if (msg.task) {
+                msg.task(tox_.get());
+            }
+            break;
+
+        case Message::Tick: {
+            // Run Tox Events
+            Tox_Err_Events_Iterate err;
+            Tox_Events *events = tox_events_iterate(tox_.get(), false, &err);
+            if (events) {
+                events_queue_.push(ToxEventsPtr(events));
+            }
+
+            uint32_t interval = tox_iteration_interval(tox_.get());
+            node_.simulation().tick_complete(interval);
+            break;
+        }
+        }
+    }
+}
+
+}  // namespace tox::test

testing/support/tox_network_test.cc

@@ -6,6 +6,8 @@
 
 #include <gtest/gtest.h>
 
+#include <atomic>
+
 namespace tox::test {
 namespace {
 
@@ -22,6 +24,8 @@ namespace {
 
         ASSERT_EQ(friends.size(), num_friends);
 
+        // Verification of connection status is done inside setup_connected_friends now,
+        // but we can double check main_tox's view.
         for (const auto &f : friends) {
             EXPECT_NE(tox_friend_get_connection_status(main_tox.get(), f.friend_number, nullptr),
                 TOX_CONNECTION_NONE);
@@ -29,7 +33,7 @@ namespace {
 
         // Verify they can actually communicate
         struct Context {
-            int count = 0;
+            std::atomic<int> count{0};
         } ctx;
 
         tox_callback_friend_message(main_tox.get(),
@@ -38,16 +42,14 @@ namespace {
             });
 
         for (const auto &f : friends) {
-            const uint8_t msg[] = "hello";
-            tox_friend_send_message(
-                f.tox.get(), 0, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), nullptr);
+            f.runner->execute([](Tox *tox) {
+                const uint8_t msg[] = "hello";
+                tox_friend_send_message(tox, 0, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), nullptr);
+            });
         }
 
         sim.run_until([&]() {
             tox_iterate(main_tox.get(), &ctx);
-            for (auto &f : friends) {
-                tox_iterate(f.tox.get(), nullptr);
-            }
             return ctx.count == num_friends;
         });
 
@@ -68,7 +70,7 @@ namespace {
         ASSERT_EQ(friends.size(), num_friends);
 
         struct Context {
-            int count = 0;
+            std::atomic<int> count{0};
         } ctx;
 
         tox_callback_friend_message(main_tox.get(),
@@ -77,17 +79,15 @@ namespace {
             });
 
         for (const auto &f : friends) {
-            const uint8_t msg[] = "hello";
-            tox_friend_send_message(
-                f.tox.get(), 0, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), nullptr);
+            f.runner->execute([](Tox *tox) {
+                const uint8_t msg[] = "hello";
+                tox_friend_send_message(tox, 0, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), nullptr);
+            });
         }
 
         sim.run_until(
             [&]() {
                 tox_iterate(main_tox.get(), &ctx);
-                for (auto &f : friends) {
-                    tox_iterate(f.tox.get(), nullptr);
-                }
                 return ctx.count == num_friends;
             },
             60000);
@@ -113,10 +113,10 @@ namespace {
         EXPECT_NE(tox_friend_get_connection_status(tox2.get(), 0, nullptr), TOX_CONNECTION_NONE);
 
         // Verify communication
-        bool received = false;
+        std::atomic<bool> received{false};
         tox_callback_friend_message(tox2.get(),
             [](Tox *, uint32_t, Tox_Message_Type, const uint8_t *, size_t, void *user_data) {
-                *static_cast<bool *>(user_data) = true;
+                *static_cast<std::atomic<bool> *>(user_data) = true;
             });
 
         const uint8_t msg[] = "hello";
@@ -125,7 +125,7 @@ namespace {
         sim.run_until([&]() {
             tox_iterate(tox1.get(), nullptr);
             tox_iterate(tox2.get(), &received);
-            return received;
+            return received.load();
         });
 
         EXPECT_TRUE(received);
@@ -148,7 +148,7 @@ namespace {
 
         // Verify we can send a group message
         struct Context {
-            int count = 0;
+            std::atomic<int> count{0};
         } ctx;
 
         tox_callback_group_message(main_tox.get(),
@@ -156,20 +156,18 @@ namespace {
                 void *user_data) { static_cast<Context *>(user_data)->count++; });
 
         for (const auto &f : friends) {
-            const uint8_t msg[] = "hello";
-            uint32_t f_gn = 0;  // It should be 0 since it's the first group.
-            Tox_Err_Group_Send_Message err_send;
-            tox_group_send_message(
-                f.tox.get(), f_gn, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), &err_send);
-            EXPECT_EQ(err_send, TOX_ERR_GROUP_SEND_MESSAGE_OK);
+            f.runner->execute([](Tox *tox) {
+                const uint8_t msg[] = "hello";
+                uint32_t f_gn = 0;  // First group
+                Tox_Err_Group_Send_Message err_send;
+                tox_group_send_message(
+                    tox, f_gn, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), &err_send);
+            });
         }
 
         sim.run_until(
             [&]() {
                 tox_iterate(main_tox.get(), &ctx);
-                for (auto &f : friends) {
-                    tox_iterate(f.tox.get(), nullptr);
-                }
                 return ctx.count == num_friends;
             },
             10000);
@@ -193,7 +191,7 @@ namespace {
         EXPECT_NE(group_number, UINT32_MAX);
 
         struct Context {
-            int count = 0;
+            std::atomic<int> count{0};
         } ctx;
 
         tox_callback_group_message(main_tox.get(),
@@ -201,20 +199,18 @@ namespace {
                 void *user_data) { static_cast<Context *>(user_data)->count++; });
 
         for (const auto &f : friends) {
-            const uint8_t msg[] = "hello";
-            uint32_t f_gn = 0;
-            Tox_Err_Group_Send_Message err_send;
-            tox_group_send_message(
-                f.tox.get(), f_gn, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), &err_send);
-            EXPECT_EQ(err_send, TOX_ERR_GROUP_SEND_MESSAGE_OK);
+            f.runner->execute([](Tox *tox) {
+                const uint8_t msg[] = "hello";
+                uint32_t f_gn = 0;
+                Tox_Err_Group_Send_Message err_send;
+                tox_group_send_message(
+                    tox, f_gn, TOX_MESSAGE_TYPE_NORMAL, msg, sizeof(msg), &err_send);
+            });
         }
 
         sim.run_until(
             [&]() {
                 tox_iterate(main_tox.get(), &ctx);
-                for (auto &f : friends) {
-                    tox_iterate(f.tox.get(), nullptr);
-                }
                 return ctx.count == num_friends;
             },
             120000);