Revise PR #318: the step-1 audit was reported as posted to the bus but was never posted, and its sender census sees 1 of 11 senders

changelog.d/tsk-6ycnvz-sender-census.md

@@ -0,0 +1,3 @@
+### Added
+
+- `a2a_sender_census()` returns a per-sender message census across all A2A channels, with per-channel breakdowns and total counts sorted by volume descending. Exposed as `GET /a2a/census` on the HTTP server.

taosmd/http_server.py

@@ -1033,6 +1033,8 @@ def _dispatch(self, method: str) -> None:
                     self._handle_a2a_send()
                 elif method == "GET" and path == "/a2a/channels":
                     self._handle_a2a_channels()
+                elif method == "GET" and path == "/a2a/census":
+                    self._handle_a2a_census()
                 elif method == "GET" and path == "/a2a/members":
                     self._handle_a2a_members(query)
                 elif method == "GET" and path == "/a2a/messages":
@@ -1490,6 +1492,10 @@ def _handle_a2a_channels(self) -> None:
             channels = runner.run(service.a2a_channels(data_dir=data_dir))
             self._send_json(200, {"channels": channels})
 
+        def _handle_a2a_census(self) -> None:
+            census = runner.run(service.a2a_sender_census(data_dir=data_dir))
+            self._send_json(200, {"census": census})
+
         def _handle_a2a_members(self, qs: dict) -> None:
             channel = (qs.get("channel") or [None])[0]
             if not channel:

taosmd/remote.py

@@ -253,6 +253,11 @@ async def a2a_channels(self, **_opts) -> list[dict]:
         resp = await self._run("GET", "/a2a/channels")
         return resp.get("channels", [])
 
+    async def a2a_sender_census(self, **_opts) -> dict:
+        """GET /a2a/census: return a per-sender message census from the remote bus."""
+        resp = await self._run("GET", "/a2a/census")
+        return resp.get("census", {})
+
     async def a2a_members(self, *, channel: str, **_opts) -> list[str]:
         """GET /a2a/members: return distinct sender names on ``channel``."""
         resp = await self._run("GET", "/a2a/members", params={"channel": channel})

taosmd/service.py

@@ -657,6 +657,67 @@ async def a2a_channels(*, data_dir=None) -> list[dict]:
     return result
 
 
