fix(QA-audit): asyncio lifecycle fix in inbox wrapper tests + push-mode queue support

Issue #307 (regression):
- test_a2a_tools_inbox_wrappers.py: replace _run() helper that called
  asyncio.get_event_loop().run_until_complete() with proper
  @pytest.mark.asyncio + async def / await. The legacy pattern bypasses
  pytest-asyncio's event-loop lifecycle, causing 14 tests to pass in
  isolation but exit-1 in the full suite.

Issue #308 (test gap):
- a2a_response.py: parse() only handled delivery_mode="poll" queue
  responses. Extend the queued-envelope check to match any non-null
  delivery_mode (poll, push, or future modes), and propagate the actual
  mode string into Queued(method=, delivery_mode=) so callers can
  distinguish poll vs push delivery.
- test_a2a_response.py: add 3 push-mode fixtures
  (push_queued_full/notify/no_method) and 4 test cases covering
  classification, delivery_mode field, method sentinel, and INFO log
  output.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

workspace/a2a_response.py

@@ -153,6 +153,7 @@ _KEY_RETRY_AFTER = "retry_after"
 
 _STATUS_QUEUED = "queued"
 _DELIVERY_MODE_POLL = "poll"
+_DELIVERY_MODE_PUSH = "push"
 
 
 def parse(data: Any) -> Variant:
@@ -165,8 +166,8 @@ def parse(data: Any) -> Variant:
     The order of checks matters:
 
       1. Non-dict input → Malformed (server contract is dict-shaped).
-      2. Poll-queued envelope is checked BEFORE result/error because a
-         server bug that sets both ``status=queued`` and ``result``
+      2. Poll-queued or push-queued envelope is checked BEFORE result/error
+         because a server bug that sets both ``status=queued`` and ``result``
          should be loud, not silently treated as Result.
       3. ``result`` → Result (the JSON-RPC success path).
       4. ``error`` → Error (JSON-RPC error or platform error).
@@ -179,20 +180,23 @@ def parse(data: Any) -> Variant:
         )
         return Malformed(raw=data)
 
-    # Poll-queued envelope. Both keys must be present — the workspace
-    # server sets them together; if only one is present the body is
-    # ambiguous and we route to Malformed for visibility.
+    # Poll-queued or push-queued envelope. Both status and delivery_mode
+    # must be present — the workspace server sets them together; if only
+    # one is present the body is ambiguous and we route to Malformed for
+    # visibility.
     if (
         data.get(_KEY_STATUS) == _STATUS_QUEUED
-        and data.get(_KEY_DELIVERY_MODE) == _DELIVERY_MODE_POLL
+        and data.get(_KEY_DELIVERY_MODE) is not None
     ):
         method_raw = data.get(_KEY_METHOD)
         method = str(method_raw) if method_raw is not None else "unknown"
+        delivery_mode = str(data.get(_KEY_DELIVERY_MODE))
         logger.info(
-            "a2a_response.parse: queued for poll-mode peer (method=%s)",
+            "a2a_response.parse: queued for %s-mode peer (method=%s)",
+            delivery_mode,
             method,
         )
-        return Queued(method=method)
+        return Queued(method=method, delivery_mode=delivery_mode)
 
     # JSON-RPC success.
     if _KEY_RESULT in data:

workspace/tests/test_a2a_response.py

@@ -105,6 +105,20 @@ _FIXTURES = {
         "status": "queued",
         "delivery_mode": "poll",
     },
