[Perf] (MG_Remote): the client counts RECORDS SENT per op beside the waits each row PAID, and the pair says item B stopped the waiting without stopping the sending

- A wait count alone cannot say whether a frame is slow because it sends too much or because it
  waits too often. The device run made that the next question: one atlas frame paid 43,232 waits
  while the upload row paid none.
- LinkMetricsNoteRecord is called at the encoder's one commit point (PipeWireCodec.cpp, beside the
  existing CallClass::WireRecords count, for the comment's reason: after chunking no emitter knows
  how many records its call produced). LinkMetricsRecordsFor reads it back, and every frame that
  emitted something now logs `P65LinkMetrics kind=frame-sent frame=N records=M by_op=...`.
- The host measurement now prints and asserts both halves for the same workload. An atlas-shaped
  load (64 sub-uploads into one level) sends 128 ResourceSubData records and waits for ZERO of
  them, while ResourceCreate 76/76, ResourceRespecify 130/130 and SetTextureParams 130/130 are
  one wait per record. That is item B stated exactly: the records still go out; the round trips do
  not. It also says what the next thing to fix is - the record COUNT of those three rows, not their
  waits, which the contract rules they must keep.

MobileGL/MG_Remote/Transport/LinkMetrics.cpp

@@ -18,6 +18,7 @@ namespace {
         // One slot per wire op, plus a last slot for anything past the table. See
         // LinkMetricsMaxOps: the point is that the per-op numbers still SUM to waitReplies.
         std::array<std::uint64_t, LinkMetricsMaxOps> waitRepliesByOp{};
+        std::array<std::uint64_t, LinkMetricsMaxOps> emittedByOp{};
     };
     // The client already has exactly one record producer. The server never writes these.
     struct Metrics {
@@ -64,6 +65,22 @@ namespace {
         }
         if (at == 0) std::snprintf(out, outSize, "(none)");
     }
+    // The same shape for the emission tally: one formatter, two arrays, so the two lines cannot
+    // drift apart in their id spelling.
+    void FormatCounts(const std::array<std::uint64_t, LinkMetricsMaxOps>& counts, char* out,
+                      std::size_t outSize) {
+        std::size_t at = 0;
+        out[0] = '\0';
+        for (std::uint32_t op = 1; op < LinkMetricsMaxOps && at + 24 < outSize; ++op) {
+            if (counts[op] == 0) continue;
+            const int n = std::snprintf(out + at, outSize - at, "%s%u=%llu", at == 0 ? "" : ",",
+                                        static_cast<unsigned>(op),
+                                        static_cast<unsigned long long>(counts[op]));
+            if (n < 0 || static_cast<std::size_t>(n) >= outSize - at) break;
+            at += static_cast<std::size_t>(n);
+        }
+        if (at == 0) std::snprintf(out, outSize, "(none)");
+    }
     void Emit(const char* kind, const Window& w, std::uint64_t wallNs, std::uint64_t cpuNs) {
         char histogram[768]{};
         std::size_t at = 0;
@@ -95,6 +112,18 @@ namespace {
                     static_cast<unsigned long long>(metrics.frame),
                     static_cast<unsigned long long>(w.waitReplies), split);
         }
+        // WHAT THE FRAME SENT, not only what it paid for. Printed for the same frames - a frame
+        // that bought no wait still SENT records, and the count is what says whether the waits are
+        // one per record (a real handshake) or fewer (a batched one).
+        std::uint64_t emitted = 0;
+        for (std::uint32_t op = 1; op < LinkMetricsMaxOps; ++op) emitted += w.emittedByOp[op];
+        if (emitted != 0) {
+            char sent[1024];
+            FormatCounts(w.emittedByOp, sent, sizeof(sent));
+            MGLOG_I("P65LinkMetrics kind=frame-sent frame=%llu records=%llu by_op=%s",
+                    static_cast<unsigned long long>(metrics.frame),
+                    static_cast<unsigned long long>(emitted), sent);
+        }
     }
 }
 void LinkMetricsBegin() {
@@ -122,6 +151,14 @@ void LinkMetricsReplyApplied(std::uint64_t startedNs) {
         ++w->samples; w->replyNs += ns; ++w->histogram[bucket];
     }
 }