+async def a2a_sender_census(*, data_dir=None) -> dict:
+    """Return a per-sender message census across every A2A channel.
+
+    Queries all :data:`~taosmd.archive.EVENT_A2A` events and aggregates them
+    by sender.  The result maps each distinct ``from`` value to a dict with:
+
+    * ``total`` -- total messages sent by that sender across all channels
+    * ``channels`` -- mapping of channel name to message count on that channel
+
+    Senders are returned in descending order of ``total``.
+
+    When a remote server URL is configured the call is forwarded to
+    :class:`~taosmd.remote.RemoteClient` transparently.
+    """
+    remote = _get_remote(data_dir)
+    if remote is not None:
+        return await remote.a2a_sender_census()
+    stores = await _api._ensure_stores(data_dir)
+    archive = stores["archive"]
+    rows = await archive.query(event_type=EVENT_A2A, limit=100_000)
+
+    deleted: set[str] = set()
+    aliases: dict[str, str] = {}
+    superseded: set[int] = set()
+    if data_dir is not None:
+        from .admin import A2AAdminState  # noqa: PLC0415
+        _admin = A2AAdminState(data_dir)
+        deleted = _admin.deleted_channels()
+        aliases = _admin.channel_aliases()
+        superseded = _admin.superseded_messages()
+
+    census: dict[str, dict] = {}
+    for row in rows:
+        try:
+            data = json.loads(row.get("data_json", "{}"))
+        except (json.JSONDecodeError, TypeError):
+            data = {}
+        if data.get("admin_action"):
+            continue
+        row_id = row["id"]
+        if row_id in superseded:
+            continue
+        sender = data.get("from") or ""
+        if not sender:
+            continue
+        thread = data.get("thread") or row.get("app_id") or "general"
+        if thread in aliases:
+            thread = aliases[thread]
+        if thread in deleted:
+            continue
+        if sender not in census:
+            census[sender] = {"total": 0, "channels": {}}
+        entry = census[sender]
+        entry["total"] += 1
+        entry["channels"][thread] = entry["channels"].get(thread, 0) + 1
+
+    return dict(
+        sorted(census.items(), key=lambda item: item[1]["total"], reverse=True)
+    )
+
+
 async def a2a_members(*, channel: str, data_dir=None) -> list[str]:
     """Return distinct sender names observed on ``channel``, sorted.
 
@@ -1262,8 +1323,8 @@ async def collections_archive(collection_id: str, *, data_dir=None) -> dict:
 
 
 __all__ = ["ingest", "search", "pending_list", "pending_resolve", "reconcile", "stats",
-           "supersede", "fetch_by_ref", "a2a_send", "a2a_feed", "a2a_channels", "a2a_members",
-           "a2a_threads", "a2a_thread_messages",
+           "supersede", "fetch_by_ref", "a2a_send", "a2a_feed", "a2a_channels", "a2a_sender_census",
+           "a2a_members", "a2a_threads", "a2a_thread_messages",
            "task_create", "task_list", "task_ready", "task_prime",
            "task_update", "task_add_edge", "task_remove_edge", "task_projects",
            "admin_shelf_create", "admin_shelf_archive", "admin_shelf_unarchive",

tests/test_a2a_channels.py

@@ -313,3 +313,68 @@ def test_a2a_setup_guide_contains_join_step():
     """Guide mentions a JOIN message."""
     text = taosmd.a2a_setup_guide()
     assert "JOIN" in text or "join" in text.lower()
+
+
+# ---------------------------------------------------------------------------
+# service.a2a_sender_census
+# ---------------------------------------------------------------------------
+
+def test_a2a_sender_census_counts_per_sender(isolated_data_dir):
+    """a2a_sender_census returns total and per-channel counts per sender."""
+    dd = str(isolated_data_dir)
+
+    asyncio.run(service.a2a_send("alice", "a1", thread="t1", data_dir=dd))
+    asyncio.run(service.a2a_send("bob", "b1", thread="t1", data_dir=dd))
+    asyncio.run(service.a2a_send("alice", "a2", thread="t2", data_dir=dd))
+    asyncio.run(service.a2a_send("alice", "a3", thread="t1", data_dir=dd))
+
+    census = asyncio.run(service.a2a_sender_census(data_dir=dd))
+    assert "alice" in census
+    assert census["alice"]["total"] == 3
+    assert census["alice"]["channels"]["t1"] == 2
+    assert census["alice"]["channels"]["t2"] == 1
+    assert census["bob"]["total"] == 1
+    assert census["bob"]["channels"]["t1"] == 1
+
+
+def test_a2a_sender_census_sorted_by_total_desc(isolated_data_dir):
+    """Senders are returned in descending order of total message count."""
+    dd = str(isolated_data_dir)
+
+    asyncio.run(service.a2a_send("zara", "z1", thread="t1", data_dir=dd))
+    asyncio.run(service.a2a_send("alice", "a1", thread="t1", data_dir=dd))
+    asyncio.run(service.a2a_send("zara", "z2", thread="t1", data_dir=dd))
+
+    census = asyncio.run(service.a2a_sender_census(data_dir=dd))
+    senders = list(census.keys())
+    assert senders == ["zara", "alice"]
+
+
+def test_a2a_sender_census_empty_store(isolated_data_dir):
+    """a2a_sender_census returns an empty dict when no messages exist."""
+    census = asyncio.run(service.a2a_sender_census(data_dir=str(isolated_data_dir)))
+    assert census == {}
+
+
+# ---------------------------------------------------------------------------
+# HTTP /a2a/census
+# ---------------------------------------------------------------------------
+
+def test_http_a2a_census_returns_census(live_server):
+    """GET /a2a/census returns a census dict with sender totals."""
+    _http_post(f"{live_server}/a2a/send", {"from": "alice", "body": "hi", "thread": "census-ch"})
+    _http_post(f"{live_server}/a2a/send", {"from": "bob", "body": "hey", "thread": "census-ch"})
+    _http_post(f"{live_server}/a2a/send", {"from": "alice", "body": "again", "thread": "census-ch"})
+
+    status, body = _get(f"{live_server}/a2a/census")
+    assert status == 200, body
+    assert "alice" in body["census"]
+    assert body["census"]["alice"]["total"] == 2
+    assert body["census"]["bob"]["total"] == 1
+
+
+def test_http_a2a_census_empty(live_server):
+    """GET /a2a/census on an empty server returns empty census."""
+    status, body = _get(f"{live_server}/a2a/census")
+    assert status == 200, body
+    assert body["census"] == {}