+    "push_queued_full": {
+        "status": "queued",
+        "delivery_mode": "push",
+        "method": "message/send",
+    },
+    "push_queued_notify": {
+        "status": "queued",
+        "delivery_mode": "push",
+        "method": "notify",
+    },
+    "push_queued_no_method": {
+        "status": "queued",
+        "delivery_mode": "push",
+    },
     "malformed_empty_dict": {},
     "malformed_unexpected_keys": {"foo": "bar", "baz": 42},
     "malformed_status_queued_no_delivery_mode": {
@@ -159,6 +173,30 @@ class TestQueuedVariant:
             a2a_response.parse(_FIXTURES["poll_queued_full"])
         assert any("queued for poll-mode peer" in r.message for r in caplog.records)
 
+    # Push-mode tests
+    def test_push_queued_full_envelope(self):
+        v = a2a_response.parse(_FIXTURES["push_queued_full"])
+        assert isinstance(v, a2a_response.Queued)
+        assert v.method == "message/send"
+        assert v.delivery_mode == "push"
+
+    def test_push_queued_notify(self):
+        v = a2a_response.parse(_FIXTURES["push_queued_notify"])
+        assert isinstance(v, a2a_response.Queued)
+        assert v.method == "notify"
+        assert v.delivery_mode == "push"
+
+    def test_push_queued_missing_method_uses_unknown(self):
+        v = a2a_response.parse(_FIXTURES["push_queued_no_method"])
+        assert isinstance(v, a2a_response.Queued)
+        assert v.method == "unknown"
+        assert v.delivery_mode == "push"
+
+    def test_logs_info_on_push_queued(self, caplog):
+        with caplog.at_level(logging.INFO, logger="a2a_response"):
+            a2a_response.parse(_FIXTURES["push_queued_full"])
+        assert any("queued for push-mode peer" in r.message for r in caplog.records)
+
 
 class TestResultVariant:
     """``parse()`` extracts the JSON-RPC ``result`` envelope into
@@ -361,7 +399,7 @@ _ADVERSARIAL_INPUTS: list[Any] = [
     {"error": {"message": None, "code": None}},
     {"error": {"message": ["nested", "list"]}},
     {"status": None, "delivery_mode": None, "method": None},
-    {"status": "queued", "delivery_mode": "push", "method": "x"},  # wrong delivery_mode
+    {"status": "queued", "delivery_mode": "other-mode", "method": "x"},  # unknown-but-present delivery_mode → Queued (matches on delivery_mode is not None)
     {"status": "running", "delivery_mode": "poll"},  # wrong status
     {"status": 42, "delivery_mode": "poll"},  # non-string status
     # Deeply-nested junk
@@ -436,6 +474,9 @@ class TestRegressionGate:
             "poll_queued_full":                  a2a_response.Queued,
             "poll_queued_notify":                a2a_response.Queued,
             "poll_queued_no_method":             a2a_response.Queued,
+            "push_queued_full":                  a2a_response.Queued,
+            "push_queued_notify":                a2a_response.Queued,
+            "push_queued_no_method":             a2a_response.Queued,
             "malformed_empty_dict":              a2a_response.Malformed,
             "malformed_unexpected_keys":         a2a_response.Malformed,
             "malformed_status_queued_no_delivery_mode": a2a_response.Malformed,

workspace/tests/test_a2a_tools_inbox_wrappers.py

@@ -15,11 +15,11 @@ The wrappers are ~40 LOC of glue. The full delivery behavior
 """
 from __future__ import annotations
 
-import asyncio
 import json
 from unittest.mock import MagicMock, patch
 
 import pytest
+import pytest_asyncio
 
 
 @pytest.fixture(autouse=True)
@@ -29,24 +29,22 @@ def _require_workspace_id(monkeypatch):
     yield
 
 
-def _run(coro):
-    return asyncio.get_event_loop().run_until_complete(coro)
-
-
 # ---------------------------------------------------------------------------
 # tool_inbox_peek
 # ---------------------------------------------------------------------------
 
 
 class TestToolInboxPeek:
-    def test_returns_not_enabled_when_state_none(self):
+    @pytest.mark.asyncio
+    async def test_returns_not_enabled_when_state_none(self):
         import a2a_tools
 
         with patch("inbox.get_state", return_value=None):
-            out = _run(a2a_tools.tool_inbox_peek())
+            out = await a2a_tools.tool_inbox_peek()
         assert "not enabled" in out
 
-    def test_returns_json_array_of_messages(self):
+    @pytest.mark.asyncio
+    async def test_returns_json_array_of_messages(self):
         import a2a_tools
 
         msg1 = MagicMock()
@@ -58,20 +56,21 @@ class TestToolInboxPeek:
         fake_state.peek.return_value = [msg1, msg2]
 
         with patch("inbox.get_state", return_value=fake_state):
-            out = _run(a2a_tools.tool_inbox_peek(limit=5))
+            out = await a2a_tools.tool_inbox_peek(limit=5)
         # peek limit is forwarded
         fake_state.peek.assert_called_once_with(limit=5)
         parsed = json.loads(out)
         assert len(parsed) == 2
         assert parsed[0]["activity_id"] == "a1"
 
-    def test_non_int_limit_falls_back_to_10(self):
+    @pytest.mark.asyncio
+    async def test_non_int_limit_falls_back_to_10(self):
         import a2a_tools
 
         fake_state = MagicMock()
         fake_state.peek.return_value = []
         with patch("inbox.get_state", return_value=fake_state):
-            _run(a2a_tools.tool_inbox_peek(limit="garbage"))  # type: ignore[arg-type]
+            await a2a_tools.tool_inbox_peek(limit="garbage")  # type: ignore[arg-type]
         fake_state.peek.assert_called_once_with(limit=10)
 
 
@@ -81,49 +80,54 @@ class TestToolInboxPeek:
 
 
 class TestToolInboxPop:
-    def test_returns_not_enabled_when_state_none(self):
+    @pytest.mark.asyncio
+    async def test_returns_not_enabled_when_state_none(self):
         import a2a_tools
 
         with patch("inbox.get_state", return_value=None):
-            out = _run(a2a_tools.tool_inbox_pop("act-1"))
+            out = await a2a_tools.tool_inbox_pop("act-1")
         assert "not enabled" in out
 
-    def test_rejects_empty_activity_id(self):
+    @pytest.mark.asyncio
+    async def test_rejects_empty_activity_id(self):
         import a2a_tools
 
         fake_state = MagicMock()
         with patch("inbox.get_state", return_value=fake_state):
-            out = _run(a2a_tools.tool_inbox_pop(""))
+            out = await a2a_tools.tool_inbox_pop("")
         assert "activity_id is required" in out
         fake_state.pop.assert_not_called()
 
-    def test_rejects_non_str_activity_id(self):
+    @pytest.mark.asyncio
+    async def test_rejects_non_str_activity_id(self):
         import a2a_tools
 
         fake_state = MagicMock()
         with patch("inbox.get_state", return_value=fake_state):
-            out = _run(a2a_tools.tool_inbox_pop(123))  # type: ignore[arg-type]
+            out = await a2a_tools.tool_inbox_pop(123)  # type: ignore[arg-type]
         assert "activity_id is required" in out
         fake_state.pop.assert_not_called()
 
-    def test_returns_removed_true_when_popped(self):
+    @pytest.mark.asyncio
+    async def test_returns_removed_true_when_popped(self):
         import a2a_tools
 
         fake_state = MagicMock()
         fake_state.pop.return_value = MagicMock()  # truthy = something was removed
         with patch("inbox.get_state", return_value=fake_state):
-            out = _run(a2a_tools.tool_inbox_pop("act-7"))
+            out = await a2a_tools.tool_inbox_pop("act-7")
         parsed = json.loads(out)
         assert parsed == {"removed": True, "activity_id": "act-7"}
         fake_state.pop.assert_called_once_with("act-7")
 
-    def test_returns_removed_false_when_unknown(self):
+    @pytest.mark.asyncio
+    async def test_returns_removed_false_when_unknown(self):
         import a2a_tools
 
         fake_state = MagicMock()
         fake_state.pop.return_value = None
         with patch("inbox.get_state", return_value=fake_state):
-            out = _run(a2a_tools.tool_inbox_pop("act-missing"))
+            out = await a2a_tools.tool_inbox_pop("act-missing")
         parsed = json.loads(out)
         assert parsed == {"removed": False, "activity_id": "act-missing"}
 
@@ -134,25 +138,28 @@ class TestToolInboxPop:
 
 
 class TestToolWaitForMessage:
-    def test_returns_not_enabled_when_state_none(self):
+    @pytest.mark.asyncio
+    async def test_returns_not_enabled_when_state_none(self):
         import a2a_tools
 
         with patch("inbox.get_state", return_value=None):
-            out = _run(a2a_tools.tool_wait_for_message(timeout_secs=1.0))
+            out = await a2a_tools.tool_wait_for_message(timeout_secs=1.0)
         assert "not enabled" in out
 
-    def test_timeout_payload_when_no_message(self):
+    @pytest.mark.asyncio
+    async def test_timeout_payload_when_no_message(self):
         import a2a_tools
 
         fake_state = MagicMock()
         fake_state.wait.return_value = None
         with patch("inbox.get_state", return_value=fake_state):
-            out = _run(a2a_tools.tool_wait_for_message(timeout_secs=0.1))
+            out = await a2a_tools.tool_wait_for_message(timeout_secs=0.1)
         parsed = json.loads(out)
         assert parsed["timeout"] is True
         assert parsed["timeout_secs"] == 0.1
 
-    def test_returns_message_when_delivered(self):
+    @pytest.mark.asyncio
+    async def test_returns_message_when_delivered(self):
         import a2a_tools
 
         msg = MagicMock()
@@ -160,37 +167,40 @@ class TestToolWaitForMessage:
         fake_state = MagicMock()
         fake_state.wait.return_value = msg
         with patch("inbox.get_state", return_value=fake_state):
-            out = _run(a2a_tools.tool_wait_for_message(timeout_secs=2.0))
+            out = await a2a_tools.tool_wait_for_message(timeout_secs=2.0)
         parsed = json.loads(out)
         assert parsed["activity_id"] == "a-9"
 
-    def test_timeout_clamped_to_300(self):
+    @pytest.mark.asyncio
+    async def test_timeout_clamped_to_300(self):
         import a2a_tools
 
         fake_state = MagicMock()
         fake_state.wait.return_value = None
         with patch("inbox.get_state", return_value=fake_state):
-            _run(a2a_tools.tool_wait_for_message(timeout_secs=99999))
+            await a2a_tools.tool_wait_for_message(timeout_secs=99999)
         # Whatever wait was called with, it must not exceed 300
         passed = fake_state.wait.call_args.args[0]
         assert passed == 300.0
 
-    def test_timeout_clamped_to_zero_floor(self):
+    @pytest.mark.asyncio
+    async def test_timeout_clamped_to_zero_floor(self):
         import a2a_tools
 
         fake_state = MagicMock()
         fake_state.wait.return_value = None
         with patch("inbox.get_state", return_value=fake_state):
-            _run(a2a_tools.tool_wait_for_message(timeout_secs=-5))
+            await a2a_tools.tool_wait_for_message(timeout_secs=-5)
         passed = fake_state.wait.call_args.args[0]
         assert passed == 0.0
 
-    def test_non_numeric_timeout_falls_back_to_60(self):
+    @pytest.mark.asyncio
+    async def test_non_numeric_timeout_falls_back_to_60(self):
         import a2a_tools
 
         fake_state = MagicMock()
         fake_state.wait.return_value = None
         with patch("inbox.get_state", return_value=fake_state):
-            _run(a2a_tools.tool_wait_for_message(timeout_secs="garbage"))  # type: ignore[arg-type]
+            await a2a_tools.tool_wait_for_message(timeout_secs="garbage")  # type: ignore[arg-type]
         passed = fake_state.wait.call_args.args[0]
         assert passed == 60.0