feat(agent, task): add concurrency limiter and message sink protocols

Implemented a concurrency registry to control concurrent task execution
for agents, allowing users to set limits per agent name. Also added message
sink protocols for managing sequenced task execution messages, enabling
replay of agent outputs and tool calls.

New files include:
- agent/concurrency.py
- task/message_sink.py

Updated the task module to include the new protocols in exports.

Total changes across 13 files with 1099 insertions.

Author: MervinPraison

src/praisonai-agents/praisonaiagents/agent/concurrency.py

@@ -0,0 +1,147 @@
+"""Per-Agent Concurrency Limiter for PraisonAI Agents.
+
+Provides a registry-based approach to limit concurrent task execution
+per agent name. No Agent constructor param bloat — uses a global registry.
+
+Usage:
+    from praisonaiagents.agent.concurrency import get_concurrency_registry
+    
+    registry = get_concurrency_registry()
+    registry.set_limit("researcher", 2)  # max 2 concurrent tasks
+    
+    # In async code:
+    async with registry.throttle("researcher"):
+        await do_work()
+    
+    # Or manual:
+    await registry.acquire("researcher")
+    try:
+        await do_work()
+    finally:
+        registry.release("researcher")
+"""
+
+import asyncio
+import threading
+from contextlib import asynccontextmanager
+from typing import Dict, Optional
+
+from praisonaiagents._logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class ConcurrencyRegistry:
+    """Registry for per-agent concurrency limits.
+    
+    Thread-safe. Each agent name maps to an asyncio.Semaphore.
+    Limit of 0 means unlimited (no throttling).
+    """
+
+    def __init__(self, default_limit: int = 0):
+        self._default_limit = default_limit
+        self._limits: Dict[str, int] = {}
+        self._semaphores: Dict[str, asyncio.Semaphore] = {}
+        self._lock = threading.Lock()
+
+    def set_limit(self, agent_name: str, max_concurrent: int) -> None:
+        """Set concurrency limit for an agent.
+        
+        Args:
+            agent_name: Agent identifier
+            max_concurrent: Max concurrent tasks (0 = unlimited)
+        """
+        with self._lock:
+            self._limits[agent_name] = max_concurrent
+            # Reset semaphore so next acquire creates a fresh one
+            self._semaphores.pop(agent_name, None)
+
+    def get_limit(self, agent_name: str) -> int:
+        """Get concurrency limit for an agent."""
+        with self._lock:
+            return self._limits.get(agent_name, self._default_limit)
+
+    def remove_limit(self, agent_name: str) -> None:
+        """Remove concurrency limit for an agent (reverts to default)."""
+        with self._lock:
+            self._limits.pop(agent_name, None)
+            self._semaphores.pop(agent_name, None)
+
+    def _get_semaphore(self, agent_name: str) -> Optional[asyncio.Semaphore]:
+        """Get or create semaphore for agent. Returns None if unlimited."""
+        with self._lock:
+            limit = self._limits.get(agent_name, self._default_limit)
+            if limit <= 0:
+                return None
+            if agent_name not in self._semaphores:
+                self._semaphores[agent_name] = asyncio.Semaphore(limit)
+            return self._semaphores[agent_name]
+
+    async def acquire(self, agent_name: str) -> None:
+        """Acquire concurrency slot for agent. No-op if unlimited."""
+        sem = self._get_semaphore(agent_name)
+        if sem is not None:
+            await sem.acquire()
+
+    def acquire_sync(self, agent_name: str) -> None:
+        """Synchronous acquire — for non-async code paths.
+        
+        Note: This creates/reuses an event loop internally.
+        Prefer async acquire() when possible.
+        """
+        sem = self._get_semaphore(agent_name)
+        if sem is None:
+            return
+        try:
+            asyncio.get_running_loop()
+            # If we're in an async context, we can't block
+            # Just try_acquire or no-op with warning
+            if not sem._value > 0:
+                logger.warning(
+                    f"Sync acquire for '{agent_name}' while async loop running and semaphore full. "
+                    f"Consider using async acquire() instead."
+                )
+            # Decrement manually for sync context
+            sem._value = max(0, sem._value - 1)
+        except RuntimeError:
+            # No running loop — safe to use asyncio.run
+            asyncio.get_event_loop().run_until_complete(sem.acquire())
+
+    def release(self, agent_name: str) -> None:
+        """Release concurrency slot for agent. No-op if unlimited."""
+        with self._lock:
+            sem = self._semaphores.get(agent_name)
+        if sem is not None:
+            try:
+                sem.release()
+            except ValueError:
+                pass  # Already fully released
+
+    @asynccontextmanager
+    async def throttle(self, agent_name: str):
+        """Async context manager for throttled execution.
+        
+        Usage:
+            async with registry.throttle("agent_name"):
+                await do_work()
+        """
+        await self.acquire(agent_name)
+        try:
+            yield
+        finally:
+            self.release(agent_name)
+
+
+# Singleton
+_registry: Optional[ConcurrencyRegistry] = None
+_registry_lock = threading.Lock()
+
+
+def get_concurrency_registry() -> ConcurrencyRegistry:
+    """Get the global concurrency registry singleton."""
+    global _registry
+    if _registry is None:
+        with _registry_lock:
+            if _registry is None:
+                _registry = ConcurrencyRegistry()
+    return _registry

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

