// Async delivery in a process that has NO Qt event loop.
//
// WHAT WAS BROKEN. Every async completion in the plain transport goes through
// one hop, and that hop was:
//
//     QCoreApplication* app = QCoreApplication::instance();
//     if (!app) return;                       // <- the callback, dropped
//
// In a Qt-free host — the deployment the plain transport exists for — that
// branch is taken for EVERY call, so `callMethodAsyncWithError` (and
// `lp_invoke_async` above it, and every generated async wrapper above that)
// promised a callback exactly once and delivered zero, forever, silently. The
// caller does not get an error; it gets nothing, which turns a bounded call
// into an unbounded hang.
//
// It could not be observed from protocol_tests, whose main() constructs a
// QCoreApplication before the first test runs — hence this second binary. See
// test_main_noqt.cpp.
//
// WHAT IS PINNED HERE:
//
//   1. The callback ARRIVES, on all four outcomes: an immediate reply, a
//      deferred ("multi") completion, a deadline, and a call cancelled by
//      release(). Each of these reaches the delivery hop from a different
//      thread, and the drop was in the hop, so every one of them was affected.
//   2. It does NOT arrive inline. The constraint on any fix here is that
//      delivery must not move onto the Asio stack: this transport delivers user
//      event callbacks inline on its single io thread already, and a user
//      callback running on an Asio read handler is the re-entrancy class that
//      produced a SIGSEGV on the QtRO twin. So the tests assert the callback
//      did not run inside the call, did not run on the caller's thread, and did
//      not run on the io thread. "Inside the call" is measured as NESTING, by
//      the delivering thread, at delivery time — see IssuingScope, and the note
//      there about why counting deliveries after the call returns is a race
//      rather than a check.
//   3. Exactly once still holds with no Qt loop in the process, including when
//      release() races a burst of arriving replies — the same shape
//      test_iofold.cpp uses, run here against the delivery thread instead of
//      the Qt loop.
//   4. It still arrives AFTER main() HAS RETURNED, from a static destructor —
//      the window in which a lazily-constructed delivery vehicle is already
//      dead. No test case can run there, so that one is a static destructor
//      whose verdict is this binary's exit code; see LateDeliveryProbe.
//
// HOW THE DETECTOR WAS VALIDATED: by running this file against the pre-fix
// tree (feat/plain-async-io-fold, cf1b9b0). Every test below that waits for a
// callback fails there — the waits run out with zero deliveries — which is the
// bug, stated as a test. The numbers are in the PR.
//
// The provider is deliberately Qt-FREE too: an IncomingCallHandler on an
// RpcServerTcp, no ModuleProxy and no QObject, because a QObject provider
// needs a thread with a Qt event loop to dispatch into and this process has
// none. That is also the honest shape — a Qt-free consumer talking to a host
// over a socket.

#include <gtest/gtest.h>

#include "logos_call_error.h"
#include "logos_object.h"
#include "logos_transport_config.h"

#include "io_context_pool.h"
#include "json_codec.h"
#include "plain_logos_object.h"
#include "plain_transport_connection.h"
#include "rpc_server.h"

#include "logos_async_dispatch.h"

#include <QCoreApplication>
#include <QString>
#include <QVariant>
#include <QVariantList>
#include <QVariantMap>

#include <atomic>
#include <chrono>
#include <cstdint>
#include <iostream>
#include <map>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <vector>

using namespace logos::plain;

namespace {

// ── a Qt-free provider ──────────────────────────────────────────────────────
//
// Handles exactly the four shapes the tests need:
//   ping   — answers immediately
//   sink   — never answers at all (the caller's deadline resolves it)
//   defer  — answers with the pending sentinel, then pushes the completion
//            event, which is the "multi" provider protocol
//   never  — answers with a sentinel it never completes (cancellation fodder)
//
// onCall runs on the connection's strand, i.e. the process's single io thread,
// which is exactly the thread the delivery hop must NOT run user callbacks on.
// It is recorded here so the tests can assert that.
class QtFreeHandler : public IncomingCallHandler {
public:
    void onCall(const CallMessage& req, CallReply reply) override
    {
        m_ioThread.store(std::this_thread::get_id());
        m_sawCall.store(true);

        if (req.method == "sink") return;      // no reply, ever

        ResultMessage res;
        res.id = req.id;
        res.ok = true;

        if (req.method == "defer" || req.method == "never") {
            const std::string callId =
                (req.method == "defer" ? "cid-" : "never-")
                + std::to_string(m_counter.fetch_add(1));
            RpcMap pending;
            pending.emplace(logos::pendingCallKey().toStdString(), RpcValue(callId));
            res.value = RpcValue(std::move(pending));
            reply(std::move(res));

            if (req.method == "never") return;

            // The completion is pushed from a worker, the way a real "multi"
            // provider does it, so it arrives as an Event frame rather than
            // inline in the reply.
            EventSink sink = eventSink();
            std::thread([sink, callId] {
                if (!sink) return;
                EventMessage evt;
                evt.object    = "omni";
                evt.eventName = logos::callCompleteEvent().toStdString();
                evt.data.push_back(RpcValue(callId));
                evt.data.push_back(RpcValue(static_cast<int64_t>(7)));
                sink(std::move(evt));
            }).detach();
            return;
        }

        res.value = req.args.empty() ? RpcValue(static_cast<int64_t>(1))
                                     : req.args.front();
        reply(std::move(res));
    }

