docs: add streaming limitations documentation and future streaming protocol

- Document current Agent.achat() streaming limitations in README
- Add StreamingRuntimeProtocol for future incremental streaming support
- Create example runtime implementations (Mock, Echo, DelayedStreaming)
- Provide clear migration path for when Agent supports AsyncIterator

Co-authored-by: Mervin Praison <[email protected]>

src/praisonai-agents/praisonaiagents/runtime/README.md

@@ -88,6 +88,16 @@ The `praisonai` runtime is the default built-in runtime that wraps the existing
 - Provides backward compatibility
 - Handles streaming and non-streaming execution
 
+##### Streaming Limitations
+
+**Important Note**: The current `Agent.achat()` implementation does not support true async streaming. When called with `stream=True`, it returns a complete string response rather than an `AsyncIterator`. As a result:
+
+- The `stream_turn()` method returns the full response as a single delta
+- True incremental streaming is not yet available
+- This is a known limitation of the underlying Agent implementation
+
+This limitation will be addressed in future versions when the Agent class is updated to support true async streaming with `AsyncIterator` returns.
+
 #### Protocol Compliance
 
 All runtimes must implement `AgentRuntimeProtocol`:
@@ -138,6 +148,38 @@ if is_runtime_available("my-runtime"):
     runtime = resolve_runtime("my-runtime")
 ```
 
+#### Enhanced Streaming Protocol (Future)
+
+For runtimes that support true incremental streaming, an optional extended protocol is available:
+
+```python
+from praisonaiagents.runtime import StreamingRuntimeProtocol
+
+class MyStreamingRuntime:
+    def supports_incremental_streaming(self) -> bool:
+        """Indicate true streaming support."""
+        return True
+    
+    async def stream_turn_incremental(self, prompt: str, **kwargs):
+        """Yield true incremental deltas."""
+        for chunk in await self.get_streaming_chunks(prompt):
+            yield RuntimeDelta(type="text", content=chunk)
+```
+
+Consumers can check for enhanced streaming support:
+
+```python
+if isinstance(runtime, StreamingRuntimeProtocol):
+    if runtime.supports_incremental_streaming():
+        # Use enhanced streaming
+        async for delta in runtime.stream_turn_incremental(prompt):
+            print(delta.content, end="", flush=True)
+    else:
+        # Fall back to single-delta streaming
+        async for delta in runtime.stream_turn(prompt):
+            print(delta.content)
+```
+
 #### Error Handling
 
 Runtimes should handle errors gracefully and return `RuntimeResult` objects with error information rather than raising exceptions:
@@ -330,6 +372,28 @@ def normalize(self, result, tool_name, ctx):
 - **Error Isolation**: Middleware failures don't break tool execution - they log warnings and continue
 - **Thread Safety**: The registry is thread-safe for concurrent agent execution
 
+## Example Implementations
+
+The module includes example runtime implementations in `examples.py`:
+
+- **MockRuntime**: Returns predefined responses for testing
+- **EchoRuntime**: Echoes back user input for debugging
+- **DelayedStreamingRuntime**: Demonstrates incremental streaming capabilities
+
+```python
+from praisonaiagents.runtime.examples import MockRuntime, EchoRuntime
+
+# Use mock runtime for testing
+mock = MockRuntime(response="Test response")
+result = await mock.run_turn("Any prompt")
+assert result.content == "Test response"
+
+# Use echo runtime for debugging
+echo = EchoRuntime()
+result = await echo.run_turn("Hello")
+assert result.content == "Echo: Hello"
+```
+
 ## Testing
 
 The runtime system includes comprehensive tests:
@@ -343,7 +407,7 @@ The runtime system includes comprehensive tests:
 Run tests with:
 
 ```bash