@@ -1,4 +1,13 @@
 """Task module for AI agent tasks"""
 from .task import Task
+from .protocols import TaskStatus, TaskLifecycleManager, InvalidTransitionError, TaskLifecycleProtocol
+from .message_sink import (
+    TaskMessage, TaskMessageSinkProtocol, NoOpTaskMessageSink,
+    InMemoryTaskMessageSink, TaskMessageEmitter,
+)
 
-__all__ = ['Task'] 
+__all__ = [
+    'Task', 'TaskStatus', 'TaskLifecycleManager', 'InvalidTransitionError', 'TaskLifecycleProtocol',
+    'TaskMessage', 'TaskMessageSinkProtocol', 'NoOpTaskMessageSink',
+    'InMemoryTaskMessageSink', 'TaskMessageEmitter',
+] 

src/praisonai-agents/praisonaiagents/task/message_sink.py

@@ -0,0 +1,146 @@
+"""Task Message Sink Protocols for PraisonAI Agents.
+
+Provides pluggable persistence for sequenced task execution messages.
+Matches Multica's ReportTaskMessages pattern but as a protocol,
+enabling replay of agent output, tool calls, errors, and status changes.
+
+Usage:
+    from praisonaiagents.task.message_sink import InMemoryTaskMessageSink, TaskMessageEmitter
+    
+    sink = InMemoryTaskMessageSink()
+    emitter = TaskMessageEmitter(task_id="t1", sink=sink, agent_name="researcher")
+    emitter.emit("agent_output", "Hello world")
+    emitter.emit("tool_call", "search_web('AI trends')")
+    
+    # Replay
+    messages = sink.replay("t1")
+"""
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Protocol, runtime_checkable, List, Optional, Dict, Any
+
+
+@dataclass
+class TaskMessage:
+    """A single sequenced message from task execution.
+    
+    Attributes:
+        task_id: Task this message belongs to
+        seq_num: Sequence number for ordering (0-indexed)
+        msg_type: Message type (agent_output, tool_call, tool_result, error, status)
+        content: Message content
+        agent_name: Agent that produced this message
+        metadata: Optional extra metadata
+        timestamp: ISO timestamp
+    """
+    task_id: str
+    seq_num: int
+    msg_type: str
+    content: str
+    agent_name: Optional[str] = None
+    metadata: Dict[str, Any] = field(default_factory=dict)
+    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary."""
+        return {
+            "task_id": self.task_id,
+            "seq_num": self.seq_num,
+            "msg_type": self.msg_type,
+            "content": self.content,
+            "agent_name": self.agent_name,
+            "metadata": self.metadata,
+            "timestamp": self.timestamp,
+        }
+
+
+@runtime_checkable
+class TaskMessageSinkProtocol(Protocol):
+    """Protocol for persisting task execution messages.
+    
+    Implementations can write to databases, files, WebSockets, etc.
+    """
+
+    def emit(self, message: TaskMessage) -> None:
+        """Persist a single task message."""
+        ...
+
+    def replay(self, task_id: str) -> List[TaskMessage]:
+        """Replay all messages for a task, ordered by seq_num."""
+        ...
+
+
+class NoOpTaskMessageSink:
+    """Default sink that does nothing. Zero overhead."""
+
+    def emit(self, message: TaskMessage) -> None:
+        pass
+
+    def replay(self, task_id: str) -> List[TaskMessage]:
+        return []
+
+
+class InMemoryTaskMessageSink:
+    """In-memory sink for testing and debugging.
+    
+    Stores all messages in a list for inspection and replay.
+    """
+
+    def __init__(self):
+        self.messages: List[TaskMessage] = []
+
+    def emit(self, message: TaskMessage) -> None:
+        self.messages.append(message)
+
+    def replay(self, task_id: str) -> List[TaskMessage]:
+        """Replay messages for a task, ordered by seq_num."""
+        task_messages = [m for m in self.messages if m.task_id == task_id]
+        return sorted(task_messages, key=lambda m: m.seq_num)
+
+    def clear(self) -> None:
+        """Clear all messages."""
+        self.messages.clear()
+
+
+class TaskMessageEmitter:
+    """Convenience emitter that auto-sequences messages for a task.
+    
+    Maintains a per-task sequence counter so callers don't need
+    to track seq_num manually.
+    
+    Usage:
+        emitter = TaskMessageEmitter(task_id="t1", sink=sink)
+        emitter.emit("agent_output", "Hello")   # seq_num=0
+        emitter.emit("tool_call", "search()")    # seq_num=1
+    """
+
+    def __init__(
+        self,
+        task_id: str,
+        sink: TaskMessageSinkProtocol,
+        agent_name: Optional[str] = None,
+    ):
+        self.task_id = task_id
+        self.sink = sink
+        self.agent_name = agent_name
+        self._seq_num = 0
+
+    def emit(
+        self,
+        msg_type: str,
+        content: str,
+        agent_name: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """Emit a message with auto-incremented sequence number."""
+        message = TaskMessage(
+            task_id=self.task_id,
+            seq_num=self._seq_num,
+            msg_type=msg_type,
+            content=content,
+            agent_name=agent_name or self.agent_name,
+            metadata=metadata or {},
+        )
+        self.sink.emit(message)
+        self._seq_num += 1