    void onMethods(const MethodsMessage& req, MethodsReply reply) override
    {
        MethodsResultMessage res;
        res.id = req.id;
        res.ok = true;
        reply(std::move(res));
    }

    void onSubscribe(const SubscribeMessage&, EventSink sink,
                     const void* connectionId) override
    {
        std::lock_guard<std::mutex> g(m_mu);
        m_sinks[connectionId] = std::move(sink);
    }

    void onUnsubscribe(const UnsubscribeMessage&, const void* connectionId) override
    {
        std::lock_guard<std::mutex> g(m_mu);
        m_sinks.erase(connectionId);
    }

    void onConnectionClosed(const void* connectionId) override
    {
        std::lock_guard<std::mutex> g(m_mu);
        m_sinks.erase(connectionId);
    }

    void onToken(const TokenMessage&) override {}

    std::thread::id ioThread() const { return m_ioThread.load(); }
    bool sawCall() const { return m_sawCall.load(); }

private:
    EventSink eventSink()
    {
        std::lock_guard<std::mutex> g(m_mu);
        return m_sinks.empty() ? EventSink{} : m_sinks.begin()->second;
    }

    std::mutex                            m_mu;
    std::map<const void*, EventSink>      m_sinks;
    std::atomic<std::thread::id>          m_ioThread{};
    std::atomic<bool>                     m_sawCall{false};
    std::atomic<std::uint64_t>            m_counter{0};
};

class QtFreeHost {
public:
    QtFreeHost()
    {
        m_server = std::make_shared<RpcServerTcp>(
            IoContextPool::shared().ioContext(), "127.0.0.1", 0,
            std::make_shared<JsonCodec>(), m_handler);
        m_started = m_server->start();
    }
    ~QtFreeHost() { if (m_server) m_server->stop(); }

    bool ok() const { return m_started && m_server->boundPort() != 0; }
    uint16_t port() const { return m_server->boundPort(); }
    QtFreeHandler& handler() { return *m_handler; }

private:
    // DELIBERATELY LEAKED, and this is a harness lifetime bug that was worth
    // finding rather than a style choice. RpcServer keeps a RAW
    // IncomingCallHandler*, hands it to every connection, and nothing joins the
    // io thread — not stop(), not the server's destruction. So a frame that has
    // already been read from the socket can be dispatched into the handler after
    // this object's members would have been destroyed, and a handler that was a
    // member died first: SIGBUS on the io thread, inside
    // RpcConnection::dispatchIncoming calling a virtual on freed storage.
    //
    // Observed at 1 run in 25 (and 1 in 5 under Guard Malloc) once the run got
    // slightly longer — a burst test leaves frames queued, and the process has to
    // stay alive long enough for the io thread to reach them. It is the test's
    // bug, not the transport's, and the fix is to let the handler outlive the io
    // thread: one small object per host, in a test binary that is about to exit.
    QtFreeHandler*                 m_handler = new QtFreeHandler();
    std::shared_ptr<RpcServerTcp>  m_server;
    bool                           m_started = false;
};

std::unique_ptr<PlainTransportConnection> connectTo(uint16_t port)
{
    LogosTransportConfig cfg;
    cfg.protocol = LogosProtocol::Tcp;
    cfg.host = "127.0.0.1";
    cfg.port = port;
    auto conn = std::make_unique<PlainTransportConnection>(cfg);
    if (!conn->connectToHost()) return nullptr;
    return conn;
}

LogosObjectErrorChannel* channelFor(LogosObject* obj)
{
    return dynamic_cast<LogosObjectErrorChannel*>(obj);
}

// ── the inline check, done properly ─────────────────────────────────────────
//
// "Did the callback run INLINE inside the call that issued it" is a question
// about NESTING, and the only place it can be answered is at the moment of
// delivery, on the delivering thread. This thread-local depth counter is that
// answer: IssuingScope raises it around an issuing call, and a callback that
// runs nested inside that call — which can only happen on the issuing thread,
// because inline means on this stack — sees its own thread's copy raised. A
// callback on the delivery thread reads that thread's copy, which is zero.
// Nothing races: no shared state, no ordering, no timing.
//
// WHAT THIS REPLACES, because it was wrong in a way worth remembering. The first
// version of test 1 read `d.total` AFTER callMethodAsyncWithError returned and
// asserted it was still zero. That is not an inline check, it is a race with the
// delivery thread — which is allowed to deliver the instant the call returns —
// and it failed 3 in 200 plain / 2 in 40 under Guard Malloc, always with
// on-caller-thread=0, i.e. always with nothing whatsoever having run inline.
// This file already says as much about tests 2 and 5 ("that measures scheduling
// luck"); test 1 was doing it anyway. A post-hoc count read cannot distinguish
// "ran inline" from "ran promptly, elsewhere". Nesting can.
thread_local int t_insideIssue = 0;

struct IssuingScope {
    IssuingScope()  { ++t_insideIssue; }
    ~IssuingScope() { --t_insideIssue; }
    IssuingScope(const IssuingScope&) = delete;
    IssuingScope& operator=(const IssuingScope&) = delete;
};

// Per-call delivery counts, plus where the delivery happened. Both 0 and 2 are
// failures and both are counted per call, because a double on one call and a
// drop on another cancel out in a total.
struct Deliveries {
    explicit Deliveries(int n) : counts(n) {}
    std::vector<std::atomic<int>> counts;
    std::atomic<int>              total{0};