-pytest praisonaiagents/runtime/tests/
+pytest tests/unit/runtime/
 ```
 
 Use the provided test utilities to verify your middleware:

src/praisonai-agents/praisonaiagents/runtime/__init__.py

@@ -41,6 +41,7 @@ __all__ = [
     "get_middleware",
     # Agent runtime APIs
     'AgentRuntimeProtocol',
+    'StreamingRuntimeProtocol',
     'RuntimeConfig', 
     'RuntimeResult',
     'RuntimeDelta',
@@ -69,6 +70,7 @@ _LAZY_GROUPS = {
         'TurnRuntimeProtocol': ('praisonaiagents.runtime.protocols', 'TurnRuntimeProtocol'),
         'TurnContextBuilderProtocol': ('praisonaiagents.runtime.protocols', 'TurnContextBuilderProtocol'),
         'AgentRuntimeProtocol': ('praisonaiagents.runtime.protocols', 'AgentRuntimeProtocol'),
+        'StreamingRuntimeProtocol': ('praisonaiagents.runtime.protocols', 'StreamingRuntimeProtocol'),
         'RuntimeConfig': ('praisonaiagents.runtime.protocols', 'RuntimeConfig'),
         'RuntimeResult': ('praisonaiagents.runtime.protocols', 'RuntimeResult'),
         'RuntimeDelta': ('praisonaiagents.runtime.protocols', 'RuntimeDelta'),

src/praisonai-agents/praisonaiagents/runtime/examples.py

@@ -0,0 +1,214 @@
+"""Example runtime implementations demonstrating extensibility.
+
+These are reference implementations showing how to create custom runtimes.
+They are not intended for production use.
+"""
+
+from typing import AsyncIterator, Optional
+import asyncio
+
+from .protocols import (
+    AgentRuntimeProtocol,
+    StreamingRuntimeProtocol,
+    RuntimeResult,
+    RuntimeDelta
+)
+
+
+class MockRuntime:
+    """Mock runtime for testing and demonstration.
+    
+    A simple runtime that returns predefined responses.
+    Useful for testing runtime registry and protocol compliance.
+    """
+    
+    def __init__(self, response: str = "Mock response"):
+        """Initialize mock runtime with predefined response."""
+        self.runtime_id = "mock"
+        self.response = response
+        
+    def supports(self, model_ref: Optional[str] = None) -> bool:
+        """Mock runtime supports all models."""
+        return True
+        
+    async def run_turn(
+        self,
+        prompt: str,
+        *,
+        system_prompt: Optional[str] = None,
+        model_ref: Optional[str] = None,
+        **kwargs
+    ) -> RuntimeResult:
+        """Return predefined response."""
+        return RuntimeResult(
+            content=self.response,
+            metadata={
+                'runtime': self.runtime_id,
+                'prompt': prompt,
+                'model': model_ref
+            }
+        )
+        
+    async def stream_turn(
+        self,
+        prompt: str,
+        **kwargs
+    ) -> AsyncIterator[RuntimeDelta]:
+        """Stream mock response as single delta."""
+        yield RuntimeDelta(
+            type="text",
+            content=self.response,
+            metadata={'runtime': self.runtime_id}
+        )
+
+
+class EchoRuntime:
+    """Echo runtime that returns the input prompt.
+    
+    Demonstrates a minimal runtime implementation that
+    simply echoes back the user's input. Useful for debugging
+    and understanding the runtime protocol.
+    """
+    
+    def __init__(self):
+        """Initialize echo runtime."""
+        self.runtime_id = "echo"
+        
+    def supports(self, model_ref: Optional[str] = None) -> bool:
+        """Echo runtime supports all models."""
+        return True
+        
+    async def run_turn(
+        self,
+        prompt: str,
+        *,
+        system_prompt: Optional[str] = None,
+        model_ref: Optional[str] = None,
+        **kwargs
+    ) -> RuntimeResult:
+        """Echo back the prompt."""
+        response = f"Echo: {prompt}"
+        if system_prompt:
+            response = f"{system_prompt}\n{response}"
+            
+        return RuntimeResult(
+            content=response,
+            metadata={
+                'runtime': self.runtime_id,
+                'echoed_length': len(prompt)
+            }
+        )
+        
+    async def stream_turn(
+        self,
+        prompt: str,
+        **kwargs
+    ) -> AsyncIterator[RuntimeDelta]:
+        """Stream echo response."""
+        system_prompt = kwargs.get('system_prompt')
+        
+        if system_prompt:
+            yield RuntimeDelta(
+                type="text",
+                content=f"{system_prompt}\n",
+                metadata={'runtime': self.runtime_id}
+            )
+            
+        yield RuntimeDelta(
+            type="text",
+            content=f"Echo: {prompt}",
+            metadata={'runtime': self.runtime_id}
+        )
+
+
+class DelayedStreamingRuntime:
+    """Example runtime with simulated incremental streaming.
+    
+    Demonstrates how a runtime would implement true incremental
+    streaming when the underlying implementation supports it.
+    This example simulates streaming by breaking the response
+    into chunks with delays.
+    """
+    
+    def __init__(self, chunk_delay: float = 0.1):
+        """Initialize with configurable chunk delay."""
+        self.runtime_id = "delayed-streaming"
+        self.chunk_delay = chunk_delay
+        
+    def supports(self, model_ref: Optional[str] = None) -> bool:
+        """Support all models for demonstration."""
+        return True
+        
+    def supports_incremental_streaming(self) -> bool:
+        """This runtime supports true incremental streaming."""
+        return True
+        
+    async def run_turn(
+        self,
+        prompt: str,
+        *,
+        system_prompt: Optional[str] = None,
+        model_ref: Optional[str] = None,
+        **kwargs
+    ) -> RuntimeResult:
+        """Execute and return full response."""
+        response = f"Processing: {prompt[:50]}..."
+        
+        return RuntimeResult(
+            content=response,
+            metadata={
+                'runtime': self.runtime_id,
+                'streaming_supported': True
+            }
+        )
+        
+    async def stream_turn(
+        self,
+        prompt: str,
+        **kwargs
+    ) -> AsyncIterator[RuntimeDelta]:
+        """Stream with simulated delays."""
+        response = f"Processing: {prompt[:50]}..."
+        
+        # Simulate incremental streaming
+        words = response.split()
+        for i, word in enumerate(words):
+            await asyncio.sleep(self.chunk_delay)
+            
+            # Add space except for first word
+            content = word if i == 0 else f" {word}"
+            
+            yield RuntimeDelta(
+                type="text",
+                content=content,
+                metadata={
+                    'runtime': self.runtime_id,
+                    'chunk_index': i
+                }
+            )
+            
+    async def stream_turn_incremental(
+        self,
+        prompt: str,
+        **kwargs
+    ) -> AsyncIterator[RuntimeDelta]:
+        """Enhanced streaming with finer granularity.
+        
+        This would be called when the consumer detects
+        StreamingRuntimeProtocol support.
+        """
+        response = f"Processing: {prompt[:50]}..."
+        
+        # Simulate character-by-character streaming
+        for i, char in enumerate(response):
+            await asyncio.sleep(self.chunk_delay / 10)  # Faster for chars
+            
+            yield RuntimeDelta(
+                type="text",
+                content=char,
+                metadata={
+                    'runtime': self.runtime_id,
+                    'char_index': i,
+                    'incremental': True
+                }
+            )

src/praisonai-agents/praisonaiagents/runtime/protocols.py

@@ -210,4 +210,53 @@ class AgentRuntimeProtocol(Protocol):
         Yields:
             RuntimeDelta objects with incremental response content
         """
+        ...
+
+
+@runtime_checkable
+class StreamingRuntimeProtocol(Protocol):
+    """Extended protocol for runtimes with enhanced streaming capabilities.
+    
+    Optional protocol that runtimes can implement to indicate support for
+    true incremental streaming. This will be the migration path when the
+    underlying Agent implementation adds AsyncIterator support.
+    
+    Example:
+        ```python
+        # Check if runtime supports enhanced streaming
+        if isinstance(runtime, StreamingRuntimeProtocol):
+            if runtime.supports_incremental_streaming():
+                # Use enhanced streaming
+                async for delta in runtime.stream_turn_incremental(prompt):
+                    process_delta(delta)
+        ```
+    """
+    
+    def supports_incremental_streaming(self) -> bool:
+        """Check if this runtime supports true incremental streaming.
+        
+        Returns:
+            True if the runtime can stream incremental deltas,
+            False if it only supports single-delta responses
+        """
+        ...
+    
+    async def stream_turn_incremental(
+        self,
+        prompt: str,
+        **kwargs
+    ) -> AsyncIterator[RuntimeDelta]:
+        """Stream true incremental response deltas.
+        
+        This method will be called when true streaming is available.
+        Unlike the base stream_turn which may return a single delta,
+        this guarantees incremental streaming of response chunks.
+        
+        Args:
+            prompt: User prompt/query
+            **kwargs: Additional options (system_prompt, model_ref, etc.)
+            
+        Yields:
+            RuntimeDelta objects with incremental response chunks
+        """
         ...