test(plain): destroy the fixtures' host on the thread that emits into it (#51)

Every LiveHost fixture in tests/protocol (five of them: test_iofold,
test_plain_object_teardown, test_plain_waiter_reaping,
test_plain_completion_sub_lifetime, test_call_error_after_acquire) puts a
ModuleProxy on a worker QThread and publishes it through a PlainTransportHost,
and every one of them freed that host from the TEST thread:

    m_host.reset();          // test thread
    m_thread->quit();
    m_thread->wait();

That is a use-after-free with a millisecond-wide window, in code five test
files share.

WHY. publishObject() connects a lambda to the proxy's eventResponse signal with
NO context object, so it is a direct connection and runs on whichever thread
emits. ModuleProxy always QUEUES that emission to its own thread — it must, or
QtRO source serialization races the reply socket — so the emitting thread is
always the worker. The lambda converts the payload and then calls fanOutEvent,
which locks the host's m_mu. Free the host on the test thread and that lock is
on a destroyed mutex. ~PlainTransportHost does disconnect the connection, which
covers an emission that has not started; a worker already inside the lambda is
not called back, it is simply running, and qvariantListToRpcList on a real
payload sits in front of the lock.

MOVING reset() AFTER quit()/wait() IS NOT THE FIX, and the obvious reason for
saying so is wrong, so here is the measured one. That order does stop the fault:
wait() joins the worker, so an in-flight lambda has finished, and it is clean
under Guard Malloc. It is clean because it throws the queue away —
QThread::quit() reaches QEventLoop::exit(), which sets the exit flag
SYNCHRONOUSLY from the calling thread instead of posting an event, so the
worker's loop stops at its next iteration and discards every emission still
queued behind it. Same specimen, same load: 273-383 of 960 events delivered,
silently. In a suite whose tests count deliveries that is the worse failure,
because nothing reports it. It also shuts the host down AFTER the proxy's
thread, which test_call_error_after_acquire needs the other way round.

THE FIX, in one shared place (live_host_teardown.h) rather than five copies,
because a copied pattern is what this was: destroy the host ON the proxy's
thread, via the SDK's own logos::runOnOwnerThread marshal. A QMetaCallEvent is
dispatched by the worker's event loop, so while it runs the worker is by
definition not inside any other slot. Qt dispatches equal-priority events FIFO,
so every emission queued before it runs first, against a live host; everything
after finds the connection already severed by ~PlainTransportHost. The io
thread, the third thread that reaches the host, is still covered by the drain
barrier ~PlainTransportHost already carries — the proxy thread is not the io
thread, so that barrier's running_in_this_thread() check still takes the
blocking path. The added blocking wait introduces no hang that was not already
there: the next two statements are quit()/wait() on the same thread, with no
timeout.