    // Deliveries that ran nested inside the call that issued them. Observed by
    // the delivering thread itself, at delivery time — see IssuingScope.
    std::atomic<int>              inlineDeliveries{0};

    // A thread no delivery may ever run on, counted rather than sampled.
    // Set before the first call is issued and only read after the last has
    // landed, so it needs no synchronisation of its own.
    //
    // Sampling the LAST delivery's thread is not enough for this claim: 300
    // calls delivered correctly and one delivered on the caller's stack is
    // still the bug, and a last-writer-wins field would miss it 299 times out
    // of 300.
    std::thread::id  forbiddenThread{};
    std::atomic<int> onForbiddenThread{0};

    std::mutex      mu;
    std::string     lastCode;
    QVariant        lastValue;
    std::thread::id lastThread;

    void record(int i, QVariant v, const logos::CallError& e)
    {
        // FIRST, and on the delivering thread: nesting is only observable from
        // inside the delivery.
        if (t_insideIssue > 0) inlineDeliveries.fetch_add(1);
        if (forbiddenThread != std::thread::id{}
            && std::this_thread::get_id() == forbiddenThread)
            onForbiddenThread.fetch_add(1);
        {
            std::lock_guard<std::mutex> g(mu);
            lastCode   = e.code;
            lastValue  = std::move(v);
            lastThread = std::this_thread::get_id();
        }
        counts[i].fetch_add(1);
        total.fetch_add(1);
    }
    std::string code() { std::lock_guard<std::mutex> g(mu); return lastCode; }
    std::thread::id thread() { std::lock_guard<std::mutex> g(mu); return lastThread; }
    int worst() const
    {
        int w = 0;
        for (const auto& c : counts) w = std::max(w, c.load());
        return w;
    }
    int missing() const
    {
        int m = 0;
        for (const auto& c : counts) if (c.load() == 0) ++m;
        return m;
    }
};

// There is no event loop to pump: the only thing to do is wait. A budget rather
// than a fixed sleep so the passing case is fast and the failing case is
// unambiguous.
bool waitFor(Deliveries& d, int target, int budgetMs)
{
    const auto deadline = std::chrono::steady_clock::now()
        + std::chrono::milliseconds(budgetMs);
    while (d.total.load() < target
           && std::chrono::steady_clock::now() < deadline)
        std::this_thread::sleep_for(std::chrono::milliseconds(2));
    return d.total.load() >= target;
}

void settle(int ms)
{
    std::this_thread::sleep_for(std::chrono::milliseconds(ms));
}

const char* kToken = "noqt-token";

// ── the delivery that happens AFTER main() has returned ─────────────────────
//
// THE DEFECT THIS PROBE EXISTS FOR. The delivery vehicle used to be an ordinary
// function-local static, constructed on the FIRST async delivery in the process.
// Anything with static storage constructed before that — which is everything
// constructed during dynamic initialisation — is therefore destroyed AFTER it,
// so a delivery issued from such a destructor posted into an io_context that had
// already run its own destructor, on a thread that had already been joined. Under
// Guard Malloc: SIGSEGV in scheduler::post_immediate_completion, reached through
// __cxa_finalize, three runs out of three. Without it: silence, delivered=0 —
// which is the exact bug this file was written to kill, moved to a later moment
// in the process's life.
//
// WHY IT IS A STATIC OBJECT AND NOT A TEST CASE. The moment under test is "after
// main() returned", and no test case runs there. So the check is a destructor,
// and its verdict is the PROCESS EXIT CODE: a failure calls _Exit with a
// distinctive status, which is what CI (and `ninja test`) sees, because gtest has
// long since printed its summary and returned. There is no other way to assert
// on this window.
//
// ORDERING, which is the whole point and is easy to break by accident: this
// object must be constructed BEFORE the first delivery in the process, so that on
// a tree that destroys the vehicle it is destroyed AFTER it. Dynamic
// initialisation gives that for free — and it is why this probe must NOT warm the
// vehicle up in its own constructor: doing so would register the vehicle's
// destructor first and hand it the LONGER life, quietly turning the reproduction
// into a no-op. TheVehicleOutlivesStaticDestructors below does the warm-up, from
// inside a test, where it lands after all dynamic initialisation.
std::atomic<int>  g_lateDeliveries{0};
std::atomic<bool> g_lateOffTheIssuingThread{false};
std::atomic<bool> g_lateProbeArmed{false};

struct LateDeliveryProbe {
    LateDeliveryProbe() { g_lateProbeArmed.store(true); }

