fix: Sending should happen in a background thread in the resubscribe test (#83)

Once we get the first evemt (the Task) the event queue will be open
during the configurable sleepi that happens in the AgentExecutor. This
is our window for when it is possible to resubscribe

# Description

Thank you for opening a Pull Request!
Before submitting your PR, there are a few things you can do to make
sure it goes smoothly:

- [x] Follow the [`CONTRIBUTING` Guide](../CONTRIBUTING.md).
- [x] Make your Pull Request title in the
<https://www.conventionalcommits.org/> specification.
- Important Prefixes for
[release-please](https://github.com/googleapis/release-please):
- `fix:` which represents bug fixes, and correlates to a
[SemVer](https://semver.org/) patch.
- `feat:` represents a new feature, and correlates to a SemVer minor.
- `feat!:`, or `fix!:`, `refactor!:`, etc., which represent a breaking
change (indicated by the `!`) and will result in a SemVer major.
- [x] Ensure the tests pass
- [x] Appropriate READMEs were updated (if necessary)

Fixes #<issue_number_goes_here> 🦕

---------

Co-authored-by: Stefano Maestri <stefano.maestri@javalinux.it>

Author: kabir

tests/optional/capabilities/test_streaming_methods.py

@@ -1,21 +1,16 @@
 import asyncio
-import json
 import logging
 import os
-from typing import Any, AsyncGenerator, Dict, List, Optional
 import uuid
 
 import pytest
-import pytest_asyncio
 
 from tck import agent_card_utils, config, message_utils
 from tests.markers import optional_capability
 from tests.capability_validator import CapabilityValidator, skip_if_capability_not_declared
 from tests.utils.transport_helpers import (
     transport_send_streaming_message,
     transport_resubscribe_task,
-    transport_send_message,
-    extract_task_id_from_response,
     generate_test_message_id,
 )
 
@@ -37,7 +32,6 @@
     "async_wait_for": BASE_TIMEOUT * 1.0,  # 2.0s default (for asyncio.wait_for)
 }
 
-
 # SimpleSSEClient class removed - now using transport-agnostic streaming
 
 
@@ -230,47 +224,129 @@ async def test_tasks_resubscribe(sut_client, agent_card_data):
     if not validator.is_capability_declared("streaming"):
         pytest.skip("Streaming capability not declared - test not applicable")
 
-    # First, create a task via message/stream to get a task ID
-    message_params = {
-        "message": {
-            "kind": "message",
-            "messageId": generate_test_message_id("resubscribe"),
-            "role": "user",
-            "parts": [{"kind": "text", "text": "Test message for resubscribe"}],
-        }
-    }
-    task_id = None  # Initialize task_id
+    # Synchronization events
+    task_id_received = asyncio.Event()
 
+    task_id = None
+    resubscribe_events = []
+    stream_error = None
+    resubscribe_error = None
+    
     try:
+        # First, create a task via message/stream to get a task ID
+        message_params = {
+            "message": {
+                "kind": "message",
+                "messageId": "test-resubscribe-message-id-" + str(uuid.uuid4()),
+                "role": "user",
+                "parts": [{"kind": "text", "text": "Test message for resubscribe"}],
+            }
+        }
+
         # Use transport-agnostic streaming message sending
         stream = transport_send_streaming_message(sut_client, message_params)
-        
-        # Collect events to get task ID
-        try:
-            event_count = 0
-            async for event in stream:
-                event_count += 1
-                logger.info(f"Processing streaming event #{event_count}: {event}")
 
-                # Look for task ID in the event
-                if isinstance(event, dict):
-                    if "id" in event and "status" in event:
-                        # This looks like a Task object
-                        task_id = event["id"]
-                        logger.info(f"Captured task ID from stream: {task_id}")
+        # Background task to process the initial stream
+        async def process_initial_stream():
+            nonlocal task_id, stream_error
+            try:
+                event_count = 0
+                async for event in stream:
+                    event_count += 1
+                    logger.info(f"Processing streaming event #{event_count}: {event}")
+
+                    # Look for task ID in the event
+                    if isinstance(event, dict):
+                        if "id" in event and "status" in event:
+                            # This looks like a Task object
+                            task_id = event["id"]
+                            logger.info(f"Captured task ID from stream: {task_id}")
+                            # Signal that task ID is available. This means that the event queue is open
+                            task_id_received.set()
+
+                    # Safety break to prevent infinite loops
+                    if event_count >= 10:
+                        logger.warning("Hit event count limit while getting task ID")
                         break
 
-                # Safety break to prevent infinite loops
-                if event_count >= 10:
-                    logger.warning("Hit event count limit while getting task ID")
-                    break
+                    # Continue processing events even after getting task ID
+                    # to keep the stream alive for resubscribe testing
 
-                # Break after getting first event with task ID
-                if task_id:
-                    break
-                    
+            except Exception as e:
+                stream_error = e
+                logger.error(f"Error in initial stream processing: {e}")
+                task_id_received.set()  # Signal completion even on error
+
+        # Background task to handle resubscription once task ID is available
+        async def process_resubscribe():
+            nonlocal resubscribe_events, resubscribe_error
+            try:
+                # Wait for task ID to be available
+                await task_id_received.wait()
+                
+                if task_id is None:
+                    logger.warning("No task ID available for resubscribe")
+                    return
+
+                logger.info(f"Starting resubscribe for task ID: {task_id}")
+                
+                # Use transport-agnostic task resubscription
+                resubscribe_stream = transport_resubscribe_task(sut_client, task_id)
+
+                event_count = 0
+                async for event in resubscribe_stream:
+                    event_count += 1
+                    logger.info(f"Processing resubscribe event #{event_count}: {event}")
+                    resubscribe_events.append(event)
+
+                    # Safety break to prevent infinite loops
+                    if event_count >= 10:
+                        logger.warning("Hit event count limit in resubscribe test, breaking")
+                        break
+
+                    # Validate this is a proper A2A object
+                    assert isinstance(event, dict), "Resubscribe events must be objects"
+
+                    # Check if this is a terminal event
+                    if "status" in event and isinstance(event["status"], dict):
+                        state = event["status"].get("state")
+                        if state in ["completed", "failed", "canceled"]:
+                            logger.info("Detected terminal event in resubscribe, ending stream processing.")
+                            break
+
+                    # Collect a few events then break
+                    if len(resubscribe_events) >= 3:
+                        break
+                        
+            except Exception as e:
+                resubscribe_error = e
+                logger.error(f"Error in resubscribe processing: {e}")
+
+        # Start both background tasks
+        initial_stream_task = asyncio.create_task(process_initial_stream())
+        resubscribe_task = asyncio.create_task(process_resubscribe())
+
+        # Wait for both tasks to complete with timeout
+        try:
+            await asyncio.wait_for(
+                asyncio.gather(initial_stream_task, resubscribe_task, return_exceptions=True),
+                timeout=TIMEOUTS["async_wait_for"] * 2)
         except asyncio.TimeoutError:
-            logger.warning("Timeout while getting task ID from initial stream")
+            logger.warning("Timeout while waiting for stream processing and resubscribe")
+            # Cancel tasks if they're still running
+            initial_stream_task.cancel()
+            resubscribe_task.cancel()
+            
+        # Check for errors from the background tasks
+        if stream_error:
+            error_msg = str(stream_error).lower()
+            if "501" in error_msg or "not implemented" in error_msg:
+                pytest.fail(
+                    "Streaming capability declared in Agent Card but SUT returned error indicating not implemented. "
+                    "This is a specification violation - declared capabilities MUST be implemented."
+                )
+            else:
+                raise stream_error
 
     except Exception as e:
         error_msg = str(e).lower()
@@ -282,49 +358,13 @@ async def test_tasks_resubscribe(sut_client, agent_card_data):
         else:
             raise
 
-    # Now try to resubscribe to this task
+    # Now validate the resubscribe results
     if task_id is None:
         pytest.skip("Could not capture task ID from initial stream - cannot test resubscribe")
 
-    try:
-        # Use transport-agnostic task resubscription
-        resubscribe_stream = transport_resubscribe_task(sut_client, task_id)
-        events = []
-
-        try:
-            event_count = 0
-            async for event in resubscribe_stream:
-                event_count += 1
-                logger.info(f"Processing resubscribe event #{event_count}: {event}")
-                events.append(event)
-
-                # Safety break to prevent infinite loops
-                if event_count >= 10:
-                    logger.warning("Hit event count limit in resubscribe test, breaking")
-                    break
-
-                # Validate this is a proper A2A object
-                assert isinstance(event, dict), "Resubscribe events must be objects"
-
-                # Check if this is a terminal event
-                if "status" in event and isinstance(event["status"], dict):
-                    state = event["status"].get("state")
-                    if state in ["completed", "failed", "canceled"]:
-                        logger.info("Detected terminal event in resubscribe, ending stream processing.")
-                        break
-
-                # Collect a few events then break
-                if len(events) >= 3:
-                    break
-                    
-        except asyncio.TimeoutError:
-            logger.warning("Timeout while processing resubscribe stream")
-
-        # Validate that we got at least some events
-        assert len(events) > 0, "Streaming capability declared but no events received from resubscribe stream"
-
-    except Exception as e:
-        error_msg = str(e).lower()
+    # Check for resubscribe errors
+    if resubscribe_error:
+        error_msg = str(resubscribe_error).lower()
         if "501" in error_msg or "not implemented" in error_msg:
             pytest.fail(
                 "Streaming capability declared but tasks/resubscribe returned error indicating not implemented. "
@@ -333,7 +373,10 @@ async def test_tasks_resubscribe(sut_client, agent_card_data):
         elif "not found" in error_msg or "404" in error_msg:
             pytest.skip("Task expired before resubscribe test - this is implementation-dependent behavior")
         else:
-            raise
+            raise resubscribe_error
+
+    # Validate that we got at least some events from resubscribe
+    assert len(resubscribe_events) > 0, "Streaming capability declared but no events received from resubscribe stream"
 
 
 @optional_capability