[Feat] (MG_Remote): P9 W3 - the server's log forward carries each line's level and never blocks a logging thread (<=WARN dropped and counted, >=ERROR lossless and paced, FATAL a bounded flush), fault injection F2, and the producerless OnLog callback deleted (9 -> 8)

CMakeLists.txt

@@ -607,6 +607,9 @@ if (MOBILEGL_BUILD_DISAGGREGATED)
         MobileGL/MG_Remote/Transport/InProcessTransport.cpp
         MobileGL/MG_Remote/Transport/SocketTransport.cpp
         MobileGL/MG_Remote/Transport/ControlInbox.cpp
+        # P9 W3: the server's log forward - a queue and a sender thread, graded by severity
+        # (<= WARN lossy, >= ERROR lossless and paced), so no logging thread writes the socket.
+        MobileGL/MG_Remote/Transport/LogForward.cpp
         # Keeps MG_Util/Debug/Log.h - and through it the GL frontend's
         # umbrella header - out of the header-only wire code (WireLog.h).
         MobileGL/MG_Remote/Transport/WireLog.cpp

MobileGL/MG_Pipe/MGPipeCallbacks.h

@@ -17,10 +17,10 @@
 // Today this traffic is 95 call sites across 17 methods poked directly into frontend
 // objects. gallium has no vocabulary for shadow writeback, GPU-write notification, texture
 // re-send requests or default-framebuffer geometry, because in Mesa the state tracker and
-// the driver share an address space. Naming them as nine callbacks plus one forward
+// the driver share an address space. Naming them as eight callbacks plus one forward
 // terminator (MGPipeContext::ResourceSubDataComplete) is the deliberate deviation (D8).
 //
-// NINE, NOT THE TEN PLAN B WROTE. The tenth was OnXfbScatterReady, and it went with the
+// EIGHT, NOT THE TEN PLAN B WROTE. The tenth was OnXfbScatterReady, and it went with the
 // design it belonged to: plan B put the XFB scatter on the CLIENT (the server would hand
 // back the packed scratch and the client would run the patch loop over its own shadow), so
 // there had to be a callback that told the client the layout. P5c/P5f went the other way -
@@ -31,6 +31,13 @@
 // reader of this file has to rule out by hand, and the size assertion below made it look
 // load-bearing.
 //
+// The ninth was OnLog, deleted in P9 W3 for the same reason: zero producers, zero consumers,
+// no EventKind. Plan B pictured the server's log lines as reverse-channel records; they never
+// were. They travel the CONTROL plane as LogLine frames (protocol.fbs), written by the server's
+// logger through its log forward (MG_Remote/Transport/LogForward.h) - which is also where the
+// severity policy this entry's comment promised ("<= WARN lossy, >= ERROR lossless and rate
+// limited") now lives, on the level LogLine has always carried.
+//
 // Installed at context creation. In a monolith these are direct calls; under split they are
 // records on the reverse channel, and their ORDER is a correctness requirement rather than
 // an optimization (section 7.4).
@@ -55,13 +62,11 @@ namespace MobileGL::MG_Pipe {
         // pDefaultFramebufferInfo.
         void (*OnSurfaceChanged)(const MGPSurfaceInfo* info);
         void (*OnCapsInvalidated)();
-        // <= WARN is lossy, >= ERROR is lossless and rate limited.
-        void (*OnLog)(Uint8 level, const char* text);
     };
 
-    // Nine, and the count is asserted so a tenth cannot be added without touching the
+    // Eight, and the count is asserted so a ninth cannot be added without touching the
     // transport's reverse-channel record table.
-    inline constexpr SizeT kMGPipeCallbackCount = 9;
+    inline constexpr SizeT kMGPipeCallbackCount = 8;
     static_assert(sizeof(MGPipeCallbacks) == kMGPipeCallbackCount * sizeof(void (*)()),
                   "MGPipeCallbacks gained or lost a callback");
 

MobileGL/MG_Remote/Client/ClientSession.h

@@ -99,6 +99,15 @@ namespace MobileGL::MG_Remote::Client {
         // m_remoteControlMutex and drains first; a caller of this does neither, which is the
         // point. Null unless spawned.
         Transport::SocketTransport* ControlSocketForTest() { return m_socketTransport.get(); }
+        // P9 W3's fault injection F2 ONLY (EventForfeitPeerTest): the control READER, so a TCP peer
+        // can stop taking anything off its control connection - the server's forwarded log lines
+        // included - and later resume (ControlInbox::PauseReadingForTest). False when there is no
+        // reader to pause (not a stream session).
+        bool PauseControlReaderForTest(bool paused) {
+            if (!m_controlInbox) return false;
+            m_controlInbox->PauseReadingForTest(paused);
+            return true;
+        }
 
         // CONTRACT-P6 4.3: THE PID THE SERVER STATED IN ITS Welcome, which is not the same fact
         // as the pid this session's launcher recorded - that one is local knowledge, this one

MobileGL/MG_Remote/Server/ServerLoop.cpp

@@ -162,6 +162,44 @@ namespace MobileGL::MG_Remote::Server {
 #endif
         }
 
+        // P9 W3, FAULT INJECTION F2 (HANDOFF-P9 §4): THE APPLY THREAD AS A LOG FIREHOSE.
+        //
+        // MOBILEGL_TEST_APPLY_LOG_FLOOD=<n> makes the apply thread log, after EVERY record it applies,
+        // `n` WARN lines of about 600 bytes and one ERROR line, each numbered in its own series
+        // ("flood warn #k", "flood error #k"). It is the one producer F2 needs and nothing in the
+        // tree otherwise is: production logging is not allowed to repeat on a per-record path
+        // (Log.h), so no real verb can be made to fill a TCP control connection with lines at will -
+        // and "the apply thread is not held by logging while its client does not read" can only be
+        // asserted about a thread that is logging a lot. The numbering is what lets the test prove
+        // every ERROR line arrived (in order) while WARN lines were dropped.
+        //
+        // Meant for a server in its OWN process (a `--serve` session child). A TEST knob: read once
+        // per Start(), 0 unless set, capped, and it says so in the log when it is set.
+        Uint32 ApplyLogFloodLinesForTest() {
+            const char* text = std::getenv("MOBILEGL_TEST_APPLY_LOG_FLOOD");
+            if (text == nullptr || *text == '\0') return 0;
+            const long lines = std::strtol(text, nullptr, 10);
+            if (lines <= 0) return 0;
+            const Uint32 capped = static_cast<Uint32>(std::min<long>(lines, 4096));
+            MGLOG_W("MG_Remote server: MOBILEGL_TEST_APPLY_LOG_FLOOD=%u - every applied record logs that many "
+                    "WARN lines and one ERROR line. A test-only lever (fault injection F2); never set it in a "
+                    "measured run",
+                    static_cast<unsigned>(capped));
+            return capped;
+        }
+
+        void FloodLogForTest(Uint32 warnLines) {
+            // One apply thread per process (ServerLoopInstance's), so plain counters are enough.
+            static unsigned long long s_warn = 0;
+            static unsigned long long s_error = 0;
+            static const std::string kFiller(560, 'w');
+            for (Uint32 i = 0; i < warnLines; ++i) {
+                MGLOG_W("MG_Remote server: MOBILEGL_TEST_APPLY_LOG_FLOOD flood warn #%llu %s", ++s_warn,
+                        kFiller.c_str());
+            }
+            MGLOG_E("MG_Remote server: MOBILEGL_TEST_APPLY_LOG_FLOOD flood error #%llu", ++s_error);
+        }
+
     } // namespace
 
     // ---------------------------------------------------------------------------------