    ~LateDeliveryProbe()
    {
        const std::thread::id issuing = std::this_thread::get_id();
        std::fprintf(stderr,
            "\n[post-main] issuing an async call from a static destructor "
            "(after main() returned)\n");
        std::fflush(stderr);

        // The null-connection early return: the smallest path that reaches the
        // delivery hop, with no socket, no host and no io thread of its own.
        PlainLogosObject obj("nobody", nullptr);
        obj.callMethodAsyncWithError(kToken, QStringLiteral("ping"), {}, 1000,
            [issuing](QVariant, const logos::CallError&) {
                if (std::this_thread::get_id() != issuing)
                    g_lateOffTheIssuingThread.store(true);
                g_lateDeliveries.fetch_add(1);
            });

        // Bounded, like every other wait in this file: the passing case is fast
        // and the failing case is unambiguous rather than a hang at exit.
        const auto deadline = std::chrono::steady_clock::now()
            + std::chrono::milliseconds(5000);
        while (g_lateDeliveries.load() == 0
               && std::chrono::steady_clock::now() < deadline)
            std::this_thread::sleep_for(std::chrono::milliseconds(2));

        const int  delivered = g_lateDeliveries.load();
        const bool offThread = g_lateOffTheIssuingThread.load();
        std::fprintf(stderr,
            "[post-main] delivered=%d off-the-issuing-thread=%d\n",
            delivered, static_cast<int>(offThread));
        std::fflush(stderr);

        if (delivered != 1 || !offThread) {
            std::fprintf(stderr,
                "\nPOST-MAIN DELIVERY FAILED: callMethodAsyncWithError promises "
                "its callback exactly once, and a delivery issued after main() "
                "returned got %d of them%s.\n"
                "  The delivery vehicle must outlive every possible caller, "
                "which for a process-wide static means it must never be "
                "destroyed. See DeliveryService in plain_logos_object.cpp.\n",
                delivered,
                (delivered == 1 && !offThread)
                    ? " (and ran on the issuing thread, which is the inline "
                      "delivery the hop exists to prevent)"
                    : "");
            std::fflush(stderr);
            std::_Exit(70);
        }
    }
};

// Constructed during dynamic initialisation; destroyed after main(). See above.
LateDeliveryProbe g_lateDeliveryProbe;

} // namespace

class NoQtLoopTest : public ::testing::Test {
protected:
    // The premise of the whole file, checked on every test rather than assumed:
    // if something ever constructs a QCoreApplication in this process, these
    // tests silently stop testing anything.
    void SetUp() override
    {
        ASSERT_EQ(QCoreApplication::instance(), nullptr)
            << "this binary must run with NO QCoreApplication — otherwise it is "
               "just protocol_tests with fewer tests";
    }
};

