Per-module concurrent dispatch: async provider seam + transports (#5)

* feat: per-module concurrent dispatch (concurrency:"multi") — zero ABI change

A "multi" module serves calls concurrently behind the ORDINARY callMethod — no
new provider/host vtable method, so LogosProviderObject's ABI is byte-identical
to before and an old host/daemon loads and forwards a multi module unmodified.

Mechanism: a multi module's generated glue returns a pending sentinel
({"__logos_pending_call__": callId}) from callMethod and pushes the real result
back later as a __logos_call_complete__ event keyed by callId, over the existing
event channel. The consumer transport detects the sentinel and awaits the
completion transparently, so generated clients are unchanged.

- logos_async_dispatch.h: shared wire constants + the contract.
- remote_transport.cpp (QtRO) / plain_logos_object.{h,cpp} (plain): consumer
  sentinel detection + await keyed by callId. The host is a pure forwarder.
- logos_protocol.h + nix/default.nix: protocol 0.2.0 (additive minor; same MAJOR
  stays compatible, so an old host accepts a 0.2 "multi" module).
- rpc_server.cpp: fix a teardown self-deadlock (stop() held m_mu while invoking a
  per-connection error handler that re-locks m_mu) that the new in-process
  subscription path exposed.
- tests/protocol/test_concurrent_dispatch.cpp: proves a multi provider overlaps
  two concurrent calls (peak 2) while single serializes (peak 1), over the plain
  transport, with the host unchanged from master.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix: coalesce concurrent async requestModule handshakes (+ async fan-out test)

A driver that fans out N async calls to an un-tokened target before any
completes used to fire N separate requestModule handshakes. Each mints a
distinct capability token and informs the target, and the later inform
OVERWRITES the earlier token there (the target stores one token per caller),
so the already-dispatched calls carried a superseded token and the target
rejected them as unauthorized ("auth token not recognized"). The sync path
never hit this — it blocks per call, so handshakes never overlap.

Coalesce in LogosAPIClient::invokeRemoteMethodAsync: the first async call to
an un-tokened target starts ONE handshake; concurrent calls to the same
target queue behind it and all drain with the single minted token when it
resolves. m_pendingHandshakes is touched only on the owner thread, so no lock
(appended last per the class's ABI note). This is what lets a concurrency:
"multi" worker actually run a single-threaded driver's fan-out concurrently —
otherwise the fanned-out calls are rejected before reaching dispatch.

Also add MultiProviderOverlapsAsync / SingleProviderSerializesAsync to the
concurrent-dispatch gtest: they fire N concurrent callMethodAsync() calls (the
fan-out pattern over the async consumer path, which the sync tests don't
exercise) and assert peak overlap 4 for "multi", 1 for "single".

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

cpp/implementations/plain/plain_logos_object.cpp

@@ -1,11 +1,13 @@
 #include "plain_logos_object.h"
 
+#include "logos_async_dispatch.h"
 #include "qvariant_rpc_value.h"
 
 #include <QCoreApplication>
 #include <QDebug>
 #include <QMetaObject>
 #include <QTimer>
+#include <QVariantMap>
 
 #include <chrono>
 #include <future>
@@ -33,6 +35,10 @@ QVariant PlainLogosObject::callMethod(const QString& authToken,
 {
     if (!m_conn || !m_conn->isOpen()) return QVariant();
 
+    // Subscribe to the completion channel BEFORE sending, so a "multi" provider's
+    // completion can't race ahead of the waiter (it's buffered either way).
+    ensureCompletionSub();
+
     CallMessage msg;
     msg.id        = m_conn->nextId();
     msg.authToken = authToken.toStdString();
@@ -52,7 +58,52 @@ QVariant PlainLogosObject::callMethod(const QString& authToken,
                    << "failed:" << QString::fromStdString(res.err);
         return QVariant();
     }
-    return rpcValueToQVariant(res.value);
+    const QVariant value = rpcValueToQVariant(res.value);
+    // A "multi" provider may have deferred: it returned a pending sentinel and
+    // pushes the real result as a completion event. Wait for it, keyed by callId.
+    if (value.typeId() == QMetaType::QVariantMap) {
+        const QVariantMap m = value.toMap();
+        if (m.contains(logos::pendingCallKey()))
+            return awaitCompletion(m.value(logos::pendingCallKey()).toString(), timeoutMs);
+    }
+    return value;
+}
+
+void PlainLogosObject::ensureCompletionSub()
+{
+    {
+        std::lock_guard<std::mutex> g(m_completionMu);
+        if (m_completionSubscribed) return;
+        m_completionSubscribed = true;
+    }
+    // Reuse the normal event subscription path (tracked in m_subs, so
+    // disconnectEvents() tears it down). The handler fires on the connection's
+    // IO thread; it buffers the result and wakes any waiter.
+    onEvent(logos::callCompleteEvent(), [this](const QString&, const QVariantList& data) {
+        if (data.size() != 2) return;
+        const QString callId = data.at(0).toString();
+        {
+            std::lock_guard<std::mutex> g(m_completionMu);
+            m_completions[callId] = data.at(1);
+        }
+        m_completionCv.notify_all();
+    });
+}
+
+QVariant PlainLogosObject::awaitCompletion(const QString& callId, int timeoutMs)
+{
+    std::unique_lock<std::mutex> lk(m_completionMu);
+    const auto deadline = std::chrono::steady_clock::now()
+        + std::chrono::milliseconds(timeoutMs > 0 ? timeoutMs : 30000);
+    const bool got = m_completionCv.wait_until(lk, deadline,
+        [&] { return m_completions.count(callId) > 0; });
+    if (!got) {
+        qWarning() << "PlainLogosObject: deferred call" << callId << "timed out";
+        return QVariant();
+    }
+    const QVariant result = m_completions[callId];
+    m_completions.erase(callId);
+    return result;
 }
 
 namespace {
@@ -95,6 +146,8 @@ void PlainLogosObject::callMethodAsync(const QString& authToken,
         return;
     }
 
+    ensureCompletionSub();
+
     CallMessage msg;
     msg.id        = m_conn->nextId();
     msg.authToken = authToken.toStdString();
@@ -110,7 +163,7 @@ void PlainLogosObject::callMethodAsync(const QString& authToken,
     // future iteration can fold this wait into the shared Asio
     // io_context (the connection already runs on it) so we don't spin
     // up a thread per pending RPC.
-    std::thread([fut, timeoutMs, callback = std::move(callback)]() mutable {
+    std::thread([this, fut, timeoutMs, callback = std::move(callback)]() mutable {
         if (fut->wait_for(std::chrono::milliseconds(timeoutMs))
             != std::future_status::ready) {
             postToQtEventLoop(std::move(callback), QVariant());
@@ -118,6 +171,13 @@ void PlainLogosObject::callMethodAsync(const QString& authToken,
         }
         auto res = fut->get();
         QVariant value = res.ok ? rpcValueToQVariant(res.value) : QVariant();
+        // Resolve a "multi" provider's deferred completion (sentinel → wait for
+        // the completion event) right here on the waiter thread.
+        if (value.typeId() == QMetaType::QVariantMap) {
+            const QVariantMap m = value.toMap();
+            if (m.contains(logos::pendingCallKey()))
+                value = awaitCompletion(m.value(logos::pendingCallKey()).toString(), timeoutMs);
+        }
         postToQtEventLoop(std::move(callback), std::move(value));
     }).detach();
 }