diff --git a/src/fast_agent/mcp/transport_tracking.py b/src/fast_agent/mcp/transport_tracking.py index bce932a8e..b6d4b575b 100644 --- a/src/fast_agent/mcp/transport_tracking.py +++ b/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 diff --git a/tests/unit/fast_agent/mcp/test_transport_tracking.py b/tests/unit/fast_agent/mcp/test_transport_tracking.py index ee9a27ac3..97053363b 100644 --- a/tests/unit/fast_agent/mcp/test_transport_tracking.py +++ b/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