// ── 1. the smallest possible statement of the bug ───────────────────────────
//
// No host, no socket, no threads of our own: a handle whose connection is not
// open takes the early-return branch in callMethodAsyncWithError, which posts
// the failure through the same delivery hop as everything else. Pre-fix the
// callback is dropped there and this waits out its whole budget.
TEST_F(NoQtLoopTest, AFailedCallDeliversItsCallbackWithNoQtLoop)
{
    PlainLogosObject obj("nobody", nullptr);

    Deliveries d(1);
    const std::thread::id caller = std::this_thread::get_id();
    d.forbiddenThread = caller;

    {
        // The nesting marker, raised for exactly the duration of the call. A
        // callback that runs inline runs INSIDE this scope, on this thread, and
        // says so from inside itself — which is a fact about the stack and not
        // about who won a race. Reading d.total after the call returns would be
        // the race; see IssuingScope.
        IssuingScope inCall;
        obj.callMethodAsyncWithError(kToken, QStringLiteral("ping"), {}, 1000,
                                     [&d](QVariant v, const logos::CallError& e) {
                                         d.record(0, std::move(v), e);
                                     });
    }

    const bool arrived = waitFor(d, 1, 5000);
    settle(50);   // a duplicate would land here

    std::cout << "  no-Qt failure path: delivered=" << d.total.load()
              << " code='" << d.code() << "'"
              << " inline=" << d.inlineDeliveries.load()
              << " on-caller-thread=" << d.onForbiddenThread.load() << std::endl;

    ASSERT_TRUE(arrived)
        << "the callback was never delivered: with no QCoreApplication the "
           "delivery hop dropped it, and callMethodAsyncWithError's "
           "exactly-once promise became exactly-never";
    EXPECT_EQ(d.total.load(), 1);
    EXPECT_EQ(d.code(), "transport_error");
    EXPECT_EQ(d.inlineDeliveries.load(), 0)
        << "the callback ran inline inside callMethodAsyncWithError";
    EXPECT_EQ(d.onForbiddenThread.load(), 0)
        << "the callback ran on the caller's thread — delivery must stay off "
           "the issuing stack";
}

// ── 2. a real call over a real socket ───────────────────────────────────────
//
// The normal path, at volume, with the delivery thread as the only place a
// callback can land. Also the place to pin the constraint that shapes the fix:
// the callback must not run on the io thread, which is where the reply is
// decoded.
TEST_F(NoQtLoopTest, RepliesDeliverExactlyOnceWithNoQtLoop)
{
    QtFreeHost host;
    ASSERT_TRUE(host.ok());
    auto conn = connectTo(host.port());
    ASSERT_NE(conn, nullptr);

    LogosObject* obj = conn->requestObject(QStringLiteral("omni"), 5000);
    ASSERT_NE(obj, nullptr);
    auto* ch = channelFor(obj);
    ASSERT_NE(ch, nullptr);

    constexpr int kCalls = 300;
    Deliveries d(kCalls);
    const std::thread::id caller = std::this_thread::get_id();
    // NOT "no callback had arrived by the time the loop finished" — with a real
    // delivery thread, callbacks for the early calls legitimately land while the
    // later ones are still being issued, so that measures scheduling luck. The
    // claim is per delivery and it is about the THREAD: no callback may run on
    // the stack that issued the call.
    d.forbiddenThread = caller;

    for (int i = 0; i < kCalls; ++i) {
        // Per call, so the nesting claim covers all 300 issue points and not
        // just the loop as a whole.
        IssuingScope inCall;
        ch->callMethodAsyncWithError(kToken, QStringLiteral("ping"),
                                     QVariantList{ QVariant(i) }, 10000,
                                     [&d, i](QVariant v, const logos::CallError& e) {
                                         d.record(i, std::move(v), e);
                                     });
    }

    const bool arrived = waitFor(d, kCalls, 20000);
    settle(200);   // duplicates would land here

    std::cout << "  no-Qt replies: " << d.total.load() << "/" << kCalls
              << " worst=" << d.worst() << " missing=" << d.missing()
              << " on-caller-thread=" << d.onForbiddenThread.load()
              << " delivery-thread!=io-thread="
              << (d.thread() != host.handler().ioThread()) << std::endl;

    ASSERT_TRUE(arrived) << "only " << d.total.load() << " of " << kCalls
                         << " callbacks were delivered with no Qt loop";
    EXPECT_EQ(d.inlineDeliveries.load(), 0)
        << d.inlineDeliveries.load() << " callbacks ran nested inside the call "
           "that issued them";
    EXPECT_EQ(d.onForbiddenThread.load(), 0)
        << d.onForbiddenThread.load() << " callbacks ran on the thread that "
           "issued the call";
    EXPECT_EQ(d.worst(), 1);
    EXPECT_EQ(d.missing(), 0);
    EXPECT_TRUE(d.code().empty());
    ASSERT_TRUE(host.handler().sawCall());
    EXPECT_NE(d.thread(), host.handler().ioThread())
        << "the callback ran on the transport's io thread — that is the "
           "re-entrancy the delivery hop exists to prevent, and delivering "
           "inline there would be a worse bug than the drop";

    obj->release();
    settle(50);
}