@@ -236,6 +274,7 @@ namespace MobileGL::MG_Remote::Server {
         m_serverOwnedSurfaces.clear();
         m_windowLostRequested.store(false, std::memory_order_release);
         m_serverWindowsLost.store(0, std::memory_order_release);
+        m_testLogFloodLines = ApplyLogFloodLinesForTest();
         {
             const std::lock_guard<std::mutex> lock(m_exitMutex);
             m_exited = false;
@@ -783,6 +822,8 @@ namespace MobileGL::MG_Remote::Server {
                             std::chrono::steady_clock::now() - applyStarted)
                             .count());
                     ++sFrameRecords;
+                    // P9 W3, F2: outside the apply timing above; 0 in every run that is not F2.
+                    if (m_testLogFloodLines != 0) FloodLogForTest(m_testLogFloodLines);
                     if (record.kind == static_cast<std::uint16_t>(MG_Pipe::MGPWireOp::Present)) {
                         const auto now = std::chrono::steady_clock::now();
                         const auto wallNs = static_cast<std::uint64_t>(

MobileGL/MG_Remote/Server/ServerLoop.h

@@ -569,6 +569,9 @@ namespace MobileGL::MG_Remote::Server {
         std::atomic<Uint64> m_parkBlocks{0};
         std::atomic<void (*)()> m_beforeRetireHook{nullptr};
         std::atomic<void (*)()> m_betweenRecordsHook{nullptr};
+        // P9 W3, fault injection F2: MOBILEGL_TEST_APPLY_LOG_FLOOD, read by Start(). Zero - the only
+        // value a production run has - makes the per-record test a load of this and nothing else.
+        Uint32 m_testLogFloodLines = 0;
 
         // C7 / ID-54: the (dpy, draw, read, ctx) currently bound on the apply thread. Written and
         // read ONLY on the apply thread inside ApplyMakeCurrent, so it needs no lock; the two

MobileGL/MG_Remote/Server/ServerMain.cpp

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: LGPL-3.0-only
 #include "../Transport/Doorbell.h"
 #include "../Transport/FdPassing.h"
+#include "../Transport/LogForward.h"
 #include "../Transport/SocketTransport.h"
 #include "../Transport/WireLog.h"
 #include "../Protocol/SurfaceOpCodec.h"
@@ -55,13 +56,42 @@ void Refuse(SocketTransport& transport, Protocol::RefuseCode code, const char* d
     WireLogError("MG_Remote server: Refuse{%s} %s", Protocol::EnumNameRefuseCode(code), detail);
 }
 
-void ForwardLog(void* user, const char* text) {
-    auto& transport = *static_cast<ITransport*>(user);
-    flatbuffers::FlatBufferBuilder builder(512);
-    auto log = Protocol::CreateLogLine(builder, Protocol::LogLevel::Info, builder.CreateString(text));
-    auto envelope = Protocol::CreateCtrlEnvelope(builder, Protocol::CtrlMsg::LogLine, log.Union());
-    Protocol::FinishCtrlEnvelopeBuffer(builder, envelope);
-    (void)transport.SendFrame({builder.GetBufferPointer(), builder.GetSize()});
+// P9 W3. The log forward's unblock hook (LogForward.h, Close): the channel's sender is still inside a
+// write to a client that has not read for the whole close budget, at the end of the session. Cutting
+// the control socket is what returns that write (EPIPE) so the thread can be joined - the session is
+// over by then (every caller is on RunSession's way out), and the socket is closed right after anyway.
+// Not through SocketTransport::Shutdown: that takes the send mutex the blocked write holds.
+void UnblockControlSend(void* user) {
+    const int fd = static_cast<SocketTransport*>(user)->StreamFd();
+    if (fd >= 0) (void)::shutdown(fd, SHUT_RDWR);
+}
+
+// The forward's counters, once, into the server's OWN log - the forwarder is already gone by the time
+// they are final, so this line is local; what the client saw of the same story is the in-stream
+// LogForward{Dropped} / {Coalesced} lines at each gap. WARN when anything was lost, delayed past a
+// deadline or cut, so it survives a production build; otherwise a D line.
+void LogForwardSummary(const Transport::LogForwardCounters& counters) {
+    const bool eventful = counters.lossyDropped != 0 || counters.errorsCoalesced != 0 || counters.abandoned != 0 ||
+                          counters.fatalFlushTimeouts != 0 || counters.transportFailed;
+    if (eventful) {
+        MGLOG_W("MG_Remote server: log forwarding ended - forwarded=%llu dropped(<=WARN)=%llu (%llu bytes) "
+                "coalesced(ERROR)=%llu paced(ERROR)=%llu notices=%llu abandoned=%llu fatal-flush-timeouts=%llu "
+                "transport-failed=%d max-offer-us=%llu",
+                static_cast<unsigned long long>(counters.forwarded),
+                static_cast<unsigned long long>(counters.lossyDropped),
+                static_cast<unsigned long long>(counters.lossyDroppedBytes),
+                static_cast<unsigned long long>(counters.errorsCoalesced),
+                static_cast<unsigned long long>(counters.errorsPaced),
+                static_cast<unsigned long long>(counters.notices),
+                static_cast<unsigned long long>(counters.abandoned),
+                static_cast<unsigned long long>(counters.fatalFlushTimeouts), counters.transportFailed ? 1 : 0,
+                static_cast<unsigned long long>(counters.maxOfferNs / 1000));
+    } else {
+        MGLOG_D("MG_Remote server: log forwarding ended - forwarded=%llu paced(ERROR)=%llu max-offer-us=%llu",
+                static_cast<unsigned long long>(counters.forwarded),
+                static_cast<unsigned long long>(counters.errorsPaced),
+                static_cast<unsigned long long>(counters.maxOfferNs / 1000));
+    }
 }
 
 // p7/spawnhang. THE PROGRESS REPORT: Wire::SurfaceProgress, sent by the thread that posted the op
@@ -192,13 +222,22 @@ Server::ServerSession::DataConnectionSource ListenerSource(int listener, int con
     };
 }
 
-struct FlushAck { ITransport* transport; std::uint64_t seq; };
+// `forward` is the session's log forward when there is one (P9 W3). The ack then goes out THROUGH it,
+// queued behind every line forwarded before the flush, because a forwarded line is no longer on the
+// wire when its MGLOG returns - sending the ack directly could overtake lines still queued, and
+// MGPipeSyncPeerLog's promise is that they are in the client's file when the ack arrives.
+struct FlushAck { ITransport* transport; Transport::LogForwardChannel* forward; std::uint64_t seq; };
 void SendLogAck(void* pointer) {
     auto& ack = *static_cast<FlushAck*>(pointer);
     flatbuffers::FlatBufferBuilder builder(64);
     auto flush = Protocol::CreateLogFlush(builder, ack.seq, true);
     auto envelope = Protocol::CreateCtrlEnvelope(builder, Protocol::CtrlMsg::LogFlush, flush.Union());
     Protocol::FinishCtrlEnvelopeBuffer(builder, envelope);
+    if (ack.forward != nullptr) {
+        ack.forward->OfferControlFrame(
+            std::vector<std::uint8_t>(builder.GetBufferPointer(), builder.GetBufferPointer() + builder.GetSize()));
+        return;
+    }
     (void)ack.transport->SendFrame({builder.GetBufferPointer(), builder.GetSize()});
 }
 
@@ -429,11 +468,25 @@ int RunSession(std::unique_ptr<SocketTransport> control, std::vector<std::uint8_
     // code, exactly as before.
     bool backendBuilt = false;
     bool acceptTried = false;
-    const auto endSession = [&](int code) -> int {
-        if (!inProcess) return code;
-        // Both point at `control`, which dies with this frame.
-        loop.SetControlProgressSink(nullptr, nullptr);
+    // P9 W3: the TCP session's log forward (LogForward.h), made where forwarding is switched on below.
+    // Declared up here because every way out has to take it down the same way: the logger stops
+    // calling it FIRST (so no line is offered to a channel being destroyed), then the channel drains
+    // within its close budget and its sender is joined, then its counters are said locally. Every
+    // shape, forked or not: a forked child's early return destroys the channel with this frame, and a
+    // forwarder left pointing at it would be a dangling pointer for however many lines remain.
+    std::unique_ptr<Transport::LogForwardChannel> logForward;
+    const auto stopLogForwarding = [&] {
         MobileGL::MG_Util::Debug::SetLogForwarder(nullptr, nullptr);
+        if (!logForward) return;
+        const auto counters = logForward->Close();
+        logForward.reset();
+        LogForwardSummary(counters);
+    };
+    const auto endSession = [&](int code) -> int {
+        stopLogForwarding();
+        if (!inProcess) return code;
+        // Points at `control`, which dies with this frame (the log forwarder did too, above).
+        loop.SetControlProgressSink(nullptr, nullptr);
         if (backendBuilt) {
             // Stop() joins a running apply thread (it destroys the backend on it), or - never
             // started - drops the backend here; either way the display lease goes with it.
@@ -575,11 +628,18 @@ int RunSession(std::unique_ptr<SocketTransport> control, std::vector<std::uint8_
 
     const char* forwarding = std::getenv("MOBILEGL_IPC_LOG_FORWARD");
     if (tcp && (!forwarding || std::strcmp(forwarding, "0") != 0)) {
+        // P9 W3: the forward is a queue with a sender thread of its own, graded by severity
+        // (LogForward.h): a client that stops reading control no longer holds any logging thread -
+        // the apply thread above all - inside a socket write.
+        logForward = std::make_unique<Transport::LogForwardChannel>(*control, &UnblockControlSend, control.get());
         // P12 review fix: in-process, only THIS session's threads (this one and its apply thread)
         // forward, and outside the log mutex - the UI thread, the listener and the next session's
         // refusals share this process and must neither reach this client nor wait on it.
-        if (inProcess) MobileGL::MG_Util::Debug::SetSessionLogForwarder(&ForwardLog, control.get());
-        else MobileGL::MG_Util::Debug::SetLogForwarder(&ForwardLog, control.get());
+        if (inProcess)
+            MobileGL::MG_Util::Debug::SetSessionLogForwarder(&Transport::LogForwardChannel::ForwardThunk,
+                                                             logForward.get());
+        else
+            MobileGL::MG_Util::Debug::SetLogForwarder(&Transport::LogForwardChannel::ForwardThunk, logForward.get());
     }
     // p7/spawnhang: this thread posts every surface op the client sends (ServerApplyWireSurfaceOp,
     // below) and is the one that says "still running" while the apply thread runs it. Cleared
@@ -700,7 +760,7 @@ int RunSession(std::unique_ptr<SocketTransport> control, std::vector<std::uint8_
                         static_cast<unsigned long long>(verbs.ExpectedApplierResetSerial()),
                         static_cast<unsigned long long>(verbs.ObjectDeaths()));
             }
-            FlushAck ack{control.get(), flush->seq()};
+            FlushAck ack{control.get(), logForward.get(), flush->seq()};
             MobileGL::MG_Util::Debug::WithLogBarrier(&SendLogAck, &ack);
             continue;
         }
@@ -749,7 +809,8 @@ int RunSession(std::unique_ptr<SocketTransport> control, std::vector<std::uint8_
                      static_cast<unsigned long long>(SessionLatchCount()), kSessionLatchedExitCode);
     }
     session.Close();
-    MobileGL::MG_Util::Debug::SetLogForwarder(nullptr, nullptr);
+    // P9 W3: the forwarder is cleared, the forward drains (bounded) and its counters are logged.
+    stopLogForwarding();
     // P12 (D5): the backend loop.Stop() destroyed is not this session's to point at any more.
     if (inProcess) session.SetBackend(nullptr);
     // The caller's exit (a forked child's ExitSessionProcess) closes control after cleanup/flush -

MobileGL/MG_Remote/Transport/ControlInbox.cpp

@@ -1,5 +1,6 @@
 #include "ControlInbox.h"
 #include "Doorbell.h"
+#include "LogForward.h"
 #include "../Protocol/SurfaceOpCodec.h"
 #include <MG_Util/Debug/Log.h>
 #include <chrono>
@@ -18,11 +19,21 @@ namespace MobileGL::MG_Remote::Transport {
         if (m_reader.joinable()) m_reader.join();
     }
 
+    void ControlInbox::PauseReadingForTest(bool paused) {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        m_pausedForTest = paused;
+        m_cv.notify_all();
+    }
+
     void ControlInbox::Read() {
         for (;;) {
             {
-                std::lock_guard<std::mutex> lock(m_mutex);
+                std::unique_lock<std::mutex> lock(m_mutex);
                 if (m_stop) return;
+                if (m_pausedForTest) {
+                    m_cv.wait(lock, [&] { return m_stop || !m_pausedForTest; });
+                    continue;
+                }
             }
             std::uint64_t size = 0;
             auto result = m_transport.ReceiveFrame({nullptr, 0}, &size, 100);
@@ -38,7 +49,9 @@ namespace MobileGL::MG_Remote::Transport {
                 else {
                     const auto* envelope = ::MobileGL::Wire::GetCtrlEnvelope(frame.data());
                     if (const auto* log = envelope->msg_as_LogLine()) {
-                        if (log->text()) MG_Util::Debug::WritePeerLog(log->text()->c_str());
+                        // P9 W3: filed with the severity the server sent (LogForward.h's mapping).
+                        if (log->text())
+                            MG_Util::Debug::WritePeerLog(LogLevelFromWire(log->level()), log->text()->c_str());
                         continue;
                     }
                 }

MobileGL/MG_Remote/Transport/ControlInbox.h

@@ -18,6 +18,11 @@ namespace MobileGL::MG_Remote::Transport {
         MobileGLResult Receive(std::vector<std::uint8_t>& frame, std::uint32_t timeoutMs);
         std::uint64_t Peek();
         bool WaitClosed(std::uint32_t timeoutMs);
+        // TEST-ONLY (P9 W3, fault injection F2). While paused the reader takes NOTHING off the
+        // socket - not a LogLine, not a reply - which is exactly a client that stopped reading its
+        // control connection, the peer the server's log forward (LogForward.h) must not be held by.
+        // Frames already queued stay queued; resuming picks up where the socket is.
+        void PauseReadingForTest(bool paused);
     private:
         void Read();
         ITransport& m_transport;
@@ -26,6 +31,7 @@ namespace MobileGL::MG_Remote::Transport {
         std::deque<std::vector<std::uint8_t>> m_frames;
         std::uint64_t m_bytes = 0;
         bool m_stop = false;
+        bool m_pausedForTest = false;
         MobileGLResult m_result = MOBILEGL_OK;
         std::thread m_reader;
     };

MobileGL/MG_Remote/Transport/LogForward.cpp

@@ -0,0 +1,343 @@
+// MobileGL - MobileGL/MG_Remote/Transport/LogForward.cpp
+// Copyright (c) 2025-2026 MobileGL-Dev
+// Licensed under the GNU Lesser General Public License v3.0:
+//   https://www.gnu.org/licenses/gpl-3.0.txt
+//   https://www.gnu.org/licenses/lgpl-3.0.txt
+// SPDX-License-Identifier: LGPL-3.0-only
+// End of Source File Header
+
+// P9 W3: the graded, non-blocking log forward. The policy and its reasons are in LogForward.h.
+
+#include "LogForward.h"
+
+#include "../Protocol/generated/protocol_generated.h"
+
+// The umbrella, for MOBILEGL_LOG_LEVEL_* and GetCurrentTime - a .cpp may pull it (WireLog.h).
+#include <MG_Util/Debug/Log.h>
+
+#include <algorithm>
+#include <cstdio>
+#include <cstring>
+
+#if defined(__linux__)
+#include <pthread.h>
+#endif
+
+namespace MobileGL::MG_Remote::Transport {
+
+    namespace Wire = ::MobileGL::Wire;
+
+    // THE TWO SCALES ARE ONE SCALE, and these are what would notice if either moved: a sixth
+    // MobileGL level, a reordered wire enum, or Log.h's 2026-08-13 inversion coming back.
+    static_assert(MOBILEGL_LOG_LEVEL_DEBUG == 0 && MOBILEGL_LOG_LEVEL_INFO == 1 && MOBILEGL_LOG_LEVEL_WARN == 2 &&
+                      MOBILEGL_LOG_LEVEL_ERROR == 3 && MOBILEGL_LOG_LEVEL_FATAL == 4,
+                  "MobileGL's log levels moved; re-read WireLogLevelFor / LogLevelFromWire");
+    static_assert(static_cast<int>(Wire::LogLevel::Debug) == 0 && static_cast<int>(Wire::LogLevel::Info) == 1 &&
+                      static_cast<int>(Wire::LogLevel::Warn) == 2 && static_cast<int>(Wire::LogLevel::Error) == 3 &&
+                      static_cast<int>(Wire::LogLevel::Fatal) == 4,
+                  "protocol.fbs's LogLevel moved; re-read WireLogLevelFor / LogLevelFromWire");
+
+    Wire::LogLevel WireLogLevelFor(int level) {
+        if (level <= MOBILEGL_LOG_LEVEL_DEBUG) return Wire::LogLevel::Debug;
+        switch (level) {
+        case MOBILEGL_LOG_LEVEL_INFO: return Wire::LogLevel::Info;
+        case MOBILEGL_LOG_LEVEL_WARN: return Wire::LogLevel::Warn;
+        case MOBILEGL_LOG_LEVEL_ERROR: return Wire::LogLevel::Error;
+        default: return Wire::LogLevel::Fatal;
+        }
+    }
+
+    int LogLevelFromWire(Wire::LogLevel level) {
+        switch (level) {
+        case Wire::LogLevel::Debug: return MOBILEGL_LOG_LEVEL_DEBUG;
+        case Wire::LogLevel::Info: return MOBILEGL_LOG_LEVEL_INFO;
+        case Wire::LogLevel::Warn: return MOBILEGL_LOG_LEVEL_WARN;
+        case Wire::LogLevel::Error: return MOBILEGL_LOG_LEVEL_ERROR;
+        case Wire::LogLevel::Fatal: return MOBILEGL_LOG_LEVEL_FATAL;
+        }
+        // A value this build's schema does not name: a newer peer's severity. Never demoted into
+        // the lossy class - an unknown severity is filed as an error.
+        return MOBILEGL_LOG_LEVEL_ERROR;
+    }
+
+    namespace {
+        // The header a notice line carries: the shape of Log()'s own ("[time] [OS thread/LEVEL]: "),
+        // so it reads like any other forwarded line and starts with '[' (WritePeerLog writes it as it
+        // came). The "OS" slot names the layer instead: GetOSName is not callable outside Log.cpp.
+        std::string NoticeHeader(int level) {
+            return "[" + MG_Util::Debug::GetCurrentTime() + "] [MG_Remote mgl-log-fwd/" +
+                   MG_Util::Debug::LogLevelTag(level) + "]: ";
+        }
+    } // namespace
+
+    LogForwardChannel::LogForwardChannel(ITransport& transport, Unblock unblock, void* unblockUser,
+                                         const LogForwardLimits& limits)
+        : m_transport(transport), m_unblock(unblock), m_unblockUser(unblockUser), m_limits(limits),
+          m_tokens(static_cast<double>(limits.errorBurstLines)), m_tokensAt(std::chrono::steady_clock::now()) {
+        m_sender = std::thread([this] { Run(); });
+    }
+
+    LogForwardChannel::~LogForwardChannel() { (void)Close(); }
+
+    void LogForwardChannel::ForwardThunk(void* user, int level, const char* line) {
+        static_cast<LogForwardChannel*>(user)->Offer(level, line);
+    }
+
+    bool LogForwardChannel::IsUrgentLocked() const { return m_urgent > 0 || m_closing; }
+
+    void LogForwardChannel::QueueGapNoticesLocked(bool lossyGapToo) {
+        // ONE NOTICE PER GAP, however many lines it swallowed, so a notice never needs room of its
+        // own worth refusing and the number of notices is bounded by the number of lines that did
+        // get in. WHERE each class's gap is closed differs, on purpose:
+        //   * an ERROR gap is marked before the very next item queued - the next ERROR line that
+        //     fits, a LogFlush ack, Close - because an error gap is rare and where it sits matters;
+        //   * a <= WARN gap is marked when the lossy class RECOVERS (the next line at or below WARN
+        //     that fits), or at a LogFlush ack or Close. Not before every ERROR line that squeezes
+        //     through in the middle of it: under a sustained stall that was one notice per ERROR
+        //     line (F2 measured 373 in one run), which buries the errors it sits between. The notice
+        //     names the time the gap began instead, so it still says where it started.
+        char text[512];
+        if (lossyGapToo && m_gapLossyLines != 0) {
+            std::snprintf(text, sizeof(text),
+                          "MG_Remote server: LogForward{Dropped} - %llu line(s) at or below WARN (%llu bytes) were "
+                          "not forwarded, from %s until here: the control connection was not taking them and the "
+                          "forward queue held its %llu-byte cap. The server's own log has every one\n",
+                          static_cast<unsigned long long>(m_gapLossyLines),
+                          static_cast<unsigned long long>(m_gapLossyBytes), m_gapLossySince.c_str(),
+                          static_cast<unsigned long long>(m_limits.lossyQueueBytes));
+            Item notice;
+            notice.kind = Kind::Notice;
+            notice.level = MOBILEGL_LOG_LEVEL_WARN;
+            notice.seq = m_nextSeq++;
+            notice.text = NoticeHeader(notice.level) + text;
+            m_queue.push_back(std::move(notice));
+            ++m_counters.notices;
+            m_gapLossyLines = 0;
+            m_gapLossyBytes = 0;
+        }
+        if (m_gapErrorLines != 0) {
+            std::snprintf(text, sizeof(text),
+                          "MG_Remote server: LogForward{Coalesced} - %llu ERROR line(s) were not forwarded here: the "
+                          "forward queue held its %llu-byte cap of error lines the control connection had not "
+                          "taken. The server's own log has every one\n",
+                          static_cast<unsigned long long>(m_gapErrorLines),
+                          static_cast<unsigned long long>(m_limits.errorQueueBytes));
+            Item notice;
+            notice.kind = Kind::Notice;
+            notice.level = MOBILEGL_LOG_LEVEL_ERROR;
+            notice.seq = m_nextSeq++;
+            notice.text = NoticeHeader(notice.level) + text;
+            m_queue.push_back(std::move(notice));
+            ++m_counters.notices;
+            m_gapErrorLines = 0;
+        }
+    }
+
+    void LogForwardChannel::Offer(int level, const char* line) {
+        if (line == nullptr) return;
+        const auto started = std::chrono::steady_clock::now();
+        const std::size_t length = std::strlen(line);
+        std::unique_lock<std::mutex> lock(m_mutex);
+        // The bound this whole file exists for, measured where it is spent: everything below FATAL
+        // returns through here with the lock still held and no wait taken.
+        const auto noteOfferTime = [this, started] {
+            const auto spent = static_cast<std::uint64_t>(
+                std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::steady_clock::now() - started)
+                    .count());
+            m_counters.maxOfferNs = std::max(m_counters.maxOfferNs, spent);
+        };
+        if (m_closing || m_counters.transportFailed) {
+            // Offered after Close() began or after the peer went: it can never be written.
+            ++m_counters.abandoned;
+            return;
+        }
+        const bool lossy = level <= MOBILEGL_LOG_LEVEL_WARN;
+        const bool fatal = level >= MOBILEGL_LOG_LEVEL_FATAL;
+        if (lossy && m_lossyBytes + length > m_limits.lossyQueueBytes) {
+            // The first drop of a gap stamps when it began; the notice that closes it names that.
+            if (m_gapLossyLines == 0) m_gapLossySince = MG_Util::Debug::GetCurrentTime();
+            ++m_counters.lossyDropped;
+            m_counters.lossyDroppedBytes += length;
+            ++m_gapLossyLines;
+            m_gapLossyBytes += length;
+            noteOfferTime();
+            return;
+        }
+        // FATAL is never refused: the process is about to end, and its last word is the one line
+        // the cap must not cost. The cap still bounds everything that can repeat.
+        if (!lossy && !fatal && m_errorBytes + length > m_limits.errorQueueBytes) {
+            ++m_counters.errorsCoalesced;
+            ++m_gapErrorLines;
+            noteOfferTime();
+            return;
+        }
+        // A lossy line that fits closes a <= WARN gap; any line closes an ERROR gap.
+        QueueGapNoticesLocked(/*lossyGapToo=*/lossy);
+        Item item;
+        item.kind = Kind::Line;
+        item.level = level;
+        item.seq = m_nextSeq++;
+        item.text.assign(line, length);
+        const std::uint64_t seq = item.seq;
+        m_queue.push_back(std::move(item));
+        if (lossy) m_lossyBytes += length;
+        else m_errorBytes += length;
+        if (fatal) ++m_urgent;
+        m_cv.notify_all();
+        if (!fatal) {
+            noteOfferTime();
+            return;
+        }
+        // FATAL: wait - bounded - until the sender has written this line and everything before it.
+        const auto deadline = started + std::chrono::milliseconds(m_limits.fatalFlushMs);
+        if (!m_cv.wait_until(lock, deadline,
+                             [this, seq] { return m_writtenSeq >= seq || m_counters.transportFailed; })) {
+            ++m_counters.fatalFlushTimeouts;
+        }
+    }
+
+    void LogForwardChannel::OfferControlFrame(std::vector<std::uint8_t> frame) {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        if (m_closing || m_counters.transportFailed) {
+            ++m_counters.abandoned;
+            return;
+        }
+        // A gap before the acknowledgement is marked before it: the client reads the file after the
+        // ack arrives, and a gap it cannot see there is a gap it will read as "nothing was said".
+        QueueGapNoticesLocked(/*lossyGapToo=*/true);
+        Item item;
+        item.kind = Kind::Control;
+        item.seq = m_nextSeq++;
+        item.frame = std::move(frame);
+        m_queue.push_back(std::move(item));
+        ++m_urgent;
+        m_cv.notify_all();
+    }
+
+    MobileGLResult LogForwardChannel::Send(const Item& item) {
+        if (item.kind == Kind::Control) {
+            return m_transport.SendFrame(MobileGLByteSpan{item.frame.data(), item.frame.size()});
+        }
+        flatbuffers::FlatBufferBuilder builder(item.text.size() + 64);
+        const auto text = builder.CreateString(item.text);
+        const auto log = Wire::CreateLogLine(builder, WireLogLevelFor(item.level), text);
+        const auto envelope = Wire::CreateCtrlEnvelope(builder, Wire::CtrlMsg::LogLine, log.Union());
+        Wire::FinishCtrlEnvelopeBuffer(builder, envelope);
+        return m_transport.SendFrame(MobileGLByteSpan{builder.GetBufferPointer(), builder.GetSize()});
+    }
+
+    void LogForwardChannel::Run() {
+#if defined(__linux__)
+        pthread_setname_np(pthread_self(), "mgl-log-fwd");
+#endif
+        // THIS THREAD NEVER LOGS. It is the logger's sink: a line it logged would come straight back
+        // to its own queue (and in the non-session shape, through the log mutex a FATAL Offer can be
+        // waiting behind). What it has to say is in the counters, which Close() hands to its owner.
+        std::unique_lock<std::mutex> lock(m_mutex);
+        for (;;) {
+            m_cv.wait(lock, [this] { return m_stop || !m_queue.empty(); });
+            if (m_queue.empty()) return; // stopped, and nothing left
+            if (m_counters.transportFailed) {
+                // The peer is gone (a write failed): nothing queued can be written any more.
+                m_counters.abandoned += m_queue.size();
+                m_writtenSeq = m_queue.back().seq;
+                m_queue.clear();
+                m_lossyBytes = 0;
+                m_errorBytes = 0;
+                m_urgent = 0;
+                m_cv.notify_all();
+                continue;
+            }
+            Item& front = m_queue.front();
+            if (front.kind == Kind::Line && front.level == MOBILEGL_LOG_LEVEL_ERROR && !IsUrgentLocked()) {
+                // THE ERROR PACE. A token bucket, refilled by the time since it was last looked at.
+                // A line with no token waits - in the queue, in order - until one accrues, or until
+                // something urgent (a FATAL line, a LogFlush ack, Close) makes the pace moot.
+                const auto now = std::chrono::steady_clock::now();
+                const double elapsed = std::chrono::duration<double>(now - m_tokensAt).count();
+                m_tokens = std::min(static_cast<double>(m_limits.errorBurstLines),
+                                    m_tokens + elapsed * static_cast<double>(m_limits.errorLinesPerSecond));
+                m_tokensAt = now;
+                if (m_tokens < 1.0) {
+                    if (!front.paced) {
+                        front.paced = true;
+                        ++m_counters.errorsPaced;
+                    }
+                    const double rate = std::max(1.0, static_cast<double>(m_limits.errorLinesPerSecond));
+                    const auto untilToken = std::chrono::duration_cast<std::chrono::steady_clock::duration>(
+                        std::chrono::duration<double>((1.0 - m_tokens) / rate));
+                    m_cv.wait_for(lock, untilToken, [this] { return IsUrgentLocked() || m_stop; });
+                    continue;
+                }
+                m_tokens -= 1.0;
+            }
+            Item item = std::move(front);
+            m_queue.pop_front();
+            if (item.kind == Kind::Line) {
+                if (item.level <= MOBILEGL_LOG_LEVEL_WARN) m_lossyBytes -= item.text.size();
+                else m_errorBytes -= item.text.size();
+            }
+            if (item.kind == Kind::Control || (item.kind == Kind::Line && item.level >= MOBILEGL_LOG_LEVEL_FATAL)) {
+                --m_urgent;
+            }
+            m_sending = true;
+            lock.unlock();
+            const MobileGLResult sent = Send(item);
+            lock.lock();
+            m_sending = false;
+            m_writtenSeq = item.seq;
+            if (sent == MOBILEGL_OK) {
+                if (item.kind == Kind::Line) ++m_counters.forwarded;
+            } else if (sent == MOBILEGL_ERR_INVALID_ARGUMENT) {
+                // Refused on its own shape (over the frame cap), not because the peer is gone.
+                ++m_counters.abandoned;
+            } else {
+                ++m_counters.abandoned;
+                m_counters.transportFailed = true;
+            }
+            m_cv.notify_all();
+        }
+    }
+
+    LogForwardCounters LogForwardChannel::Close() {
+        bool unblock = false;
+        {
+            std::unique_lock<std::mutex> lock(m_mutex);
+            if (m_closed) return m_counters;
+            if (!m_closing) {
+                m_closing = true;
+                // The last gap is marked too: a client reading the file after the session sees
+                // that the tail it has is not the whole tail.
+                QueueGapNoticesLocked(/*lossyGapToo=*/true);
+                m_cv.notify_all();
+                const auto drained = [this] {
+                    return (m_queue.empty() && !m_sending) || m_counters.transportFailed;
+                };
+                if (!m_cv.wait_for(lock, std::chrono::milliseconds(m_limits.closeDrainMs), drained)) {
+                    // Not taken within the budget: the peer is not reading. The rest is abandoned,
+                    // and a sender inside a write is blocked on that peer and needs unblocking.
+                    m_counters.abandoned += m_queue.size();
+                    if (!m_queue.empty()) m_writtenSeq = m_queue.back().seq;
+                    m_queue.clear();
+                    m_lossyBytes = 0;
+                    m_errorBytes = 0;
+                    m_urgent = 0;
+                    unblock = m_sending;
+                }
+            }
+            m_stop = true;
+            m_cv.notify_all();
+        }
+        if (unblock && m_unblock != nullptr) m_unblock(m_unblockUser);
+        if (m_sender.joinable()) m_sender.join();
+        std::lock_guard<std::mutex> lock(m_mutex);
+        m_closed = true;
+        return m_counters;
+    }
+
+    LogForwardCounters LogForwardChannel::Counters() const {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        return m_counters;
+    }
+
+} // namespace MobileGL::MG_Remote::Transport

MobileGL/MG_Remote/Transport/LogForward.h

@@ -0,0 +1,195 @@
+// MobileGL - MobileGL/MG_Remote/Transport/LogForward.h
+// Copyright (c) 2025-2026 MobileGL-Dev
+// Licensed under the GNU Lesser General Public License v3.0:
+//   https://www.gnu.org/licenses/gpl-3.0.txt
+//   https://www.gnu.org/licenses/lgpl-3.0.txt
+// SPDX-License-Identifier: LGPL-3.0-only
+// End of Source File Header
+
+// P9 W3. THE SERVER'S LOG LINES ON THE CONTROL PLANE, GRADED, AND NEVER A REASON TO STOP APPLYING.
+//
+// On TCP the server forwards every line it logs to its client as a LogLine control frame, and the
+// client files it under `<base>.server.log` (ControlInbox.cpp). Until this file the forward was a
+// blocking SocketTransport::SendFrame made by the LOGGING THREAD, inside the logger: a client that
+// stopped reading its control connection - a GL thread parked in a debugger, a ControlInbox whose
+// bounded queue filled, a peer that simply stopped - filled the socket, and the next MGLOG on the
+// apply thread never returned. Every record behind it waited; a forfeit that would have ended the
+// session by name (ServerSession::ForfeitReverseChannel) was itself a log line and never finished
+// being said. Design/04 §8.1 had written the policy down from the start (<= WARN lossy, >= ERROR
+// lossless and rate limited), and LogLine.level has been on the wire for it since P6; nothing
+// used either, and every line went out as Info.
+//
+// THE SHAPE. A LogForwardChannel owns a queue and ONE sender thread; the forwarder the logger
+// calls (Offer) only enqueues, so the logging thread never touches the socket. The sender is the
+// only thread that writes log frames, and it writes them in the order they were offered.
+//
+// THE POLICY, per severity (MOBILEGL_LOG_LEVEL_*):
+//   * DEBUG / INFO / WARN are LOSSY. A line that would take the queue's lossy bytes past
+//     LogForwardLimits::lossyQueueBytes is dropped and COUNTED. The drop is not silent to the
+//     client either: when the gap closes - the next line at or below WARN that fits, a LogFlush
+//     ack, or Close - the channel puts a WARN line of its own in the stream, `LogForward{Dropped} -
+//     N line(s) at or below WARN (B bytes) were not forwarded, from <time> until here`, so the gap
+//     in `<base>.server.log` is marked and counted. (ERROR lines that got through in the middle of
+//     the gap sit before that line; the time says where the gap began.)
+//   * ERROR is LOSSLESS, and PACED rather than dropped: at most errorLinesPerSecond leave per
+//     second after a burst of errorBurstLines, and a line over that budget WAITS in the queue, in
+//     order (it is delayed, never discarded). Memory still has to be bounded, so the error bytes
+//     queued are capped too (errorQueueBytes, eight times the lossy cap): a line past it - a peer
+//     that has not read for that long, or a flood well past the pace for that long - is not
+//     queued, and is replaced in the stream by one ERROR line `LogForward{Coalesced} - N ERROR
+//     line(s) were not forwarded here` (queued before the next item that does get in), never by
+//     nothing. The server's own log file (written before any forward, Log.cpp) has every line
+//     either way.
+//   * FATAL is ERROR plus a wait: the logging thread is about to abort, and a line still in the
+//     queue when it does is a line the client never gets - the one it most needs. So a FATAL
+//     Offer waits until the sender has written everything up to and including it, for at most
+//     fatalFlushMs, and then returns whether or not the peer took it. (Log.cpp hands a FATAL line
+//     over outside the log mutex, so that wait holds nobody else's line.)
+//   * A LogFlush acknowledgement (OfferControlFrame) is queued BEHIND every line offered before
+//     it, which is what makes MGPipeSyncPeerLog's promise ("every line before the flush is in the
+//     file when the ack arrives") still true now that a line is not on the wire when MGLOG returns.
+//     While one is queued the sender does not pace: the client is waiting for it.
+//
+// THE BOUND. An Offer below FATAL takes the channel's mutex, copies the line and returns; the
+// sender never holds that mutex across a socket write. So the time a logging thread spends
+// forwarding is a lock and a copy, whatever the peer does - LogForwardCounters::maxOfferNs is that
+// time measured, and the F2 fault injection (EventForfeitPeerTest) asserts on it.
+//
+// CLOSE. Close() stops taking lines, gives the queue closeDrainMs to go out UNPACED, and then stops
+// the sender. A sender still inside a write when that budget runs out is blocked on a peer that is
+// not reading; Close() then calls the `unblock` hook the owner supplied (ServerMain: shutdown(2) of
+// the control socket - the session is over by then) so the thread can be joined rather than
+// leaked. What was still queued is counted as abandoned.
+//
+// SCOPE. Only the TCP server shapes forward (ServerMain.cpp, MOBILEGL_IPC_LOG_FORWARD): a unix spawn
+// server and an inproc apply thread write `<base>.server.log` themselves.
+
+#pragma once
+
+#include "ITransport.h"
+
+#include <condition_variable>
+#include <chrono>
+#include <cstdint>
+#include <deque>
+#include <mutex>
+#include <string>
+#include <thread>
+#include <vector>
+
+// The wire's enum, declared opaquely so this header does not pull the generated schema (and
+// flatbuffers) into every includer; the definition is protocol_generated.h's.
+namespace MobileGL::Wire {
+    enum class LogLevel : std::uint8_t;
+}
+
+namespace MobileGL::MG_Remote::Transport {
+
+    // ---- the level mapping, both directions ------------------------------------------------
+    //
+    // MobileGL's five levels (MOBILEGL_LOG_LEVEL_DEBUG..FATAL, 0..4) and the wire's five
+    // (LogLevel::Debug..Fatal, 0..4) are the same scale, and these two say so in code rather than
+    // by a cast that would silently carry a sixth value across. Out of range: a MobileGL level
+    // below DEBUG is DEBUG and above FATAL is FATAL; a wire level this build does not know (a
+    // newer peer's) is read as ERROR - an unknown severity is never demoted into the lossy class.
+    ::MobileGL::Wire::LogLevel WireLogLevelFor(int level);
+    int LogLevelFromWire(::MobileGL::Wire::LogLevel level);
+
+    struct LogForwardLimits {
+        // Bytes of DEBUG/INFO/WARN text the queue may hold before such a line is dropped.
+        std::uint64_t lossyQueueBytes = 1ull << 20;
+        // Bytes of ERROR/FATAL text the queue may hold before such a line is coalesced.
+        std::uint64_t errorQueueBytes = 8ull << 20;
+        // The ERROR pace: a token bucket of errorBurstLines refilled at errorLinesPerSecond.
+        std::uint32_t errorLinesPerSecond = 256;
+        std::uint32_t errorBurstLines = 1024;
+        // How long a FATAL Offer waits for its line to be written.
+        std::uint32_t fatalFlushMs = 2000;
+        // How long Close() lets the queue drain before it abandons the rest.
+        std::uint32_t closeDrainMs = 2000;
+    };
+
+    struct LogForwardCounters {
+        std::uint64_t forwarded = 0;          // lines written to the transport (notices excluded)
+        std::uint64_t lossyDropped = 0;       // DEBUG/INFO/WARN lines dropped at the lossy cap
+        std::uint64_t lossyDroppedBytes = 0;
+        std::uint64_t errorsCoalesced = 0;    // ERROR/FATAL lines past the error cap
+        std::uint64_t errorsPaced = 0;        // ERROR lines that waited for the pace (delayed, sent)
+        std::uint64_t notices = 0;            // LogForward{Dropped} / {Coalesced} lines queued
+        std::uint64_t abandoned = 0;          // queued items never written (close budget, dead peer)
+        std::uint64_t fatalFlushTimeouts = 0; // FATAL Offers whose wait ran out
+        std::uint64_t maxOfferNs = 0;         // the longest non-FATAL Offer
+        bool transportFailed = false;         // a write failed; nothing more was sent
+    };
+
+    class LogForwardChannel {
+    public:
+        using Unblock = void (*)(void* user);
+
+        // `transport` must outlive the channel. `unblock` (may be null) is called by Close() at most
+        // once, from Close()'s thread, when the sender is still inside a write after closeDrainMs.
+        LogForwardChannel(ITransport& transport, Unblock unblock, void* unblockUser,
+                          const LogForwardLimits& limits = LogForwardLimits{});
+        ~LogForwardChannel();
+
+        LogForwardChannel(const LogForwardChannel&) = delete;
+        LogForwardChannel& operator=(const LogForwardChannel&) = delete;
+
+        // One line, `level` a MOBILEGL_LOG_LEVEL_* value. Never blocks on the peer below FATAL.
+        void Offer(int level, const char* line);
+        // A whole CtrlEnvelope frame, written after every line offered before it and never dropped
+        // or paced (the LogFlush acknowledgement).
+        void OfferControlFrame(std::vector<std::uint8_t> frame);
+        // Stops taking lines, drains within closeDrainMs, stops the sender. Idempotent; returns the
+        // final counters.
+        LogForwardCounters Close();
+        LogForwardCounters Counters() const;
+
+        // MG_Util::Debug::LogForwarder's shape, `user` being the channel.
+        static void ForwardThunk(void* user, int level, const char* line);
+
+    private:
+        enum class Kind : std::uint8_t { Line, Notice, Control };
+        struct Item {
+            Kind kind = Kind::Line;
+            int level = 0;
+            std::uint64_t seq = 0;
+            std::string text;               // Line, Notice
+            std::vector<std::uint8_t> frame; // Control
+            bool paced = false;             // already counted as paced once
+        };
+
+        void Run();
+        void QueueGapNoticesLocked(bool lossyGapToo);
+        bool IsUrgentLocked() const;
+        MobileGLResult Send(const Item& item);
+
+        ITransport& m_transport;
+        Unblock m_unblock;
+        void* m_unblockUser;
+        LogForwardLimits m_limits;
+
+        mutable std::mutex m_mutex;
+        std::condition_variable m_cv;
+        std::deque<Item> m_queue;
+        std::uint64_t m_lossyBytes = 0;
+        std::uint64_t m_errorBytes = 0;
+        std::uint64_t m_nextSeq = 1;
+        std::uint64_t m_writtenSeq = 0; // the seq of the last item the sender finished with
+        std::uint64_t m_urgent = 0;     // FATAL lines and control frames queued: do not pace
+        // A gap not yet marked in the stream, per class.
+        std::uint64_t m_gapLossyLines = 0;
+        std::uint64_t m_gapLossyBytes = 0;
+        std::string m_gapLossySince; // the time of the gap's first drop, for its notice
+        std::uint64_t m_gapErrorLines = 0;
+        double m_tokens = 0;
+        std::chrono::steady_clock::time_point m_tokensAt{};
+        bool m_sending = false;
+        bool m_closing = false; // no more Offers; drain unpaced
+        bool m_stop = false;    // the sender leaves once the queue is empty
+        bool m_closed = false;
+        LogForwardCounters m_counters;
+        std::thread m_sender;
+    };
+
+} // namespace MobileGL::MG_Remote::Transport

MobileGL/MG_Test/Pipe/PipeCatalogueTest.cpp

@@ -1109,17 +1109,19 @@ TEST(PipeCatalogue, PipelineSubsetMembersArePinned) {
                  "ScissorTestEnabledMask");
 }
 
-// The reverse channel is exactly NINE callbacks: plan B's section 7.1 wrote ten, and the tenth
+// The reverse channel is exactly EIGHT callbacks: plan B's section 7.1 wrote ten, and the tenth
 // (OnXfbScatterReady) went with the design it belonged to when P5c/P5f moved the XFB scatter to
 // the server's own staged shadow and its OnBufferWriteback return path. P3b/P4b espryt D1 slice 3
 // deleted the declaration; this case is what makes the struct's shrink a measured fact rather
 // than a claim, since the static_assert beside kMGPipeCallbackCount only proves the two agree.
+// P9 W3 deleted the ninth, OnLog, which never had a producer either: the server's log lines are
+// LogLine frames on the control plane (MG_Remote/Transport/LogForward.h), not reverse records.
 //
 // THE CASE NAME STAYS "…HasTenCallbacks" DELIBERATELY. G2/G14 say the ctest name set only ever
 // grows, so renaming this would delete a name the gate is watching; the count it asserts is what
 // has to be right, and the comment is where the number lives.
 TEST(PipeCatalogue, ReverseChannelHasTenCallbacks) {
-    EXPECT_EQ(kMGPipeCallbackCount, 9u);
+    EXPECT_EQ(kMGPipeCallbackCount, 8u);
     EXPECT_EQ(sizeof(MGPipeCallbacks), kMGPipeCallbackCount * sizeof(void (*)()));
 }
 

MobileGL/MG_Test/Wire/CMakeLists.txt

@@ -30,6 +30,10 @@ if (NOT WIN32)
     # exact first-frame reader, the backoff's arithmetic, the knobs). Its reader is a socket read,
     # so POSIX for the same reason.
     list(APPEND MOBILEGL_WIRE_TESTS PreAuthGateTest)
+    # P9 W3: the graded log forward - the level mapping both ways, the logger's level reaching the
+    # forwarder, and the channel against a transport whose writes can be held (and a real socket
+    # pair nobody reads). POSIX for the socket pair.
+    list(APPEND MOBILEGL_WIRE_TESTS LogForwardTest)
     # P6 sm: two MobileGL processes. Needs the real server image beside it, so
     # the dependency is declared after the loop rather than assumed.
     list(APPEND MOBILEGL_WIRE_TESTS ServerSpawnTest)

MobileGL/MG_Test/Wire/EventForfeitPeerTest.cpp

@@ -51,6 +51,21 @@
 // Everything that uses a ClientSession runs in a FORKED CHILD. The session singleton latches
 // device-lost for the rest of its process once its server goes away, so the "next connection"
 // must come from a process that never had a first one.
+//
+// P9 W3 ADDS FAULT INJECTION F2 (HANDOFF-P9 §4), stream plane only because only a TCP server
+// forwards its log: the peer STOPS READING ITS CONTROL CONNECTION (ClientSession::
+// PauseControlReaderForTest, with its receive buffer shrunk so the kernel cannot absorb the flood)
+// while the server's apply thread logs far more than any socket holds (MOBILEGL_TEST_APPLY_LOG_FLOOD:
+// per applied record, 64 WARN lines and one numbered ERROR line). Two arms:
+//   * control only - SEG_EVENT is drained. The apply thread must get through the whole flood while
+//     nobody reads (the server's own log reaches the last record's ERROR line), <= WARN lines must
+//     be dropped and counted, and once the peer reads again every ERROR line the server logged must
+//     be in the peer's `<base>.server.log`, in order, with the WARN gap marked by LogForward{Dropped}
+//     lines whose counts add up;
+//   * control and SEG_EVENT both - the reverse channel must end the way it always has, by
+//     ReverseChannelForfeit{NotDraining} after the knob and a session that exits 0, rather than an
+//     apply thread parked in a log write that never finishes.
+// Before W3 both hung: the forward was a blocking socket write made by the logging thread itself.
 
 #include <MG_Remote/Client/ClientSession.h>
 #include <MG_Remote/Protocol/SurfaceOpCodec.h>
@@ -74,6 +89,7 @@
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
+#include <functional>
 #include <memory>
 #include <string>
 #include <thread>
@@ -122,6 +138,25 @@ namespace {
     constexpr std::uint32_t kReadbacks = 5;
     constexpr MGP::MGPipeHandle kBuffer{41u, 1u};
 
+    // F2 (P9 W3). The flood: kFloodRecords records published while the peer reads nothing, each of
+    // which makes the server's apply thread log kFloodWarnLinesPerRecord WARN lines of ~650 bytes and
+    // one ERROR line (ServerLoop.cpp, MOBILEGL_TEST_APPLY_LOG_FLOOD) - about 16 MiB, against a
+    // server send buffer this host caps at 4 MiB and a forward queue of 1 MiB for <= WARN.
+    constexpr std::uint32_t kFloodWarnLinesPerRecord = 64;
+    constexpr std::uint32_t kFloodRecords = 400;
+    // THE PEER'S RECEIVE BUFFER WHILE IT IS NOT READING. Without this the kernel absorbs the flood
+    // instead of the server: this host's tcp_rmem lets a receive buffer autotune to 32 MiB, and a
+    // peer whose socket takes the whole flood is not a peer that stopped reading.
+    constexpr int kStalledReceiveBufferBytes = 16 * 1024;
+    // How long the apply thread gets to log the whole flood with nobody reading it. It needs well
+    // under a second; the budget is for a loaded runner.
+    constexpr std::uint32_t kFloodBudgetMs = 15000;
+    // The longest one forward may take a logging thread (the summary's max-offer-us). The channel
+    // spends a lock and a copy; a forward that waited on the peer would spend the whole stall.
+    constexpr unsigned long long kMaxOfferUs = 500 * 1000;
+    constexpr const char* kFloodErrorNeedle = "MOBILEGL_TEST_APPLY_LOG_FLOOD flood error #";
+    constexpr const char* kFloodWarnNeedle = "MOBILEGL_TEST_APPLY_LOG_FLOOD flood warn #";
+
     std::string ServerImage() {
         if (const char* explicitPath = std::getenv("MOBILEGL_TEST_SERVER_PATH")) return explicitPath;
         return "libMobileGLServer.so";
@@ -165,8 +200,11 @@ namespace {
     // TRANSPORT / SERVER_PATH / RING_MB / STAGE_MB outright, and nothing a CLIENT was told may
     // leak into the server - so every MOBILEGL_IPC_* goes and the few this scenario means are
     // stated. MOBILEGL_IPC_LOG_FORWARD=0 keeps the server's lines in its own role log, which is
-    // what the assertions below read.
-    std::vector<std::string> SupervisorEnvironment(const std::string& logBase, std::uint32_t waitMs) {
+    // what the assertions below read. F2 (P9 W3) turns forwarding ON - that is what it tests - and
+    // gives its peer a log base of its own, so the forwarded copy and the server's own file never
+    // share a path; and it sets the apply thread's flood.
+    std::vector<std::string> SupervisorEnvironment(const std::string& logBase, std::uint32_t waitMs,
+                                                   bool forwardLog = false, std::uint32_t floodLines = 0) {
         std::vector<std::string> env;
         for (char** e = ::environ; e != nullptr && *e != nullptr; ++e) {
             const std::string entry(*e);
@@ -178,7 +216,8 @@ namespace {
         }
         env.emplace_back("MOBILEGL_IPC_ROLE=server");
         env.emplace_back("MOBILEGL_IPC_DIAL=no");
-        env.emplace_back("MOBILEGL_IPC_LOG_FORWARD=0");
+        env.emplace_back(forwardLog ? "MOBILEGL_IPC_LOG_FORWARD=1" : "MOBILEGL_IPC_LOG_FORWARD=0");
+        if (floodLines != 0) env.emplace_back("MOBILEGL_TEST_APPLY_LOG_FLOOD=" + std::to_string(floodLines));
         env.emplace_back(WaitKnobEntry(waitMs));
         env.emplace_back("MOBILEGL_LOG_FILE_PATH=" + logBase);
         const std::string image = ServerImage();
@@ -187,9 +226,10 @@ namespace {
         return env;
     }
 
-    pid_t LaunchSupervisor(const std::string& endpoint, const std::string& logBase, std::uint32_t waitMs) {
+    pid_t LaunchSupervisor(const std::string& endpoint, const std::string& logBase, std::uint32_t waitMs,
+                           bool forwardLog = false, std::uint32_t floodLines = 0) {
         const std::string image = ServerImage();
-        std::vector<std::string> env = SupervisorEnvironment(logBase, waitMs);
+        std::vector<std::string> env = SupervisorEnvironment(logBase, waitMs, forwardLog, floodLines);
         std::vector<char*> envp;
         for (auto& entry : env) envp.push_back(entry.data());
         envp.push_back(nullptr);
@@ -253,16 +293,18 @@ namespace {
     }
 
     // Polls the supervisor's log until `needle` has appeared `times` times, or the budget runs
-    // out. Returns the milliseconds it took, or -1.
+    // out. Returns the milliseconds it took, or -1. F2's logs run to tens of MiB, so its polls are
+    // spaced wider (`every`) rather than re-reading the whole file a hundred times a second.
     long long WaitForLog(const std::string& logBase, const std::string& needle, std::size_t times,
-                         std::chrono::milliseconds budget) {
+                         std::chrono::milliseconds budget,
+                         std::chrono::milliseconds every = std::chrono::milliseconds(10)) {
         const auto start = std::chrono::steady_clock::now();
         while (std::chrono::steady_clock::now() - start < budget) {
             if (Count(ServerLog(logBase), needle) >= times) {
                 return std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() -
                                                                              start).count();
             }
-            std::this_thread::sleep_for(std::chrono::milliseconds(10));
+            std::this_thread::sleep_for(every);
         }
         return -1;
     }
@@ -305,14 +347,16 @@ namespace {
         kServerNotReaped = 19,
         kServerExitedNonZero = 20,
         kRawFrame = 21,
+        // F2 (P9 W3).
+        kNoControlReader = 22,
+        kFloodPublish = 23,
     };
 
-    // Everything a peer does between its handshake and the moment SEG_EVENT is full with the
-    // server waiting on one more writeback. Returns kChildOk or the step that failed.
-    int FillTheEventRingWithoutDraining(Client::ClientSession& client) {
-        // A pbuffer context on the server's apply thread, through the same forwarders
-        // BackendObject_Remote uses. The handles are the client's names for them; the server keys
-        // its surfaces by them and never dereferences one.
+    // A pbuffer context on the server's apply thread, through the same forwarders
+    // BackendObject_Remote uses. The handles are the client's names for them; the server keys
+    // its surfaces by them and never dereferences one. These are SURFACE OPS - they cross the
+    // control connection both ways - so a peer that is going to stop reading control does this first.
+    int BringUpPbufferContext() {
         const auto display = reinterpret_cast<EGLDisplay>(std::uintptr_t{0x4d47});
         const auto surface = reinterpret_cast<EGLSurface>(std::uintptr_t{0x5346});
         const auto context = reinterpret_cast<EGLContext>(std::uintptr_t{0x4358});
@@ -320,6 +364,39 @@ namespace {
         if (!Server::ServerInitializeEGLDisplay(display, &major, &minor)) return kEglDisplay;
         if (!Server::ServerCreateEGLPbufferSurface(surface, 16, 16)) return kEglSurface;
         if (!Server::ServerMakeEGLCurrent(display, surface, surface, context)) return kEglCurrent;
+        return kChildOk;
+    }
+
+    // F2 (P9 W3): THE PEER STOPS READING ITS CONTROL CONNECTION, then makes the server's apply thread
+    // log the flood. The receive buffer is shrunk first (kStalledReceiveBufferBytes), then the reader
+    // is paused - from here nothing is taken off the socket, not a log line, not a reply - and then
+    // kFloodRecords set_patch_state records are published without waiting. set_patch_state because it
+    // is pure applier state: no GL call, no reply, no event, so the only thing each one produces on
+    // the server is the flood's log lines.
+    int StopReadingControlAndFlood(Client::ClientSession& client) {
+        auto* control = client.ControlSocketForTest();
+        if (control == nullptr) return kNoControlReader;
+        const int receiveBuffer = kStalledReceiveBufferBytes;
+        (void)::setsockopt(control->StreamFd(), SOL_SOCKET, SO_RCVBUF, &receiveBuffer, sizeof(receiveBuffer));
+        if (!client.PauseControlReaderForTest(true)) return kNoControlReader;
+        for (std::uint32_t i = 0; i < kFloodRecords; ++i) {
+            MGP::MGPPatchState patch{};
+            patch.Vertices = 3;
+            if (!PublishWithoutWaiting(client, MGP::MGPWireOp::SetPatchState, patch)) return kFloodPublish;
+        }
+        return kChildOk;
+    }
+
+    // Everything a peer does between its handshake and the moment SEG_EVENT is full with the
+    // server waiting on one more writeback. Returns kChildOk or the step that failed. `afterContext`
+    // (F2) runs once the context is up and before the buffer is made.
+    int FillTheEventRingWithoutDraining(Client::ClientSession& client, const std::function<int()>& afterContext = {}) {
+        const int up = BringUpPbufferContext();
+        if (up != kChildOk) return up;
+        if (afterContext) {
+            const int done = afterContext();
+            if (done != kChildOk) return done;
+        }
 
         // One buffer with defined content, staged whole: create, respecify, one sub-data record.
         // These three WAIT, and that is fine - the ring holds one small surface event at most so
@@ -764,6 +841,266 @@ namespace {
         RemoveRoleLogsIfGreen(logBase);
     }
 
+    // ---- F2 (P9 W3): a peer that stops reading control while the apply thread floods its log ----
+
+    std::string ReadFileWhole(const std::string& path) {
+        std::string all;
+        if (FILE* file = std::fopen(path.c_str(), "rb")) {
+            char chunk[65536];
+            std::size_t got = 0;
+            while ((got = std::fread(chunk, 1, sizeof(chunk), file)) > 0) all.append(chunk, got);
+            std::fclose(file);
+        }
+        return all;
+    }
+
+    // The numbers after every `needle` in `text`, in file order.
+    std::vector<unsigned long long> Indices(const std::string& text, const std::string& needle) {
+        std::vector<unsigned long long> out;
+        for (auto at = text.find(needle); at != std::string::npos; at = text.find(needle, at + needle.size())) {
+            out.push_back(std::strtoull(text.c_str() + at + needle.size(), nullptr, 10));
+        }
+        return out;
+    }
+
+    // `name=<number>` on a line, or ~0 when the field is not there.
+    unsigned long long Field(const std::string& line, const std::string& name) {
+        const auto at = line.find(name + "=");
+        if (at == std::string::npos) return ~0ull;
+        return std::strtoull(line.c_str() + at + name.size() + 1, nullptr, 10);
+    }
+
+    // The sum of the counts the in-stream LogForward{Dropped} notices name.
+    unsigned long long NamedDrops(const std::string& text) {
+        unsigned long long sum = 0;
+        for (const auto count : Indices(text, "LogForward{Dropped} - ")) sum += count;
+        return sum;
+    }
+
+    // THE F2 PEER. Handshake, a pbuffer context, then StopReadingControlAndFlood - and, in the
+    // undrained arm, the event-ring fill behind it (so the apply thread meets the full SEG_EVENT only
+    // after it has logged the whole flood). Writes one byte to `published` when the last record is on
+    // the wire. The drained arm then waits for `release`, reads its control connection again, and
+    // calls SyncPeerLog - the LogFlush whose ack is queued behind every line the server forwarded
+    // before it - and writes one byte to `synced`: from then on the peer's `<base>.server.log` holds
+    // everything the server forwarded. It stops cleanly when released a second time. The undrained
+    // arm never reads again and leaves by _exit, as the other non-draining peers do.
+    [[noreturn]] void FloodPeer(const std::string& control, const std::string& peerBase, bool alsoStopDraining,
+                                int published, int release, int synced) {
+        // WritePeerLog reads MOBILEGL_LOG_FILE_PATH at each write, so this is the peer's forwarded copy.
+        ::setenv("MOBILEGL_LOG_FILE_PATH", peerBase.c_str(), 1);
+        ConfigureClient(control);
+        auto& client = Client::ClientSessionInstance();
+        if (client.StartSpawned() != MOBILEGL_OK) ::_exit(kNoWelcome);
+        if (alsoStopDraining) {
+            const int filled = FillTheEventRingWithoutDraining(client, [&client] { return StopReadingControlAndFlood(client); });
+            if (filled != kChildOk) ::_exit(filled);
+        } else {
+            const int up = BringUpPbufferContext();
+            if (up != kChildOk) ::_exit(up);
+            const int flooded = StopReadingControlAndFlood(client);
+            if (flooded != kChildOk) ::_exit(flooded);
+        }
+        const char one = 1;
+        (void)::write(published, &one, 1);
+        char go = 0;
+        (void)::read(release, &go, 1);
+        if (alsoStopDraining) ::_exit(kChildOk);
+        client.PauseControlReaderForTest(false);
+        client.SyncPeerLog();
+        (void)::write(synced, &one, 1);
+        (void)::read(release, &go, 1);
+        client.Stop();
+        ::_exit(kChildOk);
+    }
+
+    void RunLogFloodArm(bool alsoStopDraining) {
+        std::signal(SIGPIPE, SIG_IGN); // as RunArm: no write in this file may kill the test process
+        PinHeadlessEgl();
+        const std::string tag = std::string("stream-flood-") + (alsoStopDraining ? "undrained-" : "") +
+                                std::to_string(::getpid());
+        const std::string logBase = "/tmp/mgl-fz3-" + tag + ".log";
+        const std::string peerBase = "/tmp/mgl-fz3-" + tag + "-peer.log";
+        const std::string peerLogPath =
+            MobileGL::MG_Util::Debug::RoleLogPath(peerBase.c_str(), MobileGL::MG_Util::Debug::LogRole::Server);
+        MobileGL::MG_Util::Debug::TruncateRoleLogs(logBase.c_str());
+        MobileGL::MG_Util::Debug::TruncateRoleLogs(peerBase.c_str());
+        const std::uint16_t port = FreeLoopbackPort();
+        ASSERT_NE(port, 0) << "no free loopback port";
+        const std::string endpoint = "tcp://127.0.0.1:" + std::to_string(port);
+        const std::uint32_t waitMs = kEventWaitMs;
+        const pid_t supervisor =
+            LaunchSupervisor(endpoint, logBase, waitMs, /*forwardLog=*/true, kFloodWarnLinesPerRecord);
+        ASSERT_GT(supervisor, 0);
+        struct Reap {
+            pid_t pid;
+            ~Reap() { StopSupervisor(pid); }
+        } reapSupervisor{supervisor};
+        ASSERT_GE(WaitForLog(logBase, "listening on", 1, std::chrono::milliseconds(10000)), 0)
+            << "the supervisor never listened\n" << ServerLog(logBase);
+
+        int published[2] = {-1, -1}, release[2] = {-1, -1}, synced[2] = {-1, -1};
+        ASSERT_EQ(::pipe(published), 0);
+        ASSERT_EQ(::pipe(release), 0);
+        ASSERT_EQ(::pipe(synced), 0);
+        std::fflush(nullptr);
+        const pid_t peer = ::fork();
+        ASSERT_GE(peer, 0);
+        if (peer == 0) {
+            ::close(published[0]);
+            ::close(release[1]);
+            ::close(synced[0]);
+            FloodPeer(endpoint, peerBase, alsoStopDraining, published[1], release[0], synced[1]);
+        }
+        ::close(published[1]);
+        ::close(release[0]);
+        ::close(synced[1]);
+        struct ReleasePeer {
+            int fd;
+            pid_t pid;
+            ~ReleasePeer() {
+                const char go[2] = {1, 1};
+                (void)::write(fd, go, 2);
+                ::close(fd);
+                int status = 0;
+                if (pid > 0 && ::waitpid(pid, &status, WNOHANG) == 0) {
+                    ::kill(pid, SIGKILL);
+                    ::waitpid(pid, &status, 0);
+                }
+            }
+        } releasePeer{release[1], peer};
+
+        char byte = 0;
+        const ssize_t got = ::read(published[0], &byte, 1);
+        ::close(published[0]);
+        if (got != 1) {
+            const int code = WaitChild(peer, std::chrono::milliseconds(5000));
+            releasePeer.pid = -1;
+            ::close(synced[0]);
+            FAIL() << "the flood peer did not get its records out (child exit " << code << ")\n"
+                   << ServerLog(logBase).substr(0, 20000);
+        }
+        const auto every = std::chrono::milliseconds(200);
+
+        if (!alsoStopDraining) {
+            // (1) THE APPLY THREAD IS NOT HELD BY LOGGING. With the peer reading nothing, the server's
+            // own log must reach the flood's last record: every one of the kFloodRecords records is
+            // applied (and logs its ERROR line) while the control connection takes no byte. A forward
+            // that waits on the socket stalls the apply thread a few MiB in, and this never arrives.
+            const long long reachedMs =
+                WaitForLog(logBase, kFloodErrorNeedle, kFloodRecords, std::chrono::milliseconds(kFloodBudgetMs), every);
+            ASSERT_GE(reachedMs, 0) << "the apply thread did not get through the flood while its client was not "
+                                       "reading control - it was held by logging ("
+                                    << Count(ServerLog(logBase), kFloodErrorNeedle) << " of " << kFloodRecords
+                                    << " records' ERROR lines logged after " << kFloodBudgetMs << " ms)";
+            // (2) The peer reads again and synchronises its log.
+            const char go = 1;
+            ASSERT_EQ(::write(release[1], &go, 1), 1);
+            struct pollfd wait{synced[0], POLLIN, 0};
+            ASSERT_EQ(::poll(&wait, 1, 30000), 1) << "the peer never finished SyncPeerLog\n" << ServerLog(logBase).substr(0, 20000);
+            ASSERT_EQ(::read(synced[0], &byte, 1), 1) << "the peer died in SyncPeerLog (child exit "
+                                                      << WaitChild(peer, std::chrono::milliseconds(5000)) << ")";
+            ::close(synced[0]);
+            const std::string serverLog = ServerLog(logBase);
+            const std::string peerLog = ReadFileWhole(peerLogPath);
+            // (3) >= ERROR IS NOT LOST: every ERROR line the server logged is in the peer's copy, in
+            // the server's order - and that order is 1..N with no gap.
+            const auto serverErrors = Indices(serverLog, kFloodErrorNeedle);
+            const auto peerErrors = Indices(peerLog, kFloodErrorNeedle);
+            EXPECT_GE(serverErrors.size(), static_cast<std::size_t>(kFloodRecords));
+            EXPECT_EQ(peerErrors, serverErrors) << "an ERROR line the server logged did not reach the client ("
+                                                << peerErrors.size() << " of " << serverErrors.size() << ")";
+            for (std::size_t i = 0; i < peerErrors.size(); ++i) {
+                if (peerErrors[i] != i + 1) {
+                    ADD_FAILURE() << "the client's ERROR lines are not 1..N in order at " << i << ": " << peerErrors[i];
+                    break;
+                }
+            }
+            // (4) <= WARN IS LOSSY AND SAYS SO: WARN lines were dropped, and the LogForward{Dropped}
+            // notices in the peer's copy account for exactly the ones that are missing from it.
+            const std::size_t serverWarns = Count(serverLog, kFloodWarnNeedle);
+            const std::size_t peerWarns = Count(peerLog, kFloodWarnNeedle);
+            const unsigned long long namedDrops = NamedDrops(peerLog);
+            // Each applied record logs kFloodWarnLinesPerRecord WARN lines beside its one ERROR line.
+            EXPECT_EQ(serverWarns, serverErrors.size() * kFloodWarnLinesPerRecord);
+            EXPECT_LT(peerWarns, serverWarns) << "no WARN line was dropped: the flood never outran the socket, so "
+                                                 "this arm proved nothing about backpressure";
+            EXPECT_GT(namedDrops, 0ull) << "WARN lines were dropped without a LogForward{Dropped} line saying so";
+            EXPECT_EQ(peerWarns + namedDrops, serverWarns)
+                << "the WARN lines the client has and the drops it was told of do not add up to what the server logged";
+            // (5) The session ends cleanly, and the server's own summary agrees: drops counted, no
+            // ERROR coalesced, nothing abandoned, and no forward held its logging thread.
+            ASSERT_EQ(::write(release[1], &go, 1), 1);
+            const int code = WaitChild(peer, std::chrono::milliseconds(30000));
+            releasePeer.pid = -1;
+            EXPECT_EQ(code, static_cast<int>(kChildOk)) << "the flood peer did not stop cleanly";
+            ASSERT_GE(WaitForLog(logBase, "log forwarding ended", 1, std::chrono::milliseconds(15000), every), 0)
+                << "the session never logged its forward's summary";
+            const std::string summary = LineWith(ServerLog(logBase), "log forwarding ended");
+            std::printf("[f2] %.400s\n", summary.c_str());
+            EXPECT_GE(Field(summary, "dropped(<=WARN)"), namedDrops) << summary;
+            EXPECT_NE(Field(summary, "dropped(<=WARN)"), ~0ull) << summary;
+            EXPECT_EQ(Field(summary, "coalesced(ERROR)"), 0ull) << summary;
+            EXPECT_EQ(Field(summary, "abandoned"), 0ull) << summary;
+            EXPECT_EQ(Field(summary, "transport-failed"), 0ull) << summary;
+            // One notice per gap, not one per ERROR line that got through the gap (the first cut of
+            // the channel wrote 373 here, one between every pair of the flood's ERROR lines).
+            EXPECT_LE(Field(summary, "notices"), 8ull) << summary;
+            EXPECT_LT(Field(summary, "max-offer-us"), kMaxOfferUs)
+                << "one forward held its logging thread for longer than " << kMaxOfferUs << " us\n" << summary;
+            EXPECT_EQ(serverLog.find("Fatal{"), std::string::npos);
+        } else {
+            ::close(synced[0]);
+            // Control AND SEG_EVENT both undrained: the apply thread gets through the flood (the
+            // forward drops and never waits), meets the full SEG_EVENT, forfeits it by name after the
+            // knob, and the session ends with exit 0 - its log forward cut at close, since this peer
+            // never reads again. Before W3 the apply thread parked inside a log write a few MiB into
+            // the flood and none of this happened.
+            const long long budget = static_cast<long long>(waitMs) + kFloodBudgetMs + 20000;
+            const long long stoppedMs = WaitForLog(logBase, "mgl-srv-apply stops on ReverseChannelForfeit", 1,
+                                                   std::chrono::milliseconds(budget), every);
+            const long long reapedMs =
+                WaitForLog(logBase, "reaped exit=0 sessionsFaulted=0", 1, std::chrono::milliseconds(budget), every);
+            const std::string log = ServerLog(logBase);
+            ASSERT_GE(stoppedMs, 0) << "the server never reached the forfeit: its apply thread was held ("
+                                    << Count(log, kFloodErrorNeedle) << " of " << kFloodRecords
+                                    << " flood records applied)";
+            ASSERT_GE(reapedMs, 0) << "the supervisor never reaped the forfeited session with exit 0";
+            EXPECT_GE(Count(log, kFloodErrorNeedle), static_cast<std::size_t>(kFloodRecords));
+            const std::string forfeit = LineWith(log, "ReverseChannelForfeit{");
+            EXPECT_NE(forfeit.find("ReverseChannelForfeit{NotDraining} - kEventBufferWriteback"), std::string::npos)
+                << forfeit;
+            EXPECT_NE(forfeit.find(WaitKnobEntry(waitMs)), std::string::npos) << forfeit;
+            const std::string summary = LineWith(log, "log forwarding ended");
+            std::printf("[f2] %.400s\n", summary.c_str());
+            ASSERT_FALSE(summary.empty()) << "the session never logged its forward's summary";
+            EXPECT_GT(Field(summary, "dropped(<=WARN)"), 0ull) << summary;
+            EXPECT_NE(Field(summary, "dropped(<=WARN)"), ~0ull) << summary;
+            EXPECT_EQ(Field(summary, "coalesced(ERROR)"), 0ull) << summary;
+            // A peer that never reads again: the close budget ran out and the rest was cut.
+            EXPECT_GT(Field(summary, "abandoned"), 0ull) << summary;
+            EXPECT_LT(Field(summary, "max-offer-us"), kMaxOfferUs) << summary;
+            EXPECT_EQ(log.find("Fatal{"), std::string::npos);
+            EXPECT_EQ(log.find("reaped signal="), std::string::npos) << "a session died of a signal";
+            // The same supervisor welcomes the next connection.
+            std::fflush(nullptr);
+            const pid_t next = ::fork();
+            ASSERT_GE(next, 0);
+            if (next == 0) NextPeer(endpoint);
+            const int nextCode = WaitChild(next, std::chrono::milliseconds(30000));
+            EXPECT_EQ(nextCode, static_cast<int>(kChildOk)) << "the next connection was not welcomed";
+            EXPECT_GE(WaitForLog(logBase, "reaped exit=0 sessionsFaulted=0", 2, std::chrono::milliseconds(10000), every), 0)
+                << "the welcomed session did not end cleanly";
+        }
+        RemoveRoleLogsIfGreen(logBase);
+        if (!::testing::Test::HasFailure()) {
+            MobileGL::MG_Util::Debug::TruncateRoleLogs(peerBase.c_str());
+            for (const char* role : {".client.log", ".server.log"}) {
+                ::unlink((peerBase.substr(0, peerBase.size() - 4) + role).c_str());
+            }
+        }
+    }
+
 } // namespace
 
 TEST(EventForfeitPeer, ShmPeerThatStopsDrainingIsForfeitedAndTheNextConnectionIsWelcomed) {
@@ -808,4 +1145,13 @@ TEST(EventForfeitPeer, StreamControlHalfClosedWhileTheServerWaitsIsPeerGone) {
     RunArm(true, Scenario::ControlHalfClosedWhileWaiting);
 }
 
+// P9 W3, fault injection F2 (HANDOFF-P9 §4). Stream plane only: only a TCP server forwards its log.
+TEST(EventForfeitPeer, StreamLogFloodWhileControlIsNotReadHoldsNoApplyThreadAndLosesNoError) {
+    RunLogFloodArm(/*alsoStopDraining=*/false);
+}
+
+TEST(EventForfeitPeer, StreamLogFloodWithControlAndSegEventUndrainedEndsByForfeitNotAHang) {
+    RunLogFloodArm(/*alsoStopDraining=*/true);
+}
+
 #endif

MobileGL/MG_Test/Wire/LogForwardTest.cpp

@@ -0,0 +1,618 @@
+// MobileGL - MobileGL/MG_Test/Wire/LogForwardTest.cpp
+// Copyright (c) 2025-2026 MobileGL-Dev
+// Licensed under the GNU Lesser General Public License v3.0:
+//   https://www.gnu.org/licenses/gpl-3.0.txt
+//   https://www.gnu.org/licenses/lgpl-3.0.txt
+// SPDX-License-Identifier: LGPL-3.0-only
+// End of Source File Header
+
+// P9 W3: the server's graded log forward (MG_Remote/Transport/LogForward.h), without a server.
+//
+// THREE LAYERS, EACH ONE A CLAIM THE NEXT DEPENDS ON:
+//   1. THE LEVELS. MobileGL's five, the Android priorities Log() is handed, the wire's five - every
+//      level both ways, and the out-of-range answers. Before W3 every forwarded line crossed as
+//      Info, so a mapping that collapsed two levels would have been invisible; these cases are
+//      what make a collapse red.
+//   2. THE LOGGER. MGLOG_I/W/E/F on a server-role thread reach an installed forwarder with the level
+//      of the macro that logged them - the chain Log() -> LogLevelFromAndroidPriority ->
+//      LogForwarder, which is the half of "forwarding carries the level" the mapping alone does not
+//      prove. And the client's writer files a peer line under the wire's level.
+//   3. THE CHANNEL. Driven over a transport whose writes can be HELD (a peer that is not reading):
+//      a caller is never held below FATAL, <= WARN is dropped and counted and the gap is marked in
+//      the stream, ERROR is never dropped (paced, or coalesced into a named line past its own cap),
+//      a FATAL waits for its write but only for its bound, a LogFlush ack goes out behind every
+//      earlier line and unpaced, and Close() gets its sender back from a peer that never reads.
+//      The last case does the same over a real socket pair the far end never reads.
+//
+// The two-process version - a real `--serve` session child whose apply thread floods its log while
+// its TCP client stops reading control - is fault injection F2 in EventForfeitPeerTest.
+
+#include <MG_Remote/Protocol/generated/protocol_generated.h>
+#include <MG_Remote/Transport/LogForward.h>
+#include <MG_Remote/Transport/SocketTransport.h>
+#include <MG_Util/Debug/Log.h>
+
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <thread>
+#include <vector>
+
+#if !defined(_WIN32)
+#include <sys/socket.h>
+#include <unistd.h>
+#endif
+
+using namespace MobileGL::MG_Remote::Transport;
+namespace Debug = MobileGL::MG_Util::Debug;
+namespace Wire = MobileGL::Wire;
+
+namespace {
+
+    using Clock = std::chrono::steady_clock;
+
+    long long MillisecondsSince(Clock::time_point start) {
+        return std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now() - start).count();
+    }
+
+    // One frame the channel wrote, decoded.
+    struct Written {
+        bool isLog = false;
+        bool isFlushAck = false;
+        Wire::LogLevel level = Wire::LogLevel::Debug;
+        std::string text;
+        std::uint64_t flushSeq = 0;
+    };
+
+    Written Decode(const std::uint8_t* bytes, std::size_t size) {
+        Written out;
+        flatbuffers::Verifier verifier(bytes, size);
+        if (!Wire::VerifyCtrlEnvelopeBuffer(verifier)) return out;
+        const auto* envelope = Wire::GetCtrlEnvelope(bytes);
+        if (const auto* line = envelope->msg_as_LogLine()) {
+            out.isLog = true;
+            out.level = line->level();
+            if (line->text()) out.text = line->text()->str();
+        } else if (const auto* flush = envelope->msg_as_LogFlush()) {
+            out.isFlushAck = flush->ack();
+            out.flushSeq = flush->seq();
+        }
+        return out;
+    }
+
+    // A transport whose writes can be HELD: while the gate is shut, SendFrame blocks exactly as a
+    // socket write to a peer that is not reading does. Cut() ends every held and later write with
+    // TRANSPORT_CLOSED - what shutdown(2) does to the real one - and is the channel's unblock hook.
+    class GatedTransport final : public ITransport {
+    public:
+        explicit GatedTransport(bool open) : m_open(open) {}
+
+        MobileGLResult SendFrame(MobileGLByteSpan bytes) override {
+            std::unique_lock<std::mutex> lock(m_mutex);
+            ++m_attempts;
+            m_cv.notify_all();
+            m_cv.wait(lock, [this] { return m_open || m_cut; });
+            if (m_cut) return MOBILEGL_ERR_TRANSPORT_CLOSED;
+            m_written.push_back(Decode(static_cast<const std::uint8_t*>(bytes.data), static_cast<std::size_t>(bytes.size)));
+            m_cv.notify_all();
+            return MOBILEGL_OK;
+        }
+        MobileGLResult ReceiveFrame(MobileGLMutableByteSpan, std::uint64_t*, std::uint32_t) override {
+            return MOBILEGL_ERR_UNSUPPORTED;
+        }
+        std::uint64_t PeekFrameSize() override { return 0; }
+        MobileGLResult ShareFd(int, MobileGLByteSpan) override { return MOBILEGL_ERR_UNSUPPORTED; }
+        MobileGLResult ReceiveFd(int*, MobileGLMutableByteSpan, std::uint64_t*, std::uint32_t) override {
+            return MOBILEGL_ERR_UNSUPPORTED;
+        }
+        void Shutdown() override { Cut(); }
+        TransportRole Role() const override { return TransportRole::Server; }
+
+        void Open() {
+            std::lock_guard<std::mutex> lock(m_mutex);
+            m_open = true;
+            m_cv.notify_all();
+        }
+        void Cut() {
+            std::lock_guard<std::mutex> lock(m_mutex);
+            m_cut = true;
+            ++m_cuts;
+            m_cv.notify_all();
+        }
+        static void CutThunk(void* user) { static_cast<GatedTransport*>(user)->Cut(); }
+
+        // Blocks until the channel's sender is inside a held write (or `budget` runs out).
+        bool WaitForAttempt(std::chrono::milliseconds budget) {
+            std::unique_lock<std::mutex> lock(m_mutex);
+            return m_cv.wait_for(lock, budget, [this] { return m_attempts > 0; });
+        }
+        bool WaitForWritten(std::size_t count, std::chrono::milliseconds budget) {
+            std::unique_lock<std::mutex> lock(m_mutex);
+            return m_cv.wait_for(lock, budget, [this, count] { return m_written.size() >= count; });
+        }
+        std::vector<Written> Frames() {
+            std::lock_guard<std::mutex> lock(m_mutex);
+            return m_written;
+        }
+        int Cuts() {
+            std::lock_guard<std::mutex> lock(m_mutex);
+            return m_cuts;
+        }
+
+    private:
+        std::mutex m_mutex;
+        std::condition_variable m_cv;
+        bool m_open = false;
+        bool m_cut = false;
+        int m_cuts = 0;
+        std::size_t m_attempts = 0;
+        std::vector<Written> m_written;
+    };
+
+    std::string Numbered(const char* series, int index, std::size_t pad = 0) {
+        std::string line = "[00:00:00] [Linux test/X]: " + std::string(series) + " #" + std::to_string(index);
+        line.append(pad, 'p');
+        return line + "\n";
+    }
+
+    bool Contains(const std::string& text, const std::string& needle) { return text.find(needle) != std::string::npos; }
+
+    // The count a LogForward notice names ("... - N line(s) ..."), or -1.
+    long long NoticeCount(const std::string& text, const char* word) {
+        const std::string key = std::string("LogForward{") + word + "} - ";
+        const auto at = text.find(key);
+        if (at == std::string::npos) return -1;
+        return std::strtoll(text.c_str() + at + key.size(), nullptr, 10);
+    }
+
+    std::vector<int> SeriesIndices(const std::vector<Written>& written, const char* series) {
+        std::vector<int> indices;
+        const std::string key = std::string(series) + " #";
+        for (const auto& frame : written) {
+            if (!frame.isLog) continue;
+            const auto at = frame.text.find(key);
+            if (at != std::string::npos) indices.push_back(std::atoi(frame.text.c_str() + at + key.size()));
+        }
+        return indices;
+    }
+
+    // ---- 1. the levels ------------------------------------------------------------------------
+
+    TEST(LogForwardLevels, EveryMobileGLLevelCrossesTheWireAndComesBackUnchanged) {
+        const struct {
+            int level;
+            Wire::LogLevel wire;
+        } kRows[] = {
+            {MOBILEGL_LOG_LEVEL_DEBUG, Wire::LogLevel::Debug}, {MOBILEGL_LOG_LEVEL_INFO, Wire::LogLevel::Info},
+            {MOBILEGL_LOG_LEVEL_WARN, Wire::LogLevel::Warn},   {MOBILEGL_LOG_LEVEL_ERROR, Wire::LogLevel::Error},
+            {MOBILEGL_LOG_LEVEL_FATAL, Wire::LogLevel::Fatal},
+        };
+        for (const auto& row : kRows) {
+            EXPECT_EQ(WireLogLevelFor(row.level), row.wire) << "level " << row.level;
+            EXPECT_EQ(LogLevelFromWire(row.wire), row.level) << Wire::EnumNameLogLevel(row.wire);
+            EXPECT_EQ(LogLevelFromWire(WireLogLevelFor(row.level)), row.level) << "level " << row.level;
+        }
+        // Every wire value the schema has, back and forth, so a sixth appended value that nobody
+        // mapped cannot hide behind the five rows above.
+        for (const Wire::LogLevel wire : Wire::EnumValuesLogLevel()) {
+            EXPECT_EQ(WireLogLevelFor(LogLevelFromWire(wire)), wire) << Wire::EnumNameLogLevel(wire);
+        }
+    }
+
+    TEST(LogForwardLevels, OutOfRangeLevelsClampAndAnUnknownWireLevelIsNeverDemoted) {
+        EXPECT_EQ(WireLogLevelFor(-3), Wire::LogLevel::Debug);
+        EXPECT_EQ(WireLogLevelFor(MOBILEGL_LOG_LEVEL_FATAL + 5), Wire::LogLevel::Fatal);
+        // A newer peer's severity this build has no name for is filed as an error: never lossy.
+        EXPECT_EQ(LogLevelFromWire(static_cast<Wire::LogLevel>(9)), MOBILEGL_LOG_LEVEL_ERROR);
+        EXPECT_EQ(LogLevelFromWire(static_cast<Wire::LogLevel>(255)), MOBILEGL_LOG_LEVEL_ERROR);
+    }
+
+    TEST(LogForwardLevels, TheAndroidPriorityLogIsHandedCarriesEachMgLogLevel) {
+        // What MGLOG_D..MGLOG_F pass (Log.h), each to its own level.
+        EXPECT_EQ(Debug::LogLevelFromAndroidPriority(ANDROID_LOG_DEBUG), MOBILEGL_LOG_LEVEL_DEBUG);
+        EXPECT_EQ(Debug::LogLevelFromAndroidPriority(ANDROID_LOG_INFO), MOBILEGL_LOG_LEVEL_INFO);
+        EXPECT_EQ(Debug::LogLevelFromAndroidPriority(ANDROID_LOG_WARN), MOBILEGL_LOG_LEVEL_WARN);
+        EXPECT_EQ(Debug::LogLevelFromAndroidPriority(ANDROID_LOG_ERROR), MOBILEGL_LOG_LEVEL_ERROR);
+        EXPECT_EQ(Debug::LogLevelFromAndroidPriority(ANDROID_LOG_FATAL), MOBILEGL_LOG_LEVEL_FATAL);
+        // What no MGLOG_* passes.
+        EXPECT_EQ(Debug::LogLevelFromAndroidPriority(ANDROID_LOG_VERBOSE), MOBILEGL_LOG_LEVEL_DEBUG);
+        EXPECT_EQ(Debug::LogLevelFromAndroidPriority(ANDROID_LOG_UNKNOWN), MOBILEGL_LOG_LEVEL_INFO);
+        EXPECT_EQ(Debug::LogLevelFromAndroidPriority(ANDROID_LOG_DEFAULT), MOBILEGL_LOG_LEVEL_INFO);
+        EXPECT_EQ(Debug::LogLevelFromAndroidPriority(ANDROID_LOG_SILENT), MOBILEGL_LOG_LEVEL_INFO);
+        EXPECT_STREQ(Debug::LogLevelTag(MOBILEGL_LOG_LEVEL_DEBUG), "DEBUG");
+        EXPECT_STREQ(Debug::LogLevelTag(MOBILEGL_LOG_LEVEL_INFO), "INFO");
+        EXPECT_STREQ(Debug::LogLevelTag(MOBILEGL_LOG_LEVEL_WARN), "WARN");
+        EXPECT_STREQ(Debug::LogLevelTag(MOBILEGL_LOG_LEVEL_ERROR), "ERROR");
+        EXPECT_STREQ(Debug::LogLevelTag(MOBILEGL_LOG_LEVEL_FATAL), "FATAL");
+        EXPECT_STREQ(Debug::LogLevelTag(-1), "DEBUG");
+        EXPECT_STREQ(Debug::LogLevelTag(99), "FATAL");
+    }
+
+    // ---- 2. the logger ------------------------------------------------------------------------
+
+    // One log base for the whole binary: the file sinks latch their path on first use, and gtest
+    // may run every case in one process.
+    const std::string& LogBase() {
+        static const std::string base = [] {
+            const std::string path = "/tmp/mgl-logforward-" + std::to_string(::getpid()) + ".log";
+            ::setenv("MOBILEGL_LOG_FILE_PATH", path.c_str(), 1);
+            Debug::TruncateRoleLogs(path.c_str());
+            return path;
+        }();
+        return base;
+    }
+
+    std::string ReadWhole(const std::string& path) {
+        std::string all;
+        if (FILE* file = std::fopen(path.c_str(), "rb")) {
+            char chunk[4096];
+            std::size_t got = 0;
+            while ((got = std::fread(chunk, 1, sizeof(chunk), file)) > 0) all.append(chunk, got);
+            std::fclose(file);
+        }
+        return all;
+    }
+
+    struct Captured {
+        int level;
+        std::string line;
+    };
+    std::mutex g_capturedMutex;
+    std::vector<Captured> g_captured;
+    void Capture(void*, int level, const char* line) {
+        std::lock_guard<std::mutex> lock(g_capturedMutex);
+        g_captured.push_back({level, line});
+    }
+
+    int CapturedLevelOf(const char* marker) {
+        std::lock_guard<std::mutex> lock(g_capturedMutex);
+        for (const auto& entry : g_captured) {
+            if (Contains(entry.line, marker)) return entry.level;
+        }
+        return -1;
+    }
+
+    TEST(LogForwardLogger, EachMgLogMacroReachesTheForwarderWithItsOwnLevel) {
+        (void)LogBase();
+        // A thread of its own, so the server role this stamps does not leak into another case.
+        std::thread server([] {
+            Debug::SetThreadLogRole(Debug::LogRole::Server);
+            Debug::SetLogForwarder(&Capture, nullptr);
+            MGLOG_I("w3-logger-marker info");
+            MGLOG_W("w3-logger-marker warn");
+            MGLOG_E("w3-logger-marker error");
+            // FATAL goes to the forwarder OUTSIDE the log mutex (Log.cpp) - still with its level.
+            MGLOG_F("w3-logger-marker fatal");
+#if MOBILEGL_LOG_ACTIVE_LEVEL <= MOBILEGL_LOG_LEVEL_DEBUG
+            MGLOG_D("w3-logger-marker debug");
+#endif
+            Debug::SetLogForwarder(nullptr, nullptr);
+        });
+        server.join();
+        EXPECT_EQ(CapturedLevelOf("w3-logger-marker info"), MOBILEGL_LOG_LEVEL_INFO);
+        EXPECT_EQ(CapturedLevelOf("w3-logger-marker warn"), MOBILEGL_LOG_LEVEL_WARN);
+        EXPECT_EQ(CapturedLevelOf("w3-logger-marker error"), MOBILEGL_LOG_LEVEL_ERROR);
+        EXPECT_EQ(CapturedLevelOf("w3-logger-marker fatal"), MOBILEGL_LOG_LEVEL_FATAL);
+#if MOBILEGL_LOG_ACTIVE_LEVEL <= MOBILEGL_LOG_LEVEL_DEBUG
+        EXPECT_EQ(CapturedLevelOf("w3-logger-marker debug"), MOBILEGL_LOG_LEVEL_DEBUG);
+#endif
+        // A client-role thread forwards nothing: the forwarder is the SERVER's.
+        std::thread client([] {
+            Debug::SetThreadLogRole(Debug::LogRole::Client);
+            Debug::SetLogForwarder(&Capture, nullptr);
+            MGLOG_E("w3-logger-marker client-side");
+            Debug::SetLogForwarder(nullptr, nullptr);
+        });
+        client.join();
+        EXPECT_EQ(CapturedLevelOf("w3-logger-marker client-side"), -1);
+    }
+
+    TEST(LogForwardLogger, APeerLineIsFiledUnderTheLevelTheWireCarried) {
+        const std::string serverLog = Debug::RoleLogPath(LogBase().c_str(), Debug::LogRole::Server);
+        // A line the server formatted is written as it came - its header already names its level.
+        Debug::WritePeerLog(MOBILEGL_LOG_LEVEL_ERROR, "[01:02:03] [Linux mgl-srv-apply/ERROR]: w3-peer headed\n");
+        // Bare text gets the wire's level as its tag, and a line end.
+        Debug::WritePeerLog(MOBILEGL_LOG_LEVEL_WARN, "w3-peer bare warn");
+        Debug::WritePeerLog(MOBILEGL_LOG_LEVEL_FATAL, "w3-peer bare fatal\n");
+        Debug::WritePeerLog(LogLevelFromWire(Wire::LogLevel::Debug), "w3-peer bare debug");
+        const std::string log = ReadWhole(serverLog);
+        EXPECT_TRUE(Contains(log, "[01:02:03] [Linux mgl-srv-apply/ERROR]: w3-peer headed\n")) << log;
+        EXPECT_TRUE(Contains(log, "[peer/WARN]: w3-peer bare warn\n")) << log;
+        EXPECT_TRUE(Contains(log, "[peer/FATAL]: w3-peer bare fatal\n")) << log;
+        EXPECT_TRUE(Contains(log, "[peer/DEBUG]: w3-peer bare debug\n")) << log;
+        EXPECT_FALSE(Contains(log, "[peer/ERROR]: [01:02:03]")) << "a headed line was tagged twice\n" << log;
+    }
+
+    // ---- 3. the channel -----------------------------------------------------------------------
+
+    TEST(LogForwardChannel, LinesGoOutInOrderWithTheirLevelOnTheWire) {
+        GatedTransport transport(/*open=*/true);
+        LogForwardChannel channel(transport, &GatedTransport::CutThunk, &transport);
+        channel.Offer(MOBILEGL_LOG_LEVEL_DEBUG, "order debug\n");
+        channel.Offer(MOBILEGL_LOG_LEVEL_INFO, "order info\n");
+        channel.Offer(MOBILEGL_LOG_LEVEL_WARN, "order warn\n");
+        channel.Offer(MOBILEGL_LOG_LEVEL_ERROR, "order error\n");
+        channel.Offer(MOBILEGL_LOG_LEVEL_FATAL, "order fatal\n");
+        const auto counters = channel.Close();
+        const auto written = transport.Frames();
+        ASSERT_EQ(written.size(), 5u);
+        const Wire::LogLevel levels[] = {Wire::LogLevel::Debug, Wire::LogLevel::Info, Wire::LogLevel::Warn,
+                                         Wire::LogLevel::Error, Wire::LogLevel::Fatal};
+        const char* texts[] = {"order debug\n", "order info\n", "order warn\n", "order error\n", "order fatal\n"};
+        for (std::size_t i = 0; i < 5; ++i) {
+            EXPECT_TRUE(written[i].isLog);
+            EXPECT_EQ(written[i].level, levels[i]) << i;
+            EXPECT_EQ(written[i].text, texts[i]) << i;
+        }
+        EXPECT_EQ(counters.forwarded, 5u);
+        EXPECT_EQ(counters.lossyDropped, 0u);
+        EXPECT_EQ(counters.abandoned, 0u);
+        EXPECT_EQ(transport.Cuts(), 0) << "a drained Close must not cut the transport";
+    }
+
+    // THE CASE THE WHOLE FILE IS FOR. The peer stops reading (the sender's first write is held);
+    // the caller keeps logging WARN lines. Every Offer must return at once - timed, on a worker,
+    // under a watchdog so that a channel that DOES hold its caller reds this case instead of hanging
+    // it - the lines past the lossy cap are dropped and counted, and once the peer reads again the
+    // gap is marked in the stream, before the next line, with the count.
+    TEST(LogForwardChannel, APeerThatStopsReadingCostsWarnLinesNotTheCallersTime) {
+        GatedTransport transport(/*open=*/false);
+        LogForwardLimits limits;
+        limits.lossyQueueBytes = 16 * 1024;
+        LogForwardChannel channel(transport, &GatedTransport::CutThunk, &transport, limits);
+        constexpr int kLines = 2000;
+        std::atomic<bool> done{false};
+        long long offeringMs = -1;
+        std::thread caller([&] {
+            const auto start = Clock::now();
+            for (int i = 1; i <= kLines; ++i) channel.Offer(MOBILEGL_LOG_LEVEL_WARN, Numbered("stall warn", i, 100).c_str());
+            offeringMs = MillisecondsSince(start);
+            done = true;
+        });
+        const auto start = Clock::now();
+        while (!done && MillisecondsSince(start) < 5000) std::this_thread::sleep_for(std::chrono::milliseconds(5));
+        const bool heldTheCaller = !done;
+        transport.Open(); // lets a held caller finish, so a red is a red and not a hang
+        caller.join();
+        ASSERT_FALSE(heldTheCaller) << "2000 WARN Offers did not return within 5 s while the peer was not "
+                                       "reading: the forward held its caller";
+        EXPECT_LT(offeringMs, 2000) << "the Offers took " << offeringMs << " ms against a peer that was not reading";
+        auto counters = channel.Counters();
+        EXPECT_GT(counters.lossyDropped, 0u) << "nothing was dropped past a 16 KiB cap with ~230 KiB offered";
+        EXPECT_LT(counters.maxOfferNs, 50ull * 1000 * 1000) << "one Offer took " << counters.maxOfferNs << " ns";
+        // The peer reads again; once what was queued is out, the next line goes out behind a notice
+        // naming the gap. (Waited for first: offered into a queue still full, it would be dropped.)
+        ASSERT_TRUE(transport.WaitForWritten(kLines - counters.lossyDropped, std::chrono::milliseconds(5000)));
+        channel.Offer(MOBILEGL_LOG_LEVEL_WARN, "after the gap\n");
+        counters = channel.Close();
+        const auto written = transport.Frames();
+        ASSERT_FALSE(written.empty());
+        EXPECT_EQ(written.back().text, "after the gap\n");
+        ASSERT_GE(written.size(), 2u);
+        const Written& notice = written[written.size() - 2];
+        EXPECT_EQ(notice.level, Wire::LogLevel::Warn);
+        EXPECT_EQ(NoticeCount(notice.text, "Dropped"), static_cast<long long>(counters.lossyDropped)) << notice.text;
+        EXPECT_EQ(notice.text[0], '[') << "a notice must carry a header, so the client files it as it came";
+        EXPECT_EQ(counters.notices, 1u);
+        // Every line is accounted for: written, or counted as dropped.
+        EXPECT_EQ(SeriesIndices(written, "stall warn").size() + counters.lossyDropped, static_cast<std::size_t>(kLines));
+        EXPECT_EQ(counters.forwarded, SeriesIndices(written, "stall warn").size() + 1);
+    }
+
+    TEST(LogForwardChannel, ErrorLinesAreNeverDroppedWhileWarnLinesAre) {
+        GatedTransport transport(/*open=*/false);
+        LogForwardLimits limits;
+        limits.lossyQueueBytes = 8 * 1024;
+        LogForwardChannel channel(transport, &GatedTransport::CutThunk, &transport, limits);
+        constexpr int kEach = 500;
+        for (int i = 1; i <= kEach; ++i) {
+            channel.Offer(MOBILEGL_LOG_LEVEL_WARN, Numbered("mixed warn", i, 100).c_str());
+            channel.Offer(MOBILEGL_LOG_LEVEL_ERROR, Numbered("mixed error", i, 100).c_str());
+        }
+        auto counters = channel.Counters();
+        EXPECT_GT(counters.lossyDropped, 0u);
+        EXPECT_EQ(counters.errorsCoalesced, 0u);
+        transport.Open();
+        counters = channel.Close();
+        const auto errors = SeriesIndices(transport.Frames(), "mixed error");
+        ASSERT_EQ(errors.size(), static_cast<std::size_t>(kEach)) << "an ERROR line was lost while WARN lines were dropped";
+        for (int i = 0; i < kEach; ++i) EXPECT_EQ(errors[i], i + 1) << "ERROR lines out of order at " << i;
+        EXPECT_EQ(counters.errorsCoalesced, 0u);
+        EXPECT_EQ(counters.abandoned, 0u);
+    }
+
+    // Memory is bounded for ERROR too - but past its cap an ERROR line becomes a COUNT in a named
+    // ERROR line, never nothing: every line offered is written or counted by a notice.
+    TEST(LogForwardChannel, ErrorLinesPastTheirCapAreCoalescedIntoANamedLineNotLost) {
+        GatedTransport transport(/*open=*/false);
+        LogForwardLimits limits;
+        limits.errorQueueBytes = 4 * 1024;
+        LogForwardChannel channel(transport, &GatedTransport::CutThunk, &transport, limits);
+        constexpr int kLines = 200;
+        for (int i = 1; i <= kLines; ++i) channel.Offer(MOBILEGL_LOG_LEVEL_ERROR, Numbered("capped error", i, 60).c_str());
+        auto counters = channel.Counters();
+        EXPECT_GT(counters.errorsCoalesced, 0u);
+        transport.Open();
+        ASSERT_TRUE(transport.WaitForWritten(kLines - counters.errorsCoalesced, std::chrono::milliseconds(5000)));
+        channel.Offer(MOBILEGL_LOG_LEVEL_ERROR, "after the error gap\n");
+        counters = channel.Close();
+        const auto written = transport.Frames();
+        long long coalescedNamed = 0;
+        for (const auto& frame : written) {
+            const long long named = NoticeCount(frame.text, "Coalesced");
+            if (named < 0) continue;
+            EXPECT_EQ(frame.level, Wire::LogLevel::Error) << "a coalesced ERROR gap must be named at ERROR";
+            coalescedNamed += named;
+        }
+        EXPECT_EQ(coalescedNamed, static_cast<long long>(counters.errorsCoalesced));
+        EXPECT_EQ(SeriesIndices(written, "capped error").size() + counters.errorsCoalesced, static_cast<std::size_t>(kLines))
+            << "an ERROR line was neither written nor counted";
+        EXPECT_EQ(written.back().text, "after the error gap\n");
+    }
+
+    TEST(LogForwardChannel, ErrorLinesPastThePaceWaitInOrderAndAreNotDropped) {
+        GatedTransport transport(/*open=*/true);
+        LogForwardLimits limits;
+        limits.errorLinesPerSecond = 40;
+        limits.errorBurstLines = 4;
+        LogForwardChannel channel(transport, &GatedTransport::CutThunk, &transport, limits);
+        constexpr int kLines = 24;
+        const auto start = Clock::now();
+        for (int i = 1; i <= kLines; ++i) channel.Offer(MOBILEGL_LOG_LEVEL_ERROR, Numbered("paced error", i).c_str());
+        EXPECT_LT(MillisecondsSince(start), 200) << "pacing held the caller";
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+        // A burst of 4 and ~40/s: about 8 by now, and far fewer than all 24.
+        EXPECT_LT(transport.Frames().size(), 18u) << "the pace did not hold anything back";
+        ASSERT_TRUE(transport.WaitForWritten(kLines, std::chrono::milliseconds(5000))) << "paced lines never all went out";
+        const auto counters = channel.Close();
+        const auto indices = SeriesIndices(transport.Frames(), "paced error");
+        ASSERT_EQ(indices.size(), static_cast<std::size_t>(kLines));
+        for (int i = 0; i < kLines; ++i) EXPECT_EQ(indices[i], i + 1);
+        EXPECT_GE(counters.errorsPaced, 10u);
+        EXPECT_EQ(counters.errorsCoalesced, 0u);
+    }
+
+    // MGPipeSyncPeerLog's promise, kept through a queue: the ack is written after every line
+    // offered before it - and at once, not at the ERROR pace, because the client is waiting on it.
+    TEST(LogForwardChannel, AFlushAckGoesOutBehindEveryEarlierLineAndIsNotPaced) {
+        GatedTransport transport(/*open=*/true);
+        LogForwardLimits limits;
+        limits.errorLinesPerSecond = 1;
+        limits.errorBurstLines = 1;
+        LogForwardChannel channel(transport, &GatedTransport::CutThunk, &transport, limits);
+        for (int i = 1; i <= 10; ++i) channel.Offer(MOBILEGL_LOG_LEVEL_ERROR, Numbered("before ack", i).c_str());
+        flatbuffers::FlatBufferBuilder builder(64);
+        const auto flush = Wire::CreateLogFlush(builder, 77, true);
+        const auto envelope = Wire::CreateCtrlEnvelope(builder, Wire::CtrlMsg::LogFlush, flush.Union());
+        Wire::FinishCtrlEnvelopeBuffer(builder, envelope);
+        const auto start = Clock::now();
+        channel.OfferControlFrame(
+            std::vector<std::uint8_t>(builder.GetBufferPointer(), builder.GetBufferPointer() + builder.GetSize()));
+        ASSERT_TRUE(transport.WaitForWritten(11, std::chrono::milliseconds(2000)))
+            << "the ack (and the lines before it) waited for a 1-line-per-second pace";
+        EXPECT_LT(MillisecondsSince(start), 1500);
+        (void)channel.Close();
+        const auto written = transport.Frames();
+        ASSERT_EQ(written.size(), 11u);
+        EXPECT_EQ(SeriesIndices(written, "before ack").size(), 10u);
+        EXPECT_TRUE(written.back().isFlushAck);
+        EXPECT_EQ(written.back().flushSeq, 77u);
+    }
+
+    TEST(LogForwardChannel, AFatalLineWaitsForItsWriteButOnlyForItsBound) {
+        {
+            // The peer reads: the Offer returns with the line (and the one before it) written.
+            GatedTransport transport(/*open=*/true);
+            LogForwardChannel channel(transport, &GatedTransport::CutThunk, &transport);
+            channel.Offer(MOBILEGL_LOG_LEVEL_WARN, "before fatal\n");
+            channel.Offer(MOBILEGL_LOG_LEVEL_FATAL, "fatal written\n");
+            const auto written = transport.Frames();
+            ASSERT_EQ(written.size(), 2u) << "a FATAL Offer returned before its line was written";
+            EXPECT_EQ(written[1].level, Wire::LogLevel::Fatal);
+            EXPECT_EQ(channel.Counters().fatalFlushTimeouts, 0u);
+        }
+        {
+            // The peer does not read: the Offer returns after its bound, not never.
+            GatedTransport transport(/*open=*/false);
+            LogForwardLimits limits;
+            limits.fatalFlushMs = 300;
+            limits.closeDrainMs = 100;
+            LogForwardChannel channel(transport, &GatedTransport::CutThunk, &transport, limits);
+            const auto start = Clock::now();
+            channel.Offer(MOBILEGL_LOG_LEVEL_FATAL, "fatal held\n");
+            const long long waited = MillisecondsSince(start);
+            EXPECT_GE(waited, 250);
+            EXPECT_LT(waited, 3000);
+            EXPECT_EQ(channel.Counters().fatalFlushTimeouts, 1u);
+        }
+    }
+
+    // A peer that never reads again: Close() gives the queue its budget, abandons the rest, cuts
+    // the transport to get the sender out of its held write, and joins it.
+    TEST(LogForwardChannel, CloseGetsItsSenderBackFromAPeerThatNeverReads) {
+        GatedTransport transport(/*open=*/false);
+        LogForwardLimits limits;
+        limits.closeDrainMs = 200;
+        LogForwardChannel channel(transport, &GatedTransport::CutThunk, &transport, limits);
+        for (int i = 1; i <= 50; ++i) channel.Offer(MOBILEGL_LOG_LEVEL_ERROR, Numbered("never read", i).c_str());
+        ASSERT_TRUE(transport.WaitForAttempt(std::chrono::milliseconds(2000)));
+        const auto start = Clock::now();
+        const auto counters = channel.Close();
+        EXPECT_LT(MillisecondsSince(start), 3000);
+        EXPECT_EQ(transport.Cuts(), 1) << "the held sender was not unblocked";
+        EXPECT_EQ(counters.abandoned, 50u) << "49 queued + the one inside the held write";
+        EXPECT_TRUE(counters.transportFailed);
+        EXPECT_EQ(counters.forwarded, 0u);
+        // After Close, an Offer is refused (counted), never queued behind a dead sender.
+        channel.Offer(MOBILEGL_LOG_LEVEL_ERROR, "after close\n");
+        EXPECT_EQ(channel.Counters().abandoned, 51u);
+    }
+
+#if !defined(_WIN32)
+    // The same stall over the real thing: a SocketTransport pair whose far end does not read until
+    // the caller is done. The socket fills, the sender blocks inside SendFrame (as it would on TCP),
+    // and the caller - 20,000 WARN lines of ~600 bytes, ~12 MB, far past any socket buffer - never
+    // waits on it. Then the far end reads, and what it gets is levelled LogLines with the gap
+    // marked. Close() needs no cut: the reader drains.
+    TEST(LogForwardChannel, OverARealSocketThePeerNotReadingNeverHoldsTheCaller) {
+        std::unique_ptr<SocketTransport> client, server;
+        ASSERT_EQ(SocketTransport::CreatePair(client, server), MOBILEGL_OK);
+        LogForwardChannel channel(*server, nullptr, nullptr);
+        constexpr int kLines = 20000;
+        const auto start = Clock::now();
+        for (int i = 1; i <= kLines; ++i) {
+            channel.Offer(i % 100 == 0 ? MOBILEGL_LOG_LEVEL_ERROR : MOBILEGL_LOG_LEVEL_WARN,
+                          Numbered(i % 100 == 0 ? "socket error" : "socket warn", i, 560).c_str());
+        }
+        const long long offeringMs = MillisecondsSince(start);
+        EXPECT_LT(offeringMs, 5000) << "offering took " << offeringMs << " ms against a socket nobody read";
+        EXPECT_GT(channel.Counters().lossyDropped, 0u);
+        EXPECT_EQ(channel.Counters().errorsCoalesced, 0u);
+        // Now the far end reads, until the ack the channel queues behind everything arrives.
+        flatbuffers::FlatBufferBuilder builder(64);
+        const auto flush = Wire::CreateLogFlush(builder, 9, true);
+        const auto envelope = Wire::CreateCtrlEnvelope(builder, Wire::CtrlMsg::LogFlush, flush.Union());
+        Wire::FinishCtrlEnvelopeBuffer(builder, envelope);
+        channel.OfferControlFrame(
+            std::vector<std::uint8_t>(builder.GetBufferPointer(), builder.GetBufferPointer() + builder.GetSize()));
+        std::vector<Written> received;
+        std::vector<std::uint8_t> frame(1 << 20);
+        const auto readStart = Clock::now();
+        for (;;) {
+            ASSERT_LT(MillisecondsSince(readStart), 20000) << "the ack never arrived";
+            std::uint64_t size = 0;
+            const auto got = client->ReceiveFrame({frame.data(), frame.size()}, &size, 1000);
+            if (got == MOBILEGL_ERR_TIMEOUT) continue;
+            ASSERT_EQ(got, MOBILEGL_OK);
+            received.push_back(Decode(frame.data(), static_cast<std::size_t>(size)));
+            if (received.back().isFlushAck) break;
+        }
+        const auto counters = channel.Close();
+        const auto errors = SeriesIndices(received, "socket error");
+        ASSERT_EQ(errors.size(), static_cast<std::size_t>(kLines / 100)) << "an ERROR line was lost";
+        for (std::size_t i = 0; i < errors.size(); ++i) EXPECT_EQ(errors[i], static_cast<int>((i + 1) * 100));
+        long long droppedNamed = 0;
+        for (const auto& line : received) {
+            if (!line.isLog) continue;
+            const long long named = NoticeCount(line.text, "Dropped");
+            if (named >= 0) droppedNamed += named;
+            else if (Contains(line.text, "socket warn")) EXPECT_EQ(line.level, Wire::LogLevel::Warn);
+            else if (Contains(line.text, "socket error")) EXPECT_EQ(line.level, Wire::LogLevel::Error);
+        }
+        EXPECT_EQ(droppedNamed, static_cast<long long>(counters.lossyDropped));
+        EXPECT_EQ(SeriesIndices(received, "socket warn").size() + counters.lossyDropped,
+                  static_cast<std::size_t>(kLines - kLines / 100));
+        EXPECT_EQ(counters.abandoned, 0u);
+        EXPECT_FALSE(counters.transportFailed);
+    }
+#endif
+
+} // namespace

MobileGL/MG_Util/Debug/Log.cpp

@@ -60,14 +60,43 @@ namespace MobileGL {
             action(user);
         }
 
-        void WritePeerLog(const char* message) {
+        int LogLevelFromAndroidPriority(int priority) {
+            switch (priority) {
+            case ANDROID_LOG_VERBOSE:
+            case ANDROID_LOG_DEBUG: return MOBILEGL_LOG_LEVEL_DEBUG;
+            case ANDROID_LOG_INFO: return MOBILEGL_LOG_LEVEL_INFO;
+            case ANDROID_LOG_WARN: return MOBILEGL_LOG_LEVEL_WARN;
+            case ANDROID_LOG_ERROR: return MOBILEGL_LOG_LEVEL_ERROR;
+            case ANDROID_LOG_FATAL: return MOBILEGL_LOG_LEVEL_FATAL;
+            default: return MOBILEGL_LOG_LEVEL_INFO;
+            }
+        }
+
+        const char* LogLevelTag(int level) {
+            // The same five words MGLOG_D..MGLOG_F put in a header (Log.h), so a tagged peer line
+            // reads like a local one.
+            static const char* const kTags[] = {"DEBUG", "INFO", "WARN", "ERROR", "FATAL"};
+            if (level < MOBILEGL_LOG_LEVEL_DEBUG) level = MOBILEGL_LOG_LEVEL_DEBUG;
+            if (level > MOBILEGL_LOG_LEVEL_FATAL) level = MOBILEGL_LOG_LEVEL_FATAL;
+            return kTags[level];
+        }
+
+        void WritePeerLog(int level, const char* message) {
             std::lock_guard<std::mutex> lock(LogMutex());
             const char* path = std::getenv("MOBILEGL_LOG_FILE_PATH");
             if (!path || !*path) path = MOBILEGL_LOG_FILE_PATH;
             if (!path || !*path) return;
             if (!s_serverLogFile) s_serverLogFile = std::fopen(RoleLogPath(path, LogRole::Server).c_str(), "w");
             if (s_serverLogFile) {
+                // A server-formatted line starts with its own "[time] [OS thread/LEVEL]: " header,
+                // level tag included, and is written untouched. Anything else gets the wire's level
+                // as its tag and a line end if it lacks one: the severity must survive into the file
+                // whatever the peer put in the text.
+                const bool headed = message[0] == '[';
+                if (!headed) std::fprintf(s_serverLogFile, "[peer/%s]: ", LogLevelTag(level));
                 std::fputs(message, s_serverLogFile);
+                const std::size_t length = std::strlen(message);
+                if (!headed && (length == 0 || message[length - 1] != '\n')) std::fputc('\n', s_serverLogFile);
                 std::fflush(s_serverLogFile);
             }
         }
@@ -315,10 +344,21 @@ namespace MobileGL {
 
             WriteToFile(out.c_str());
 #if MOBILEGL_BUILD_DISAGGREGATED
+            // P9 W3: the forwarder is told the line's severity (LogForwarder, Log.h).
+            const int level = LogLevelFromAndroidPriority(androidLogLevel);
             bool forwardOutsideLock = false;
+            bool forwardWasSessionScoped = false;
             if (s_forwarder && ThreadIsServerRole() && !t_forwarding) {
-                if (!s_forwardSessionThreadsOnly) s_forwarder(s_forwarderUser, out.c_str());
-                else forwardOutsideLock = t_forwardsToPeer;
+                forwardWasSessionScoped = s_forwardSessionThreadsOnly;
+                if (!s_forwardSessionThreadsOnly) {
+                    // A FATAL line goes out after the log mutex is released, like a session-scoped
+                    // one: the forwarder may wait (bounded) for it to reach the peer before the
+                    // process aborts, and must not hold every other thread's line while it does.
+                    if (level >= MOBILEGL_LOG_LEVEL_FATAL) forwardOutsideLock = true;
+                    else s_forwarder(s_forwarderUser, level, out.c_str());
+                } else {
+                    forwardOutsideLock = t_forwardsToPeer;
+                }
             }
 #endif
 
@@ -327,11 +367,13 @@ namespace MobileGL {
             if (forwardOutsideLock) {
                 // SetSessionLogForwarder: this thread belongs to the live session. The line goes out
                 // after the log mutex is released, so a stalled client holds only the forward mutex.
+                // (And a FATAL line of either kind, for the reason above.) Re-asked under the
+                // forward mutex, because the forwarder may have been replaced or cleared since.
                 lock.unlock();
                 std::lock_guard<std::mutex> forwardLock(ForwardMutex());
-                if (s_forwarder != nullptr && s_forwardSessionThreadsOnly) {
+                if (s_forwarder != nullptr && s_forwardSessionThreadsOnly == forwardWasSessionScoped) {
                     t_forwarding = true;
-                    s_forwarder(s_forwarderUser, out.c_str());
+                    s_forwarder(s_forwarderUser, level, out.c_str());
                     t_forwarding = false;
                 }
             }

MobileGL/MG_Util/Debug/Log.h

@@ -136,7 +136,19 @@ namespace MobileGL {
             void Close();
 
 #if MOBILEGL_BUILD_DISAGGREGATED
-            using LogForwarder = void (*)(void*, const char*);
+            // P9 W3: THE FORWARDER IS TOLD EACH LINE'S SEVERITY, as a MOBILEGL_LOG_LEVEL_* value.
+            // It used to get the text alone, so the server stamped every forwarded line
+            // LogLevel::Info on the wire - and the one policy the wire's LogLine.level exists for
+            // (<= WARN lossy, >= ERROR lossless and rate limited; design/04 §8.1) had nothing to
+            // decide on. The text still carries its own "/LEVEL]" tag; the number is what a
+            // forwarder can act on without parsing it back out.
+            //
+            // A forwarder is called for every line a forwarding thread logs, so it must not block
+            // on its peer (MG_Remote/Transport/LogForward.h is the one ServerMain installs). A
+            // FATAL line is the exception it may make, for a bounded time: it is handed over
+            // OUTSIDE the log mutex (Log.cpp), so a forwarder that waits for it to reach the peer
+            // before the process aborts holds no other thread's log line hostage.
+            using LogForwarder = void (*)(void* user, int level, const char* line);
             void SetLogForwarder(LogForwarder forwarder, void* user);
             // P12 review fix (log forwarding). THE IN-PROCESS DISPLAY SERVER'S FORWARDER, scoped to
             // the live session's own threads. SetLogForwarder forwards every server-role thread's
@@ -148,11 +160,30 @@ namespace MobileGL {
             // the log mutex, under a forward mutex of its own, so a client that stops reading stalls
             // the session's threads and not every thread that logs (the UI thread's surface
             // callbacks). SetLogForwarder(nullptr, nullptr) clears either kind and waits for a send in
-            // flight.
+            // flight. (Since P9 W3 the forwarder ServerMain installs does not wait on the client at
+            // all - it queues, and a sender thread of its own does the socket write - so "a client
+            // that stops reading" stalls neither; the forward mutex is kept for the barrier.)
             void SetSessionLogForwarder(LogForwarder forwarder, void* user);
             void SetThreadForwardsLogToPeer(bool forwards);
-            void WritePeerLog(const char* message);
+            // The CLIENT's writer for a line its server forwarded, into `<base>.server.log`.
+            // `level` is the severity the server sent (a MOBILEGL_LOG_LEVEL_* value, mapped back
+            // from the wire by MG_Remote::Transport::LogLevelFromWire). A line the server formatted
+            // carries its own "[time] [OS thread/LEVEL]: " header and is written as it came; a line
+            // without one - anything that does not start with '[' - is written behind a
+            // "[peer/LEVEL]: " tag, so the severity the wire carried is in the file either way.
+            void WritePeerLog(int level, const char* message);
             void WithLogBarrier(void (*action)(void*), void* user);
+
+            // THE SEVERITY OF A LINE, from what Log() is actually given. Log()'s signature is part
+            // of the pull build's symbol set (G1), so it cannot grow a level argument; the Android
+            // priority it already takes is a faithful carrier of the five MGLOG_* levels (Log.h's
+            // MGLOG_D..MGLOG_F pass ANDROID_LOG_DEBUG..ANDROID_LOG_FATAL). VERBOSE reads as DEBUG,
+            // and the three priorities no MGLOG_* passes (UNKNOWN, DEFAULT, SILENT) as INFO - a
+            // guess, but one that neither drops a real diagnostic's severity nor invents one.
+            int LogLevelFromAndroidPriority(int priority);
+            // "DEBUG" / "INFO" / "WARN" / "ERROR" / "FATAL": the tag MGLOG_* writes into a line's
+            // header, for a writer that has only the number. Out of range clamps to the nearest end.
+            const char* LogLevelTag(int level);
             // P6: ONE LOG PER ROLE, and the role is a per-THREAD fact because under inproc both
             // roles live in one process. The client keeps MOBILEGL_LOG_FILE_PATH unchanged so
             // every existing reader keeps its path; the server's lines go to the same path with

docs/Disaggregated/ARCHITECTURE.md

@@ -26,7 +26,7 @@
 | 1–2 | 边界：两张表、三种拓扑；对象模型：句柄、两种代数、CSO | [`design/01-boundary-and-objects.md`](design/01-boundary-and-objects.md) |
 | 3–4 | 调用目录、生成器 G1–G8、能力位；记录与 payload 格式 | [`design/02-catalogue-and-records.md`](design/02-catalogue-and-records.md) |
 | 5–7 | 前端 tracker（何时推、推什么、怎么去重）；纹理上传；着色器 | [`design/03-frontend.md`](design/03-frontend.md) |
-| 8 | 反向通道：九个回调、有序性、错误与 ack、纹理重铸拉取、XFB | [`design/04-reverse-channel.md`](design/04-reverse-channel.md) |
+| 8 | 反向通道：八个回调、有序性、错误与 ack、纹理重铸拉取、XFB | [`design/04-reverse-channel.md`](design/04-reverse-channel.md) |
 | 9–10 | 后端改造（`PipeInputs`、memo 重键、A/B 臂）；server 侧 applier | [`design/05-backend-and-server.md`](design/05-backend-and-server.md) |
 | 11 | 传输：共享段、环、门铃、控制面、背压、等待规则、事件、两根轴 | [`design/06-transport.md`](design/06-transport.md) |
 | 12–13 | persistent map 与大缓冲采纳；回读、往返清单、五部分验证门 | [`design/07-memory-readback-verification.md`](design/07-memory-readback-verification.md) |

docs/Disaggregated/design/03-frontend.md

@@ -60,5 +60,5 @@ shader 看见哪个纹理取决于 sampler uniform 类型、mipmap 完备性与
 
 - `CreateShaderState` 过线的是逐 stage SPIR-V + 反射归档，**不是源码**：glslang 全在 client、SPIRV-Cross 全在 server，文件级切割；server 侧无编译池。归档序列化器 `ProgramArtifactsCodec`，monolith 只在 verify 构建里调它（D-H3）。
 - server 在 verb 时刻从已推送状态**惰性特化**（D-B2：draw FBO clamp mask、fragColor 广播数、storage-block 绑定签名、image 格式、patch 参数等）——正是两个后端原来的做法。
-- link / compile 失败不需要同步返回：`GL_LINK_STATUS` 由 client 从 `ProgramObject` 回答，后端失败以 `OnLog` ≥ERROR 无损呈现。
+- link / compile 失败不需要同步返回：`GL_LINK_STATUS` 由 client 从 `ProgramObject` 回答，后端失败以 server 日志的 ≥ERROR 行（控制面 `LogLine`，无损，`MG_Remote/Transport/LogForward.h`）呈现。
 - 内部 shader（颜色 blit、深度 mip、multisample resolve）烘焙成签进树的 SPIR-V（`Wire*Spirv.h`），新鲜度门 `MOBILEGL_BAKED_INTERNAL_SHADERS`（`MG_Test/SelfTest/BakedInternalShadersTest.cpp`）重跑树内 glslang 逐字节比对（P7）。

docs/Disaggregated/design/04-reverse-channel.md

@@ -6,7 +6,7 @@
 
 ### 8.1 `MGPipeCallbacks`（`MG_Pipe/MGPipeCallbacks.h`）
 
-九个具名回调 + 一个正向终止符（`ResourceSubDataComplete`），取代后端直接 poke 前端对象的 95 个调用点（具名化是有意偏离 D8）。monolith 下直调，split 下是 `SEG_EVENT` 上的记录。
+八个具名回调 + 一个正向终止符（`ResourceSubDataComplete`），取代后端直接 poke 前端对象的 95 个调用点（具名化是有意偏离 D8）。monolith 下直调，split 下是 `SEG_EVENT` 上的记录。
 
 | 回调 | 作用 |
 |---|---|
@@ -18,7 +18,8 @@
 | `OnMipLevelsGenerated` | 只带形状 |
 | `OnSurfaceChanged` | 默认帧缓冲的格式与尺寸（server 拥有显示时 client 靠它得知窗口尺寸） |
 | `OnCapsInvalidated` | 取代 `InvalidateCompileEnv` |
-| `OnLog` | ≤WARN 有损，≥ERROR 无损 + 速率限制 |
+
+`OnLog` 已删除（P9 W3，回调 9 → 8）：零生产者、零消费者、无 `EventKind`。server 日志走控制面 `LogLine`（带 `level`），由 `MG_Remote/Transport/LogForward.h` 的队列 + 发送线程转发：≤WARN 有损（队列满即丢、计数、流内 `LogForward{Dropped}` 标出缺口），≥ERROR 无损 + 限速（超速排队延迟；超出错误字节上限时合并成一条 `LogForward{Coalesced}`），FATAL 有界等待写出；日志线程永不等 socket。
 
 后端凭空造的前端对象（Magma 占位纹理、swapchain 默认 FB 占位）改为 server 原生；pull 构建里的 `m_backend` 类成员真删会动 `sizeof`（G1），随 P13 退役（D-K）。