tx.report coverage attribution: fw emission ceiling, interleaved drops, dead MISSED_RPT_NUM (#363 step 1)

docs/logging.md

@@ -109,7 +109,7 @@ Emitters: L = library, RX/TX/... = demo. Optional fields in [brackets];
 | `tx.frame` | TX | n, rc — precoder demo variant: n, ok |
 | `tx.stats` | TX | submitted, failed, was_timeout, last_rc |
 | `tx.agg` | L (`DEVOURER_TX_USB_AGG`, send_packets) | frames, bytes, shim, ok — one per multi-frame bulk-OUT URB |
-| `tx.report` | L (`DEVOURER_TX_REPORT`, CCX decode) | state (0=delivered, 1=retry-drop), ok, retries, final_rate, queue_time_raw, bmc, macid, fmt ("8812"\|"halmac"); halmac adds tag (SW_DEFINE echo), rts_retries, missed |
+| `tx.report` | L (`DEVOURER_TX_REPORT`, CCX decode) | t, state (0=delivered, 1=retry-drop), ok, retries, final_rate, queue_time_raw, bmc, macid, fmt ("8812"\|"halmac"); halmac adds tag (SW_DEFINE echo), rts_retries, missed (fw-stuffed constant on Jaguar3 — tag gaps are the drop signal; `tests/txrpt_coverage_attrib.py`) — t is the achieved-report-rate timebase (the CCX emission ceiling is reports/s) |
 | `tx.status` | RX, duplex (C2H TX_RPT decode) | hoff, queue, retry, airtime_us, rate |
 | `tx.queue` | RX (`DEVOURER_QUEUE_POLL_MS`, 8814) | q1…q5 "0x%08x" |
 | `tx.contx` | TX (continuous mode) | mcs, t_ms |

docs/scheduled-mac.md

@@ -207,6 +207,18 @@ separately proven (`tests/ack_responder_check.sh`).
    HalMAC tag gaps (interior losses); the reliability layer must tolerate a
    ~5–15% report-less frame tail (treat missing report as "unknown", not
    "delivered").
+
+   Coverage is rate-bounded (`tests/txrpt_coverage_attrib.py`, 8812CU TX):
+   the CCX emission path saturates at ~1.3–1.4k reports/s — full coverage to
+   ~1.25 k fps, then `coverage ≈ ceiling/fps` (measured 99.4% @ 1.26 k,
+   77.2% @ 1.82 k, 53–56% @ 2.39 k fps). The excess drops per-report and
+   interleaved (99.3% of unreported frames sit in tag gaps ≤ 2 — an
+   emission-time rate limiter, not transport-batch loss), and the CCX
+   MISSED_RPT_NUM field is stuffed with a constant on this fw (verified
+   against the 8822B/C/E vendor headers — parse is exact, the fw just
+   doesn't populate it), so tag gaps are the only drop signal. Above the
+   ceiling, either sample SPE_RPT 1-in-N to keep the demanded rate under
+   ~1.3 k/s or account report-less frames as "unknown".
 2. **Closed-loop hardware ACK + autonomous retry is GO on Jaguar1 and
    Jaguar3** (100% delivery, retries ≈ 0.2–0.3) including retargeting an
    arbitrary UE MAC mid-session (re-arm `SetAckResponder`, change the

src/TxReport.h

@@ -44,7 +44,13 @@ struct TxReport {
   uint16_t queue_time_raw = 0; /* Jaguar1: 256 µs units; halmac: raw fw units */
   uint8_t final_rate = 0;  /* DESC_RATE* hw index of the final attempt */
   uint8_t sw_define = 0;   /* halmac: descriptor SW_DEFINE byte-0 echo */
-  uint8_t missed_rpt = 0;  /* halmac: reports the fw dropped before this one */
+  uint8_t missed_rpt = 0;  /* halmac MISSED_RPT_NUM — bench-measured on the
+                            * 8812CU: the fw stuffs a CONSTANT (4) here on
+                            * every report, drops or no drops, so the field
+                            * carries no drop accounting on Jaguar3. Detect
+                            * emission drops from SW_DEFINE tag gaps instead;
+                            * the measured emission ceiling and gap shape are
+                            * in docs/scheduled-mac.md. */
 };
 
 /* Jaguar1 (8812/8821) CCX payload: `p` points AFTER the 2-byte C2H envelope
@@ -103,7 +109,11 @@ inline TxReport parse_ccx_halmac(const uint8_t *c2h, size_t len) {
 inline void emit_tx_report(EventSink &sink, const TxReport &r,
                            const char *fmt) {
   Ev ev(sink, "tx.report");
-  ev.f("state", static_cast<unsigned>(r.state))
+  /* t (host monotonic ms): the achieved-report-rate instrument — the CCX
+   * emission ceiling is a reports-per-second quantity, unmeasurable from a
+   * stream with no timebase. */
+  ev.t()
+      .f("state", static_cast<unsigned>(r.state))
       .f("ok", r.state == 0)
       .f("retries", static_cast<unsigned>(r.data_retries))
       .f("final_rate", static_cast<unsigned>(r.final_rate))

tests/txrpt_coverage_attrib.py

@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+"""tx.report coverage attribution — where do the missing reports go?
+
+Per recorded arq_e2e run: coverage (reports/submitted), the achieved report
+rate, and the RUN-LENGTH histogram of unreported frames between consecutive
+reports (tag-unwrap deltas; the unwrap is trustworthy whenever the run's
+ledger join matched — see arq_e2e_analyze).
+
+How to read the gap shape: interleaved 1–2-frame gaps are per-report
+emission-time drops at the fw/C2H-mailbox (a rate limiter); large contiguous
+gaps would be transport-batch loss (a dropped C2H aggregate loses dozens of
+reports at once). Bench-measured on the 8812CU (Jaguar3): the fw's CCX
+emission saturates at ~1.3k reports/s — above that, excess reports are
+dropped individually and interleaved (99.96% of unreported frames in gaps
+<= 2), and the CCX MISSED_RPT_NUM field is stuffed with a constant (4), so
+the fw offers no drop accounting of its own (layout verified identical
+across the 8822B/8822C/8822E vendor headers; the constant is fw behaviour,
+not a parse bug). Consequence for accounting layers: coverage is
+~min(1, ceiling/fps) — either keep the demanded report rate under the
+ceiling (sample SPE_RPT 1-in-N) or treat report-less frames as "unknown".
+
+  python3 tests/txrpt_coverage_attrib.py /tmp/arq-e2e/<run> [...]
+"""
+import json
+import sys
+from collections import Counter
+
+
+def analyze(rundir):
+    n = 0
+    submitted = 0
+    prev_tag = None
+    gaps = Counter()
+    missed_vals = Counter()
+    t_first = t_last = None
+    with open(f"{rundir}/drone.jsonl", errors="replace") as f:
+        for line in f:
+            if line.startswith('{"ev":"tx.stats"'):
+                try:
+                    submitted = max(submitted,
+                                    json.loads(line).get("submitted", 0))
+                except Exception:
+                    pass
+                continue
+            if not line.startswith('{"ev":"tx.report"'):
+                continue
+            try:
+                ev = json.loads(line)
+            except Exception:
+                continue
+            tag = ev.get("tag")
+            if tag is None:
+                continue
+            n += 1
+            missed_vals[int(ev.get("missed", 0))] += 1
+            t = ev.get("t")
+            if t is not None:
+                t_first = t if t_first is None else t_first
+                t_last = t
+            if prev_tag is not None:
+                d = (tag - prev_tag) % 256
+                if d > 1:
+                    gaps[d - 1] += 1
+            prev_tag = tag
+    if n == 0:
+        raise SystemExit(f"{rundir}: no tagged tx.report events "
+                         f"(J1-format reports carry no tag)")
+    name = rundir.rstrip("/").split("/")[-1]
+    cov = 100.0 * n / max(1, submitted)
+    dur_s = (t_last - t_first) / 1000.0 if (t_first is not None and
+                                            t_last and t_last > t_first) else 0
+    rate = n / dur_s if dur_s else 0
+    unrep = sum(k * v for k, v in gaps.items())
+    small = sum(k * v for k, v in gaps.items() if k <= 2)
+    top = ", ".join(f"{k}x{v}" for k, v in sorted(gaps.items())[:8])
+    print(f"\n== {name}: submitted={submitted} reports={n} "
+          f"coverage={cov:.1f}%"
+          + (f" achieved={rate:.0f} rpt/s" if rate else ""))
+    print(f"   unreported={unrep} gap-hist [{top}"
+          f"{', ...' if len(gaps) > 8 else ''}] "
+          f"max={max(gaps) if gaps else 0} "
+          f"in-gaps<=2: {100.0 * small / max(1, unrep):.1f}%")
+    if len(missed_vals) == 1:
+        (mv, _), = missed_vals.items()
+        print(f"   missed field: constant {mv} on every report "
+              f"(fw drop-accounting unavailable)")
+    else:
+        # A fw that populates the field gets the literal reconciliation:
+        # summed fw-acknowledged drops vs the tag-gap ground truth. The
+        # 3-bit field saturates at 7, so a shortfall with gaps > 7 present
+        # is saturation, not necessarily transport loss.
+        fw_sum = sum(mv * c for mv, c in missed_vals.items())
+        print(f"   missed field values: {dict(sorted(missed_vals.items()))} "
+              f"— fw-acknowledged drops {fw_sum} vs unreported {unrep} "
+              f"(delta {unrep - fw_sum})")
+
+
+def main():
+    for rd in sys.argv[1:]:
+        analyze(rd)
+
+
+if __name__ == "__main__":
+    main()