// ── 3. the deferred ("multi") completion ────────────────────────────────────
//
// Reaches the hop from the completion-event handler on the io thread, which is
// a different resolver from the reply path above and was dropped just as hard.
TEST_F(NoQtLoopTest, DeferredCompletionsDeliverWithNoQtLoop)
{
    QtFreeHost host;
    ASSERT_TRUE(host.ok());
    auto conn = connectTo(host.port());
    ASSERT_NE(conn, nullptr);

    LogosObject* obj = conn->requestObject(QStringLiteral("omni"), 5000);
    ASSERT_NE(obj, nullptr);
    auto* ch = channelFor(obj);
    ASSERT_NE(ch, nullptr);

    constexpr int kCalls = 40;
    Deliveries d(kCalls);
    for (int i = 0; i < kCalls; ++i) {
        ch->callMethodAsyncWithError(kToken, QStringLiteral("defer"), {}, 10000,
                                     [&d, i](QVariant v, const logos::CallError& e) {
                                         d.record(i, std::move(v), e);
                                     });
    }
    const bool arrived = waitFor(d, kCalls, 20000);
    settle(200);

    std::cout << "  no-Qt deferred completions: " << d.total.load() << "/"
              << kCalls << " worst=" << d.worst() << " code='" << d.code()
              << "'" << std::endl;

    ASSERT_TRUE(arrived) << "only " << d.total.load() << " of " << kCalls
                         << " deferred completions were delivered";
    EXPECT_EQ(d.worst(), 1);
    EXPECT_EQ(d.missing(), 0);
    EXPECT_TRUE(d.code().empty());
    {
        std::lock_guard<std::mutex> g(d.mu);
        EXPECT_EQ(d.lastValue.toInt(), 7)
            << "the deferred call delivered the sentinel, not the completion";
    }

    obj->release();
    settle(50);
}

// ── 4. the deadline ─────────────────────────────────────────────────────────
//
// Reaches the hop from the DeadlineService thread — the third distinct
// resolver. A dropped timeout is the worst of the four: the call is never going
// to be answered, so the caller waits forever on a callback that was the only
// thing that could have told it so.
TEST_F(NoQtLoopTest, TimeoutsDeliverWithNoQtLoop)
{
    QtFreeHost host;
    ASSERT_TRUE(host.ok());
    auto conn = connectTo(host.port());
    ASSERT_NE(conn, nullptr);

    LogosObject* obj = conn->requestObject(QStringLiteral("omni"), 5000);
    ASSERT_NE(obj, nullptr);
    auto* ch = channelFor(obj);
    ASSERT_NE(ch, nullptr);

    constexpr int kCalls = 20;
    Deliveries d(kCalls);
    for (int i = 0; i < kCalls; ++i) {
        ch->callMethodAsyncWithError(kToken, QStringLiteral("sink"), {}, 200,
                                     [&d, i](QVariant v, const logos::CallError& e) {
                                         d.record(i, std::move(v), e);
                                     });
    }
    const bool arrived = waitFor(d, kCalls, 10000);
    settle(200);

    std::cout << "  no-Qt timeouts: " << d.total.load() << "/" << kCalls
              << " worst=" << d.worst() << " code='" << d.code() << "'"
              << std::endl;

    ASSERT_TRUE(arrived) << "only " << d.total.load() << " of " << kCalls
                         << " deadlines were delivered";
    EXPECT_EQ(d.worst(), 1);
    EXPECT_EQ(d.missing(), 0);
    EXPECT_EQ(d.code(), "timeout");

    obj->release();
    settle(50);
}

