From 306c32d60e3459484bb6644b5529fa36890729d4 Mon Sep 17 00:00:00 2001
From: l5y <220195275+l5yth@users.noreply.github.com>
Date: Sat, 25 Jul 2026 23:10:50 +0200
Subject: [PATCH] data,web: mesh activity reporting & announcements

---
 ACCEPTANCE.md                                 | 154 ++++++
 SPEC.md                                       |  58 ++
 data/ingestor_activity.sql                    |  35 ++
 data/mesh.sh                                  |   4 +-
 data/mesh_ingestor/CONTRACTS.md               |  15 +
 data/mesh_ingestor/activity.py                | 127 +++++
 data/mesh_ingestor/announce.py                | 404 ++++++++++++++
 data/mesh_ingestor/config.py                  |   6 +-
 data/mesh_ingestor/daemon.py                  |  38 +-
 data/mesh_ingestor/handlers/_state.py         |  11 +-
 data/mesh_ingestor/ingestors.py               |   7 +-
 .../protocols/meshcore/_constants.py          |   9 +
 .../protocols/meshcore/provider.py            |  41 +-
 .../protocols/meshcore/telemetry.py           |  28 +-
 data/mesh_ingestor/protocols/meshtastic.py    |  28 +-
 tests/test_activity_unit.py                   | 200 +++++++
 tests/test_announce_unit.py                   | 506 ++++++++++++++++++
 tests/test_daemon_unit.py                     |  27 +
 tests/test_provider_unit.py                   | 228 ++++++++
 .../application/data_processing/ingestors.rb  |  37 ++
 web/lib/potato_mesh/application/database.rb   |  13 +-
 web/lib/potato_mesh/application/queries.rb    |   1 +
 .../application/queries/ingestor_queries.rb   |  90 ++++
 web/lib/potato_mesh/application/retention.rb  |   1 +
 web/lib/potato_mesh/application/routes/api.rb |  11 +-
 web/spec/activity_spec.rb                     | 149 ++++++
 web/spec/app_spec.rb                          |  40 ++
 web/spec/queries_spec.rb                      |  80 +++
 web/spec/retention_spec.rb                    |  31 ++
 29 files changed, 2362 insertions(+), 17 deletions(-)
 create mode 100644 data/ingestor_activity.sql
 create mode 100644 data/mesh_ingestor/activity.py
 create mode 100644 data/mesh_ingestor/announce.py
 create mode 100644 tests/test_activity_unit.py
 create mode 100644 tests/test_announce_unit.py
 create mode 100644 web/lib/potato_mesh/application/queries/ingestor_queries.rb
 create mode 100644 web/spec/activity_spec.rb

diff --git a/ACCEPTANCE.md b/ACCEPTANCE.md
index 9b6d1e0..f78b67c 100644
--- a/ACCEPTANCE.md
+++ b/ACCEPTANCE.md
@@ -3979,3 +3979,157 @@ survive the swatch-shape change), **FU-A2** (the region toggle) and **UX-A9**
 (chat hanging indent — the `overflow-wrap` addition rides the same D-033 rule),
 and the `buildRoleButtons` filter specs (swatch/dataset/compound-key behaviour
 unchanged). No Ruby/Python/Rust/Flutter surface is touched.
+
+---
+
+## Feature: Mesh activity reporting & announcements
+
+Maps to SPEC decisions **MA1–MA10**. Each ingestor counts **every** frame it
+handles (all RX, incl. ignored/errored/unimplemented, plus its own TX) as one
+merged `packets` figure, appends the per-interval delta to its hourly
+`POST /api/ingestors` heartbeat (MA1–MA2); the web app persists a per-ingestor
+`ingestor_activity` time-series (MA3) from which `GET /api/stats` derives a
+`MAX`-per-protocol 24 h packets/hour moving average (MA4–MA5). Each ingestor then
+periodically broadcasts a one-line activity summary — numbers **dogfed from the
+target instance's own API** — on its protocol's default channel (MA6), gated by
+`RX_ONLY` (reused as the transmit gate), the target's `/version` privacy flag
+(fail-closed), and a ≥ 24 h post-start delay (MA7–MA8), via an **optional**
+duck-typed provider send
+that leaves `MeshProtocol` conformance intact (MA9). Unless a check says
+otherwise, start the server in **public** mode
+(`API_TOKEN=acctest PRIVATE=0 FEDERATION=0 bundle exec ruby app.rb`).
+
+### MA-A1 — Every frame is counted, including drops; TX too — MA1
+```bash
+( . .venv/bin/activate && pytest -q tests/test_activity_unit.py -k "count" )
+```
+**Expected:** pass. The merged `packets` counter increments once per received
+frame at the earliest seam (`handlers/_state._mark_packet_seen`) — verified for a
+**stored** packet, an **ignored** packet (`unsupported-port` / `no-message-payload`),
+and an **errored** packet (one that raises inside `store_packet_dict`) — and once
+per ingestor **transmission** (the announcement send and a MeshCore
+telemetry/status poll). The count is taken *before* any drop/dispatch decision, so
+no receive or transmit path bypasses it ("we don't want to under-report").
+
+### MA-A2 — Heartbeat carries a per-interval delta that resets — MA2
+```bash
+( . .venv/bin/activate && pytest -q tests/test_activity_unit.py -k "heartbeat_delta" )
+( . .venv/bin/activate && pytest -q tests/test_mesh.py -k "ingestor" )
+```
+**Expected:** pass. `queue_ingestor_heartbeat` includes `packets` = frames counted
+since the previous heartbeat and **zeroes** the running counter afterward: two
+heartbeats bracketing N then M frames report N then M (never N then N+M). A
+heartbeat with no traffic sends `0` (or omits the field). A pre-feature payload
+without `packets` is still accepted by `POST /api/ingestors` (additive, D8).
+
+### MA-A3 — Activity time-series: append per heartbeat, pruned by retention — MA3
+```bash
+( cd web && bundle exec rspec spec/activity_spec.rb )
+( cd web && bundle exec rspec spec/retention_spec.rb -e "ingestor_activity" )
+git grep -nE 'ingestor_activity' -- data/ingestor_activity.sql web/lib
+```
+**Expected:** pass, and the grep shows a new **append-only** `ingestor_activity`
+table (`ingestor_id`, `at`, `packets`, `protocol`, index on `at`). Each
+`POST /api/ingestors` carrying `packets` appends exactly one row (the `ingestors`
+snapshot row is still upserted, one per node); two ingestors of one protocol write
+**independent** rows (not pre-summed). The retention worker deletes activity rows
+older than the configured window (≥ 24 h), so the table cannot grow unbounded.
+
+### MA-A4 — MAX-per-protocol packets/hour over 24 h — MA4
+```bash
+( cd web && bundle exec rspec spec/queries_spec.rb -e "packets_per_hour" )
+```
+**Expected:** pass. With two `meshcore` ingestors reporting different 24 h packet
+totals, `packets_per_hour.meshcore` = `MAX(total_A, total_B) ÷ 24` — the busiest
+single vantage, so the quieter ingestor and any overlap never inflate it. `total`
+is the `MAX` across **all** ingestors regardless of protocol; a protocol with no
+active ingestor reads `0`. Rows older than 24 h do not contribute.
+
+### MA-A5 — `/api/stats` exposes `packets_per_hour` additively — MA5
+```bash
+curl -s http://127.0.0.1:41447/api/stats \
+  | python3 -c 'import sys,json; d=json.load(sys.stdin); p=d["packets_per_hour"]; \
+SC=("total","meshcore","meshtastic","reticulum"); \
+print(all(isinstance(p[s],(int,float)) for s in SC) and p["reticulum"]==0 \
+and set(SC).issubset(d) and all(m in d["total"] for m in ("nodes","messages","telemetry")))'
+```
+**Expected:** prints `True` — the response carries an additive top-level
+`packets_per_hour` map keyed `{total, meshcore, meshtastic, reticulum}`
+(`reticulum` a `0` stub), **and** the pre-existing scope × metric × window tree
+(S1: each scope still carrying `nodes`/`messages`/`telemetry`) is unchanged and
+still present. No version bump — **S-A1** still passes.
+
+### MA-A6 — Announcement content is dogfed from the instance API — MA6
+```bash
+( . .venv/bin/activate && pytest -q tests/test_announce_unit.py -k "message or dogfeed" )
+```
+**Expected:** pass. The announcement string is exactly
+`"<Protocol> activity in the last 24h: <N> active nodes, <M> packets/hour. https://<domain>"`,
+where `<N>` = the target's `GET /api/stats` `<protocol>.nodes.day` and `<M>` =
+`GET /api/stats` `packets_per_hour.<protocol>` — both fetched over HTTP from
+`<domain>`, never computed from the ingestor's local counters — and the rendered
+line is truncated to the protocol's character limit. `<domain>` = the configured
+`INSTANCE_DOMAIN`.
+
+### MA-A7 — Announcement gates: RX_ONLY, privacy fail-closed, 24 h — MA7
+```bash
+( . .venv/bin/activate && pytest -q tests/test_announce_unit.py \
+    -k "gate or private or rx_only or elapsed" )
+```
+**Expected:** pass. **No** announcement is sent when any gate fails: `RX_ONLY=1`
+(the **reused** receive-only flag — its existing MeshCore-poll TX suppression now
+also covers the announcement, so it is the sole opt-out; no separate `ENABLE_TX`
+env exists); the target `/version` reports `private_mode: true`; the `/version`
+fetch **errors or is unparseable** (fail-closed — treated as private/skip); or
+`< 24 h` have elapsed since ingestor start. With `RX_ONLY` unset (the default
+`0`), `private_mode: false`, and `≥ 24 h` elapsed, exactly one announcement per
+24 h per domain is transmitted.
+
+### MA-A8 — Default channel/scope + 24 h cadence — MA8
+```bash
+( . .venv/bin/activate && pytest -q tests/test_announce_unit.py \
+    -k "channel or cadence or interval or domains" )
+```
+**Expected:** pass. The announcement is sent on Meshtastic channel `CHANNEL_INDEX`
+(default `0`) / MeshCore's public channel; the first fires no earlier than 24 h
+post-start and subsequent ones no more often than every 24 h; an ingestor with
+several `INSTANCE_DOMAIN` targets announces each once per cycle with that domain's
+own numbers and link.
+
+### MA-A9 — Send is optional and duck-typed; MeshProtocol conformance intact — MA9
+```bash
+( . .venv/bin/activate && pytest -q tests/test_provider_unit.py \
+    -k "send_channel_announcement or MeshProtocol" )
+```
+**Expected:** pass. Both `MeshtasticProvider` and `MeshcoreProvider` expose
+`send_channel_announcement(...)`; the announce scheduler resolves it via
+`getattr(provider, "send_channel_announcement", None)` and **no-ops when absent**
+(e.g. a receive-only transport). The `@runtime_checkable MeshProtocol` interface is
+unchanged — **A4b**'s `isinstance(provider, MeshProtocol)` conformance still passes
+and `send_channel_announcement` is **not** a required member.
+
+### MA-A10 — Additive contract is documented — MA10 / D8
+```bash
+git grep -nE 'packets|ingestor_activity|packets_per_hour' -- data/mesh_ingestor/CONTRACTS.md
+```
+**Expected:** the additive heartbeat `packets` field, the `ingestor_activity`
+schema, and the `GET /api/stats` `packets_per_hour` addition are all documented in
+`CONTRACTS.md` (Layer C source of truth). The engineering bar (100 % tests/docs/
+headers/lint) is enforced by Layer **B** (B1–B5); behavior is covered by
+MA-A1…MA-A9.
+
+### MA-R1 — Regression: prior acceptance still holds
+```bash
+( . .venv/bin/activate && pytest -q tests/ )
+( cd web && bundle exec rspec ) && ( cd web && npm test )
+```
+**Expected:** every prior check still passes. At risk and explicitly required to
+remain green: **A1a/A1b** (apex — the new ingestor→instance GET and the LoRa
+announcement add no broker term or dependency); **A4b** (MeshProtocol isinstance
+conformance — send stays optional/duck-typed, MA9); **S-A1** (the `/api/stats`
+scope × metric × window tree is unchanged; `packets_per_hour` is an additive
+sibling — no version bump, so `test_version_sync.py` is unaffected); **A2a/A2b**
+(privacy — the message API still 404s under `PRIVATE`, and the announcement
+fail-closes on the same flag, MA7); **C2** (`tests/test_mesh.py` — the
+`POST /api/ingestors` `packets` field is additive and old payloads still validate);
+and **B1** (all suites). No `/api/*` response shape is broken by construction.
diff --git a/SPEC.md b/SPEC.md
index 5bc28fb..9c66836 100644
--- a/SPEC.md
+++ b/SPEC.md
@@ -1045,3 +1045,61 @@ contradicted.
 | **LC2** | **Legend pressed state paints (specificity fix).** A role chip is a `<button>`; the reset `button:not(.chat-tab):not(.sort-button)` (0,2,1) outranked the bare `.legend-item[aria-pressed="true"]` (0,2,0), so every chip computed to `#333` and only `font-weight` distinguished selected from filtered. The selector is now `button.legend-item[aria-pressed="true"]` (0,2,1), which wins on source order — the selected blue paints. | review |
 | **LC3** | **Chat log never scrolls horizontally (extends UX11/D-033).** `.chat-tabpanel` is `overflow-y: auto`, so its x-axis is a scroll container; nothing broke long tokens. `overflow-wrap: anywhere` on `.chat-entry-msg, .chat-entry-node` wraps unbreakable tokens under the 19ch hang and lowers the entry's min-content width so it can never exceed the panel. `anywhere`, not `break-word`; no `overflow-x: hidden` (which would only mask the next regression). | review |
 | **LC4** | **Engineering bar (D9).** Each fix shipped with a fail-first check (Phase 2), full JSDoc/comments, the exact Apache header, clean linters; all prior suites stay green; the `buildRoleButtons` filter specs gain the swatch-shape assertion, none removed. | CLAUDE.md |
