fix(mcp): route listen-channel messages through the shared ping tally

src/fast_agent/mcp/transport_tracking.py

@@ -424,8 +424,7 @@ def _handle_listen_event(self, event: ChannelEvent, now: datetime) -> None:
                 summary = event.detail or "subscription event"
                 classification = ActivityState.NOTIFICATION
             else:
-                classification = self._classify_message(event.message)
-                self._listen_counts.increment(classification)
+                classification = self._tally_message_counts("listen", event.message, now)
                 summary = _summarise_classified_message(classification, event.message)
             self._listen_last_summary = summary
             self._listen_last_at = now
@@ -593,6 +592,8 @@ def _tally_classification(
             self._tally_resumption_classification(classification)
         elif channel_key == "stdio":
             self._tally_stdio_classification(classification)
+        elif channel_key == "listen":
+            self._tally_listen_classification(classification)
 
     def _tally_post_classification(
         self,
@@ -620,6 +621,9 @@ def _tally_resumption_classification(self, classification: ActivityState) -> Non
     def _tally_stdio_classification(self, classification: ActivityState) -> None:
         self._stdio_counts.increment(classification)
 
+    def _tally_listen_classification(self, classification: ActivityState) -> None:
+        self._listen_counts.increment(classification)
+
     def _register_ping(self, timestamp: datetime) -> None:
         self._get_ping_count += 1
         self._get_last_ping_at = timestamp

tests/unit/fast_agent/mcp/test_transport_tracking.py

@@ -209,6 +209,68 @@ def test_listen_channel_tracks_requests_notifications_and_state() -> None:
     assert snapshot.listen.activity_buckets == ["none", "request"]
 
 
+def test_listen_ping_reply_clears_pending_ping_request() -> None:
+    """A ping reply arriving on `listen` must resolve the request parked by another channel."""
+    metrics = TransportChannelMetrics()
+
+    for index in range(3):
+        metrics.record_event(
+            ChannelEvent(
+                channel="post-json",
+                event_type="message",
+                message=JSONRPCRequest(jsonrpc="2.0", id=index, method="ping"),
+            )
+        )
+        metrics.record_event(
+            ChannelEvent(
+                channel="listen",
+                event_type="message",
+                message=JSONRPCResponse(jsonrpc="2.0", id=index, result={}),
+            )
+        )
+
+    snapshot = metrics.snapshot()
+    assert snapshot.listen is not None
+    # the replies are pings, not ordinary responses
+    assert snapshot.listen.response_count == 0
+    assert snapshot.listen.last_message_summary == "ping"
+    # and nothing is left parked once every ping has been answered
+    assert metrics._ping_request_ids == set()
+
+
+def test_listen_channel_still_tallies_after_ping_routing() -> None:
+    """Routing `listen` through the shared tally must not stop it counting."""
+    metrics = TransportChannelMetrics()
+
+    metrics.record_event(
+        ChannelEvent(
+            channel="listen",
+            event_type="message",
+            message=JSONRPCRequest(jsonrpc="2.0", id="l-1", method="subscriptions/listen"),
+        )
+    )
+    metrics.record_event(
+        ChannelEvent(
+            channel="listen",
+            event_type="message",
+            message=JSONRPCNotification(jsonrpc="2.0", method="notifications/tools/list_changed"),
+        )
+    )
+    metrics.record_event(
+        ChannelEvent(
+            channel="listen",
+            event_type="message",
+            message=JSONRPCResponse(jsonrpc="2.0", id="l-1", result={}),
+        )
+    )
+
+    snapshot = metrics.snapshot()
+    assert snapshot.listen is not None
+    assert snapshot.listen.request_count == 1
+    assert snapshot.listen.notification_count == 1
+    assert snapshot.listen.response_count == 1
+
+
 def test_unsupported_listen_channel_is_hidden() -> None:
     metrics = TransportChannelMetrics()
     metrics.record_event(
@@ -327,3 +389,84 @@ def test_ping_request_variants_are_classified_as_ping(method: str) -> None:
     assert snapshot.stdio.last_message_summary == "ping"
     assert snapshot.stdio.activity_buckets[-1] == "ping"
     assert snapshot.stdio.request_count == 0
+
+
+def _ping_request(channel: ChannelName, request_id: object) -> ChannelEvent:
+    return ChannelEvent(
+        channel=channel,
+        event_type="message",
+        message=JSONRPCRequest(jsonrpc="2.0", id=request_id, method="ping"),
+    )
+
+
+def _response(channel: ChannelName, request_id: object) -> ChannelEvent:
+    return ChannelEvent(
+        channel=channel,
+        event_type="message",
+        message=JSONRPCResponse(jsonrpc="2.0", id=request_id, result={}),
+    )
+
+
+def _build_bucket(scenario: str, reply_channel: ChannelName) -> TransportChannelMetrics:
+    """Compose one activity bucket on ``reply_channel``, per the scenario table."""
+    metrics = TransportChannelMetrics()
+
+    if scenario in {"A", "B", "F"}:
+        # the ping is issued elsewhere; the reply is the cross-channel case
+        metrics.record_event(_ping_request("post-json", "p-1"))
+    if scenario == "B":
+        metrics.record_event(
+            ChannelEvent(
+                channel=reply_channel,
+                event_type="message",
+                message=JSONRPCNotification(
+                    jsonrpc="2.0", method="notifications/tools/list_changed"
+                ),
+            )
+        )
+    if scenario in {"C", "F"}:
+        # a genuine response that must not be masked by the ping reply
+        metrics.record_event(_response(reply_channel, "r-1"))
+    if scenario in {"A", "B", "F"}:
+        metrics.record_event(_response(reply_channel, "p-1"))
+
+    return metrics
+
+
+@pytest.mark.parametrize(
+    ("scenario", "expected_bucket", "expected_response_count"),
+    [
+        ("A", "ping", 0),
+        ("B", "ping", 0),
+        ("C", "response", 1),
+        ("F", "response", 1),
+    ],
+)
+def test_ping_exchange_rendering_matches_scenario_table(
+    scenario: str, expected_bucket: str, expected_response_count: int
+) -> None:
+    """Pin the four bucket compositions from the #926 review.
+
+    A and B are the fix: a bucket whose only content was a miscounted ping now
+    renders as ``ping`` instead of ``response``. C and F are the guard rails --
+    ``_history_priority`` ranks ``RESPONSE`` above ``PING``, so a genuine
+    response sharing the bucket keeps the summary, and only the ping is
+    discounted from ``response_count``.
+    """
+    snapshot = _build_bucket(scenario, "listen").snapshot()
+
+    assert snapshot.listen is not None
+    assert snapshot.listen.activity_buckets[-1] == expected_bucket
+    assert snapshot.listen.response_count == expected_response_count
+
+
+@pytest.mark.parametrize("scenario", ["A", "B", "C", "F"])
+def test_listen_matches_get_on_the_scenario_table(scenario: str) -> None:
+    """``listen`` was the outlier: ``get`` already rendered these buckets this way."""
+    listen = _build_bucket(scenario, "listen").snapshot().listen
+    get = _build_bucket(scenario, "get").snapshot().get
+
+    assert listen is not None
+    assert get is not None
+    assert listen.activity_buckets[-1] == get.activity_buckets[-1]
+    assert listen.response_count == get.response_count