// ── 5. cancellation by release() ────────────────────────────────────────────
//
// The fourth resolver: teardown, on the caller's own thread. This one is the
// reason the fix cannot be "call it inline when there is no Qt loop" — inline
// here means running user code from inside release(), i.e. from inside a
// destructor path, which is exactly the re-entrancy that has already produced a
// SIGSEGV in this codebase on the QtRO twin.
TEST_F(NoQtLoopTest, CancelledCallsDeliverWithNoQtLoopAndNotInsideRelease)
{
    QtFreeHost host;
    ASSERT_TRUE(host.ok());
    auto conn = connectTo(host.port());
    ASSERT_NE(conn, nullptr);

    LogosObject* obj = conn->requestObject(QStringLiteral("omni"), 5000);
    ASSERT_NE(obj, nullptr);
    auto* ch = channelFor(obj);
    ASSERT_NE(ch, nullptr);

    constexpr int kCalls = 20;
    Deliveries d(kCalls);
    const std::thread::id caller = std::this_thread::get_id();
    d.forbiddenThread = caller;
    for (int i = 0; i < kCalls; ++i) {
        IssuingScope inCall;
        ch->callMethodAsyncWithError(kToken, QStringLiteral("never"), {}, 30000,
                                     [&d, i](QVariant v, const logos::CallError& e) {
                                         d.record(i, std::move(v), e);
                                     });
    }
    settle(300);   // let every sentinel come back, so the calls are parked
    ASSERT_EQ(d.total.load(), 0) << "the calls resolved before the release";

    {
        // The same nesting marker, around release() this time: a cancellation
        // callback that ran from inside release() would run inside this scope,
        // on this thread, and would count itself.
        IssuingScope inRelease;
        obj->release();
    }

    const bool arrived = waitFor(d, kCalls, 10000);
    settle(200);

    std::cout << "  no-Qt cancellations: " << d.total.load() << "/" << kCalls
              << " worst=" << d.worst() << " code='" << d.code()
              << "' inside-release=" << d.inlineDeliveries.load()
              << " on-releasing-thread=" << d.onForbiddenThread.load()
              << std::endl;

    ASSERT_TRUE(arrived) << "only " << d.total.load() << " of " << kCalls
                         << " cancellations were delivered";
    // "How many had arrived by the time release() returned" is NOT the inline
    // check, and measuring it that way is how this test first went red under
    // Guard Malloc: with everything slowed down, the delivery thread finished
    // all twenty before the releasing thread executed its next statement, which
    // is correct behaviour and reads as a violation. The claim is about NESTING,
    // and the marker above measures exactly that — plus the thread, since the
    // releasing thread is the forbidden one here.
    EXPECT_EQ(d.inlineDeliveries.load(), 0)
        << "a cancellation callback ran from inside release()";
    EXPECT_EQ(d.worst(), 1);
    EXPECT_EQ(d.missing(), 0);
    EXPECT_EQ(d.code(), "transport_error");
    EXPECT_EQ(d.onForbiddenThread.load(), 0)
        << "a cancellation callback ran on the releasing thread";
}