+
+---
+
+## Feature: Mesh activity reporting & announcements
+
+Turns each ingestor from a pure listener into a reporter **and** an announcer.
+(1) **Reporting:** every ingestor counts *every* frame it handles — all received
+frames (including ignored / errored / unimplemented) plus its own transmissions —
+as one merged `packets` figure and appends the per-interval delta to its hourly
+`POST /api/ingestors` heartbeat; the web app persists a per-ingestor activity
+time-series so a **packets/hour moving average** is computable across
+time × protocol × multiple ingestors. (2) **Announcing** ("we stop listening and
+start talking"): each ingestor periodically broadcasts a one-line activity summary
+on its protocol's default channel, drawing the numbers **back from the target
+instance's own API** (dogfeeding, so one radio's partial view never
+under-represents the mesh). Integrates with
+`data/mesh_ingestor/handlers/_state.py` (RX/TX counting seam), `ingestors.py`
+(heartbeat payload), `config.py` (reuse `RX_ONLY`), `daemon.py`
+(announce schedule), `mesh_protocol.py` + `protocols/{meshtastic,meshcore,meshtastic_udp}`
+(optional send), a new announce + instance-stats-GET module, `data/ingestors.sql`
+plus a new `data/ingestor_activity.sql` table,
+`web/lib/potato_mesh/application/routes/{ingest,api}.rb`, the query + retention
+layers, and `data/mesh_ingestor/CONTRACTS.md`.
+
+**Conflict check against existing decisions.** *Invariant I (apex / local-LoRa)*
+— **consistent, extends §4.2**: the announcement is a LoRa transmission on the
+local aether and the dogfeed is an HTTP GET to the ingestor's *own* instance; no
+MQTT/broker/cloud dependency or connection is added (A1 stays clean), and ingestor
+TX already exists (`RX_ONLY`, MeshCore telemetry polls), so this widens an existing
+capability rather than crossing the apex line. *Invariant II (privacy)* —
+**extends**: a new gate suppresses the announcement unless the target instance
+reports non-private, read authoritatively from its `/version` and **fail-closed**
+on any fetch error; only public aggregates (active nodes, packet rate) are ever
+announced. *Invariant III (federation) & §5 (no analytics / phone-home)* —
+**consistent, new authorized behavior**: the announcement publishes an *unsigned
+public activity line on the community's local mesh* — not the signed federation
+wire, not cloud egress, not third-party analytics — and is opt-out via
+`RX_ONLY=1`. *Invariant IV (parity & pluggability)* — **extends**: sending is an
+*optional, duck-typed* provider capability, not a new formal `MeshProtocol` member,
+so the `A4b` conformance contract and every existing provider are untouched; both
+protocols announce identically. *D8 (stable contract)* — **extends**: the heartbeat
+`packets` field, the `/api/stats` `packets_per_hour` map, and the activity table
+are all additive; no version bump. *§3.3 (web is POST-only intake)* —
+**consistent**: the dogfeed is a read; no new ingest path. No invariant is
+contradicted.
+
+| # | Decision | Source |
+| --- | --- | --- |
+| **MA1** | **Merged packet counter — count everything, at the earliest seam.** Each ingestor maintains one merged `packets` counter incremented on **every received frame** — *all of them, including ignored / errored / unimplemented / unsupported-port packets* — counted at the earliest common receive seam (`handlers/_state._mark_packet_seen`, already invoked by both the Meshtastic `on_receive` path and every MeshCore handler/telemetry path *before* any dispatch or filtering), **plus every ingestor-initiated transmission** (the announcement itself and the existing MeshCore telemetry/status polls). RX and TX are deliberately **merged** into a single figure (no separate breakdown). "We don't want to under-report" is satisfied at source: counting precedes every drop/ignore decision. A build-phase check confirms each protocol's RX entry funnels through the seam so no family is missed. | interview |
+| **MA2** | **Heartbeat carries the per-interval delta.** `POST /api/ingestors` gains an additive optional `packets` field = frames counted since the previous heartbeat (a per-interval **delta**, not a since-boot cumulative), reset to zero each time a heartbeat is queued. Per-interval deltas map one-to-one onto time-series rows and are restart-safe (a reboot simply starts a fresh interval). Absent ⇒ treated as `0`, so pre-feature ingestors and the existing `tests/test_mesh.py` fixtures are unaffected (D8). | interview |
+| **MA3** | **Per-ingestor activity time-series (the moving-average schema).** A new append-only table `ingestor_activity` (`ingestor_id TEXT`, `at INTEGER`, `packets INTEGER`, `protocol TEXT`, indexed on `at`) records one row per heartbeat delta. This is the schema that makes a packets/hour moving average computable while distinguishing **multiple protocols and multiple ingestors per protocol** (each ingestor's contribution is kept separate rather than pre-summed). The `ingestors` snapshot table (one row per node) is unchanged. Rows are pruned by the existing retention worker (window ≥ the 24 h the announcement needs; sized with the other activity floors). | proposed |
+| **MA4** | **Aggregation = MAX per protocol (dedup-free "in the air").** The mesh-wide packets/hour for a protocol is `MAX` over that protocol's ingestors of *(that ingestor's total `packets` reported in the last 24 h ÷ 24)*. A single radio can only hear ≤ what is actually transmitted, so the busiest single vantage is the best dedup-free estimate of unique air traffic and can never double-count a frame heard by two radios (true per-frame dedup is impossible anyway — ignored/errored frames carry no id). The fixed `÷ 24` denominator (not ÷ elapsed) keeps the rate stable and avoids divide-by-small spikes; an ingestor with < 24 h of data simply reads lower and ramps up (moot in practice — the announcement fires only ≥ 24 h after start, MA7). *Accepted limitation:* MAX under-estimates the true union when different radios are busiest in different hours. | interview |
+| **MA5** | **Exposure via an additive `/api/stats` field.** `GET /api/stats` gains a top-level additive `packets_per_hour` map — `{ total, meshcore, meshtastic, reticulum }` — carrying the MA4 24 h MAX-aggregated moving average (`reticulum` a forward-looking `0` stub, consistent with S6; `total` is the MAX across all ingestors regardless of protocol). The existing scope × metric × window tree (S1) is untouched, so this needs **no** version bump. The announcement's active-node figure reuses the existing `<protocol>.nodes.day` count (already deduped by `node_id`); only packets/hour is new. | proposed |
+| **MA6** | **Announcement content, drawn from the instance's own API (dogfeeding).** Each ingestor broadcasts, for its own configured protocol, a single line formatted to that protocol's character limit: `"<Protocol> activity in the last 24h: <N> active nodes, <M> packets/hour. https://<domain>"`. `<N>` and `<M>` are fetched **from the target instance** (`GET /api/stats` → `<protocol>.nodes.day` and `packets_per_hour.<protocol>`), never computed from the ingestor's local view — because one ingestor may not see the whole mesh. `<domain>` is the configured `INSTANCE_DOMAIN`. Reporting (MA1–MA4) is independent and continues regardless of announcement state. | interview |
+| **MA7** | **Announcement is triple-gated.** An announcement is transmitted only when **all** hold: (a) `RX_ONLY` is unset — the **reused** receive-only flag (default `0`) is the single transmit gate; `RX_ONLY=1` forbids *every* ingestor TX (the MeshCore polls and the announcement alike), so it is the sole opt-out and **no separate `ENABLE_TX` env is added**; (b) the target instance reports **non-private** — the ingestor GETs `<domain>/version` and honors `config.private_mode`, re-checked every cycle, and **fails closed** (skips) on any fetch/parse error, so a privacy signal is never missed (Invariant II); (c) **≥ 24 h have elapsed since ingestor start** — so the first numbers are accurate over a full window and restarts cannot spam the channel. | interview |
+| **MA8** | **Default channel/scope + 24 h cadence.** The announcement is broadcast on the protocol's **default channel and scope** — Meshtastic channel `CHANNEL_INDEX` (default `0`) via the interface text-send; MeshCore its public/default channel — honoring existing `ALLOWED_CHANNELS`/`HIDDEN_CHANNELS` intent. After the initial ≥ 24 h wait it repeats **every 24 h**, per configured instance domain (an ingestor with several `INSTANCE_DOMAIN` targets announces each with its own numbers and link). TX volume is negligible (~1 frame/day/domain). | proposed |
+| **MA9** | **Optional duck-typed provider send.** Transmission is exposed as a new **optional** provider method (e.g. `send_channel_announcement(iface, text)`) accessed via `getattr(provider, …, None)` — mirroring the existing optional `self_node_item` extension — so the `@runtime_checkable MeshProtocol` interface and its `A4b` isinstance conformance are unchanged, and a provider that cannot transmit (or a receive-only transport) simply omits it. Both `meshtastic` and `meshcore` providers implement it; neither protocol is privileged. | interview |
+| **MA10** | **Engineering bar (D9).** Every new/changed unit ships with 100 % unit tests (counter increment across RX families + TX; heartbeat delta + reset; activity-table write + retention; MAX-aggregation math incl. multi-ingestor; the `/api/stats` field; the four announcement gates incl. fail-closed privacy and the 24 h wait; per-protocol send), full PDoc/RDoc, Apache headers, `black`/`rufo` clean; `pytest`/`rspec`/`npm test` stay green; `CONTRACTS.md` documents the additive heartbeat field, the activity schema, and the `/api/stats` addition. | CLAUDE.md |
diff --git a/data/ingestor_activity.sql b/data/ingestor_activity.sql
new file mode 100644
index 0000000..6ff38d9
--- /dev/null
+++ b/data/ingestor_activity.sql
@@ -0,0 +1,35 @@
+-- Copyright © 2025-26 l5yth & contributors
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+--
+--     http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+PRAGMA journal_mode=WAL;
+
+-- Per-heartbeat mesh-activity time-series (SPEC MA3).
+--
+-- Append-only: one row per ingestor heartbeat that carries a `packets` count
+-- (the merged RX+TX frames observed since the previous heartbeat, MA1/MA2).
+-- Each ingestor's contribution is kept as its own rows rather than pre-summed,
+-- so a packets/hour moving average can be computed across time, protocol, and
+-- multiple ingestors per protocol (the read-side MAX aggregation, MA4). Rows
+-- are pruned by the retention worker on `at`.
+CREATE TABLE IF NOT EXISTS ingestor_activity (
+  id           INTEGER PRIMARY KEY,
+  ingestor_id  TEXT NOT NULL,
+  at           INTEGER NOT NULL,
+  packets      INTEGER NOT NULL,
+  protocol     TEXT NOT NULL DEFAULT 'meshtastic'
+);
+
+CREATE INDEX IF NOT EXISTS idx_ingestor_activity_at ON ingestor_activity(at);
+CREATE INDEX IF NOT EXISTS idx_ingestor_activity_protocol_at
+  ON ingestor_activity(protocol, at);
diff --git a/data/mesh.sh b/data/mesh.sh
index 6f7140d..0993a81 100755
--- a/data/mesh.sh
+++ b/data/mesh.sh
@@ -32,4 +32,6 @@ if ! .venv/bin/python -c "import sys; exit(0 if '.venv' in sys.prefix else 1)" 2
 fi
 .venv/bin/python -m pip install -U pip
 .venv/bin/python -m pip install -r "$(dirname "$0")/requirements.txt"
-exec .venv/bin/python mesh.py
+
+# -u keeps stdout unbuffered
+exec .venv/bin/python -u mesh.py
diff --git a/data/mesh_ingestor/CONTRACTS.md b/data/mesh_ingestor/CONTRACTS.md
index 784d8c5..9b8027e 100644
--- a/data/mesh_ingestor/CONTRACTS.md
+++ b/data/mesh_ingestor/CONTRACTS.md
@@ -218,6 +218,9 @@ Heartbeat payload:
 - `version` (string)
 - Optional: `lora_freq`, `modem_preset`
 - Optional: `protocol` (string; e.g. `"meshtastic"`, `"meshcore"`) — declares the mesh backend for this ingestor; defaults to `"meshtastic"` when absent
+- Optional: `packets` (int ≥ 0) — **mesh-activity delta (SPEC MA1/MA2).** The merged count of *every* frame this ingestor handled since its previous heartbeat: all received frames (including ignored / errored / unimplemented) **plus** its own transmissions (announcement + MeshCore telemetry polls), counted at the earliest receive/transmit seam so nothing is under-reported. It is a **per-interval delta** (reset on each send), **not** a since-boot cumulative. Additive and backward-compatible: an absent or negative value records no activity, so pre-feature ingestors are unaffected.
+
+**Mesh-activity time-series (SPEC MA3).** Each heartbeat carrying a non-negative `packets` value appends one **append-only** row to the `ingestor_activity` table (`ingestor_id`, `at`, `packets`, `protocol`; `data/ingestor_activity.sql`); the `ingestors` snapshot row is upserted as before. Each ingestor's contribution is stored separately (never pre-summed) so a packets/hour moving average is computable across time × protocol × multiple ingestors. The row is best-effort — a failed activity insert never sinks the liveness heartbeat (still `201`). Rows are pruned by the retention worker on `at`. The read-side aggregate is served by `GET /api/stats` (`packets_per_hour`, below).
 
 **Protocol propagation**: all event records (`messages`, `positions`, `telemetry`, `traces`, `neighbors`) that reference this ingestor via their `ingestor` field inherit its `protocol` value at write time when no explicit per-record `protocol` stamp is present. Per-record stamps take precedence — the ingestor heartbeat default only kicks in when the per-record field is absent or malformed.
 
@@ -303,6 +306,7 @@ do **not** accept `before`.
   "meshcore":   { "nodes": {…}, "messages": {…}, "telemetry": {…} },
   "meshtastic": { "nodes": {…}, "messages": {…}, "telemetry": {…} },
   "reticulum":  { "nodes": {…}, "messages": {…}, "telemetry": {…} },  // stub: always 0