+void LinkMetricsNoteRecord(std::uint32_t op) {
+    if (!metrics.active) return;
+    const std::uint32_t slot = op < LinkMetricsMaxOps ? op : LinkMetricsMaxOps - 1;
+    ++metrics.current.emittedByOp[slot]; ++metrics.total.emittedByOp[slot];
+}
+std::uint64_t LinkMetricsRecordsFor(std::uint32_t op) {
+    return metrics.total.emittedByOp[op < LinkMetricsMaxOps ? op : LinkMetricsMaxOps - 1];
+}
 std::uint64_t LinkMetricsReplyWaits() { return metrics.total.waitReplies; }
 std::uint64_t LinkMetricsReplyWaitsFor(std::uint32_t op) {
     return metrics.total.waitRepliesByOp[op < LinkMetricsMaxOps ? op : LinkMetricsMaxOps - 1];

MobileGL/MG_Remote/Transport/LinkMetrics.h

@@ -26,6 +26,14 @@ namespace MobileGL::MG_Remote::Transport {
     // still counts, or it would pass on a session where nothing was counted at all.
     std::uint64_t LinkMetricsReplyWaits();
     std::uint64_t LinkMetricsReplyWaitsFor(std::uint32_t op);
+
+    // RECORDS EMITTED, PER OP - the other half of the same question. A wait count says what a frame
+    // PAID; this says what it SENT, and the device run that motivated it showed one atlas frame
+    // paying 43,232 waits while the upload row paid none - so the next thing to know is how many
+    // records those waits were bought for and by whom. Counted at the encoder's one commit point
+    // (PipeWireCodec.cpp, ++m_emitSeq), so no row can be emitted without being counted.
+    void LinkMetricsNoteRecord(std::uint32_t op);
+    std::uint64_t LinkMetricsRecordsFor(std::uint32_t op);
     void LinkMetricsPresent();
     void LinkMetricsServerPresent(std::uint64_t serial);
 }

MobileGL/MG_Remote/Wire/PipeWireCodec.cpp

@@ -1266,6 +1266,12 @@ namespace MobileGL::MG_Remote::Wire {
         if (MG_Util::PipeStats::Enabled()) {
             MG_Util::PipeStats::AddCalls(MG_Util::PipeStats::CallClass::WireRecords, 1);
         }
+        // THE SAME COMMIT, SPLIT BY OP (P12). The CallClass above answers "how many records"; this
+        // answers "which rows", which is the question a wait count cannot: the device run had one
+        // atlas frame pay 43,232 waits while the upload row paid none, so the next thing to know is
+        // how many records those waits were bought for and by whom. Counted here for the reason the
+        // comment above gives - after chunking, no emitter knows how many records its call made.
+        Transport::LinkMetricsNoteRecord(static_cast<std::uint32_t>(op));
 
         ++m_emitSeq;
         // The stage mark: where SEG_STAGE stood once everything this record names had been

MobileGL/MG_Test/Wire/RemoteClientControls.inc

@@ -972,6 +972,20 @@ TEST(RemoteClientControls, ReplyWaitsByOpForAnAtlasShapedLoad) {
         }
         // PRINT FIRST, ASSERT SECOND: the table is the point of the case, and a gate that fired
         // before it would hide the numbers on exactly the run that needs them - the red one.
+        //
+        // BOTH HALVES OF THE SAME FRAME: what it PAID (waits) and what it SENT (records). They are
+        // different questions and the pair is the point - item B did not stop the upload records
+        // from being sent, it stopped them from being waited for.
+        std::printf("records sent by op (whole session):\n");
+        for (Uint32 op = 1; op < Transport::LinkMetricsMaxOps; ++op) {
+            const Uint64 records = Transport::LinkMetricsRecordsFor(op);
+            if (records != 0)
+                std::printf("  op=%-24s records=%llu waits=%llu\n",
+                            Codec::WireOpName(static_cast<MGPWireOp>(op)),
+                            static_cast<unsigned long long>(records),
+                            static_cast<unsigned long long>(Transport::LinkMetricsReplyWaitsFor(op)));
+        }
+        std::fflush(nullptr);
         std::printf("shape 2 a texture per sprite: %d textures -> wait_replies=%llu\n", kSprites,
                     static_cast<unsigned long long>(Transport::LinkMetricsReplyWaits() - atlasWaitsBefore));
         for (Uint32 op = 1; op < Transport::LinkMetricsMaxOps; ++op) {
@@ -985,6 +999,11 @@ TEST(RemoteClientControls, ReplyWaitsByOpForAnAtlasShapedLoad) {
         // drives must NOT be, or the case would also pass on a session that counted nothing.
         if (Transport::LinkMetricsReplyWaitsFor(static_cast<Uint32>(MGPWireOp::ResourceSubData)) != 0)
             ::_exit(143);
+        // AND THE UPLOAD RECORDS WERE STILL SENT: the shape 1 stitch is 64 sub-uploads, so the row
+        // must show 64 records with 0 waits. A pair that went to 0/0 together would mean the
+        // emitter stopped sending, which is a different (and much worse) change than item B.
+        if (Transport::LinkMetricsRecordsFor(static_cast<Uint32>(MGPWireOp::ResourceSubData)) < 64)
+            ::_exit(148);
         if (Transport::LinkMetricsReplyWaitsFor(static_cast<Uint32>(MGPWireOp::ResourceCreate)) == 0)
             ::_exit(144);
         if (Transport::LinkMetricsReplyWaitsFor(static_cast<Uint32>(MGPWireOp::ResourceRespecify)) == 0)