// ── 6. exactly once, with the delivery thread as the hop ────────────────────
//
// The release-racing-replies shape from test_iofold.cpp — the only one that
// actually detects a broken exactly-once gate — re-run here so that the gate is
// pinned against the OTHER delivery vehicle too. Teardown snapshots the
// in-flight map and then delivers one call at a time with the lock released,
// so the io thread has a window N deliveries wide in which to answer a call
// teardown has already claimed.
//
// VALIDATED AS A DETECTOR HERE, and the validation turned up something about the
// mechanism that the note in tests/protocol/CMakeLists.txt does not say. There
// are TWO gates in AsyncCall, not one: claim()'s compare-exchange, and the
// swap in takeCallback() which leaves a second caller holding an empty
// std::function. Removing the CAS alone changes nothing measurable — this test,
// its Qt twin and PlainCancelPendingRaceTest all stay green, 0 doubles — because
// the swap still absorbs the duplicate. With BOTH removed this test reports 22
// doubled deliveries in 20 rounds x 500 calls, while the three PER-PATH tests
// above stay green, which is the difference between a detector and a pin.
//
// So: a validation that removes one of the two gates proves nothing, and the
// exactly-once guarantee is stronger than the CAS on its own. Neither half is
// redundant — the CAS is what stops the second caller from also erasing
// registries and cancelling timers — but the CALLBACK is protected by the swap.
TEST_F(NoQtLoopTest, ReleaseRacingRepliesDeliversEachCallOnceWithNoQtLoop)
{
    QtFreeHost host;
    ASSERT_TRUE(host.ok());
    auto conn = connectTo(host.port());
    ASSERT_NE(conn, nullptr);

    constexpr int kRounds = 20;
    constexpr int kCalls  = 500;
    int doubled = 0;
    int dropped = 0;
    int byReply = 0;
    int byTeardown = 0;

    for (int r = 0; r < kRounds; ++r) {
        LogosObject* obj = conn->requestObject(QStringLiteral("omni"), 5000);
        ASSERT_NE(obj, nullptr);
        auto* ch = channelFor(obj);
        ASSERT_NE(ch, nullptr);

        auto d = std::make_shared<Deliveries>(kCalls);
        auto codes = std::make_shared<std::vector<std::atomic<int>>>(2);
        for (int i = 0; i < kCalls; ++i) {
            ch->callMethodAsyncWithError(kToken, QStringLiteral("ping"),
                                         QVariantList{ QVariant(i) }, 20000,
                                         [d, codes, i](QVariant v, const logos::CallError& e) {
                                             (*codes)[e.code.empty() ? 0 : 1].fetch_add(1);
                                             d->record(i, std::move(v), e);
                                         });
        }
        // WAIT FOR THE FIRST REPLY, then release — and this is not a tidy-up,
        // it is what makes the test a race at all.
        //
        // In this binary the provider and the consumer share the process's ONE
        // io thread (the QtFreeHandler runs on the same strand that writes the
        // client's frames), unlike the Qt twin in test_iofold.cpp whose provider
        // sits on its own QThread. So the issuing thread can enqueue all 500
        // calls and release before the io thread has drained a single one: on a
        // slow sandbox that produced answered-by-reply=0,
        // cancelled-by-teardown=10000 — one resolver, no race, and the
        // exactly-once assertions below reduced to decoration. (It still
        // reported 0 doubles and 0 drops, which is exactly why the
        // "both resolvers were live" guards further down have to exist.)
        //
        // Waiting for one delivery proves the reply path is running; with 500
        // calls in the burst, hundreds are still outstanding for teardown to
        // cancel. The jittered nudge then sweeps where in the burst it lands.
        ASSERT_TRUE(waitFor(*d, 1, 20000))
            << "round " << r << ": no reply came back at all before the release";
        std::this_thread::sleep_for(std::chrono::microseconds((r % 6) * 120));
        obj->release();

        // Per round rather than in aggregate, so a run against a tree that
        // drops callbacks stops on the first round instead of waiting out
        // twenty budgets.
        ASSERT_TRUE(waitFor(*d, kCalls, 20000))
            << "round " << r << ": only " << d->total.load() << " of " << kCalls
            << " callbacks were delivered";
        settle(50);

        for (const auto& c : d->counts) {
            if (c.load() > 1) doubled += c.load() - 1;
            if (c.load() == 0) ++dropped;
        }
        byReply    += (*codes)[0].load();
        byTeardown += (*codes)[1].load();
    }

    std::cout << "  no-Qt: " << kRounds << " rounds x " << kCalls
              << " calls released mid-burst -> answered-by-reply=" << byReply
              << " cancelled-by-teardown=" << byTeardown
              << " DOUBLE deliveries=" << doubled << " dropped=" << dropped
              << std::endl;

    EXPECT_GT(byReply, 0)    << "no call was answered by its reply";
    EXPECT_GT(byTeardown, 0) << "no call was cancelled by teardown — the race "
                                "did not happen";
    EXPECT_EQ(doubled, 0) << doubled << " calls were delivered more than once";
    EXPECT_EQ(dropped, 0) << dropped << " calls were never delivered at all";
}

// ── 7. the delivery vehicle outlives static destruction ─────────────────────
//
// The visible half of the post-main probe above: this test WARMS THE VEHICLE UP
// — constructing it here, i.e. after all dynamic initialisation, is what puts it
// on the wrong side of the destruction order from the probe — and states where
// the verdict will appear. The assertion itself cannot live in a test case,
// because the moment it is about is after main() returns; it lives in
// LateDeliveryProbe::~LateDeliveryProbe and reports through the process exit
// code. See the note over that struct.
//
// On the tree this was written against (the delivery service as an ordinary
// function-local static) the run ends in `[post-main] delivered=0` followed by
// exit 70, or SIGSEGV under Guard Malloc. Both are what a failure looks like
// here, and both were observed before the fix.
TEST_F(NoQtLoopTest, TheVehicleOutlivesStaticDestructors)
{
    ASSERT_TRUE(g_lateProbeArmed.load())
        << "the post-main probe was not constructed during dynamic "
           "initialisation, so nothing will check the after-main() window";

    PlainLogosObject obj("nobody", nullptr);
    Deliveries d(1);
    {
        IssuingScope inCall;
        obj.callMethodAsyncWithError(kToken, QStringLiteral("ping"), {}, 1000,
                                     [&d](QVariant v, const logos::CallError& e) {
                                         d.record(0, std::move(v), e);
                                     });
    }
    ASSERT_TRUE(waitFor(d, 1, 5000))
        << "the warm-up delivery never arrived, so the vehicle was never "
           "constructed and the probe below tests nothing";
    EXPECT_EQ(d.inlineDeliveries.load(), 0);

    std::cout << "  delivery vehicle constructed inside the run; the after-main "
                 "delivery is checked by LateDeliveryProbe (exit code 70 on "
                 "failure)" << std::endl;
}