+  "packets_per_hour": { "total": 50, "meshcore": 50, "meshtastic": 30, "reticulum": 0 },
   "sampled": false
 }
 ```
@@ -320,6 +324,17 @@ do **not** accept `before`.
 - **Privacy.** Every metric honors the node opt-out marker. When `PRIVATE=1`, all
   `messages` counts are forced to `0` (mirroring the disabled message API);
   `nodes`/`telemetry` counts remain.
+- **`packets_per_hour`** (additive, SPEC MA4/MA5) is a flat top-level map keyed
+  `{ total, meshcore, meshtastic, reticulum }` carrying the 24-hour packets/hour
+  moving average as a rounded integer. It is aggregated **MAX-per-protocol**:
+  `MAX` over that protocol's ingestors of *(the ingestor's `packets` total in the
+  last 24 h ÷ 24)* — a single radio hears ≤ what is actually transmitted, so the
+  busiest vantage is the best dedup-free estimate of air traffic and never
+  double-counts a frame heard by two radios. `total` is the same MAX over **every**
+  ingestor regardless of protocol; `reticulum` is the always-zero forward-looking
+  stub. Unlike `messages`, it is **not** privacy-gated (packets are a public
+  aggregate, no message content). Additive to the 0.7.x `/api/stats` tree — no
+  version bump; the ingestor dogfeeds it for the activity announcement (MA6).
 - **`sampled`** is unchanged: always `false` (the counts are exact, not sampled).
 
 ### GET /api/events live-update stream (SSE)
diff --git a/data/mesh_ingestor/activity.py b/data/mesh_ingestor/activity.py
new file mode 100644
index 0000000..06d49ce
--- /dev/null
+++ b/data/mesh_ingestor/activity.py
@@ -0,0 +1,127 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Merged mesh-activity packet counter shared across the ingestor.
+
+SPEC decision **MA1** (``## Feature: Mesh activity reporting & announcements``):
+the ingestor counts **every** frame it handles — all received frames (including
+ignored / errored / unimplemented) *plus* its own transmissions — as one merged
+figure so the dashboard can report "what is in the air" without under-reporting.
+
+This module owns that single counter so the receive seam
+(:func:`data.mesh_ingestor.handlers._state._mark_packet_seen`), the transmit
+sites (the MeshCore telemetry/status polls and, later, the activity
+announcement), and the heartbeat sender
+(:func:`data.mesh_ingestor.ingestors.queue_ingestor_heartbeat`) all share one
+consistent view. It deliberately depends on nothing but :mod:`threading`, so it
+can be imported from any layer (handlers, protocols, ingestors) without risking
+an import cycle.
+
+The counter is a *per-interval delta* (**MA2**): :func:`take_packet_count`
+returns the running total and atomically resets it to zero, so each heartbeat
+carries only the frames observed since the previous heartbeat. Access is
+serialised by a lock because frames are recorded on the mesh library's receive
+thread (or the MeshCore asyncio loop) while the daemon thread drains the count
+on its heartbeat cycle.
+"""
+
+from __future__ import annotations
+
+import threading
+
+_lock = threading.Lock()
+"""Serialises access to :data:`_packet_count` across the record/drain threads."""
+
+_packet_count: int = 0
+"""Merged count of frames handled (RX + TX) since the last :func:`take_packet_count`."""
+
+
+def _add(count: int) -> None:
+    """Add *count* to the merged activity counter under the shared lock.
+
+    Parameters:
+        count: Number of frames to add to the running total.
+    """
+
+    global _packet_count
+    with _lock:
+        _packet_count += count
+
+
+def record_packet(count: int = 1) -> None:
+    """Count received frame(s) toward the merged activity total (MA1).
+
+    Invoked from the earliest common receive seam
+    (:func:`~data.mesh_ingestor.handlers._state._mark_packet_seen`) so *every*
+    received frame is counted before any dispatch, filter, or drop decision —
+    including ignored, errored, and unimplemented packets.
+
+    Parameters:
+        count: Number of received frames to record; defaults to ``1``.
+    """
+
+    _add(count)
+
+
+def record_tx(count: int = 1) -> None:
+    """Count ingestor transmission(s) toward the merged activity total (MA1).
+
+    Invoked at each transmit site (the MeshCore telemetry/status polls and the
+    activity announcement) so the ingestor's own airtime is reflected in the
+    reported figure, merged with the received count.
+
+    Parameters:
+        count: Number of transmitted frames to record; defaults to ``1``.
+    """
+
+    _add(count)
+
+
+def take_packet_count() -> int:
+    """Return the merged activity count and atomically reset it to zero.
+
+    This yields the *per-interval delta* (MA2): the number of frames recorded
+    since the previous call. The read and the reset happen under the shared
+    lock so a frame recorded concurrently is never double-counted or lost.
+
+    Returns:
+        The number of frames (RX + TX) recorded since the last call.
+    """
+
+    global _packet_count
+    with _lock:
+        value = _packet_count
+        _packet_count = 0
+        return value
+
+
+def reset() -> None:
+    """Reset the merged activity counter to zero.
+
+    Provided for test isolation and for any future "clear on reconnect" need;
+    unlike :func:`take_packet_count` it discards the running total without
+    returning it.
+    """
+
+    global _packet_count
+    with _lock:
+        _packet_count = 0
+
+
+__all__ = [
+    "record_packet",
+    "record_tx",
+    "take_packet_count",
+    "reset",
+]
diff --git a/data/mesh_ingestor/announce.py b/data/mesh_ingestor/announce.py
new file mode 100644
index 0000000..cdfb8c7
--- /dev/null
+++ b/data/mesh_ingestor/announce.py
@@ -0,0 +1,404 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Activity announcement: dogfeed the instance API and build the broadcast text.
+
+SPEC decision **MA6** — the ingestor periodically broadcasts a one-line activity
+summary on its protocol's default channel, drawing the numbers **back from the
+target instance's own API** (dogfeeding: one radio may not see the whole mesh).
+
+This module owns the read-only *dogfeed* HTTP client (GET ``/version`` and
+``/api/stats`` — both public, no auth) and the character-limited message builder.
+The transmit primitive lives on each provider (``send_channel_announcement``,
+MA9); the scheduling and the ``RX_ONLY`` / privacy / 24-hour gates live in the
+daemon (MA7/MA8). Every fetch fails **soft** — any network or
+shape error returns ``None`` so the caller can fail closed.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+import urllib.request
+
+from . import config
+
+#: Conservative single-frame text limits per protocol (SPEC MA6). ASCII bytes ≈
+#: characters; the announcement template is short, so truncation is only a
+#: safety net for an unusually long instance domain.
+ANNOUNCE_CHAR_LIMITS = {"meshtastic": 200, "meshcore": 140}
+
+#: Fallback character limit for an unrecognised protocol.
+_DEFAULT_CHAR_LIMIT = 140
+
+#: Human-facing protocol labels used in the announcement line.
+_PROTOCOL_DISPLAY_NAMES = {"meshtastic": "Meshtastic", "meshcore": "MeshCore"}
+
+#: Timeout (seconds) for each dogfeed GET.
+_DOGFEED_TIMEOUT_SECS = 10
+
+#: Browser-like headers mirroring :mod:`~data.mesh_ingestor.queue` so an instance
+#: behind Cloudflare does not block the dogfeed request.
+_HTTP_HEADERS = {
+    "User-Agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+        "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
+    ),
+    "Accept": "application/json",
+    "Accept-Language": "en-US,en;q=0.9",
+}
+
+
+def _get_json(url: str, *, timeout: float = _DOGFEED_TIMEOUT_SECS) -> dict | None:
+    """GET *url* and parse a JSON object.
+
+    Parameters:
+        url: Absolute URL to fetch.
+        timeout: Socket timeout in seconds.
+
+    Returns:
+        The decoded mapping, or ``None`` on any network / decode error or when
+        the payload is not a JSON object.
+    """
+
+    req = urllib.request.Request(url, headers=dict(_HTTP_HEADERS))
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            payload = json.loads(resp.read().decode("utf-8"))
+    except Exception as exc:
+        config._debug_log(
+            "dogfeed GET failed",
+            context="announce.get",
+            severity="warn",
+            url=url,
+            error_class=exc.__class__.__name__,
+            error_message=str(exc),
+        )
+        return None
+    return payload if isinstance(payload, dict) else None
+
+
+def fetch_private_mode(
+    instance_url: str, *, timeout: float = _DOGFEED_TIMEOUT_SECS
+) -> bool | None:
+    """Return the target instance's ``private_mode`` flag from ``GET /version``.
+
+    Parameters:
+        instance_url: Base URL of the PotatoMesh instance (e.g. ``https://mesh``).
+        timeout: Socket timeout in seconds.
+
+    Returns:
+        ``True`` / ``False`` for a well-formed response, or ``None`` when the
+        fetch fails or the flag is absent/malformed. The caller treats ``None``
+        as "assume private, skip" (fail-closed — SPEC MA7 / Invariant II).
+    """
+
+    data = _get_json(f"{instance_url}/version", timeout=timeout)
+    if not isinstance(data, dict):
+        return None
+    config_block = data.get("config")
+    if not isinstance(config_block, dict):
+        return None
+    value = config_block.get("private_mode")
+    return value if isinstance(value, bool) else None
+
+
+def fetch_activity(
+    instance_url: str, protocol: str, *, timeout: float = _DOGFEED_TIMEOUT_SECS
+) -> tuple[int, float | int] | None:
+    """Return ``(active_nodes, packets_per_hour)`` for *protocol* from ``/api/stats``.
+
+    These are the mesh-wide numbers the announcement quotes (SPEC MA6):
+    ``active_nodes`` = ``<protocol>.nodes.day`` and ``packets_per_hour`` =
+    ``packets_per_hour.<protocol>``.
+
+    Parameters:
+        instance_url: Base URL of the PotatoMesh instance.
+        protocol: ``"meshtastic"`` or ``"meshcore"``.
+        timeout: Socket timeout in seconds.
+
+    Returns:
+        A ``(active_nodes, packets_per_hour)`` tuple, or ``None`` on any
+        fetch/parse error or malformed shape.
+    """
+
+    data = _get_json(f"{instance_url}/api/stats", timeout=timeout)
+    if not isinstance(data, dict):
+        return None
+    scope = data.get(protocol)
+    packets_map = data.get("packets_per_hour")
+    if not isinstance(scope, dict) or not isinstance(packets_map, dict):
+        return None
+    nodes = scope.get("nodes")
+    if not isinstance(nodes, dict):
+        return None
+    active_nodes = nodes.get("day")
+    packets = packets_map.get(protocol)
+    if not isinstance(active_nodes, int) or not isinstance(packets, (int, float)):
+        return None
+    return int(active_nodes), packets
+
+
+def protocol_display_name(protocol: str | None) -> str:
+    """Return the human-facing protocol label used in the announcement.
+
+    Parameters:
+        protocol: Protocol key (e.g. ``"meshcore"``).
+
+    Returns:
+        A display label (``"MeshCore"``), the capitalised key for an unknown
+        protocol, or ``""`` when *protocol* is empty.
+    """
+
+    if not protocol:
+        return ""
+    return _PROTOCOL_DISPLAY_NAMES.get(protocol, protocol.capitalize())
+
+
+def build_announcement(
+    protocol: str,
+    active_nodes: int,
+    packets_per_hour: float | int,
+    instance_url: str,
+    *,
+    char_limit: int | None = None,
+) -> str:
+    """Build the activity announcement line (SPEC MA6).
+
+    Format: ``"<Protocol> activity in the last 24h: <N> active nodes, <M>
+    packets/hour. <instance_url>"``, truncated to the protocol's character limit.
+
+    Parameters:
+        protocol: ``"meshtastic"`` or ``"meshcore"``.
+        active_nodes: 24-hour active-node count for *protocol*.
+        packets_per_hour: 24-hour packets/hour moving average for *protocol*.
+        instance_url: Base URL of the instance (already carries the scheme).
+        char_limit: Optional explicit limit; defaults to the protocol's entry in
+            :data:`ANNOUNCE_CHAR_LIMITS`.
+
+    Returns:
+        The announcement string, truncated to fit the character limit.
+    """
+
+    text = (
+        f"{protocol_display_name(protocol)} activity in the last 24h: "
+        f"{active_nodes} active nodes, {packets_per_hour} packets/hour. "
+        f"{instance_url}"
+    )
+    limit = (
+        char_limit
+        if char_limit is not None
+        else ANNOUNCE_CHAR_LIMITS.get(protocol, _DEFAULT_CHAR_LIMIT)
+    )
+    if limit is not None and len(text) > limit:
+        text = text[:limit]
+    return text
+
+
+# ---------------------------------------------------------------------------
+# Scheduling & gating (SPEC MA7/MA8)
+# ---------------------------------------------------------------------------
+
+#: The first announcement is withheld until the ingestor has been running this
+#: long, so the dogfed 24-hour numbers are accurate over a full window and a
+#: restart storm cannot spam the channel (SPEC MA7 d).
+ANNOUNCE_INITIAL_DELAY_SECS = 24 * 60 * 60
+
+#: Minimum spacing between announcement cycles once eligible (SPEC MA8).
+ANNOUNCE_INTERVAL_SECS = 24 * 60 * 60
+
+
+def announcements_enabled() -> bool:
+    """Return whether announcements may be transmitted at all (SPEC MA7 a).
+
+    ``True`` only when :data:`~data.mesh_ingestor.config.RX_ONLY` is off — the
+    reused receive-only flag (default off) is the single transmit gate; there is
+    no separate enable switch.
+
+    Returns:
+        ``True`` when the transmit gate permits an announcement.
+    """
+
+    return not getattr(config, "RX_ONLY", False)
+
+
+def announce_due(*, start_time: float, last_announce: float | None, now: float) -> bool:
+    """Return whether an announcement cycle is due (SPEC MA7 d / MA8).
+
+    Parameters:
+        start_time: Unix time the ingestor started.
+        last_announce: Unix time of the previous cycle, or ``None`` if none yet.
+        now: Current unix time.
+
+    Returns:
+        ``True`` when at least :data:`ANNOUNCE_INITIAL_DELAY_SECS` have elapsed
+        since *start_time* **and** at least :data:`ANNOUNCE_INTERVAL_SECS` since
+        *last_announce*.
+    """
+
+    if now - start_time < ANNOUNCE_INITIAL_DELAY_SECS:
+        return False
+    if last_announce is not None and now - last_announce < ANNOUNCE_INTERVAL_SECS:
+        return False
+    return True
+
+
+def send_announcement_to_instance(
+    provider: object, iface: object, instance_url: str, protocol: str
+) -> bool:
+    """Dogfeed one instance and, unless it is private, build and send the line.
+
+    The privacy gate is **fail-closed** (SPEC MA7 c / Invariant II): the
+    announcement is sent only when ``/version`` explicitly reports
+    ``private_mode == False``; a ``True`` flag or any fetch/parse error skips it.
+
+    Parameters:
+        provider: Active mesh provider exposing ``send_channel_announcement``.
+        iface: Live mesh interface to transmit through.
+        instance_url: Base URL of the target PotatoMesh instance.
+        protocol: The ingestor's protocol (``"meshtastic"`` / ``"meshcore"``).
+
+    Returns:
+        ``True`` when an announcement was transmitted, ``False`` otherwise.
+    """
+
+    if fetch_private_mode(instance_url) is not False:
+        config._debug_log(
+            "Activity announcement skipped: instance private or unreachable",
+            context="announce.tx",
+            url=instance_url,
+        )
+        return False
+    numbers = fetch_activity(instance_url, protocol)
+    if numbers is None:
+        config._debug_log(
+            "Activity announcement skipped: no activity numbers from instance",
+            context="announce.tx",
+            url=instance_url,
+        )
+        return False
+    active_nodes, packets_per_hour = numbers
+    send = getattr(provider, "send_channel_announcement", None)
+    if not callable(send):
+        config._debug_log(
+            "Activity announcement skipped: provider cannot transmit",
+            context="announce.tx",
+            url=instance_url,
+            protocol=protocol,
+        )
+        return False
+    text = build_announcement(protocol, active_nodes, packets_per_hour, instance_url)
+    # Emitted for every outbound announcement so each TX is traceable end-to-end
+    # (the provider then logs the mesh-layer send).
+    config._debug_log(
+        "Activity announcement transmitting",
+        context="announce.tx",
+        url=instance_url,
+        protocol=protocol,
+        active_nodes=active_nodes,
+        packets_per_hour=packets_per_hour,
+        text=text,
+    )
+    send(iface, text)
+    return True
+
+
+def run_announcement_cycle(
+    provider: object, iface: object, *, protocol: str | None = None
+) -> bool:
+    """Announce to **every** configured instance domain (SPEC MA6/MA8 per-domain).
+
+    Each instance is dogfed and sent its own numbers/link independently; a
+    failure against one instance never aborts the others.
+
+    Parameters:
+        provider: Active mesh provider.
+        iface: Live mesh interface.
+        protocol: Override protocol; defaults to
+            :data:`~data.mesh_ingestor.config.PROTOCOL`.
+
+    Returns:
+        ``True`` when at least one announcement was transmitted.
+    """
+
+    protocol = protocol or getattr(config, "PROTOCOL", "meshtastic")
+    sent_any = False
+    for instance_url, _api_token in getattr(config, "INSTANCES", ()):
+        try:
+            if send_announcement_to_instance(provider, iface, instance_url, protocol):
+                sent_any = True
+        except Exception as exc:
+            config._debug_log(
+                "activity announcement failed",
+                context="announce.cycle",
+                severity="warn",
+                url=instance_url,
+                error_class=exc.__class__.__name__,
+                error_message=str(exc),
+            )
+    return sent_any
+
+
+def maybe_run_announcements(
+    provider: object,
+    iface: object,
+    *,
+    start_time: float,
+    last_announce: float | None,
+    now: float | None = None,
+) -> float | None:
+    """Daemon entry point: run an announcement cycle when scheduled (SPEC MA7/MA8).
+
+    Applies the enable / RX-only gate and the 24-hour schedule, then dogfeeds and
+    announces to each configured instance. The cycle timestamp advances whenever
+    the schedule fires (whether or not any instance was actually announced to),
+    so a private or unreachable instance is retried on the next 24-hour tick
+    rather than on every loop iteration.
+
+    Parameters:
+        provider: Active mesh provider.
+        iface: Live mesh interface.
+        start_time: Unix time the ingestor started.
+        last_announce: Unix time of the previous cycle, or ``None``.
+        now: Current unix time; defaults to :func:`time.time`.
+
+    Returns:
+        The updated ``last_announce`` timestamp — *now* when a cycle ran, else
+        the unchanged *last_announce*.
+    """
+
+    if now is None:
+        now = time.time()
+    if not announcements_enabled():
+        return last_announce
+    if not announce_due(start_time=start_time, last_announce=last_announce, now=now):
+        return last_announce
+    run_announcement_cycle(provider, iface)
+    return now
+
+
+__all__ = [
+    "ANNOUNCE_CHAR_LIMITS",
+    "ANNOUNCE_INITIAL_DELAY_SECS",
+    "ANNOUNCE_INTERVAL_SECS",
+    "announce_due",
+    "announcements_enabled",
+    "build_announcement",
+    "fetch_activity",
+    "fetch_private_mode",
+    "maybe_run_announcements",
+    "protocol_display_name",
+    "run_announcement_cycle",
+    "send_announcement_to_instance",
+]
diff --git a/data/mesh_ingestor/config.py b/data/mesh_ingestor/config.py
index 2d7ba73..62116cc 100644
--- a/data/mesh_ingestor/config.py
+++ b/data/mesh_ingestor/config.py
@@ -144,9 +144,9 @@ RX_ONLY = os.environ.get("RX_ONLY") == "1"
 """Receive-only mode: forbid every ingestor-initiated mesh transmission.
 
 Some operators run listening posts where any TX is undesired.  When set, the
-ingestor never transmits on the mesh: currently this disables the MeshCore
-contact telemetry/status polls (the only ingestor-initiated RF traffic), and
-any future TX feature must honour it too.  Local companion-link reads (host
+ingestor never transmits on the mesh: this disables the MeshCore contact
+telemetry/status polls and the periodic activity announcement (SPEC MA7), the
+only ingestor-initiated RF traffic.  Local companion-link reads (host
 self-telemetry, contact roster, channel queries) are not transmissions and
 continue to work."""
 
diff --git a/data/mesh_ingestor/daemon.py b/data/mesh_ingestor/daemon.py
index 7f2febc..678b664 100644
--- a/data/mesh_ingestor/daemon.py
+++ b/data/mesh_ingestor/daemon.py
@@ -24,7 +24,7 @@ import time
 
 from pubsub import pub
 
-from . import config, handlers, ingestors, interfaces, queue
+from . import announce, config, handlers, ingestors, interfaces, queue
 from .mesh_protocol import MeshProtocol
 from .utils import _retry_dict_snapshot
 
@@ -265,6 +265,7 @@ class _DaemonState:
     ingestor_announcement_sent: bool = False
     announced_target: bool = False
     last_self_node_report: float | None = None
+    last_announce: float | None = None
 
 
 # ---------------------------------------------------------------------------
@@ -573,6 +574,39 @@ def _try_send_self_node(state: _DaemonState) -> None:
         )
 
 
+# ---------------------------------------------------------------------------
+# Periodic activity-announcement helper
+# ---------------------------------------------------------------------------
+
+
+def _process_announcements(state: _DaemonState) -> float | None:
+    """Run the periodic activity-announcement cycle when scheduled (SPEC MA6-MA8).
+
+    Delegates the transmit gate (``RX_ONLY``), the per-instance fail-closed
+    privacy check, the >=24h post-start delay, and the 24h cadence to
+    :func:`~data.mesh_ingestor.announce.maybe_run_announcements`, which dogfeeds
+    each configured instance's own API for the numbers it announces. A no-op
+    until the interface is connected (the announcement needs a live radio to
+    transmit).
+
+    Parameters:
+        state: Current daemon loop state (provides the provider, interface, and
+            the last-announce timestamp).
+
+    Returns:
+        The updated ``last_announce`` timestamp for :class:`_DaemonState`.
+    """
+
+    if state.iface is None:
+        return state.last_announce
+    return announce.maybe_run_announcements(
+        state.provider,
+        state.iface,
+        start_time=ingestors.ingestor_start_time(),
+        last_announce=state.last_announce,
+    )
+
+
 # ---------------------------------------------------------------------------
 # Loop iteration helper
 # ---------------------------------------------------------------------------
@@ -610,6 +644,7 @@ def _loop_iteration(state: _DaemonState) -> bool:
         or _now - state.last_self_node_report >= config._SELF_NODE_REPORT_INTERVAL_SECS
     ):
         _try_send_self_node(state)
+    state.last_announce = _process_announcements(state)
     state.retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS)
     return False
 
@@ -727,6 +762,7 @@ __all__ = [
     "_event_wait_allows_default_timeout",
     "_is_ble_interface",
     "_node_items_snapshot",
+    "_process_announcements",
     "_process_ingestor_heartbeat",
     "_subscribe_receive_topics",
     "_try_connect",
diff --git a/data/mesh_ingestor/handlers/_state.py b/data/mesh_ingestor/handlers/_state.py
index 420c5f1..b9d0240 100644
--- a/data/mesh_ingestor/handlers/_state.py
+++ b/data/mesh_ingestor/handlers/_state.py
@@ -24,7 +24,7 @@ from __future__ import annotations
 import math
 import time
 
-from .. import config
+from .. import activity, config
 from ..serialization import _canonical_node_id
 
 # ---------------------------------------------------------------------------
@@ -182,10 +182,17 @@ def last_packet_monotonic() -> float | None:
 
 
 def _mark_packet_seen() -> None:
-    """Record that a packet has been processed by updating the monotonic clock."""
+    """Record that a packet has been processed.
+
+    Updates the monotonic activity clock (used for inactivity-reconnect) and
+    increments the merged mesh-activity counter (SPEC MA1) so that *every*
+    received frame — including one a downstream handler later ignores or fails
+    to store — is counted here, before any dispatch or drop decision.
+    """
 
     global _last_packet_monotonic
     _last_packet_monotonic = time.monotonic()
+    activity.record_packet()
 
 
 __all__ = [
diff --git a/data/mesh_ingestor/ingestors.py b/data/mesh_ingestor/ingestors.py
index 9516a5a..e4899d4 100644
--- a/data/mesh_ingestor/ingestors.py
+++ b/data/mesh_ingestor/ingestors.py
@@ -21,7 +21,7 @@ from dataclasses import dataclass, field
 from typing import Callable
 
 from .. import VERSION as INGESTOR_VERSION
-from . import config, queue
+from . import activity, config, queue
 from .serialization import _canonical_node_id
 
 HEARTBEAT_INTERVAL_SECS = 60 * 60
@@ -108,12 +108,17 @@ def queue_ingestor_heartbeat(
     if not force and last is not None and now - last < interval:
         return False
 
+    # Drain the merged activity counter only once we are committed to sending
+    # (past every early-return guard above), so a suppressed heartbeat never
+    # discards an interval's packet count.  The value is the per-interval delta
+    # since the previous heartbeat (SPEC MA2); ``take_packet_count`` resets it.
     payload = {
         "node_id": canonical,
         "start_time": STATE.start_time,
         "last_seen_time": now,
         "version": INGESTOR_VERSION,
         "protocol": getattr(config, "PROTOCOL", "meshtastic") or "meshtastic",
+        "packets": activity.take_packet_count(),
     }
     if getattr(config, "LORA_FREQ", None) is not None:
         payload["lora_freq"] = config.LORA_FREQ
diff --git a/data/mesh_ingestor/protocols/meshcore/_constants.py b/data/mesh_ingestor/protocols/meshcore/_constants.py
index a1ce379..86c2dc5 100644
--- a/data/mesh_ingestor/protocols/meshcore/_constants.py
+++ b/data/mesh_ingestor/protocols/meshcore/_constants.py
@@ -35,6 +35,15 @@ deferred until contacts have been fetched so that the daemon's first
 (issue #788).
 """
 
+_ANNOUNCE_SEND_TIMEOUT_SECS: float = 15.0
+"""Seconds to wait for an activity announcement (``send_chan_msg``) to complete.
+
+The announcement is scheduled from the daemon's synchronous thread onto the
+MeshCore asyncio loop via ``run_coroutine_threadsafe``; this bounds the blocking
+wait on that cross-thread future so a wedged radio cannot stall the daemon loop
+(SPEC MA6/MA9).
+"""
+
 _DEFAULT_BAUDRATE: int = 115200
 """Default baud rate for MeshCore serial connections."""
 
diff --git a/data/mesh_ingestor/protocols/meshcore/provider.py b/data/mesh_ingestor/protocols/meshcore/provider.py
index cd4b16a..9cc4b05 100644
--- a/data/mesh_ingestor/protocols/meshcore/provider.py
+++ b/data/mesh_ingestor/protocols/meshcore/provider.py
@@ -20,8 +20,8 @@ import asyncio
 import sys
 import threading
 
-from ... import config
-from ._constants import _CONNECT_TIMEOUT_SECS
+from ... import activity, config
+from ._constants import _ANNOUNCE_SEND_TIMEOUT_SECS, _CONNECT_TIMEOUT_SECS
 from .decode import _self_info_to_node_dict
 from .identity import _meshcore_node_id
 from .interface import _MeshcoreInterface
@@ -200,3 +200,40 @@ class MeshcoreProvider:
         if self_item is not None:
             items.append(self_item)
         return items
+
+    def send_channel_announcement(self, iface: object, text: str) -> None:
+        """Broadcast an activity announcement on the default channel (SPEC MA6/MA9).
+
+        Schedules ``send_chan_msg`` on channel
+        :data:`~data.mesh_ingestor.config.CHANNEL_INDEX` onto the MeshCore
+        asyncio loop (which the provider runs in a background thread) via
+        :func:`asyncio.run_coroutine_threadsafe`, blocking up to
+        :data:`_ANNOUNCE_SEND_TIMEOUT_SECS` for completion, and counts the
+        transmission toward the merged activity total (MA1). This is an
+        **optional**, duck-typed provider method (not a formal
+        :class:`~data.mesh_ingestor.mesh_protocol.MeshProtocol` member); the
+        daemon resolves it via ``getattr`` and skips it when absent. It is a
+        no-op when the interface has no live event loop / handle.
+
+        Parameters:
+            iface: Active :class:`_MeshcoreInterface`.
+            text: Announcement string to transmit.
+        """
+
+        if not isinstance(iface, _MeshcoreInterface):
+            return
+        mc = getattr(iface, "_mc", None)
+        loop = getattr(iface, "_loop", None)
+        if mc is None or loop is None or loop.is_closed():
+            return
+        activity.record_tx()
+        future = asyncio.run_coroutine_threadsafe(
+            mc.commands.send_chan_msg(config.CHANNEL_INDEX, text), loop
+        )
+        future.result(timeout=_ANNOUNCE_SEND_TIMEOUT_SECS)
+        config._debug_log(
+            "MeshCore activity announcement transmitted",
+            context="meshcore.tx",
+            channel=config.CHANNEL_INDEX,
+            chars=len(text),
+        )
diff --git a/data/mesh_ingestor/protocols/meshcore/telemetry.py b/data/mesh_ingestor/protocols/meshcore/telemetry.py
index 42722d6..392ce79 100644
--- a/data/mesh_ingestor/protocols/meshcore/telemetry.py
+++ b/data/mesh_ingestor/protocols/meshcore/telemetry.py
@@ -35,7 +35,7 @@ import asyncio
 import time
 from collections.abc import Mapping
 
-from ... import config
+from ... import activity, config
 from .interface import _MeshcoreInterface
 from .messages import _derive_message_id
 
@@ -401,7 +401,12 @@ async def _poll_contact_telemetry(
     Falls back to a status request when the telemetry pull yields nothing, so
     sensor-less nodes still report battery/uptime.  One contact per call keeps
     airtime bounded to a single request per poll interval regardless of roster
-    size; the meshcore library serialises mesh requests internally.
+    size; the meshcore library serialises mesh requests internally.  Both ends
+    of the attempt emit a ``meshcore.telemetry.poll`` debug line — one when the
+    request is initiated, one when neither the telemetry nor the status request
+    returned usable data — because ``req_*_sync`` return ``None`` on timeout
+    without raising, which would otherwise leave an unanswered poll
+    indistinguishable from a disabled poll loop.
 
     Parameters:
         mc: Connected MeshCore instance.
@@ -415,7 +420,16 @@ async def _poll_contact_telemetry(
     node_id = iface.lookup_node_id((contact.get("public_key") or "")[:12])
     if node_id is None:
         return
+    # Log before the request goes out: req_*_sync return None on timeout without
+    # raising, so an unanswered poll is otherwise silent.
+    config._debug_log(
+        "MeshCore contact telemetry poll initiated",
+        context="meshcore.telemetry.poll",
+        node_id=node_id,
+    )
     try:
+        # An on-air pull is an ingestor transmission — count it (SPEC MA1).
+        activity.record_tx()
         lpp = await mc.commands.req_telemetry_sync(contact)
     except Exception as exc:
         config._debug_log(
@@ -430,6 +444,8 @@ async def _poll_contact_telemetry(
     ):
         return
     try:
+        # The status fallback is a second on-air pull — count it too (MA1).
+        activity.record_tx()
         status = await mc.commands.req_status_sync(contact)
     except Exception as exc:
         config._debug_log(
@@ -439,8 +455,14 @@ async def _poll_contact_telemetry(
             error=str(exc),
         )
         return
-    _queue_meshcore_telemetry(
+    if _queue_meshcore_telemetry(
         handlers, node_id, _status_to_telemetry_section(status), "status"
+    ):
+        return
+    config._debug_log(
+        "MeshCore contact telemetry poll returned no data",
+        context="meshcore.telemetry.poll",
+        node_id=node_id,
     )
 
 
diff --git a/data/mesh_ingestor/protocols/meshtastic.py b/data/mesh_ingestor/protocols/meshtastic.py
index afd75c9..d276ed9 100644
--- a/data/mesh_ingestor/protocols/meshtastic.py
+++ b/data/mesh_ingestor/protocols/meshtastic.py
@@ -18,7 +18,7 @@ from __future__ import annotations
 
 from pubsub import pub
 
-from .. import config, daemon as _daemon, handlers, interfaces
+from .. import activity, config, daemon as _daemon, handlers, interfaces
 from ..utils import _retry_dict_snapshot
 
 
@@ -96,5 +96,31 @@ class MeshtasticProvider:
             return []
         return result
 
+    def send_channel_announcement(self, iface: object, text: str) -> None:
+        """Broadcast an activity announcement on the default channel (SPEC MA6/MA9).
+
+        Sends *text* on channel :data:`~data.mesh_ingestor.config.CHANNEL_INDEX`
+        via the Meshtastic interface and counts the transmission toward the
+        merged activity total (MA1). This is an **optional**, duck-typed provider
+        method (not a formal :class:`MeshProtocol` member); the daemon resolves
+        it via ``getattr`` and skips it when absent.
+
+        Parameters:
+            iface: Active Meshtastic interface exposing ``sendText``.
+            text: Announcement string to transmit.
+        """
+
+        send_text = getattr(iface, "sendText", None)
+        if not callable(send_text):
+            return
+        activity.record_tx()
+        send_text(text, channelIndex=config.CHANNEL_INDEX)
+        config._debug_log(
+            "Meshtastic activity announcement transmitted",
+            context="meshtastic.tx",
+            channel=config.CHANNEL_INDEX,
+            chars=len(text),
+        )
+
 
 __all__ = ["MeshtasticProvider"]
diff --git a/tests/test_activity_unit.py b/tests/test_activity_unit.py
new file mode 100644
index 0000000..fece0cf
--- /dev/null
+++ b/tests/test_activity_unit.py
@@ -0,0 +1,200 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Unit tests for the merged mesh-activity counter (SPEC MA1/MA2).
+
+Covers :mod:`data.mesh_ingestor.activity` (the counter API), the receive-seam
+wiring in :func:`data.mesh_ingestor.handlers._state._mark_packet_seen` (so every
+received frame — stored, ignored, or errored — is counted), and the
+per-interval ``packets`` delta carried by
+:func:`data.mesh_ingestor.ingestors.queue_ingestor_heartbeat`.
+"""
+
+from __future__ import annotations
+
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+import data.mesh_ingestor.activity as activity
+import data.mesh_ingestor.ingestors as ingestors_mod
+from data.mesh_ingestor.handlers import _state
+from data.mesh_ingestor.ingestors import (
+    _IngestorState,
+    queue_ingestor_heartbeat,
+    set_ingestor_node_id,
+)
+
+
+@pytest.fixture(autouse=True)
+def reset_activity_counter():
+    """Zero the shared activity counter around every test for isolation."""
+    activity.reset()
+    yield
+    activity.reset()
+
+
+@pytest.fixture(autouse=True)
+def reset_ingestor_state():
+    """Swap in fresh ingestor identity state around every test."""
+    original = ingestors_mod.STATE
+    ingestors_mod.STATE = _IngestorState()
+    yield
+    ingestors_mod.STATE = original
+
+
+# ---------------------------------------------------------------------------
+# Counter API
+# ---------------------------------------------------------------------------
+
+
+class TestCounterApi:
+    """Tests for the :mod:`data.mesh_ingestor.activity` counter primitives."""
+
+    def test_record_packet_counts_one_by_default(self):
+        """A bare ``record_packet()`` adds a single frame."""
+        activity.record_packet()
+        assert activity.take_packet_count() == 1
+
+    def test_record_packet_counts_a_custom_amount(self):
+        """``record_packet(n)`` adds ``n`` frames at once."""
+        activity.record_packet(4)
+        assert activity.take_packet_count() == 4
+
+    def test_record_tx_counts_transmissions(self):
+        """``record_tx()`` counts an ingestor transmission (MA1)."""
+        activity.record_tx()
+        activity.record_tx(2)
+        assert activity.take_packet_count() == 3
+
+    def test_rx_and_tx_share_one_merged_count(self):
+        """RX and TX accumulate into a single merged figure (MA1)."""
+        activity.record_packet(5)
+        activity.record_tx(2)
+        assert activity.take_packet_count() == 7
+
+    def test_take_packet_count_resets_to_zero(self):
+        """Reading the count via ``take_packet_count`` clears it (MA2)."""
+        activity.record_packet(3)
+        assert activity.take_packet_count() == 3
+        assert activity.take_packet_count() == 0
+
+    def test_reset_discards_the_running_total(self):
+        """``reset()`` zeroes the counter without returning it."""
+        activity.record_packet(9)
+        activity.reset()
+        assert activity.take_packet_count() == 0
+
+
+# ---------------------------------------------------------------------------
+# Receive seam — every frame counted, before any dispatch/drop
+# ---------------------------------------------------------------------------
+
+
+class TestReceiveSeam:
+    """The RX seam counts every received frame regardless of its fate."""
+
+    def test_mark_packet_seen_counts_the_frame(self):
+        """``_mark_packet_seen`` increments the merged counter (MA1)."""
+        _state._mark_packet_seen()
+        assert activity.take_packet_count() == 1
+
+    def test_on_receive_counts_stored_ignored_and_errored(self, monkeypatch):
+        """Every ``on_receive`` outcome is counted at the seam (MA1).
+
+        The count is taken before dispatch, so a packet the downstream handler
+        stores, ignores (returns without queuing), or errors on (raises) is
+        counted identically.
+        """
+        from data.mesh_ingestor.handlers import generic
+
+        monkeypatch.setattr(generic, "_pkt_to_dict", lambda p: dict(p))
+
+        # Stored / ignored: store returns without raising.
+        monkeypatch.setattr(generic, "store_packet_dict", lambda pkt: None)
+        generic.on_receive({"id": 1}, object())
+        assert activity.take_packet_count() == 1
+
+        # Errored: store raises; on_receive swallows it, but the frame was
+        # already counted at the seam.
+        def _boom(pkt):
+            raise RuntimeError("boom")
+
+        monkeypatch.setattr(generic, "store_packet_dict", _boom)
+        generic.on_receive({"id": 2}, object())
+        assert activity.take_packet_count() == 1
+
+    def test_on_receive_dedup_counts_once(self, monkeypatch):
+        """A packet already flagged ``_potatomesh_seen`` is not re-counted."""
+        from data.mesh_ingestor.handlers import generic
+
+        monkeypatch.setattr(generic, "_pkt_to_dict", lambda p: dict(p))
+        monkeypatch.setattr(generic, "store_packet_dict", lambda pkt: None)
+
+        packet = {"id": 1}
+        generic.on_receive(packet, object())
+        generic.on_receive(packet, object())  # same dict → deduped
+        assert activity.take_packet_count() == 1
+
+
+# ---------------------------------------------------------------------------
+# Heartbeat per-interval delta (MA2)
+# ---------------------------------------------------------------------------
+
+
+class TestHeartbeatDelta:
+    """The heartbeat carries — and resets — the per-interval packet delta."""
+
+    def test_heartbeat_delta_included_in_payload(self):
+        """The heartbeat payload carries the accumulated ``packets`` count."""
+        set_ingestor_node_id("!aabbccdd")
+        activity.record_packet(7)
+        sent = []
+        queue_ingestor_heartbeat(force=True, send=lambda p, pl: sent.append(pl))
+        assert sent[0]["packets"] == 7
+
+    def test_heartbeat_delta_resets_between_sends(self):
+        """Consecutive heartbeats report their own interval, not the cumulative."""
+        set_ingestor_node_id("!aabbccdd")
+        sent = []
+        activity.record_packet(3)
+        queue_ingestor_heartbeat(force=True, send=lambda p, pl: sent.append(pl))
+        activity.record_packet(2)
+        queue_ingestor_heartbeat(force=True, send=lambda p, pl: sent.append(pl))
+        assert [pl["packets"] for pl in sent] == [3, 2]
+
+    def test_heartbeat_delta_zero_when_no_traffic(self):
+        """A heartbeat with no observed frames reports ``0``."""
+        set_ingestor_node_id("!aabbccdd")
+        sent = []
+        queue_ingestor_heartbeat(force=True, send=lambda p, pl: sent.append(pl))
+        assert sent[0]["packets"] == 0
+
+    def test_heartbeat_delta_not_drained_when_suppressed(self):
+        """A suppressed (interval-guarded) heartbeat must not discard the count."""
+        set_ingestor_node_id("!aabbccdd")
+        ingestors_mod.STATE.last_heartbeat = int(time.time())
+        activity.record_packet(5)
+        sent = []
+        result = queue_ingestor_heartbeat(send=lambda p, pl: sent.append(pl))
+        assert result is False
+        assert sent == []
+        # The interval-guarded heartbeat returned before building the payload,
+        # so the 5 frames survive for the next real heartbeat.
+        assert activity.take_packet_count() == 5
diff --git a/tests/test_announce_unit.py b/tests/test_announce_unit.py
new file mode 100644
index 0000000..519353a
--- /dev/null
+++ b/tests/test_announce_unit.py
@@ -0,0 +1,506 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Unit tests for :mod:`data.mesh_ingestor.announce` (SPEC MA6).
+
+Covers the character-limited announcement message builder and the read-only
+*dogfeed* HTTP client (fetch the instance's own ``/version`` and ``/api/stats``),
+including the fail-soft behaviour on network / shape errors.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+import urllib.error
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+import data.mesh_ingestor.announce as announce
+
+
+class _FakeResponse:
+    """Minimal context-manager stand-in for an ``http.client.HTTPResponse``."""
+
+    def __init__(self, body: str):
+        self._body = body.encode("utf-8")
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_exc):
+        return False
+
+    def read(self):
+        return self._body
+
+
+def _install_fake_http(monkeypatch, url_to_payload):
+    """Patch ``announce``'s ``urlopen`` to serve *url_to_payload* by URL."""
+
+    def _fake_urlopen(req, timeout=None):
+        url = getattr(req, "full_url", req)
+        if url not in url_to_payload:
+            raise urllib.error.URLError(f"no fake response for {url}")
+        return _FakeResponse(json.dumps(url_to_payload[url]))
+
+    monkeypatch.setattr(announce.urllib.request, "urlopen", _fake_urlopen)
+
+
+# ---------------------------------------------------------------------------
+# Message builder
+# ---------------------------------------------------------------------------
+
+
+class TestBuildAnnouncementMessage:
+    """Tests for :func:`announce.build_announcement` (MA6)."""
+
+    def test_message_format_matches_spec(self):
+        """The announcement line matches the SPEC MA6 template exactly."""
+        text = announce.build_announcement("meshcore", 12, 50, "https://mesh.example")
+        assert text == (
+            "MeshCore activity in the last 24h: 12 active nodes, "
+            "50 packets/hour. https://mesh.example"
+        )
+
+    def test_message_uses_protocol_display_name(self):
+        """Meshtastic renders its capitalised display name."""
+        text = announce.build_announcement("meshtastic", 3, 7, "https://m")
+        assert text.startswith("Meshtastic activity in the last 24h:")
+
+    def test_message_truncates_to_protocol_char_limit(self):
+        """An over-long line is clipped to the protocol's character limit."""
+        long_url = "https://" + ("a" * 300) + ".example"
+        text = announce.build_announcement("meshtastic", 5, 10, long_url)
+        assert len(text) == announce.ANNOUNCE_CHAR_LIMITS["meshtastic"]
+
+    def test_message_honours_explicit_char_limit(self):
+        """An explicit ``char_limit`` overrides the per-protocol default."""
+        text = announce.build_announcement(
+            "meshcore", 5, 10, "https://x", char_limit=10
+        )
+        assert len(text) == 10
+
+    def test_message_unknown_protocol_uses_fallback_limit(self):
+        """An unrecognised protocol falls back to the default char limit."""
+        long_url = "https://" + ("z" * 300)
+        text = announce.build_announcement("reticulum", 1, 2, long_url)
+        assert text.startswith("Reticulum activity")
+        assert len(text) == announce._DEFAULT_CHAR_LIMIT
+
+
+class TestProtocolDisplayName:
+    """Tests for :func:`announce.protocol_display_name`."""
+
+    def test_known_protocols(self):
+        """Known keys map to their branded labels."""
+        assert announce.protocol_display_name("meshcore") == "MeshCore"
+        assert announce.protocol_display_name("meshtastic") == "Meshtastic"
+
+    def test_unknown_protocol_is_capitalised(self):
+        """An unknown key is capitalised as a best effort."""
+        assert announce.protocol_display_name("reticulum") == "Reticulum"
+
+    def test_empty_protocol_is_blank(self):
+        """An empty/``None`` protocol yields an empty label."""
+        assert announce.protocol_display_name("") == ""
+        assert announce.protocol_display_name(None) == ""
+
+
+# ---------------------------------------------------------------------------
+# Dogfeed HTTP client
+# ---------------------------------------------------------------------------
+
+
+class TestDogfeedFetchActivity:
+    """Tests for :func:`announce.fetch_activity` (MA6)."""
+
+    def test_dogfeed_reads_nodes_and_packets_per_hour(self, monkeypatch):
+        """Returns ``(<protocol>.nodes.day, packets_per_hour.<protocol>)``."""
+        _install_fake_http(
+            monkeypatch,
+            {
+                "https://mesh.example/api/stats": {
+                    "meshcore": {"nodes": {"day": 12, "hour": 3}},
+                    "meshtastic": {"nodes": {"day": 99}},
+                    "packets_per_hour": {"meshcore": 50, "meshtastic": 30},
+                },
+            },
+        )
+        assert announce.fetch_activity("https://mesh.example", "meshcore") == (12, 50)
+
+    def test_dogfeed_returns_none_on_malformed_shape(self, monkeypatch):
+        """A response missing the packets_per_hour map yields ``None``."""
+        _install_fake_http(
+            monkeypatch,
+            {"https://mesh.example/api/stats": {"meshcore": {"nodes": {"day": 1}}}},
+        )
+        assert announce.fetch_activity("https://mesh.example", "meshcore") is None
+
+    def test_dogfeed_returns_none_when_nodes_object_missing(self, monkeypatch):
+        """A scope present but lacking a ``nodes`` sub-object yields ``None``."""
+        _install_fake_http(
+            monkeypatch,
+            {
+                "https://mesh.example/api/stats": {
+                    "meshcore": {},
+                    "packets_per_hour": {"meshcore": 50},
+                },
+            },
+        )
+        assert announce.fetch_activity("https://mesh.example", "meshcore") is None
+
+    def test_dogfeed_returns_none_on_non_integer_counts(self, monkeypatch):
+        """Non-numeric counts are rejected as malformed."""
+        _install_fake_http(
+            monkeypatch,
+            {
+                "https://mesh.example/api/stats": {
+                    "meshcore": {"nodes": {"day": "lots"}},
+                    "packets_per_hour": {"meshcore": 50},
+                },
+            },
+        )
+        assert announce.fetch_activity("https://mesh.example", "meshcore") is None
+
+    def test_dogfeed_returns_none_on_network_error(self, monkeypatch):
+        """A transport error fails soft to ``None``."""
+
+        def _boom(req, timeout=None):
+            raise urllib.error.URLError("boom")
+
+        monkeypatch.setattr(announce.urllib.request, "urlopen", _boom)
+        assert announce.fetch_activity("https://mesh.example", "meshtastic") is None
+
+
+class TestDogfeedFetchPrivateMode:
+    """Tests for :func:`announce.fetch_private_mode` (MA7 privacy gate)."""
+
+    @pytest.mark.parametrize("flag", [True, False])
+    def test_dogfeed_reads_private_mode_flag(self, monkeypatch, flag):
+        """Reads ``config.private_mode`` from ``/version``."""
+        _install_fake_http(
+            monkeypatch,
+            {"https://mesh.example/version": {"config": {"private_mode": flag}}},
+        )
+        assert announce.fetch_private_mode("https://mesh.example") is flag
+
+    def test_dogfeed_missing_flag_returns_none(self, monkeypatch):
+        """A response without the flag yields ``None`` (caller fails closed)."""
+        _install_fake_http(
+            monkeypatch,
+            {"https://mesh.example/version": {"config": {"site_name": "x"}}},
+        )
+        assert announce.fetch_private_mode("https://mesh.example") is None
+
+    def test_dogfeed_missing_config_block_returns_none(self, monkeypatch):
+        """A response without a ``config`` block yields ``None``."""
+        _install_fake_http(
+            monkeypatch, {"https://mesh.example/version": {"version": "0.7.3"}}
+        )
+        assert announce.fetch_private_mode("https://mesh.example") is None
+
+    def test_dogfeed_network_error_returns_none(self, monkeypatch):
+        """A transport error fails soft to ``None`` (fail-closed input)."""
+
+        def _boom(req, timeout=None):
+            raise urllib.error.URLError("boom")
+
+        monkeypatch.setattr(announce.urllib.request, "urlopen", _boom)
+        assert announce.fetch_private_mode("https://mesh.example") is None
+
+
+class TestDogfeedGetJson:
+    """Tests for the low-level :func:`announce._get_json` helper."""
+
+    def test_dogfeed_non_object_json_returns_none(self, monkeypatch):
+        """A JSON array (not an object) is rejected."""
+        _install_fake_http(monkeypatch, {"https://mesh.example/x": [1, 2, 3]})
+        assert announce._get_json("https://mesh.example/x") is None
+
+
+# ---------------------------------------------------------------------------
+# Scheduling & gating (SPEC MA7/MA8)
+# ---------------------------------------------------------------------------
+
+
+class _RecordingProvider:
+    """Fake provider that records ``send_channel_announcement`` calls."""
+
+    def __init__(self):
+        self.sent = []
+
+    def send_channel_announcement(self, iface, text):
+        self.sent.append((iface, text))
+
+
+def _stub_dogfeed(monkeypatch, *, private=False, numbers=(12, 50)):
+    """Patch the dogfeed fetchers to controlled return values (no HTTP)."""
+    monkeypatch.setattr(announce, "fetch_private_mode", lambda url, **_kw: private)
+    monkeypatch.setattr(announce, "fetch_activity", lambda url, proto, **_kw: numbers)
+
+
+class TestAnnouncementGates:
+    """Tests for :func:`announce.announcements_enabled` (SPEC MA7 a)."""
+
+    def test_gate_enabled_by_default(self, monkeypatch):
+        """Enabled when RX_ONLY is off (the default)."""
+        monkeypatch.setattr(announce.config, "RX_ONLY", False)
+        assert announce.announcements_enabled() is True
+
+    def test_gate_disabled_when_rx_only(self, monkeypatch):
+        """RX_ONLY (reused as the single transmit gate) forbids the announcement."""
+        monkeypatch.setattr(announce.config, "RX_ONLY", True)
+        assert announce.announcements_enabled() is False
+
+
+class TestAnnounceDue:
+    """Tests for :func:`announce.announce_due` (SPEC MA7 d / MA8)."""
+
+    def test_gate_not_due_before_24h_elapsed(self):
+        """Withheld until 24 h after start."""
+        now = 1_000_000
+        assert (
+            announce.announce_due(start_time=now - 1000, last_announce=None, now=now)
+            is False
+        )
+
+    def test_due_after_24h_elapsed(self):
+        """Eligible once 24 h have elapsed since start."""
+        now = 1_000_000
+        assert (
+            announce.announce_due(start_time=now - 90_000, last_announce=None, now=now)
+            is True
+        )
+
+    def test_cadence_interval_blocks_second_within_24h(self):
+        """A recent cycle blocks the next until the 24 h interval passes."""
+        now = 1_000_000
+        assert (
+            announce.announce_due(
+                start_time=now - 200_000, last_announce=now - 3600, now=now
+            )
+            is False
+        )
+
+    def test_cadence_due_again_after_24h_interval(self):
+        """Due again once 24 h have elapsed since the previous cycle."""
+        now = 1_000_000
+        assert (
+            announce.announce_due(
+                start_time=now - 200_000, last_announce=now - 90_000, now=now
+            )
+            is True
+        )
+
+
+class TestSendAnnouncementToInstance:
+    """Tests for :func:`announce.send_announcement_to_instance` (MA6/MA7)."""
+
+    def test_sends_the_built_line_when_public(self, monkeypatch):
+        """A public instance receives the dogfed announcement line."""
+        _stub_dogfeed(monkeypatch, private=False, numbers=(12, 50))
+        provider = _RecordingProvider()
+        sent = announce.send_announcement_to_instance(
+            provider, "IFACE", "https://mesh.example", "meshcore"
+        )
+        assert sent is True
+        assert provider.sent == [
+            (
+                "IFACE",
+                "MeshCore activity in the last 24h: 12 active nodes, "
+                "50 packets/hour. https://mesh.example",
+            )
+        ]
+
+    def test_private_instance_is_skipped_fail_closed(self, monkeypatch):
+        """A private instance is skipped (nothing transmitted)."""
+        _stub_dogfeed(monkeypatch, private=True)
+        provider = _RecordingProvider()
+        assert (
+            announce.send_announcement_to_instance(
+                provider, "I", "https://m", "meshcore"
+            )
+            is False
+        )
+        assert provider.sent == []
+
+    def test_private_fetch_error_fails_closed(self, monkeypatch):
+        """An unknown privacy state (None) is treated as private (fail-closed)."""
+        _stub_dogfeed(monkeypatch, private=None)
+        provider = _RecordingProvider()
+        assert (
+            announce.send_announcement_to_instance(
+                provider, "I", "https://m", "meshcore"
+            )
+            is False
+        )
+        assert provider.sent == []
+
+    def test_skips_when_activity_numbers_unavailable(self, monkeypatch):
+        """No announcement when the stats dogfeed yields nothing."""
+        monkeypatch.setattr(announce, "fetch_private_mode", lambda url, **_kw: False)
+        monkeypatch.setattr(announce, "fetch_activity", lambda url, proto, **_kw: None)
+        provider = _RecordingProvider()
+        assert (
+            announce.send_announcement_to_instance(
+                provider, "I", "https://m", "meshcore"
+            )
+            is False
+        )
+        assert provider.sent == []
+
+    def test_skips_when_provider_cannot_send(self, monkeypatch):
+        """A provider without send_channel_announcement is a no-op."""
+        _stub_dogfeed(monkeypatch, private=False, numbers=(1, 2))
+        assert (
+            announce.send_announcement_to_instance(
+                object(), "I", "https://m", "meshcore"
+            )
+            is False
+        )
+
+    def test_logs_transmitting_with_full_context(self, monkeypatch):
+        """Every outbound announcement emits an ``announce.tx`` debug line so the
+        TX is traceable end-to-end (target, protocol, numbers, text)."""
+        _stub_dogfeed(monkeypatch, private=False, numbers=(12, 50))
+        logs = []
+        monkeypatch.setattr(
+            announce.config,
+            "_debug_log",
+            lambda message, **meta: logs.append((message, meta)),
+        )
+        announce.send_announcement_to_instance(
+            _RecordingProvider(), "IFACE", "https://mesh.example", "meshcore"
+        )
+        transmit = [
+            meta for msg, meta in logs if msg == "Activity announcement transmitting"
+        ]
+        assert len(transmit) == 1
+        meta = transmit[0]
+        assert meta["context"] == "announce.tx"
+        assert meta["url"] == "https://mesh.example"
+        assert meta["protocol"] == "meshcore"
+        assert meta["active_nodes"] == 12
+        assert meta["packets_per_hour"] == 50
+        assert "packets/hour" in meta["text"]
+
+    def test_logs_skip_reason_when_private(self, monkeypatch):
+        """A skipped (private/unreachable) instance emits a skip debug line."""
+        _stub_dogfeed(monkeypatch, private=True)
+        logs = []
+        monkeypatch.setattr(
+            announce.config,
+            "_debug_log",
+            lambda message, **meta: logs.append((message, meta)),
+        )
+        announce.send_announcement_to_instance(
+            _RecordingProvider(), "I", "https://m", "meshcore"
+        )
+        assert any("skipped: instance private" in msg for msg, _meta in logs)
+
+
+class TestRunAnnouncementCycle:
+    """Tests for :func:`announce.run_announcement_cycle` (SPEC MA8 per-domain)."""
+
+    def test_domains_announces_each_configured_instance(self, monkeypatch):
+        """Every configured instance is announced to, with its own link."""
+        monkeypatch.setattr(
+            announce.config, "INSTANCES", (("https://a", ""), ("https://b", ""))
+        )
+        monkeypatch.setattr(announce.config, "PROTOCOL", "meshtastic")
+        _stub_dogfeed(monkeypatch, private=False, numbers=(3, 7))
+        provider = _RecordingProvider()
+        assert announce.run_announcement_cycle(provider, "IFACE") is True
+        urls = [text.rsplit(" ", 1)[-1] for _iface, text in provider.sent]
+        assert urls == ["https://a", "https://b"]
+
+    def test_cycle_survives_one_instance_error(self, monkeypatch):
+        """A failure against one instance never aborts the others."""
+        monkeypatch.setattr(
+            announce.config, "INSTANCES", (("https://bad", ""), ("https://good", ""))
+        )
+        monkeypatch.setattr(announce.config, "PROTOCOL", "meshcore")
+
+        def _priv(url, **_kw):
+            if url == "https://bad":
+                raise RuntimeError("boom")
+            return False
+
+        monkeypatch.setattr(announce, "fetch_private_mode", _priv)
+        monkeypatch.setattr(
+            announce, "fetch_activity", lambda url, proto, **_kw: (1, 1)
+        )
+        provider = _RecordingProvider()
+        assert announce.run_announcement_cycle(provider, "I") is True
+        assert [t.rsplit(" ", 1)[-1] for _i, t in provider.sent] == ["https://good"]
+
+
+class TestMaybeRunAnnouncements:
+    """Tests for the daemon entry point :func:`announce.maybe_run_announcements`."""
+
+    def test_gate_noop_when_not_enabled(self, monkeypatch):
+        """No cycle runs and last_announce is unchanged when RX_ONLY forbids TX."""
+        monkeypatch.setattr(announce.config, "RX_ONLY", True)
+        ran = []
+        monkeypatch.setattr(
+            announce, "run_announcement_cycle", lambda *a, **k: ran.append(True)
+        )
+        result = announce.maybe_run_announcements(
+            "P", "I", start_time=0, last_announce=42.0, now=1_000_000
+        )
+        assert result == 42.0
+        assert ran == []
+
+    def test_cadence_noop_when_not_due(self, monkeypatch):
+        """No cycle runs before the 24 h delay; last_announce is unchanged."""
+        monkeypatch.setattr(announce.config, "RX_ONLY", False)
+        ran = []
+        monkeypatch.setattr(
+            announce, "run_announcement_cycle", lambda *a, **k: ran.append(True)
+        )
+        now = 1_000_000
+        result = announce.maybe_run_announcements(
+            "P", "I", start_time=now - 1000, last_announce=None, now=now
+        )
+        assert result is None
+        assert ran == []
+
+    def test_cadence_runs_cycle_and_advances_timestamp(self, monkeypatch):
+        """When due, the cycle runs and last_announce advances to now."""
+        monkeypatch.setattr(announce.config, "RX_ONLY", False)
+        ran = []
+        monkeypatch.setattr(
+            announce,
+            "run_announcement_cycle",
+            lambda provider, iface, **k: ran.append((provider, iface)) or True,
+        )
+        now = 1_000_000
+        result = announce.maybe_run_announcements(
+            "P", "IFACE", start_time=now - 90_000, last_announce=None, now=now
+        )
+        assert result == now
+        assert ran == [("P", "IFACE")]
+
+    def test_cadence_defaults_now_to_wall_clock(self, monkeypatch):
+        """With ``now`` omitted the current wall clock is used (gate-off path)."""
+        monkeypatch.setattr(announce.config, "RX_ONLY", True)
+        result = announce.maybe_run_announcements(
+            "P", "I", start_time=0, last_announce=7.0
+        )
+        assert result == 7.0
diff --git a/tests/test_daemon_unit.py b/tests/test_daemon_unit.py
index 5116542..c1674e5 100644
--- a/tests/test_daemon_unit.py
+++ b/tests/test_daemon_unit.py
@@ -829,6 +829,33 @@ def test_loop_iteration_full_pass_returns_false(monkeypatch):
     assert daemon._loop_iteration(state) is False
 
 
+# ---------------------------------------------------------------------------
+# _process_announcements
+# ---------------------------------------------------------------------------
+
+
+def test_process_announcements_noop_without_iface():
+    """Returns the unchanged last_announce when no interface is connected."""
+    state = _make_state(iface=None, last_announce=123.0)
+    assert daemon._process_announcements(state) == 123.0
+
+
+def test_process_announcements_delegates_when_connected(monkeypatch):
+    """Delegates to announce.maybe_run_announcements with the daemon's provider,
+    interface, ingestor start time, and last_announce, and returns its result."""
+    calls = []
+
+    def _fake(provider, iface, *, start_time, last_announce):
+        calls.append((provider, iface, start_time, last_announce))
+        return 999.0
+
+    monkeypatch.setattr(daemon.announce, "maybe_run_announcements", _fake)
+    monkeypatch.setattr(daemon.ingestors, "ingestor_start_time", lambda: 555)
+    state = _make_state(iface="IFACE", provider="PROV", last_announce=None)
+    assert daemon._process_announcements(state) == 999.0
+    assert calls == [("PROV", "IFACE", 555, None)]
+
+
 # ---------------------------------------------------------------------------
 # PROTOCOL env-var selection
 # ---------------------------------------------------------------------------
diff --git a/tests/test_provider_unit.py b/tests/test_provider_unit.py
index eeea5b8..6467fa8 100644
--- a/tests/test_provider_unit.py
+++ b/tests/test_provider_unit.py
@@ -1957,6 +1957,118 @@ def test_poll_contact_telemetry_paths(monkeypatch):
     assert captured == []
 
 
+def test_poll_contact_telemetry_counts_tx(monkeypatch):
+    """Each on-air poll request counts as a transmission (SPEC MA1).
+
+    The stubbed handlers' ``_mark_packet_seen`` is a no-op, so the merged
+    activity counter moves only via the ``activity.record_tx()`` calls inside
+    ``_poll_contact_telemetry`` — one per on-air request. This pins the exact
+    TX count for the telemetry pull, the status fallback, an error path, and
+    the no-contact short-circuit.
+    """
+    import types
+
+    import data.mesh_ingestor.activity as activity
+
+    mc_tel, iface, stub, _captured = _telemetry_env(
+        monkeypatch, contacts=[{"public_key": _TEST_CONTACT_KEY, "adv_name": "Sensor"}]
+    )
+
+    def _mc(
+        telemetry_result=None,
+        telemetry_error=None,
+        status_result=None,
+        status_error=None,
+    ):
+        class _Commands:
+            async def req_telemetry_sync(self, contact):
+                if telemetry_error:
+                    raise telemetry_error
+                return telemetry_result
+
+            async def req_status_sync(self, contact):
+                if status_error:
+                    raise status_error
+                return status_result
+
+        return types.SimpleNamespace(commands=_Commands())
+
+    # LPP success → exactly one on-air request (telemetry pull), no fallback.
+    activity.take_packet_count()  # drain any activity from earlier tests
+    asyncio.run(
+        mc_tel._poll_contact_telemetry(
+            _mc(telemetry_result=[{"type": "temperature", "value": 21.5}]),
+            iface,
+            stub,
+            {},
+        )
+    )
+    assert activity.take_packet_count() == 1
+
+    # Empty telemetry → status fallback: two on-air requests counted.
+    asyncio.run(
+        mc_tel._poll_contact_telemetry(
+            _mc(telemetry_result=None, status_result={"bat": 4056}), iface, stub, {}
+        )
+    )
+    assert activity.take_packet_count() == 2
+
+    # Telemetry request raises → the attempt is still counted (one TX), no fallback.
+    asyncio.run(
+        mc_tel._poll_contact_telemetry(
+            _mc(telemetry_error=RuntimeError("timeout")), iface, stub, {}
+        )
+    )
+    assert activity.take_packet_count() == 1
+
+    # No resolvable contact → returns before any request → nothing counted.
+    empty = _MeshcoreInterface(target=None)
+    asyncio.run(mc_tel._poll_contact_telemetry(_mc(), empty, stub, {}))
+    assert activity.take_packet_count() == 0
+
+
+def test_poll_contact_telemetry_logs_initiation_and_empty_result(monkeypatch):
+    """Each poll logs initiation; an all-empty poll also logs 'returned no data'.
+
+    ``req_telemetry_sync``/``req_status_sync`` return ``None`` on timeout without
+    raising, so these two ``meshcore.telemetry.poll`` debug lines are the only
+    signal separating an unanswered on-air poll from a disabled poll loop.
+    """
+    import types
+
+    mc_tel, iface, stub, captured = _telemetry_env(
+        monkeypatch, contacts=[{"public_key": _TEST_CONTACT_KEY, "adv_name": "Sensor"}]
+    )
+    logs: list = []
+    monkeypatch.setattr(
+        mc_tel.config,
+        "_debug_log",
+        lambda message, **meta: logs.append((message, meta)),
+    )
+
+    class _Silent:
+        async def req_telemetry_sync(self, contact):
+            return None
+
+        async def req_status_sync(self, contact):
+            return None
+
+    asyncio.run(
+        mc_tel._poll_contact_telemetry(
+            types.SimpleNamespace(commands=_Silent()), iface, stub, {}
+        )
+    )
+
+    assert captured == []
+    node_id = iface.lookup_node_id(_TEST_CONTACT_KEY[:12])
+    assert [message for message, _meta in logs] == [
+        "MeshCore contact telemetry poll initiated",
+        "MeshCore contact telemetry poll returned no data",
+    ]
+    assert all(meta["context"] == "meshcore.telemetry.poll" for _msg, meta in logs)
+    assert all(meta["node_id"] == node_id for _msg, meta in logs)
+
+
 def test_telemetry_poll_loop_disabled_and_ticking(monkeypatch):
     """The poll loop exits when disabled and fires both poll kinds when enabled."""
     import types
@@ -4884,3 +4996,119 @@ def test_run_meshcore_autoadd_set_rejected_logs_warning(monkeypatch):
         and kw.get("autoadd_config") == 0x01
         for _msg, kw in logs
     )
+
+
+# ---------------------------------------------------------------------------
+# send_channel_announcement (SPEC MA6/MA9): optional duck-typed provider TX
+# ---------------------------------------------------------------------------
+
+
+def test_send_channel_announcement_meshtastic_sends_and_counts():
+    """Meshtastic sends on CHANNEL_INDEX and counts the transmission (MA1)."""
+    import data.mesh_ingestor.activity as activity
+    import data.mesh_ingestor.config as config
+
+    calls = []
+
+    class _Iface:
+        def sendText(self, text, channelIndex=0):
+            calls.append((text, channelIndex))
+
+    activity.take_packet_count()  # drain
+    MeshtasticProvider().send_channel_announcement(_Iface(), "hello mesh")
+    assert calls == [("hello mesh", config.CHANNEL_INDEX)]
+    assert activity.take_packet_count() == 1
+
+
+def test_send_channel_announcement_meshtastic_noop_without_sendtext():
+    """An interface lacking sendText is a no-op that counts no TX."""
+    import data.mesh_ingestor.activity as activity
+
+    activity.take_packet_count()
+    MeshtasticProvider().send_channel_announcement(object(), "hi")
+    assert activity.take_packet_count() == 0
+
+
+def test_send_channel_announcement_meshcore_sends_and_counts():
+    """MeshCore schedules send_chan_msg on its loop and counts the TX (MA1)."""
+    import asyncio as _asyncio
+    import threading as _threading
+    import types as _types
+    import data.mesh_ingestor.activity as activity
+    import data.mesh_ingestor.config as config
+
+    iface = _MeshcoreInterface(target=None)
+    sent = []
+
+    class _Commands:
+        async def send_chan_msg(self, chan, msg, timestamp=None):
+            sent.append((chan, msg))
+            return _types.SimpleNamespace()
+
+    iface._mc = _types.SimpleNamespace(commands=_Commands())
+
+    loop = _asyncio.new_event_loop()
+    thread = _threading.Thread(target=loop.run_forever, daemon=True)
+    thread.start()
+    iface._loop = loop
+    try:
+        activity.take_packet_count()  # drain
+        MeshcoreProvider().send_channel_announcement(iface, "hello mesh")
+        assert sent == [(config.CHANNEL_INDEX, "hello mesh")]
+        assert activity.take_packet_count() == 1
+    finally:
+        loop.call_soon_threadsafe(loop.stop)
+        thread.join(timeout=5)
+        loop.close()
+
+
+def test_send_channel_announcement_meshcore_noop_guards():
+    """No-op (no TX) for a wrong iface type or a missing/closed loop or handle."""
+    import asyncio as _asyncio
+    import types as _types
+    import data.mesh_ingestor.activity as activity
+
+    provider = MeshcoreProvider()
+    activity.take_packet_count()  # drain
+
+    # Wrong interface type.
+    provider.send_channel_announcement(object(), "x")
+    # Missing mc / loop (fresh interface).
+    provider.send_channel_announcement(_MeshcoreInterface(target=None), "x")
+    # Closed loop.
+    closed = _MeshcoreInterface(target=None)
+    loop = _asyncio.new_event_loop()
+    loop.close()
+    closed._mc = _types.SimpleNamespace(commands=_types.SimpleNamespace())
+    closed._loop = loop
+    provider.send_channel_announcement(closed, "x")
+
+    assert activity.take_packet_count() == 0
+
+
+def test_send_channel_announcement_is_optional_MeshProtocol_member():
+    """The send method is an optional duck-typed extension: both providers expose
+    it and still satisfy MeshProtocol, but it is NOT a required member — a minimal
+    conforming provider without it still passes isinstance (MA9 / A4b)."""
+    assert hasattr(MeshtasticProvider(), "send_channel_announcement")
+    assert hasattr(MeshcoreProvider(), "send_channel_announcement")
+    assert isinstance(MeshtasticProvider(), MeshProtocol)
+    assert isinstance(MeshcoreProvider(), MeshProtocol)
+
+    class _Minimal:
+        name = "minimal"
+
+        def subscribe(self):
+            return []
+
+        def connect(self, *, active_candidate):
+            return (None, None, None)
+
+        def extract_host_node_id(self, iface):
+            return None
+
+        def node_snapshot_items(self, iface):
+            return []
+
+    assert isinstance(_Minimal(), MeshProtocol)
+    assert not hasattr(_Minimal(), "send_channel_announcement")
diff --git a/web/lib/potato_mesh/application/data_processing/ingestors.rb b/web/lib/potato_mesh/application/data_processing/ingestors.rb
index f0802a5..d82b57d 100644
--- a/web/lib/potato_mesh/application/data_processing/ingestors.rb
+++ b/web/lib/potato_mesh/application/data_processing/ingestors.rb
@@ -67,6 +67,8 @@ module PotatoMesh
                      SQL
         end
 
+        record_ingestor_activity(db, node_id, last_seen_time, payload["packets"], protocol)
+
         true
       rescue SQLite3::SQLException => e
         warn_log(
@@ -78,6 +80,41 @@ module PotatoMesh
         )
         false
       end
+
+      # Append a per-heartbeat activity row for the moving-average time-series
+      # (SPEC MA2/MA3), called after the ingestor liveness upsert.
+      #
+      # A missing, non-numeric, or negative +packets+ value records **no** row:
+      # the field is additive, so older ingestors that never send it simply
+      # contribute nothing to the time-series. A failure here is logged and
+      # swallowed rather than propagated — the supplementary activity write must
+      # never sink a heartbeat whose liveness upsert has already committed.
+      #
+      # @param db [SQLite3::Database] open database handle.
+      # @param ingestor_id [String] canonical ingestor node id.
+      # @param at [Integer] heartbeat timestamp used as the activity bucket time.
+      # @param raw_packets [Object] the payload's +packets+ value (may be nil).
+      # @param protocol [String] the ingestor's declared protocol.
+      # @return [void]
+      def record_ingestor_activity(db, ingestor_id, at, raw_packets, protocol)
+        packets = coerce_integer(raw_packets)
+        return if packets.nil? || packets.negative?
+
+        with_busy_retry do
+          db.execute(
+            "INSERT INTO ingestor_activity(ingestor_id, at, packets, protocol) VALUES(?,?,?,?)",
+            [ingestor_id, at, packets, protocol],
+          )
+        end
+      rescue SQLite3::SQLException => e
+        warn_log(
+          "Failed to record ingestor activity",
+          context: "data_processing.ingestors",
+          node_id: ingestor_id,
+          error_class: e.class.name,
+          error_message: e.message,
+        )
+      end
     end
   end
 end
diff --git a/web/lib/potato_mesh/application/database.rb b/web/lib/potato_mesh/application/database.rb
index fac8292..4520503 100644
--- a/web/lib/potato_mesh/application/database.rb
+++ b/web/lib/potato_mesh/application/database.rb
@@ -110,7 +110,7 @@ module PotatoMesh
       def init_db
         FileUtils.mkdir_p(File.dirname(PotatoMesh::Config.db_path))
         db = open_database
-        %w[nodes messages positions telemetry neighbors instances traces ingestors].each do |schema|
+        %w[nodes messages positions telemetry neighbors instances traces ingestors ingestor_activity].each do |schema|
           sql_file = File.expand_path("../../../../data/#{schema}.sql", __dir__)
           db.execute_batch(File.read(sql_file))
         end
@@ -537,6 +537,17 @@ module PotatoMesh
             db.execute("UPDATE ingestors SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''")
           end
         end
+
+        # The per-heartbeat activity time-series (SPEC MA3) is a standalone,
+        # append-only table; older installations gain it here without any data
+        # backfill (the moving average simply starts populating from the next
+        # heartbeat that carries a `packets` count).
+        activity_tables =
+          db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='ingestor_activity'").flatten
+        if activity_tables.empty?
+          activity_schema = File.expand_path("../../../../data/ingestor_activity.sql", __dir__)
+          db.execute_batch(File.read(activity_schema))
+        end
       rescue SQLite3::SQLException, Errno::ENOENT => e
         warn_log(
           "Failed to apply schema upgrade",
diff --git a/web/lib/potato_mesh/application/queries.rb b/web/lib/potato_mesh/application/queries.rb
index baa80ed..4b5f373 100644
--- a/web/lib/potato_mesh/application/queries.rb
+++ b/web/lib/potato_mesh/application/queries.rb
@@ -16,6 +16,7 @@
 
 require_relative "queries/common"
 require_relative "queries/node_queries"
+require_relative "queries/ingestor_queries"
 require_relative "queries/chat_queries"
 require_relative "queries/telemetry_queries"
 require_relative "queries/federation_queries"
diff --git a/web/lib/potato_mesh/application/queries/ingestor_queries.rb b/web/lib/potato_mesh/application/queries/ingestor_queries.rb
new file mode 100644
index 0000000..9672569
--- /dev/null
+++ b/web/lib/potato_mesh/application/queries/ingestor_queries.rb
@@ -0,0 +1,90 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# frozen_string_literal: true
+
+module PotatoMesh
+  module App
+    module Queries
+      # Rolling window used for the packets/hour moving average (SPEC MA4): each
+      # ingestor's packet total over the last 24 hours.
+      PACKETS_PER_HOUR_WINDOW_SECONDS = 86_400
+
+      # Fixed denominator that turns a window total into an hourly rate. Derived
+      # from the window (24 h) rather than the *elapsed* span so the rate stays
+      # stable and never spikes for a freshly-started ingestor with only a few
+      # hours of data (SPEC MA4). The announcement only fires ≥ 24 h after start
+      # (MA7), so by then a live ingestor has a full window anyway.
+      PACKETS_PER_HOUR_DIVISOR = PACKETS_PER_HOUR_WINDOW_SECONDS / 3600.0
+
+      # Compute the mesh-wide packets/hour moving average per protocol scope
+      # (SPEC MA4/MA5), aggregated **MAX-per-protocol** across ingestors.
+      #
+      # A single radio can only hear ≤ what is actually transmitted, so the
+      # busiest single vantage is the best dedup-free estimate of unique air
+      # traffic and can never double-count a frame heard by two radios (true
+      # per-frame dedup is impossible — ignored/errored frames carry no id).
+      # For each protocol the rate is
+      # +MAX(ingestor's 24 h packet total) ÷ 24+, rounded; +total+ is the same
+      # MAX taken over **every** ingestor regardless of protocol (so an ingestor
+      # reporting several protocols contributes its combined total). +reticulum+
+      # is a forward-looking always-zero stub (SPEC S6/MA5).
+      #
+      # @param now [Integer] reference unix timestamp in seconds.
+      # @param db [SQLite3::Database, nil] optional open database handle to reuse.
+      # @return [Hash{String => Integer}] +{ "total", "meshcore", "meshtastic",
+      #   "reticulum" }+ => rounded packets/hour.
+      def query_packets_per_hour(now: Time.now.to_i, db: nil)
+        handle = db || open_database(readonly: true)
+        handle.results_as_hash = true
+        reference_now = coerce_integer(now) || Time.now.to_i
+        cutoff = reference_now - PACKETS_PER_HOUR_WINDOW_SECONDS
+
+        rows = with_busy_retry do
+          handle.execute(
+            "SELECT ingestor_id, protocol AS p, SUM(packets) AS total " \
+            "FROM ingestor_activity WHERE at >= ? GROUP BY ingestor_id, protocol",
+            [cutoff],
+          )
+        end
+
+        per_protocol_max = Hash.new(0)
+        per_ingestor_total = Hash.new(0)
+        rows.each do |row|
+          protocol = row["p"]
+          total = row["total"].to_i
+          per_protocol_max[protocol] = [per_protocol_max[protocol], total].max if protocol
+          per_ingestor_total[row["ingestor_id"]] += total
+        end
+
+        {
+          "total" => packets_per_hour_rate(per_ingestor_total.values.max || 0),
+          "meshcore" => packets_per_hour_rate(per_protocol_max["meshcore"]),
+          "meshtastic" => packets_per_hour_rate(per_protocol_max["meshtastic"]),
+          "reticulum" => 0,
+        }
+      ensure
+        handle&.close unless db
+      end
+
+      # Convert a 24-hour packet total into a rounded packets/hour rate (MA4).
+      #
+      # @param total_packets [Integer] packets observed over the 24 h window.
+      # @return [Integer] rounded hourly rate.
+      def packets_per_hour_rate(total_packets)
+        (total_packets / PACKETS_PER_HOUR_DIVISOR).round
+      end
+    end
+  end
+end
diff --git a/web/lib/potato_mesh/application/retention.rb b/web/lib/potato_mesh/application/retention.rb
index d0761d7..0b2b86b 100644
--- a/web/lib/potato_mesh/application/retention.rb
+++ b/web/lib/potato_mesh/application/retention.rb
@@ -51,6 +51,7 @@ module PotatoMesh
         ["telemetry", "rx_time"],
         ["traces", "rx_time"],
         ["ingestors", "last_seen_time"],
+        ["ingestor_activity", "at"],
         ["nodes", "last_heard"],
       ].freeze
 
diff --git a/web/lib/potato_mesh/application/routes/api.rb b/web/lib/potato_mesh/application/routes/api.rb
index a9be0d9..49e06ea 100644
--- a/web/lib/potato_mesh/application/routes/api.rb
+++ b/web/lib/potato_mesh/application/routes/api.rb
@@ -115,9 +115,16 @@ module PotatoMesh
             content_type :json
             priv = private_mode? ? 1 : 0
             cached = PotatoMesh::App::ApiCache.fetch("api:stats:#{priv}", ttl_seconds: 15) do
-              # Scope → metric → window tree (SPEC S1). +sampled+ stays last and
+              # Scope → metric → window tree (SPEC S1) plus the additive
+              # +packets_per_hour+ moving-average map (SPEC MA5); both are drawn
+              # from independent read-side queries. +sampled+ stays last and
               # +false+ for backward continuity with the prior payload.
-              query_active_node_stats.merge("sampled" => false).to_json
+              query_active_node_stats
+                .merge(
+                  "packets_per_hour" => query_packets_per_hour,
+                  "sampled" => false,
+                )
+                .to_json
             end
 
             etag cached[:etag], kind: :weak
diff --git a/web/spec/activity_spec.rb b/web/spec/activity_spec.rb
new file mode 100644
index 0000000..fe026d1
--- /dev/null
+++ b/web/spec/activity_spec.rb
@@ -0,0 +1,149 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# frozen_string_literal: true
+
+require "spec_helper"
+require "json"
+require "time"
+
+# Web-side coverage for the per-heartbeat activity time-series (SPEC MA3) and
+# the additive +packets+ delta on the ingestor heartbeat (MA2). A heartbeat
+# carrying a non-negative integer +packets+ appends exactly one append-only
+# +ingestor_activity+ row; anything else records nothing.
+RSpec.describe "Ingestor activity time-series (MA3)" do
+  let(:app) { Sinatra::Application }
+  let(:api_token) { "secret-token" }
+  let(:auth_headers) do
+    {
+      "CONTENT_TYPE" => "application/json",
+      "HTTP_AUTHORIZATION" => "Bearer #{api_token}",
+    }
+  end
+
+  before do
+    @original_token = ENV["API_TOKEN"]
+    ENV["API_TOKEN"] = api_token
+    clear_tables
+  end
+
+  after do
+    ENV["API_TOKEN"] = @original_token
+    clear_tables
+  end
+
+  def clear_tables
+    with_db do |db|
+      db.execute("DELETE FROM ingestor_activity")
+      db.execute("DELETE FROM ingestors")
+    end
+  end
+
+  def with_db(readonly: false)
+    db = PotatoMesh::Application.open_database(readonly: readonly)
+    db.busy_timeout = PotatoMesh::Config.db_busy_timeout_ms
+    db.execute("PRAGMA foreign_keys = ON")
+    yield db
+  ensure
+    db&.close
+  end
+
+  def post_heartbeat(overrides = {})
+    now = Time.now.to_i
+    payload = {
+      node_id: "!abc12345",
+      start_time: now - 3600,
+      last_seen_time: now - 60,
+      version: "0.6.0",
+      protocol: "meshtastic",
+    }.merge(overrides)
+    post "/api/ingestors", payload.to_json, auth_headers
+  end
+
+  def activity_rows
+    with_db(readonly: true) do |db|
+      db.execute(
+        "SELECT ingestor_id, at, packets, protocol FROM ingestor_activity ORDER BY id",
+      )
+    end
+  end
+
+  describe "POST /api/ingestors" do
+    it "records ingestor activity for a heartbeat carrying packets" do
+      now = Time.now.to_i
+      post_heartbeat(last_seen_time: now - 60, packets: 42, protocol: "meshcore")
+      expect(last_response.status).to eq(201)
+
+      rows = activity_rows
+      expect(rows.length).to eq(1)
+      ingestor_id, at, packets, protocol = rows.first
+      expect(ingestor_id).to eq("!abc12345")
+      expect(at).to eq(now - 60)
+      expect(packets).to eq(42)
+      expect(protocol).to eq("meshcore")
+    end
+
+    it "records ingestor activity of zero for an idle heartbeat" do
+      post_heartbeat(packets: 0)
+      expect(last_response.status).to eq(201)
+      rows = activity_rows
+      expect(rows.length).to eq(1)
+      expect(rows.first[2]).to eq(0)
+    end
+
+    it "records ingestor activity independently per ingestor" do
+      post_heartbeat(node_id: "!abc12345", packets: 10)
+      post_heartbeat(node_id: "!def67890", packets: 25)
+      expect(activity_rows.map { |r| [r[0], r[2]] }).to contain_exactly(
+        ["!abc12345", 10],
+        ["!def67890", 25],
+      )
+    end
+
+    it "appends a fresh activity row on every heartbeat (append-only)" do
+      post_heartbeat(node_id: "!abc12345", packets: 3)
+      post_heartbeat(node_id: "!abc12345", packets: 4)
+      expect(activity_rows.map { |r| r[2] }).to eq([3, 4])
+    end
+
+    it "records no activity when packets is absent (older ingestor)" do
+      post_heartbeat
+      expect(last_response.status).to eq(201)
+      expect(activity_rows).to be_empty
+    end
+
+    it "records no activity when packets is negative or non-numeric" do
+      post_heartbeat(node_id: "!abc12345", packets: -5)
+      post_heartbeat(node_id: "!def67890", packets: "abc")
+      expect(activity_rows).to be_empty
+    end
+
+    it "still records the heartbeat when the activity insert fails" do
+      # Force the supplementary activity INSERT to raise by removing its table,
+      # then prove the liveness heartbeat still returns 201 (graceful
+      # degradation). The table is restored so the surrounding suite is
+      # unaffected.
+      with_db { |db| db.execute("ALTER TABLE ingestor_activity RENAME TO ingestor_activity_bak") }
+      begin
+        post_heartbeat(packets: 99)
+        expect(last_response.status).to eq(201)
+      ensure
+        with_db do |db|
+          db.execute("ALTER TABLE ingestor_activity_bak RENAME TO ingestor_activity")
+        end
+      end
+      expect(activity_rows).to be_empty
+    end
+  end
+end
diff --git a/web/spec/app_spec.rb b/web/spec/app_spec.rb
index 59d7ed6..3bce1d1 100644
--- a/web/spec/app_spec.rb
+++ b/web/spec/app_spec.rb
@@ -108,6 +108,7 @@ RSpec.describe "Potato Mesh Sinatra app" do
       db.execute("DELETE FROM positions")
       db.execute("DELETE FROM telemetry")
       db.execute("DELETE FROM ingestors")
+      db.execute("DELETE FROM ingestor_activity")
     end
     ensure_self_instance_record!
   end
@@ -6709,6 +6710,45 @@ RSpec.describe "Potato Mesh Sinatra app" do
   end
 
   describe "GET /api/stats" do
+    it "exposes the additive packets_per_hour MAX-per-protocol map" do
+      clear_database
+      now = reference_time.to_i
+      allow(Time).to receive(:now).and_return(reference_time)
+
+      with_db do |db|
+        # meshcore: busiest ingestor 1200 pkts/24h ⇒ 50/h; a quieter second
+        # ingestor of the same protocol must not inflate the MAX.
+        db.execute(
+          "INSERT INTO ingestor_activity(ingestor_id, at, packets, protocol) VALUES (?,?,?,?)",
+          ["!core0001", now - 100, 1200, "meshcore"],
+        )
+        db.execute(
+          "INSERT INTO ingestor_activity(ingestor_id, at, packets, protocol) VALUES (?,?,?,?)",
+          ["!core0002", now - 100, 720, "meshcore"],
+        )
+        # meshtastic: 720 pkts/24h ⇒ 30/h.
+        db.execute(
+          "INSERT INTO ingestor_activity(ingestor_id, at, packets, protocol) VALUES (?,?,?,?)",
+          ["!tast0001", now - 100, 720, "meshtastic"],
+        )
+      end
+
+      get "/api/stats"
+
+      expect(last_response).to be_ok
+      payload = JSON.parse(last_response.body)
+      pph = payload["packets_per_hour"]
+      expect(pph.keys).to contain_exactly("total", "meshcore", "meshtastic", "reticulum")
+      expect(pph["meshcore"]).to eq(50)
+      expect(pph["meshtastic"]).to eq(30)
+      expect(pph["total"]).to eq(50) # MAX over every ingestor = 1200 / 24
+      expect(pph["reticulum"]).to eq(0)
+      # The additive field leaves the S1 scope × metric × window tree intact.
+      expect(payload["sampled"]).to eq(false)
+      expect(payload["total"]).to have_key("nodes")
+      expect(payload).not_to have_key("active_nodes")
+    end
+
     it "returns exact SQL-backed activity counts with per-protocol breakdowns" do
       clear_database
       now = reference_time.to_i
diff --git a/web/spec/queries_spec.rb b/web/spec/queries_spec.rb
index 9a2137c..3a5a4f2 100644
--- a/web/spec/queries_spec.rb
+++ b/web/spec/queries_spec.rb
@@ -1263,6 +1263,86 @@ RSpec.describe PotatoMesh::App::Queries do
     end
   end
 
+  describe "#query_packets_per_hour" do
+    before do
+      with_db { |db| db.execute("DELETE FROM ingestor_activity") }
+    end
+
+    after do
+      with_db { |db| db.execute("DELETE FROM ingestor_activity") }
+    end
+
+    def seed_activity(rows)
+      with_db do |db|
+        rows.each do |ingestor_id, at, packets, protocol|
+          db.execute(
+            "INSERT INTO ingestor_activity(ingestor_id, at, packets, protocol) VALUES (?,?,?,?)",
+            [ingestor_id, at, packets, protocol],
+          )
+        end
+      end
+    end
+
+    it "returns all zeros when there is no activity" do
+      expect(queries.query_packets_per_hour(now: now)).to eq(
+        "total" => 0, "meshcore" => 0, "meshtastic" => 0, "reticulum" => 0,
+      )
+    end
+
+    it "takes the MAX per protocol of each ingestor's 24h total over 24" do
+      seed_activity(
+        [
+          # meshcore ingestor A: 1200 over the window (two heartbeats).
+          ["!coreaaaa", now - 100, 700, "meshcore"],
+          ["!coreaaaa", now - 3700, 500, "meshcore"],
+          # meshcore ingestor B: 900 — the quieter vantage must not inflate MAX.
+          ["!corebbbb", now - 200, 900, "meshcore"],
+          # meshtastic ingestor C: 720.
+          ["!tastcccc", now - 300, 720, "meshtastic"],
+        ],
+      )
+      result = queries.query_packets_per_hour(now: now)
+      expect(result["meshcore"]).to eq(50) # MAX(1200, 900) / 24
+      expect(result["meshtastic"]).to eq(30) # 720 / 24
+      expect(result["total"]).to eq(50) # MAX over every ingestor = 1200 / 24
+      expect(result["reticulum"]).to eq(0)
+    end
+
+    it "reports zero for a protocol with no active ingestor" do
+      seed_activity([["!coreaaaa", now - 100, 240, "meshcore"]])
+      result = queries.query_packets_per_hour(now: now)
+      expect(result["meshcore"]).to eq(10) # 240 / 24
+      expect(result["meshtastic"]).to eq(0)
+      expect(result["reticulum"]).to eq(0)
+    end
+
+    it "excludes activity older than the 24h window" do
+      seed_activity(
+        [
+          ["!coreaaaa", now - 100, 480, "meshcore"],        # inside → counts
+          ["!coreaaaa", now - 90_000, 100_000, "meshcore"], # >24h → excluded
+        ],
+      )
+      expect(queries.query_packets_per_hour(now: now)["meshcore"]).to eq(20) # 480 / 24
+    end
+
+    it "keeps reticulum a zero stub even when reticulum activity exists" do
+      seed_activity([["!reti0001", now - 100, 720, "reticulum"]])
+      result = queries.query_packets_per_hour(now: now)
+      # The reticulum scope is always emitted as zero (forward-looking stub)…
+      expect(result["reticulum"]).to eq(0)
+      # …but a reticulum ingestor still contributes to the protocol-agnostic
+      # total (MAX across every ingestor).
+      expect(result["total"]).to eq(30) # 720 / 24
+    end
+
+    it "rounds the hourly rate to the nearest integer" do
+      seed_activity([["!coreaaaa", now - 100, 100, "meshcore"]])
+      # 100 / 24 = 4.166… → 4
+      expect(queries.query_packets_per_hour(now: now)["meshcore"]).to eq(4)
+    end
+  end
+
   describe "#query_telemetry_buckets" do
     it "clamps oversized window_seconds to the 28-day visibility cap" do
       huge_window = PotatoMesh::Config.four_weeks_seconds * 50
diff --git a/web/spec/retention_spec.rb b/web/spec/retention_spec.rb
index 31dcc8b..239c777 100644
--- a/web/spec/retention_spec.rb
+++ b/web/spec/retention_spec.rb
@@ -208,6 +208,37 @@ RSpec.describe PotatoMesh::App::Retention do
       expect(removed["ingestors"]).to eq(1)
     end
 
+    it "prunes ingestor_activity rows past the retention window" do
+      fresh = now - 100
+      stale = now - PotatoMesh::Config.year_seconds - 86_400
+
+      db = SQLite3::Database.new(PotatoMesh::Config.db_path)
+      begin
+        db.execute(
+          "INSERT INTO ingestor_activity(ingestor_id, at, packets, protocol) VALUES (?,?,?,?)",
+          ["!fffff001", fresh, 42, "meshtastic"],
+        )
+        db.execute(
+          "INSERT INTO ingestor_activity(ingestor_id, at, packets, protocol) VALUES (?,?,?,?)",
+          ["!aaaaa001", stale, 7, "meshcore"],
+        )
+      ensure
+        db&.close
+      end
+
+      removed = harness_class.purge_old_data!(now: now)
+      expect(removed["ingestor_activity"]).to eq(1)
+
+      db = SQLite3::Database.new(PotatoMesh::Config.db_path, readonly: true)
+      begin
+        remaining =
+          db.execute("SELECT ingestor_id FROM ingestor_activity ORDER BY ingestor_id").flatten
+        expect(remaining).to eq(["!fffff001"])
+      ensure
+        db&.close
+      end
+    end
+
     it "keeps every row when the database is empty" do
       removed = harness_class.purge_old_data!(now: now)
       expect(removed.values.uniq).to eq([0])