EVIDENCE. test_plain_host_event_teardown.cpp is the specimen, and it is a
detector: 24 wide events queued per round, teardown aimed at the trailing edge
of the first so the worker is inside the host's lambda. Validated the way this
directory validates detectors — against a real checkout of the code it replaces
(cf1b9b0), with the fixture's teardown as it was:

  pre-fix, no detector:    SIGABRT 3/3 runs   ("mutex lock failed: Invalid
                           argument" out of a Qt event handler)
  pre-fix, Guard Malloc:   SIGSEGV 3/3 runs   at plain_transport_host.cpp:354
                           in fanOutEvent, on the thread named "QThread", under
                           ModuleProxy::eventResponse
  naive (quit/wait/reset): NO fault, either detector — and 273-383 of 960
                           emissions delivered; the rest silently dropped
  fixed, no detector:      960/960 emitted, 0 after the free, 40/40 rounds
                           still draining when teardown began
  fixed, Guard Malloc:     same counts, clean

WHAT THIS IS NOT. The five fixtures do not fault on their own today: pre-fix,
their suites are clean under Guard Malloc (2/2 runs, 33 tests) because their
test bodies drain the burst before teardown. So this closes a live trap in
shared fixture code — one that two separate probes fell into by copying the
pattern — rather than a reproduced CI failure. It remains a candidate for the
suite's unexplained SIGSEGVs, not a proven cause, and it is stated that way in
the test file.

Full suite: 304/304 (302 before, +2 new). `nix build .#tests`: 304/304 via
ctest, 95s. The five affected suites are clean under Guard Malloc on the fixed
tree.

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

tests/protocol/CMakeLists.txt

@@ -59,6 +59,12 @@ find_package(Qt${QT_VERSION_MAJOR} REQUIRED COMPONENTS Test)
 # the failure mode that would make the whole thing unshippable. Every one of
 # those was checked by running it — against the pre-fix commit for the first two
 # groups, against a throwaway local edit for the last.
+#   PlainHostEventTeardownTest.QueuedEmissionsNeverOutliveTheHost
+#
+# The last one is a detector for the HARNESS rather than for the transport, and
+# it is checked the same way: put a plain `m_host.reset()` back in its fixture
+# and it takes the process down with SIGSEGV on the first round, with or without
+# a memory detector. The backtrace it produces is recorded in the file.
 #
 # Note what is NOT on that list: the PER-PATH exactly-once tests stay GREEN with
 # the CAS removed. They are PINS, NOT DETECTORS — a call resolved once calls
@@ -216,6 +222,16 @@ add_executable(protocol_tests
     # early-return path, many threads at once, and the shipped
     # release()-from-an-io-thread-event-callback shape.
     test_sync_call_release_race.cpp
+    # A test of the HARNESS the five tests above run on. Each of them puts a
+    # ModuleProxy on a worker QThread and publishes it through a
+    # PlainTransportHost, and each of them used to free that host from the TEST
+    # thread — while the worker was inside the direct-connection lambda
+    # publishObject() hooked onto eventResponse, which locks the host's m_mu
+    # after converting the payload. Shared fixture code, so a fault there can
+    # corrupt any run in this binary. Pins the ordering that live_host_teardown.h
+    # now owns: the host is destroyed ON the proxy's thread, every emission
+    # queued before that runs against a live host, and none runs after.
+    test_plain_host_event_teardown.cpp
     # Component tests that moved here with their code (from logos-cpp-sdk)
     test_token_manager.cpp
     test_mock_store.cpp

tests/protocol/live_host_teardown.h

@@ -0,0 +1,118 @@
+#ifndef LOGOS_TESTS_LIVE_HOST_TEARDOWN_H
+#define LOGOS_TESTS_LIVE_HOST_TEARDOWN_H
+
+// Destroying a live PlainTransportHost that publishes a ModuleProxy living on
+// its own QThread.
+//
+// Every "LiveHost" fixture in this directory is the same shape: a
+// PlainTransportHost built on the test thread, a ModuleProxy moved onto a
+// worker QThread, and publishObject() wiring the two together. They all tore
+// down the same way too, and it was wrong the same way:
+//
+//     m_host.reset();          // test thread
+//     m_thread->quit();
+//     m_thread->wait();
+//
+// WHY THAT FAULTS. publishObject() connects a lambda to the proxy's
+// eventResponse signal with NO context object, so the connection is direct and
+// the lambda runs on whichever thread emits. ModuleProxy always QUEUES that
+// emission to its own thread (module_proxy.cpp — it must, or QtRO source
+// serialization races the reply socket), so the emitting thread is always the
+// proxy's worker thread. The lambda converts the payload and then calls
+// PlainTransportHost::fanOutEvent, which locks the host's m_mu.
+//
+// A reset() on the test thread therefore frees the host underneath a lambda the
+// worker thread is already inside. ~PlainTransportHost does disconnect the
+// connection, and that closes the window for an emission that has not STARTED —
+// but a worker already past the connection lookup is not called back, it is
+// simply running, and the payload conversion between lambda entry and the
+// fanOutEvent lock is milliseconds wide on a real event. Observed, on this
+// tree, as a hard SIGSEGV on the worker thread:
+//
+//     pthread_mutex_lock
+//     std::mutex::lock
+//     logos::plain::PlainTransportHost::fanOutEvent      plain_transport_host.cpp:354
+//     PlainTransportHost::publishObject(...)::$_0        plain_transport_host.cpp:334
+//     doActivate<false>
+//     ModuleProxy::eventResponse
+//     ModuleProxy::ModuleProxy(...)::$_0                 module_proxy.cpp:38
+//
+// (`EXC_BAD_ACCESS ... at 0x71de13fa0`, thread named "QThread"; under Guard
+// Malloc the freed page is unmapped so it faults every time instead of
+// sometimes corrupting. Without a detector the same run reports
+// `mutex lock failed: Invalid argument` out of a Qt event handler.)
+//
+// WHY MOVING reset() AFTER quit()/wait() IS NOT THE FIX — measured, because the
+// obvious argument for rejecting it turns out to be wrong. That order DOES stop
+// the fault: wait() joins the worker, so a direct-connection slot already
+// running there has finished before reset(), and an emission posted after the
+// loop exits is never dispatched. It is clean under Guard Malloc, twice over.
+//
+// It is clean because it throws the queue away. QThread::quit() reaches
+// QEventLoop::exit(), which stores an exit flag SYNCHRONOUSLY from the calling
+// thread instead of posting an event, so the worker's loop stops at its next
+// iteration and every QMetaCallEvent still behind it — every pending
+// eventResponse emission — is discarded. Measured on the specimen in
+// test_plain_host_event_teardown.cpp (24 events queued per round, 40 rounds):
+// 273-383 of 960 delivered, the rest dropped, and dropped silently. In a suite
+// whose tests COUNT deliveries that is a worse failure than the crash, because
+// nothing reports it.
+//
+// It is a cross-thread deletion either way, safe only because wait() happened
+// to join first; and it shuts the host down AFTER the proxy's thread, which
+// test_call_error_after_acquire needs the other way round — the host must stop
+// serving before the proxy it dispatches to is dismantled.
+//
+// THE FIX: do the deletion ON the thread that emits. A QMetaCallEvent is
+// dispatched by the worker's event loop, so while it runs the worker is by
+// definition not inside any other slot — no emission can be in flight. And Qt
+// posts equal-priority events FIFO, so every emission queued before this call
+// runs first, against a host that is still alive — 960 of 960 on the same
+// specimen, where the quit()-first order delivered a third of them. Everything
+// queued after it finds the connection already severed by ~PlainTransportHost.
+// The io thread, the third thread that reaches the host, is covered by the drain
+// barrier ~PlainTransportHost already carries.
+//
+// The blocking wait adds no hang that was not already there: the caller's very
+// next statements are quit()/wait(), which block on the same worker thread with
+// no timeout. As before, park nothing in the provider across teardown — release
+// the provider's gate first, as the fixtures do.
+//
+// The marshal itself is the SDK's own logos::runOnOwnerThread — the same
+// primitive every inbound Logos call already uses to reach a module's thread,
+// including its same-thread short circuit. Only the "that thread is gone"
+// guard is added, because a fixture may tear down after its worker has
+// finished and a blocking marshal onto a dead event loop never returns.
+
+#include "logos_thread_marshal.h"
+#include "plain_transport_host.h"
+
+#include <QObject>
+#include <QThread>
+
+#include <memory>
+
+namespace logos::testing {
+
+inline void destroyHostOnProxyThread(
+    std::unique_ptr<logos::plain::PlainTransportHost>& host,
+    QObject* proxy)
+{
+    if (!host) return;
+
+    // No worker to serialize against: no proxy, or its thread has stopped, so
+    // nothing can be inside the fan-out lambda and nothing ever will be.
+    QThread* worker = proxy ? proxy->thread() : nullptr;
+    if (!worker || !worker->isRunning()) {
+        host.reset();
+        return;
+    }
+
+    // Capturing `host` by reference is safe precisely because this blocks until
+    // the lambda has returned (and runs it inline when we are already there).
+    logos::runOnOwnerThread(proxy, [&host] { host.reset(); });
+}
+
+} // namespace logos::testing
+
+#endif // LOGOS_TESTS_LIVE_HOST_TEARDOWN_H

tests/protocol/test_call_error_after_acquire.cpp

@@ -44,6 +44,8 @@
 
 #include "plain_transport_host.h"
 
+#include "live_host_teardown.h"
+
 #include <QCoreApplication>
 #include <QElapsedTimer>
 #include <QJsonArray>
@@ -154,8 +156,9 @@ public:
     {
         // Order matters: tear the host down FIRST so no inbound frame can be
         // dispatched to the proxy while we are dismantling it, then stop the
-        // proxy's thread, then delete the proxy it was serving.
-        m_host.reset();
+        // proxy's thread, then delete the proxy it was serving. The teardown
+        // itself runs on the PROXY's thread — see live_host_teardown.h.
+        logos::testing::destroyHostOnProxyThread(m_host, m_proxy);
         QCoreApplication::processEvents(QEventLoop::AllEvents, 50);
         m_thread->quit();
         m_thread->wait();

tests/protocol/test_iofold.cpp

@@ -84,6 +84,8 @@
 #include "plain_transport_connection.h"
 #include "plain_transport_host.h"
 
+#include "live_host_teardown.h"
+
 #include <boost/asio/ip/tcp.hpp>
 
 #include <QCoreApplication>
@@ -399,7 +401,8 @@ public:
     {
         m_provider.letGo();
         QCoreApplication::processEvents(QEventLoop::AllEvents, 200);
-        m_host.reset();
+        // On the PROXY's thread, not this one — see live_host_teardown.h.
+        logos::testing::destroyHostOnProxyThread(m_host, m_proxy);
         QCoreApplication::processEvents(QEventLoop::AllEvents, 50);
         m_thread->quit();
         m_thread->wait();

tests/protocol/test_plain_completion_sub_lifetime.cpp

@@ -72,6 +72,8 @@
 #include "plain_transport_connection.h"
 #include "plain_transport_host.h"
 
+#include "live_host_teardown.h"
+
 #include <QCoreApplication>
 #include <QJsonArray>
 #include <QString>
@@ -169,7 +171,8 @@ public:
     ~LiveHost()
     {
         QCoreApplication::processEvents(QEventLoop::AllEvents, 100);
-        m_host.reset();
+        // On the PROXY's thread, not this one — see live_host_teardown.h.
+        logos::testing::destroyHostOnProxyThread(m_host, m_proxy);
         QCoreApplication::processEvents(QEventLoop::AllEvents, 50);
         m_thread->quit();
         m_thread->wait();

tests/protocol/test_plain_host_event_teardown.cpp

@@ -0,0 +1,340 @@
+// Tearing down a live PlainTransportHost while its proxy thread is emitting.
+//
+// This is a test of the TEST HARNESS. Every LiveHost fixture in this directory
+// (test_iofold, test_plain_object_teardown, test_plain_waiter_reaping,
+// test_plain_completion_sub_lifetime, test_call_error_after_acquire) puts a
+// ModuleProxy on a worker QThread and publishes it through a
+// PlainTransportHost, and all of them used to free that host from the TEST
+// thread. That is a use-after-free with a wide window, in shared fixture code,
+// so it could corrupt any run in this binary — which makes it a candidate for
+// the unexplained SIGSEGVs this suite has seen. live_host_teardown.h carries
+// the mechanism and the reasoning; this file is the specimen that proves the
+// mechanism is needed and that it works.
+//
+// THE PATH, which is three threads deep and none of them obvious:
+//
+//   provider emits  ->  ModuleProxy queues the emission to ITS OWN thread
+//                       (module_proxy.cpp:38 — it must, or QtRO source
+//                       serialization races the reply socket)
+//                   ->  worker thread: emit eventResponse
+//                   ->  the lambda publishObject() connected with NO context
+//                       object, so it is a DIRECT connection and runs right
+//                       there on the worker
+//                   ->  qvariantListToRpcList(payload)      <-- the window
+//                   ->  PlainTransportHost::fanOutEvent -> lock m_mu
+//
+// Free the host on the test thread and the last step locks a destroyed mutex.
+// ~PlainTransportHost does disconnect the connection, which covers an emission
+// that has not started; it does nothing for a worker already inside the lambda,
+// and the payload conversion in front of the lock is milliseconds wide.
+//
+// HOW THIS TEST AIMS AT THAT WINDOW rather than sleeping towards it:
+//
+//   * WIDENED. Each event carries kElems elements, so the conversion between
+//     lambda entry and the m_mu lock is real work rather than nanoseconds.
+//
+//   * AIMED. The round queues a burst and then waits for the FIRST emission to
+//     complete before tearing down. Emissions run back to back on the worker,
+//     so an observation is the trailing edge of one and the worker is inside
+//     the next — i.e. inside the host's lambda — when teardown starts.
+//
+//   * MEASURED, so it cannot pass vacuously. pendingRounds counts rounds that
+//     began teardown with emissions still queued; if a timing change ever makes
+//     that zero the test fails instead of silently stopping to test anything.
+//
+// WHAT IT ASSERTS, and why each is structural on fixed code:
+//
+//   * afterHostGone == 0. No emission may run on the worker at or after the
+//     moment the host is freed. With the free ON the worker (a QMetaCallEvent)
+//     the worker is by definition not inside any other slot while it happens.
+//
+//   * emitted == every event pushed. Qt dispatches equal-priority events FIFO,
+//     so a teardown queued behind a burst runs after the whole burst — against
+//     a host that is still alive. Nothing is dropped and nothing is late.
+//
+// NO SUBSCRIBER is connected, and that is deliberate: the fault is the m_mu
+// lock at the TOP of fanOutEvent, which precedes the sink lookup, so a consumer
+// would add a second connection's teardown to a teardown test without widening
+// what is being tested.
+//
+// PRE-FIX EVIDENCE (this file with logos::testing::destroyHostOnProxyThread
+// replaced by a plain m_host.reset(), everything else identical): a hard
+// SIGSEGV on the worker thread, on the very first round, with or without a
+// memory detector.
+//
+//   EXC_BAD_ACCESS (SIGSEGV) KERN_INVALID_ADDRESS at 0x71de13fa0, thread "QThread"
+//     pthread_mutex_lock
+//     std::mutex::lock
+//     logos::plain::PlainTransportHost::fanOutEvent   plain_transport_host.cpp:354
+//     PlainTransportHost::publishObject(...)::$_0     plain_transport_host.cpp:334
+//     doActivate<false>
+//     ModuleProxy::eventResponse
+//     ModuleProxy::ModuleProxy(...)::$_0              module_proxy.cpp:38
+//
+// Undetected, the same run reports `mutex lock failed: Invalid argument` out of
+// a Qt event handler and aborts. The detector used here is macOS Guard Malloc
+// (DYLD_INSERT_LIBRARIES=/usr/lib/libgmalloc.dylib), which unmaps freed pages so
+// the dangling lock faults every time; ASan/TSan are unusable on this toolchain
+// (libclang_rt livelocks in its own initializer before main), the same note that
+// sits on test_plain_object_teardown.cpp. Fixed, this file is clean under Guard
+// Malloc with pendingRounds == kRounds.
+//
+// THE THIRD ORDER, and why `emitted` is asserted rather than just printed. The
+// obvious repair — quit(), wait(), THEN reset() — is memory-clean: wait() joins
+// the worker, so an in-flight lambda has finished, and Guard Malloc says nothing
+// (afterHostGone 0, twice). It is clean because QThread::quit() reaches
+// QEventLoop::exit(), which sets the exit flag synchronously from the CALLING
+// thread rather than posting an event, so the worker's loop stops at its next
+// iteration and discards every emission still queued behind it. Same specimen,
+// same load: 273-383 of 960 delivered. Nothing reports that on its own — which
+// is why the count is an assertion here.
+
+#include <gtest/gtest.h>
+
+#include "logos_provider_interface.h"
+#include "logos_transport_config.h"
+#include "module_proxy.h"
+
+#include "plain_transport_host.h"
+
+#include "live_host_teardown.h"
+
+#include <QCoreApplication>
+#include <QJsonArray>
+#include <QString>
+#include <QThread>
+#include <QVariant>
+#include <QVariantList>
+
+#include <atomic>
+#include <chrono>
+#include <cstdint>
+#include <iostream>
+#include <memory>
+#include <thread>
+
+using namespace logos::plain;
+
+namespace {
+
+// Bookkeeping shared between the test thread and the proxy's worker thread.
+struct Watch {
+    // Emissions that reached the worker's eventResponse.
+    std::atomic<int>  emitted{0};
+    // Set on the worker, immediately after the host is destroyed there.
+    std::atomic<bool> hostGone{false};
+    // Emissions that ran with the host already gone. Must stay 0.
+    std::atomic<int>  afterHostGone{0};
+};
+
+// Pushes events on demand through the real route: the EventCallback handed to
+// setEventListener is ModuleProxy's, so every push takes the queued path a real
+// provider's event takes.
+class Pusher : public LogosProviderObject {
+public:
+    QVariant callMethod(const QString&, const QVariantList& args) override
+    {
+        return args.value(0, QVariant(1));
+    }
+    QJsonArray getMethods() override { return QJsonArray{}; }
+    bool informModuleToken(const QString&, const QString&) override { return true; }
+    void setEventListener(EventCallback cb) override { m_emit = std::move(cb); }
+    void init(void*) override {}
+    QString providerName() const override { return QStringLiteral("pusher"); }
+    QString providerVersion() const override { return QStringLiteral("1.0.0"); }
+
+    void push(const QVariantList& data)
+    {
+        if (m_emit) m_emit(QStringLiteral("teardown_probe"), data);
+    }
+
+private:
+    EventCallback m_emit;
+};
+
+// The fixture shape this whole file is about, reduced to the parts that matter.
+class LiveHost {
+public:
+    explicit LiveHost(Watch& w) : m_watch(w)
+    {
+        LogosTransportConfig cfg;
+        cfg.protocol = LogosProtocol::Tcp;
+        cfg.host = "127.0.0.1";
+        cfg.port = 0;   // ephemeral
+        m_host = std::make_unique<PlainTransportHost>(cfg);
+        m_started = m_host->start();
+
+        m_proxy = new ModuleProxy(&m_provider);
+        m_proxy->saveToken(QStringLiteral("origin"), QStringLiteral("live-token"));
+        m_thread = new QThread;
+        m_proxy->moveToThread(m_thread);
+        m_thread->start();
+
+        m_published = m_host->publishObject("pusher_module", m_proxy);
+
+        // Connected AFTER publishObject, so on each emission this runs on the
+        // worker thread immediately after the host's fan-out lambda — i.e. it
+        // observes the same emission that would have touched a freed host.
+        QObject::connect(m_proxy, &ModuleProxy::eventResponse,
+                         [this](const QString&, const QVariantList&) {
+                             if (m_watch.hostGone.load(std::memory_order_acquire))
+                                 m_watch.afterHostGone.fetch_add(1);
+                             m_watch.emitted.fetch_add(1, std::memory_order_release);
+                         });
+
+        const QString endpoint = m_host->endpoint();
+        m_port = endpoint.mid(endpoint.lastIndexOf(':') + 1).toUShort();
+    }
+
+    ~LiveHost()
+    {
+        // The host dies on the proxy's thread. hostGone is set THERE, right
+        // after the free, so the observer above reads it with no race of its
+        // own: both run on that one thread.
+        QMetaObject::invokeMethod(m_proxy, [this] {
+            m_host.reset();
+            m_watch.hostGone.store(true, std::memory_order_release);
+        }, Qt::BlockingQueuedConnection);
+
+        m_thread->quit();
+        m_thread->wait();
+        delete m_proxy;
+        delete m_thread;
+    }
+
+    bool ok() const { return m_started && m_published && m_port != 0; }
+    Pusher& provider() { return m_provider; }
+
+private:
+    Watch& m_watch;
+    Pusher m_provider;
+    std::unique_ptr<PlainTransportHost> m_host;
+    ModuleProxy* m_proxy = nullptr;
+    QThread* m_thread = nullptr;
+    bool m_started = false;
+    bool m_published = false;
+    uint16_t m_port = 0;
+};
+
+// The same teardown, expressed through the shared helper the five real fixtures
+// call. Same object, same ordering — this one is here so the helper itself is
+// exercised rather than only the shape it encodes.
+class LiveHostViaHelper {
+public:
+    LiveHostViaHelper()
+    {
+        LogosTransportConfig cfg;
+        cfg.protocol = LogosProtocol::Tcp;
+        cfg.host = "127.0.0.1";
+        cfg.port = 0;
+        m_host = std::make_unique<PlainTransportHost>(cfg);
+        m_started = m_host->start();
+
+        m_proxy = new ModuleProxy(&m_provider);
+        m_proxy->saveToken(QStringLiteral("origin"), QStringLiteral("live-token"));
+        m_thread = new QThread;
+        m_proxy->moveToThread(m_thread);
+        m_thread->start();
+
+        m_published = m_host->publishObject("pusher_module", m_proxy);
+    }
+
+    ~LiveHostViaHelper()
+    {
+        logos::testing::destroyHostOnProxyThread(m_host, m_proxy);
+        m_thread->quit();
+        m_thread->wait();
+        delete m_proxy;
+        delete m_thread;
+    }
+
+    bool ok() const { return m_started && m_published; }
+    Pusher& provider() { return m_provider; }
+
+private:
+    Pusher m_provider;
+    std::unique_ptr<PlainTransportHost> m_host;
+    ModuleProxy* m_proxy = nullptr;
+    QThread* m_thread = nullptr;
+    bool m_started = false;
+    bool m_published = false;
+};
+
+// Width of the window: the fan-out lambda runs qvariantListToRpcList() over
+// this before it touches the host.
+constexpr int kElems  = 8000;
+// Events per round. Enough that the burst is still draining when teardown
+// starts, which is what pendingRounds checks.
+constexpr int kBurst  = 24;
+constexpr int kRounds = 40;
+
+QVariantList widePayload()
+{
+    QVariantList l;
+    l.reserve(kElems);
+    for (int i = 0; i < kElems; ++i)
+        l.append(QVariant(QStringLiteral("xxxxxxxxxxxxxxxx")));
+    return l;
+}
+
+TEST(PlainHostEventTeardownTest, QueuedEmissionsNeverOutliveTheHost)
+{
+    const QVariantList payload = widePayload();
+
+    int totalEmitted = 0;
+    int totalAfter   = 0;
+    int pendingRounds = 0;
+
+    for (int r = 0; r < kRounds; ++r) {
+        Watch w;
+        auto host = std::make_unique<LiveHost>(w);
+        ASSERT_TRUE(host->ok()) << "round " << r;
+
+        for (int i = 0; i < kBurst; ++i) host->provider().push(payload);
+
+        // Aim: tear down on the trailing edge of the first emission, while the
+        // worker is inside the next one.
+        const auto deadline =
+            std::chrono::steady_clock::now() + std::chrono::seconds(5);
+        while (w.emitted.load(std::memory_order_acquire) < 1 &&
+               std::chrono::steady_clock::now() < deadline)
+            std::this_thread::yield();
+        ASSERT_GE(w.emitted.load(), 1) << "round " << r << ": nothing emitted";
+
+        if (w.emitted.load(std::memory_order_acquire) < kBurst) ++pendingRounds;
+        host.reset();
+
+        totalEmitted += w.emitted.load();
+        totalAfter   += w.afterHostGone.load();
+    }
+
+    std::cout << "  rounds=" << kRounds << " burst=" << kBurst
+              << " emitted=" << totalEmitted
+              << " afterHostGone=" << totalAfter
+              << " pendingAtTeardown=" << pendingRounds << " rounds\n";
+
+    // The window was really open — otherwise the two assertions below are free.
+    EXPECT_GT(pendingRounds, 0)
+        << "no round began teardown with emissions still queued; this test "
+           "stopped exercising the race it exists for";
+    EXPECT_EQ(totalAfter, 0)
+        << "an event emission ran on the proxy thread after the host was freed";
+    EXPECT_EQ(totalEmitted, kRounds * kBurst)
+        << "teardown swallowed queued emissions instead of running behind them";
+}
+
+// The helper on the path the fixtures use it on. Nothing to measure here beyond
+// "a burst in flight at teardown is survivable"; the measured version is above.
+TEST(PlainHostEventTeardownTest, TheSharedHelperTearsDownUnderTheSameLoad)
+{
+    const QVariantList payload = widePayload();
+    for (int r = 0; r < 10; ++r) {
+        LiveHostViaHelper host;
+        ASSERT_TRUE(host.ok()) << "round " << r;
+        for (int i = 0; i < kBurst; ++i) host.provider().push(payload);
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+    SUCCEED();
+}
+
+} // namespace

tests/protocol/test_plain_object_teardown.cpp

@@ -52,6 +52,8 @@
 #include "plain_transport_connection.h"
 #include "plain_transport_host.h"
 
+#include "live_host_teardown.h"
+
 #include <QCoreApplication>
 #include <QElapsedTimer>
 #include <QJsonArray>
@@ -164,7 +166,8 @@ public:
         // below; let every blocked (and queued) call finish first.
         m_provider.letGo();
         QCoreApplication::processEvents(QEventLoop::AllEvents, 200);
-        m_host.reset();
+        // On the PROXY's thread, not this one — see live_host_teardown.h.
+        logos::testing::destroyHostOnProxyThread(m_host, m_proxy);
         QCoreApplication::processEvents(QEventLoop::AllEvents, 50);
         m_thread->quit();
         m_thread->wait();

tests/protocol/test_plain_waiter_reaping.cpp

@@ -84,9 +84,10 @@
 
 #include "plain_transport_connection.h"
 #include "plain_transport_host.h"
-
 #include "plain_logos_object.h"
 
+#include "live_host_teardown.h"
+
 #include <QCoreApplication>
 #include <QElapsedTimer>
 #include <QJsonArray>
@@ -310,7 +311,8 @@ public:
         // below; let every blocked (and queued) call finish first.
         m_provider.letGo();
         QCoreApplication::processEvents(QEventLoop::AllEvents, 100);
-        m_host.reset();
+        // On the PROXY's thread, not this one — see live_host_teardown.h.
+        logos::testing::destroyHostOnProxyThread(m_host, m_proxy);
         QCoreApplication::processEvents(QEventLoop::AllEvents, 50);
         m_thread->quit();
         m_thread->wait();