Release v4.6.50

* fix: wrapper layer gaps - case normalization, lazy loading, async persistence (fixes #1766)

- Gap 1: Fix PluginRegistry case normalization inconsistency by normalizing names during storage and using lazy loaders
- Gap 2: Implement lazy loading in PluginRegistry by storing loaders instead of immediately executing them
- Gap 3: Add async persistence support with aon_* methods and achat wrapping in agent hooks

Co-authored-by: MervinPraison <[email protected]>

* fix: critical bugs in wrapper layer - registry consistency and async event loop blocking

- Fix registry inconsistency: register() now writes to both _loaders and _items
- Fix event loop blocking: PersistentAgent.achat uses asyncio.to_thread for sync fallback
- Fix threading deadlock: use RLock and release lock before calling loader()
- Add async session end methods (aend_session) for proper async lifecycle

Addresses critical P1/P2 issues identified by code reviewers.

🤖 Generated with [Claude Code](https://claude.ai/code)

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

---------

Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>
Co-authored-by: MervinPraison <[email protected]>

README.md

@@ -382,7 +382,7 @@ praisonai flow
 
 Open **http://localhost:7861** — use the **Agent** and **Agent Team** components to create sequential or parallel workflows. Connect Chat Input → Agent Team → Chat Output for instant multi-agent pipelines.
 
-> 📖 [Full Flow docs](https://docs.praison.ai/docs/concepts/flow) — visual agent building, component reference, and deployment
+> 📖 [Full Flow docs](https://docs.praison.ai/docs/concepts/agentflow) — visual agent building, component reference, and deployment
 
 ### 8. PraisonAI UI 🤖 (Clean Chat)
 

api.md

@@ -911,6 +911,18 @@ Methods:
 * <code title="cli">praisonai hooks add <a href="./src/praisonai/praisonai/cli/commands/hooks.py">--help</a></code>
 * <code title="cli">praisonai hooks list <a href="./src/praisonai/praisonai/cli/commands/hooks.py">--help</a></code>
 * <code title="cli">praisonai hooks remove <a href="./src/praisonai/praisonai/cli/commands/hooks.py">--help</a></code>
+* <code title="cli">praisonai kanban block <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban boards <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban comment <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban complete <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban create <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban dispatch <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban link <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban list <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban move <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban reclaim <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban show <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban unblock <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
 * <code title="cli">praisonai knowledge add <a href="./src/praisonai/praisonai/cli/commands/knowledge.py">--help</a></code>
 * <code title="cli">praisonai knowledge index <a href="./src/praisonai/praisonai/cli/commands/knowledge.py">--help</a></code>
 * <code title="cli">praisonai knowledge list <a href="./src/praisonai/praisonai/cli/commands/knowledge.py">--help</a></code>

docker/Dockerfile.chat

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=4.6.48" \
+    "praisonai>=4.6.50" \
     "praisonai[chat]" \
     "embedchain[github,youtube]"
 

docker/Dockerfile.dev

@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=4.6.48" \
+    "praisonai>=4.6.50" \
     "praisonai[ui]" \
     "praisonai[chat]" \
     "praisonai[realtime]" \

docker/Dockerfile.ui

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=4.6.48" \
+    "praisonai>=4.6.50" \
     "praisonai[ui]" \
     "praisonai[crewai]"
 

src/praisonai-agents/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "praisonaiagents"
-version = "1.6.48"
+version = "1.6.50"
 description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
 readme = "README.md"
 requires-python = ">=3.10"

src/praisonai-agents/uv.lock

@@ -3125,7 +3125,7 @@ wheels = [
 
 [[package]]
 name = "praisonaiagents"
-version = "1.6.48"
+version = "1.6.50"
 source = { editable = "." }
 dependencies = [
     { name = "aiohttp" },

src/praisonai/README.md

@@ -382,7 +382,7 @@ praisonai flow
 
 Open **http://localhost:7861** — use the **Agent** and **Agent Team** components to create sequential or parallel workflows. Connect Chat Input → Agent Team → Chat Output for instant multi-agent pipelines.
 
-> 📖 [Full Flow docs](https://docs.praison.ai/docs/concepts/flow) — visual agent building, component reference, and deployment
+> 📖 [Full Flow docs](https://docs.praison.ai/docs/concepts/agentflow) — visual agent building, component reference, and deployment
 
 ### 8. PraisonAI UI 🤖 (Clean Chat)
 

src/praisonai/praisonai.rb

@@ -3,8 +3,8 @@ class Praisonai < Formula
   
     desc "AI tools for various AI applications"
     homepage "https://github.com/MervinPraison/PraisonAI"
-    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.6.48.tar.gz"
-    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.6.48.tar.gz | shasum -a 256`.split.first
+    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.6.50.tar.gz"
+    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.6.50.tar.gz | shasum -a 256`.split.first
     license "MIT"
   
     depends_on "[email protected]"

src/praisonai/praisonai/_async_bridge.py

@@ -31,12 +31,8 @@ class _BackgroundLoop:
         server processes should call :func:`shutdown` explicitly to cancel
         in-flight tasks before process exit.
         """
-        # Enforce the documented invariant. ``threading.Lock.locked()``
-        # is process-wide, not thread-local, but combined with the fact
-        # that the only callers are :py:meth:`get` and :py:meth:`get_unlocked`
-        # (the latter only invoked from inside ``run_sync`` while holding
-        # the lock), this catches the "forgot to hold the lock" mistake.
-        assert self._lock.locked(), "_spawn_locked() requires self._lock to be held"
+        # Create the loop+thread; caller must hold self._lock.
+        # The lock contract is now enforced structurally by submit() method.
         if self._loop is None or self._loop.is_closed():
             self._loop = asyncio.new_event_loop()
             self._thread = threading.Thread(
@@ -51,9 +47,11 @@ class _BackgroundLoop:
         with self._lock:
             return self._spawn_locked()
 
-    def get_unlocked(self) -> asyncio.AbstractEventLoop:
-        """Get loop assuming caller holds _lock. For run_sync() use only."""
-        return self._spawn_locked()
+    def submit(self, coro):
+        """Atomically (re)spawn loop if needed and submit coro. Returns concurrent.futures.Future."""
+        with self._lock:
+            loop = self._spawn_locked()
+            return asyncio.run_coroutine_threadsafe(coro, loop)
 
     def shutdown(self, timeout: float = 5.0) -> None:
         with self._lock:
@@ -77,7 +75,16 @@ class _BackgroundLoop:
                 self._loop = None
                 self._thread = None
 
-_BG = _BackgroundLoop()
+_BG: "_BackgroundLoop | None" = None
+_BG_LOCK = threading.Lock()
+
+def _get_bg() -> "_BackgroundLoop":
+    global _BG
+    if _BG is None:
+        with _BG_LOCK:
+            if _BG is None:
+                _BG = _BackgroundLoop()
+    return _BG
 
 
 def run_sync(coro: Awaitable[T], *, timeout: float | None = _DEFAULT_TIMEOUT) -> T:
@@ -109,10 +116,8 @@ def run_sync(coro: Awaitable[T], *, timeout: float | None = _DEFAULT_TIMEOUT) ->
             "await the coroutine directly instead."
         )
 
-    # Submit the coroutine inside the lock to prevent shutdown races
-    with _BG._lock:
-        loop = _BG.get_unlocked()
-        fut: Future = asyncio.run_coroutine_threadsafe(coro, loop)
+    # Submit the coroutine atomically
+    fut = _get_bg().submit(coro)
     
     try:
         return fut.result(timeout=timeout)
@@ -139,4 +144,6 @@ def run_sync(coro: Awaitable[T], *, timeout: float | None = _DEFAULT_TIMEOUT) ->
 
 def shutdown() -> None:
     """Public hook for long-running server processes to stop the bridge cleanly."""
-    _BG.shutdown()
+    bg = _BG
+    if bg is not None:
+        bg.shutdown()

src/praisonai/praisonai/_registry.py

@@ -41,39 +41,28 @@ class PluginRegistry(Generic[T]):
             builtins: Dict of name -> loader function for built-in plugins
         """
         self._entry_point_group = entry_point_group
-        self._items: Dict[str, Type[T]] = {}
+        self._loaders: Dict[str, Callable[[], Type[T]]] = {}  # lazy loaders
+        self._items: Dict[str, Type[T]] = {}  # resolved cache
         self._aliases: Dict[str, str] = {}  # alias -> canonical name
-        self._lock = threading.Lock()
+        self._lock = threading.RLock()  # Use RLock for re-entrant access
         
-        # Load built-in plugins with error handling
+        # Store built-in loaders (normalized) without calling them
         if builtins:
             for name, loader in builtins.items():
-                try:
-                    self._items[name] = loader()
-                except ImportError:
-                    # Built-in plugin dependencies not available, skip
-                    pass
-                except Exception:
-                    # Log other errors but don't crash initialization
-                    logger.warning("Failed to load built-in plugin %r", name, exc_info=True)
+                self._add_loader(name, loader)
         
-        self._load_entry_points()
+        self._discover_entry_points()
 
-    def _load_entry_points(self) -> None:
-        """Load plugins from entry points."""
+    def _add_loader(self, name: str, loader: Callable[[], Type[T]]) -> None:
+        """Add a loader with normalized name."""
+        normalized_name = name.lower()
+        self._loaders[normalized_name] = loader
+
+    def _discover_entry_points(self) -> None:
+        """Discover entry point loaders without loading them."""
         try:
             for ep in entry_points(group=self._entry_point_group):
-                try:
-                    plugin_class = ep.load()
-                    with self._lock:
-                        self._items[ep.name] = plugin_class
-                except Exception:
-                    # Do not break plugin dispatch because one plugin is broken
-                    logger.warning(
-                        "Failed to load plugin %r from entry point",
-                        ep.name,
-                        exc_info=True,
-                    )
+                self._add_loader(ep.name, ep.load)
         except Exception:
             # entry_points() might not be available in older Python versions
             logger.debug("Entry points not available for group %s", self._entry_point_group)
@@ -94,6 +83,8 @@ class PluginRegistry(Generic[T]):
         """
         with self._lock:
             canonical_name = name.lower()
+            # Store both in loaders (for consistency with discovery) and items (for immediate access)
+            self._loaders[canonical_name] = lambda: cls
             self._items[canonical_name] = cls
             
             # Register aliases
@@ -120,7 +111,7 @@ class PluginRegistry(Generic[T]):
                 return True
             
             # Check if it's a canonical name
-            if normalized_name in self._items:
+            if normalized_name in self._loaders:
                 # Remove all aliases pointing to this plugin
                 aliases_to_remove = [
                     alias for alias, canonical in self._aliases.items()
@@ -129,7 +120,9 @@ class PluginRegistry(Generic[T]):
                 for alias in aliases_to_remove:
                     del self._aliases[alias]
                 
-                del self._items[normalized_name]
+                # Remove from both loaders and items cache
+                del self._loaders[normalized_name]
+                self._items.pop(normalized_name, None)
                 return True
             
             return False
@@ -146,28 +139,44 @@ class PluginRegistry(Generic[T]):
         Raises:
             ValueError: If plugin is not found
         """
+        normalized_name = name.lower()
+        
         with self._lock:
-            normalized_name = name.lower()
-            
-            # Resolve alias to canonical name
+            # Check cache first
             canonical_name = self._aliases.get(normalized_name, normalized_name)
             cls = self._items.get(canonical_name)
             
-            # Capture available plugins snapshot while holding lock
-            # to avoid race condition between check and error message
-            if cls is None:
-                available_plugins = sorted(self._items.keys())
+            if cls is not None:
+                return cls
+            
+            # Try to load from loaders
+            loader = self._loaders.get(canonical_name)
+            if loader is None:
+                # Capture available plugins snapshot while holding lock
+                available_loaders = sorted(self._loaders.keys())
                 available_aliases = sorted(self._aliases.keys())
-                available_snapshot = available_plugins + available_aliases
-            else:
-                available_snapshot = None
+                available_snapshot = available_loaders + available_aliases
+                raise ValueError(
+                    f"Unknown {self._entry_point_group} plugin: {name!r}. "
+                    f"Available: {available_snapshot}"
+                )
+            
+            # Load and cache the plugin (release lock during loading to prevent deadlock)
+            pass
         
-        if cls is None:
+        # Load outside the lock to prevent deadlock if loader calls back into registry
+        try:
+            cls = loader()
+        except ImportError as e:
             raise ValueError(
-                f"Unknown {self._entry_point_group} plugin: {name!r}. "
-                f"Available: {available_snapshot}"
-            )
-        return cls
+                f"Plugin {name!r} is registered but its dependencies "
+                f"are not installed: {e}"
+            ) from e
+        
+        # Re-acquire lock to cache the result
+        with self._lock:
+            self._items[canonical_name] = cls
+            return cls
 
     def create(self, name: str, *args, **kwargs) -> T:
         """Create an instance of the specified plugin.
@@ -192,7 +201,7 @@ class PluginRegistry(Generic[T]):
             Sorted list of plugin names
         """
         with self._lock:
-            return sorted(self._items.keys())
+            return sorted(self._loaders.keys())
 
     def is_available(self, name: str) -> bool:
         """Check if a plugin is available.
@@ -225,4 +234,4 @@ class PluginRegistry(Generic[T]):
             Sorted list of all registered names and aliases
         """
         with self._lock:
-            return sorted(list(self._items.keys()) + list(self._aliases.keys()))
+            return sorted(list(self._loaders.keys()) + list(self._aliases.keys()))

src/praisonai/praisonai/agents_generator.py

@@ -28,14 +28,33 @@ from .tool_registry import ToolRegistry
 
 # BaseTool import is now handled centrally by ToolResolver
 
-# Check for additional framework availability using centralized detection
+# Framework availability detection (lazy via __getattr__)
 from ._framework_availability import is_available
-PRAISONAI_TOOLS_AVAILABLE = is_available("praisonai_tools")
-CREWAI_AVAILABLE          = is_available("crewai")
-AUTOGEN_AVAILABLE         = is_available("autogen")
-AG2_AVAILABLE             = is_available("ag2")
-PRAISONAI_AVAILABLE       = is_available("praisonaiagents")
-AGENTOPS_AVAILABLE        = is_available("agentops")
+
+# Lazy constants mapping for backward compatibility
+_AVAIL = {
+    "PRAISONAI_TOOLS_AVAILABLE": "praisonai_tools",
+    "CREWAI_AVAILABLE": "crewai",
+    "AUTOGEN_AVAILABLE": "autogen",
+    "AG2_AVAILABLE": "ag2",
+    "PRAISONAI_AVAILABLE": "praisonaiagents",
+    "AGENTOPS_AVAILABLE": "agentops",
+}
+
+__all__ = list(_AVAIL.keys())
+
+def __getattr__(name):
+    """Lazy attribute access for framework availability constants.
+    
+    This allows backward compatibility while avoiding import-time probing.
+    Only probes the framework when the constant is actually accessed.
+    """
+    if name in _AVAIL:
+        return is_available(_AVAIL[name])
+    raise AttributeError(f"module '{__name__}' has no attribute '{name}'")
+
+def __dir__():
+    return sorted(set(globals()) | set(_AVAIL))
 
 # Framework adapter registry - now uses proper registry pattern
 # This replaces the hardcoded FRAMEWORK_ADAPTERS dict
@@ -282,7 +301,7 @@ class AgentsGenerator:
         
         # Keep tool registry for backward compatibility with autogen adapters
         self.tool_registry = ToolRegistry()
-        self.tool_registry.register_builtin_autogen_adapters()
+        self.tool_registry.register_builtin_autogen_adapters(_suppress_deprecation_warning=True)
         
         # Initialize tool resolver with the registry wired in (single source of truth for tool resolution)
         from .tool_resolver import ToolResolver
@@ -576,7 +595,7 @@ class AgentsGenerator:
         tools_dict = {}
         
         # Demand-driven tool resolution - only resolve tools actually used in YAML
-        if CREWAI_AVAILABLE or AUTOGEN_AVAILABLE or PRAISONAI_AVAILABLE or AG2_AVAILABLE:
+        if is_available("crewai") or is_available("autogen") or is_available("praisonaiagents") or is_available("ag2"):
             try:
                 # Collect all tool names mentioned in the YAML config
                 needed_tools: set[str] = set()
@@ -696,7 +715,7 @@ class AgentsGenerator:
         Returns:
             str: Result of the workflow execution
         """
-        if not PRAISONAI_AVAILABLE:
+        if not is_available("praisonaiagents"):
             raise ImportError("PraisonAI is not installed. Please install it with 'pip install praisonaiagents'")
         
         try:

src/praisonai/praisonai/auto.py

@@ -65,37 +65,7 @@ def _load_optional(key: str, loader):
 # Availability checks now delegated to centralized module
 from ._framework_availability import is_available
 
-def _check_crewai_available() -> bool:
-    """Check if crewai is available (cached, thread-safe)."""
-    return is_available("crewai")
-
-def _check_autogen_available() -> bool:
-    """Check if autogen v0.2 is available (cached, thread-safe)."""
-    return is_available("autogen")
-
-def _check_autogen_v4_available() -> bool:
-    """Check if autogen v0.4 is available (cached, thread-safe)."""
-    return is_available("autogen_v4")
-
-def _check_ag2_available() -> bool:
-    """Check if AG2 (community fork of AutoGen) is available (cached, thread-safe)."""
-    return is_available("ag2")
-
-def _check_praisonai_available() -> bool:
-    """Check if praisonaiagents is available (cached, thread-safe)."""
-    return is_available("praisonaiagents")
-
-def _check_praisonai_tools_available() -> bool:
-    """Check if praisonai_tools is available (cached, thread-safe)."""
-    return is_available("praisonai_tools")
-
-def _check_litellm_available() -> bool:
-    """Check if litellm is available (cached, thread-safe)."""
-    return is_available("litellm")
-
-def _check_openai_available() -> bool:
-    """Check if openai is available (cached, thread-safe)."""
-    return is_available("openai")
+# Redundant wrapper functions removed - use is_available() directly
 
 
 def _get_crewai():
@@ -504,7 +474,7 @@ class BaseAutoGenerator:
         model_name = self.config_list[0]['model']
         
         # Try LiteLLM first (preferred - supports 100+ providers)
-        if _check_litellm_available():
+        if is_available("litellm"):
             litellm = _get_litellm()
             response = litellm.completion(
                 model=model_name,
@@ -516,7 +486,7 @@ class BaseAutoGenerator:
             return response_model.model_validate_json(content)
         
         # Fallback to OpenAI SDK (uses beta.chat.completions.parse)
-        if _check_openai_available():
+        if is_available("openai"):
             client = self._get_openai_client()
             response = client.beta.chat.completions.parse(
                 model=model_name,
@@ -554,7 +524,7 @@ class BaseAutoGenerator:
         model_name = self.config_list[0]['model']
         
         # Try LiteLLM async first (preferred - supports 100+ providers)
-        if _check_litellm_available():
+        if is_available("litellm"):
             litellm = _get_litellm()
             response = await litellm.acompletion(
                 model=model_name,
@@ -566,7 +536,7 @@ class BaseAutoGenerator:
             return response_model.model_validate_json(content)
         
         # Fallback to OpenAI AsyncSDK (uses beta.chat.completions.parse)
-        if _check_openai_available():
+        if is_available("openai"):
             if self._async_openai_client is None:
                 with self._client_lock:
                     if self._async_openai_client is None:
@@ -728,7 +698,7 @@ class AutoGenerator(BaseAutoGenerator):
             raise ImportError(f"{adapter.name} is not installed. Please install with:\n    {install_hint}")
         
         # Check tools availability if required by this framework
-        if requires_tools_extra and not _check_praisonai_tools_available():
+        if requires_tools_extra and not is_available("praisonai_tools"):
             logger.warning(f"Tools are not available for {framework}. To use tools, install:\n    {install_hint}")
 
         self.topic = topic

src/praisonai/praisonai/bots/_config_schema.py

@@ -23,11 +23,14 @@ class AgentConfigSchema(BaseModel):
 
 class ChannelConfigSchema(BaseModel):
     """Schema for a single channel configuration."""
+    platform: Optional[str] = None
     token: str = ""
     mode: str = "poll"  # poll | ws | webhook | hybrid
     group_policy: str = "respond_all"  # respond_all, mention_only, command_only
     allowlist: List[str] = Field(default_factory=list)
     blocklist: List[str] = Field(default_factory=list)
+    allowed_users: Optional[str] = None
+    routes: Dict[str, str] = Field(default_factory=dict)
     webhook_url: Optional[str] = None
     webhook_port: int = 8080
     
@@ -102,12 +105,15 @@ class BotYamlSchema(BaseModel):
                 "No channels configured. Add at least one channel "
                 "(telegram, discord, slack, whatsapp) to your bot.yaml"
             )
-        # Validate channel names
+        # Channel keys may be platform names (telegram) or custom ids (telegram_cfo)
+        # when ``platform`` is set on the channel block (gateway / multi-bot YAML).
         valid_platforms = {"telegram", "discord", "slack", "whatsapp", "email", "agentmail"}
-        for name in self.channels:
-            if name not in valid_platforms:
+        for name, channel in self.channels.items():
+            platform = (channel.platform or name).lower()
+            if platform not in valid_platforms:
                 raise ValueError(
-                    f"Unknown channel '{name}'. Supported: {', '.join(sorted(valid_platforms))}"
+                    f"Unknown channel '{name}' (platform '{platform}'). "
+                    f"Supported platforms: {', '.join(sorted(valid_platforms))}"
                 )
         return self
 

src/praisonai/praisonai/claw/default_app.py

@@ -72,6 +72,8 @@ configure_host(
         "guardrails",
         "sessions",
         "usage",
+        "kanban",        # ADD
+        "jobs-board",    # ADD
         "config",
         "logs",
         "debug",

src/praisonai/praisonai/cli/app.py

@@ -305,6 +305,7 @@ def register_commands():
     from .commands.mcp import app as mcp_app
     from .commands.serve import app as serve_app
     from .commands.schedule import app as schedule_app
+    from .commands.kanban import app as kanban_app
     from .commands.run import app as run_app
     from .commands.profile import app as profile_app
     from .commands.benchmark import app as benchmark_app
@@ -387,6 +388,7 @@ def register_commands():
     app.add_typer(mcp_app, name="mcp", help="MCP server management")
     app.add_typer(serve_app, name="serve", help="API server management")
     app.add_typer(schedule_app, name="schedule", help="Scheduler management")
+    app.add_typer(kanban_app, name="kanban", help="Kanban task management")
     app.add_typer(run_app, name="run", help="Run agents")
     app.add_typer(profile_app, name="profile", help="Performance profiling and diagnostics")
     app.add_typer(benchmark_app, name="benchmark", help="Comprehensive performance benchmarking")

src/praisonai/praisonai/cli/commands/kanban.py

@@ -0,0 +1,452 @@
+"""
+Kanban command group for PraisonAI CLI.
+
+Provides kanban task management for multi-agent coordination.
+"""
+
+import json
+from typing import Optional, List
+
+import typer
+
+from ..output.console import get_output_controller
+
+app = typer.Typer(help="Kanban task management")
+
+
+def _get_kanban_store():
+    """Get kanban store instance."""
+    try:
+        from praisonai.kanban.sqlite_store import SQLiteKanbanStore
+        return SQLiteKanbanStore()
+    except ImportError as e:
+        output = get_output_controller()
+        output.print_error(f"Kanban module not available: {e}")
+        raise typer.Exit(4)
+
+
+@app.command("list")
+def kanban_list_cmd(
+    status: Optional[str] = typer.Option(None, "--status", "-s", help="Filter by status (todo, ready, running, blocked, done)"),
+    assignee: Optional[str] = typer.Option(None, "--assignee", "-a", help="Filter by assignee"),
+    board: str = typer.Option("default", "--board", "-b", help="Board name"),
+    limit: int = typer.Option(50, "--limit", "-l", help="Maximum number of tasks to show"),
+    json_output: bool = typer.Option(False, "--json", help="Output as JSON"),
+):
+    """List kanban tasks with optional filtering."""
+    output = get_output_controller()
+    
+    try:
+        store = _get_kanban_store()
+        
+        # Build filters
+        filters = {'board': board}
+        if status:
+            filters['status'] = status
+        if assignee:
+            filters['assignee'] = assignee
+        
+        tasks = store.list_tasks(filters)
+        
+        if len(tasks) > limit:
+            tasks = tasks[:limit]
+        
+        if json_output:
+            result = {
+                'tasks': [task.to_dict() for task in tasks],
+                'count': len(tasks),
+                'filters': filters,
+                'limited': len(tasks) == limit
+            }
+            output.print_json(result)
+        else:
+            # Pretty print table
+            if not tasks:
+                output.print_info("No tasks found")
+                return
+            
+            output.print_info(f"Found {len(tasks)} tasks{' (limited)' if len(tasks) == limit else ''}")
+            output.print("")
+            
+            # Table header
+            print(f"{'ID':<15} {'Status':<10} {'Title':<40} {'Assignee':<15} {'Priority':<8}")
+            print("-" * 90)
+            
+            # Task rows
+            for task in tasks:
+                title = task.title[:37] + "..." if len(task.title) > 40 else task.title
+                assignee = task.assignee[:12] + "..." if len(task.assignee) > 15 else task.assignee
+                print(f"{task.id:<15} {task.status.value:<10} {title:<40} {assignee:<15} {task.priority:<8}")
+    
+    except Exception as e:
+        output.print_error(f"Failed to list tasks: {e}")
+        raise typer.Exit(1)
+
+
+@app.command("create")
+def kanban_create_cmd(
+    title: str = typer.Argument(..., help="Task title"),
+    body: str = typer.Option("", "--body", "-b", help="Task description"),
+    assignee: str = typer.Option("", "--assignee", "-a", help="Username to assign to"),
+    status: str = typer.Option("todo", "--status", "-s", help="Initial status"),
+    priority: int = typer.Option(0, "--priority", "-p", help="Task priority (higher = more important)"),
+    board: str = typer.Option("default", "--board", help="Board name"),
+    json_output: bool = typer.Option(False, "--json", help="Output as JSON"),
+):
+    """Create a new kanban task."""
+    output = get_output_controller()
+    
+    try:
+        store = _get_kanban_store()
+        
+        task_data = {
+            'title': title,
+            'body': body,
+            'assignee': assignee,
+            'status': status,
+            'priority': priority,
+            'board': board
+        }
+        
+        task = store.create_task(task_data)
+        
+        if json_output:
+            output.print_json(task.to_dict())
+        else:
+            output.print_success(f"Created task {task.id}")
+            output.print_info(f"Title: {task.title}")
+            output.print_info(f"Status: {task.status.value}")
+            if assignee:
+                output.print_info(f"Assignee: {task.assignee}")
+    
+    except Exception as e:
+        output.print_error(f"Failed to create task: {e}")
+        raise typer.Exit(1)
+
+
+@app.command("show")
+def kanban_show_cmd(
+    task_id: str = typer.Argument(..., help="Task ID to show"),
+    json_output: bool = typer.Option(False, "--json", help="Output as JSON"),
+):
+    """Show detailed task information."""
+    output = get_output_controller()
+    
+    try:
+        store = _get_kanban_store()
+        
+        task = store.get_task(task_id)
+        if not task:
+            output.print_error(f"Task {task_id} not found")
+            raise typer.Exit(1)
+        
+        comments = store.get_comments(task_id)
+        
+        if json_output:
+            result = {
+                'task': task.to_dict(),
+                'comments': [comment.to_dict() for comment in comments],
+                'comment_count': len(comments)
+            }
+            output.print_json(result)
+        else:
+            # Pretty print task details
+            output.print_info(f"Task: {task.id}")
+            output.print("")
+            output.print_info(f"Title: {task.title}")
+            output.print_info(f"Status: {task.status.value}")
+            output.print_info(f"Assignee: {task.assignee or '(unassigned)'}")
+            output.print_info(f"Priority: {task.priority}")
+            output.print_info(f"Board: {task.board}")
+            output.print_info(f"Created: {task.created_at.strftime('%Y-%m-%d %H:%M:%S')}")
+            output.print_info(f"Updated: {task.updated_at.strftime('%Y-%m-%d %H:%M:%S')}")
+            
+            if task.body:
+                output.print("")
+                output.print_info("Description:")
+                output.print(task.body)
+            
+            if comments:
+                output.print("")
+                output.print_info(f"Comments ({len(comments)}):")
+                for comment in comments:
+                    output.print(f"  [{comment.created_at.strftime('%m-%d %H:%M')}] {comment.author}: {comment.text}")
+    
+    except Exception as e:
+        output.print_error(f"Failed to show task: {e}")
+        raise typer.Exit(1)
+
+
+@app.command("move")
+def kanban_move_cmd(
+    task_id: str = typer.Argument(..., help="Task ID to move"),
+    status: str = typer.Argument(..., help="New status (todo, ready, running, blocked, done)"),
+    json_output: bool = typer.Option(False, "--json", help="Output as JSON"),
+):
+    """Move task to a new status."""
+    output = get_output_controller()
+    
+    try:
+        store = _get_kanban_store()
+        
+        task = store.move_task(task_id, status)
+        
+        if json_output:
+            output.print_json(task.to_dict())
+        else:
+            output.print_success(f"Moved task {task_id} to {status}")
+    
+    except Exception as e:
+        output.print_error(f"Failed to move task: {e}")
+        raise typer.Exit(1)
+
+
+@app.command("comment")
+def kanban_comment_cmd(
+    task_id: str = typer.Argument(..., help="Task ID to comment on"),
+    text: str = typer.Argument(..., help="Comment text"),
+    author: str = typer.Option("cli", "--author", "-a", help="Comment author"),
+    json_output: bool = typer.Option(False, "--json", help="Output as JSON"),
+):
+    """Add a comment to a task."""
+    output = get_output_controller()
+    
+    try:
+        store = _get_kanban_store()
+        
+        comment = store.add_comment(task_id, author, text)
+        
+        if json_output:
+            output.print_json(comment.to_dict())
+        else:
+            output.print_success(f"Added comment to task {task_id}")
+    
+    except Exception as e:
+        output.print_error(f"Failed to add comment: {e}")
+        raise typer.Exit(1)
+
+
+@app.command("link")
+def kanban_link_cmd(
+    parent_id: str = typer.Argument(..., help="Parent task ID"),
+    child_id: str = typer.Argument(..., help="Child task ID"),
+    json_output: bool = typer.Option(False, "--json", help="Output as JSON"),
+):
+    """Create dependency link between tasks."""
+    output = get_output_controller()
+    
+    try:
+        store = _get_kanban_store()
+        
+        link = store.add_link(parent_id, child_id)
+        
+        if json_output:
+            result = {
+                'parent_id': link.parent_id,
+                'child_id': link.child_id,
+                'created_at': link.created_at.isoformat()
+            }
+            output.print_json(result)
+        else:
+            output.print_success(f"Linked {parent_id} -> {child_id}")
+    
+    except Exception as e:
+        output.print_error(f"Failed to link tasks: {e}")
+        raise typer.Exit(1)
+
+
+@app.command("boards")
+def kanban_boards_cmd(
+    action: str = typer.Argument("list", help="Action: list, switch"),
+    board: Optional[str] = typer.Argument(None, help="Board name for switch action"),
+    json_output: bool = typer.Option(False, "--json", help="Output as JSON"),
+):
+    """Manage kanban boards."""
+    output = get_output_controller()
+    
+    try:
+        if action == "list":
+            from praisonai.kanban.paths import list_available_boards
+            boards = list_available_boards()
+            
+            if json_output:
+                output.print_json({"boards": boards})
+            else:
+                output.print_info(f"Available boards:")
+                for board_name in boards:
+                    output.print(f"  - {board_name}")
+        
+        elif action == "switch":
+            if not board:
+                output.print_error("Board name required for switch action")
+                raise typer.Exit(1)
+            
+            import os
+            from pathlib import Path
+            import json
+            
+            # Set for current session
+            os.environ['PRAISONAI_KANBAN_BOARD'] = board
+            
+            # Persist to config file for future sessions
+            config_dir = Path.home() / ".praisonai"
+            config_dir.mkdir(exist_ok=True)
+            config_file = config_dir / "kanban_config.json"
+            
+            try:
+                config = {"active_board": board}
+                with open(config_file, 'w') as f:
+                    json.dump(config, f, indent=2)
+                output.print_success(f"Switched to board '{board}' (persisted)")
+            except Exception as e:
+                output.print_success(f"Switched to board '{board}' (session only)")
+                output.print_info(f"Could not persist setting: {e}")
+        
+        else:
+            output.print_error(f"Unknown action: {action}")
+            raise typer.Exit(1)
+    
+    except Exception as e:
+        output.print_error(f"Board operation failed: {e}")
+        raise typer.Exit(1)
+
+
+@app.command("dispatch")
+def kanban_dispatch_cmd(
+    max_spawn: int = typer.Option(3, "--max-spawn", "-m", help="Maximum tasks to spawn"),
+    json_output: bool = typer.Option(False, "--json", help="Output as JSON"),
+):
+    """Manually trigger kanban task dispatcher."""
+    output = get_output_controller()
+    
+    try:
+        import asyncio
+        from praisonai.gateway.kanban_dispatcher import dispatch_once
+        
+        # Run dispatcher once
+        spawned = asyncio.run(dispatch_once(max_spawn))
+        
+        if json_output:
+            output.print_json({"spawned": spawned})
+        else:
+            if spawned > 0:
+                output.print_success(f"Spawned {spawned} task workers")
+            else:
+                output.print_info("No ready tasks to spawn")
+    
+    except Exception as e:
+        output.print_error(f"Dispatch failed: {e}")
+        raise typer.Exit(1)
+
+
+@app.command("block")
+def kanban_block_cmd(
+    task_id: str = typer.Argument(..., help="Task ID to block"),
+    reason: str = typer.Argument(..., help="Reason for blocking"),
+    json_output: bool = typer.Option(False, "--json", help="Output as JSON"),
+):
+    """Block a task with a reason."""
+    output = get_output_controller()
+    
+    try:
+        store = _get_kanban_store()
+        
+        # Move to blocked and add comment
+        task = store.move_task(task_id, 'blocked')
+        store.add_comment(task_id, 'cli', f"Blocked: {reason}")
+        
+        if json_output:
+            output.print_json(task.to_dict())
+        else:
+            output.print_success(f"Blocked task {task_id}: {reason}")
+    
+    except Exception as e:
+        output.print_error(f"Failed to block task: {e}")
+        raise typer.Exit(1)
+
+
+@app.command("unblock")
+def kanban_unblock_cmd(
+    task_id: str = typer.Argument(..., help="Task ID to unblock"),
+    json_output: bool = typer.Option(False, "--json", help="Output as JSON"),
+):
+    """Unblock a task (move to ready)."""
+    output = get_output_controller()
+    
+    try:
+        store = _get_kanban_store()
+        
+        task = store.move_task(task_id, 'ready')
+        store.add_comment(task_id, 'cli', "Unblocked by CLI")
+        
+        if json_output:
+            output.print_json(task.to_dict())
+        else:
+            output.print_success(f"Unblocked task {task_id}")
+    
+    except Exception as e:
+        output.print_error(f"Failed to unblock task: {e}")
+        raise typer.Exit(1)
+
+
+@app.command("complete")
+def kanban_complete_cmd(
+    task_id: str = typer.Argument(..., help="Task ID to complete"),
+    comment: str = typer.Option("", "--comment", "-c", help="Completion comment"),
+    json_output: bool = typer.Option(False, "--json", help="Output as JSON"),
+):
+    """Mark a task as completed."""
+    output = get_output_controller()
+    
+    try:
+        store = _get_kanban_store()
+        
+        task = store.move_task(task_id, 'done')
+        
+        if comment:
+            store.add_comment(task_id, 'cli', f"Completed: {comment}")
+        
+        if json_output:
+            output.print_json(task.to_dict())
+        else:
+            output.print_success(f"Completed task {task_id}")
+    
+    except Exception as e:
+        output.print_error(f"Failed to complete task: {e}")
+        raise typer.Exit(1)
+
+
+@app.command("reclaim")
+def kanban_reclaim_cmd(
+    task_id: str = typer.Argument(..., help="Task ID to reclaim"),
+    json_output: bool = typer.Option(False, "--json", help="Output as JSON"),
+):
+    """Reclaim a stuck/abandoned task."""
+    output = get_output_controller()
+    
+    try:
+        store = _get_kanban_store()
+        
+        # Force release any claim and move to ready
+        store.update_task(task_id, {'claim_lock': None})
+        task = store.move_task(task_id, 'ready')
+        store.add_comment(task_id, 'cli', "Reclaimed from stuck state")
+        
+        if json_output:
+            output.print_json(task.to_dict())
+        else:
+            output.print_success(f"Reclaimed task {task_id}")
+    
+    except Exception as e:
+        output.print_error(f"Failed to reclaim task: {e}")
+        raise typer.Exit(1)
+
+
+# Add to main CLI app if this module is being registered
+def register_kanban_commands(main_app):
+    """Register kanban commands with main CLI app."""
+    main_app.add_typer(app, name="kanban")
+
+
+if __name__ == "__main__":
+    app()

src/praisonai/praisonai/deploy.py

@@ -57,7 +57,7 @@ class CloudDeployer:
             file.write("FROM python:3.11-slim\n")
             file.write("WORKDIR /app\n")
             file.write("COPY . .\n")
-            file.write("RUN pip install flask praisonai==4.6.48 gunicorn markdown\n")
+            file.write("RUN pip install flask praisonai==4.6.50 gunicorn markdown\n")
             file.write("EXPOSE 8080\n")
             file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
             

src/praisonai/praisonai/gateway/kanban_dispatcher.py

@@ -0,0 +1,417 @@
+"""
+Kanban task dispatcher for gateway integration.
+
+This module implements the background dispatcher that claims ready kanban tasks
+and spawns agent runs to execute them. It integrates with the PraisonAI gateway
+system and follows Hermes patterns for worker management.
+"""
+
+import asyncio
+import logging
+import os
+import subprocess
+import time
+from typing import Dict, Any, List, Optional
+
+logger = logging.getLogger(__name__)
+
+# Global dispatcher state
+_dispatcher_running = False
+_dispatcher_task = None
+
+
+class KanbanDispatcher:
+    """Kanban task dispatcher that claims ready tasks and spawns workers."""
+    
+    def __init__(self, max_concurrent: int = 3, poll_interval: float = 5.0):
+        """
+        Initialize kanban dispatcher.
+        
+        Args:
+            max_concurrent: Maximum concurrent task executions
+            poll_interval: Seconds between task polling
+        """
+        self.max_concurrent = max_concurrent
+        self.poll_interval = poll_interval
+        self.running_tasks: Dict[str, subprocess.Popen] = {}
+        self.worker_id = f"gateway_{os.getpid()}"
+        
+    def _get_kanban_store(self):
+        """Get kanban store instance."""
+        try:
+            from praisonai.kanban.sqlite_store import SQLiteKanbanStore
+            return SQLiteKanbanStore()
+        except ImportError:
+            logger.error("Kanban store not available")
+            return None
+    
+    def _fire_hook_event(self, event_type: str, task_data: Dict[str, Any]):
+        """Fire kanban hook events (when available)."""
+        try:
+            # Try to import and fire hook events from praisonaiagents
+            from praisonaiagents.hooks import fire_hook
+            fire_hook(event_type, task_data)
+        except (ImportError, AttributeError):
+            # Hook system not available, just log
+            logger.debug(f"Hook event {event_type}: {task_data.get('task_id', 'unknown')}")
+    
+    async def dispatch_once(self) -> int:
+        """
+        Single dispatch cycle: claim ready tasks and spawn workers.
+        
+        Returns:
+            Number of tasks spawned in this cycle
+        """
+        store = self._get_kanban_store()
+        if not store:
+            return 0
+        
+        # Clean up completed processes first
+        self._cleanup_completed_tasks(store)
+        
+        # Check how many slots we have available
+        available_slots = self.max_concurrent - len(self.running_tasks)
+        if available_slots <= 0:
+            logger.debug(f"No available slots (running: {len(self.running_tasks)})")
+            return 0
+        
+        # Get ready tasks
+        ready_tasks = store.list_tasks({'status': 'ready'})
+        if not ready_tasks:
+            logger.debug("No ready tasks found")
+            return 0
+        
+        spawned = 0
+        for task in ready_tasks[:available_slots]:
+            try:
+                # Try to claim the task atomically
+                claimed = store.claim_task(task.id, self.worker_id)
+                if not claimed:
+                    logger.debug(f"Failed to claim task {task.id} (already claimed)")
+                    continue
+                
+                # Spawn worker process
+                success = await self._spawn_worker(task, store)
+                if success:
+                    spawned += 1
+                    
+                    # Fire claimed hook event
+                    self._fire_hook_event('KANBAN_TASK_CLAIMED', {
+                        'task_id': task.id,
+                        'worker_id': self.worker_id,
+                        'task': task.to_dict()
+                    })
+                else:
+                    # Failed to spawn, release claim
+                    store.release_claim(task.id, self.worker_id)
+                    
+            except Exception as e:
+                logger.error(f"Error processing task {task.id}: {e}")
+                # Release claim on error
+                try:
+                    store.release_claim(task.id, self.worker_id)
+                except:
+                    pass
+        
+        if spawned > 0:
+            logger.info(f"Spawned {spawned} kanban task workers")
+        
+        return spawned
+    
+    async def _spawn_worker(self, task: Any, store: Any) -> bool:
+        """
+        Spawn a worker process for the task.
+        
+        Args:
+            task: Task object to execute
+            store: Kanban store instance
+            
+        Returns:
+            True if worker was spawned successfully
+        """
+        try:
+            # Prepare environment variables
+            env = os.environ.copy()
+            env.update({
+                'PRAISONAI_KANBAN_TASK': task.id,
+                'PRAISONAI_KANBAN_BOARD': task.board,
+                'PRAISONAI_KANBAN_WORKER': self.worker_id,
+            })
+            
+            # Build command to execute task
+            # This could be configurable, but for now use a simple approach
+            cmd = self._build_execution_command(task)
+            
+            logger.info(f"Spawning worker for task {task.id}: {' '.join(cmd)}")
+            
+            # Start process with output redirect to avoid deadlock
+            import tempfile
+            with tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix='.log') as temp_log:
+                temp_log_path = temp_log.name
+            
+            process = subprocess.Popen(
+                cmd,
+                env=env,
+                stdout=open(temp_log_path, 'w'),
+                stderr=subprocess.STDOUT,
+                text=True
+            )
+            
+            # Store log path for later cleanup
+            if not hasattr(self, '_temp_logs'):
+                self._temp_logs = {}
+            self._temp_logs[task.id] = temp_log_path
+            
+            # Track the running task
+            self.running_tasks[task.id] = process
+            
+            return True
+            
+        except Exception as e:
+            logger.error(f"Failed to spawn worker for task {task.id}: {e}")
+            return False
+    
+    def _build_execution_command(self, task: Any) -> List[str]:
+        """
+        Build the command to execute a kanban task.
+        
+        This is a simplified implementation. In a full system, this would:
+        1. Parse task.body for specific execution instructions
+        2. Route to different executors based on task metadata
+        3. Support various agent configurations
+        
+        Args:
+            task: Task to build command for
+            
+        Returns:
+            Command as list of strings
+        """
+        # Default: use praisonai agent with the task as a prompt
+        return [
+            'praisonai',
+            'agent',
+            'run',
+            '--name', f'kanban_worker_{task.id}',
+            '--prompt', f'Task: {task.title}\n\nDetails: {task.body}',
+            '--tools', 'kanban_heartbeat,kanban_comment',
+            '--output-mode', 'quiet'
+        ]
+    
+    def _cleanup_completed_tasks(self, store: Any):
+        """
+        Clean up completed task processes and update their status.
+        
+        Args:
+            store: Kanban store instance
+        """
+        completed = []
+        
+        for task_id, process in self.running_tasks.items():
+            poll_result = process.poll()
+            
+            if poll_result is not None:
+                # Process completed
+                completed.append(task_id)
+                
+                try:
+                    # Get return code
+                    return_code = process.returncode
+                    
+                    # Read output from temp log file
+                    stdout_data = ""
+                    if hasattr(self, '_temp_logs') and task_id in self._temp_logs:
+                        try:
+                            with open(self._temp_logs[task_id], 'r') as f:
+                                stdout_data = f.read()
+                        except Exception as e:
+                            logger.warning(f"Failed to read log for task {task_id}: {e}")
+                            stdout_data = f"<log read error: {e}>"
+                    else:
+                        stdout_data = "<no log available>"
+                    
+                    # Update task based on exit code
+                    if return_code == 0:
+                        # Success - mark as done
+                        store.move_task(task_id, 'done')
+                        store.add_comment(
+                            task_id, 
+                            self.worker_id, 
+                            f"Task completed successfully\n\nOutput:\n{stdout_data[:500]}"
+                        )
+                        
+                        # Fire completion hook
+                        task = store.get_task(task_id)
+                        self._fire_hook_event('KANBAN_TASK_COMPLETED', {
+                            'task_id': task_id,
+                            'worker_id': self.worker_id,
+                            'return_code': return_code,
+                            'task': task.to_dict() if task else {}
+                        })
+                        
+                        logger.info(f"Task {task_id} completed successfully")
+                    else:
+                        # Failed - mark as blocked
+                        store.move_task(task_id, 'blocked')
+                        store.add_comment(
+                            task_id,
+                            self.worker_id,
+                            f"Task failed with exit code {return_code}\n\nOutput:\n{stdout_data[:500]}"
+                        )
+                        
+                        # Fire failure hook
+                        task = store.get_task(task_id)
+                        self._fire_hook_event('KANBAN_TASK_FAILED', {
+                            'task_id': task_id,
+                            'worker_id': self.worker_id,
+                            'return_code': return_code,
+                            'error': stdout_data[:500],
+                            'task': task.to_dict() if task else {}
+                        })
+                        
+                        logger.warning(f"Task {task_id} failed with code {return_code}")
+                
+                except Exception as e:
+                    logger.error(f"Error processing completed task {task_id}: {e}")
+                    # Release claim as fallback
+                    try:
+                        store.release_claim(task_id, self.worker_id)
+                    except:
+                        pass
+        
+        # Remove completed tasks from tracking and clean up temp logs
+        for task_id in completed:
+            del self.running_tasks[task_id]
+            
+            # Clean up temporary log file
+            if hasattr(self, '_temp_logs') and task_id in self._temp_logs:
+                try:
+                    import os
+                    os.unlink(self._temp_logs[task_id])
+                except Exception as e:
+                    logger.warning(f"Failed to cleanup temp log for task {task_id}: {e}")
+                finally:
+                    del self._temp_logs[task_id]
+    
+    async def run_forever(self):
+        """
+        Run the dispatcher loop forever.
+        """
+        logger.info(f"Starting kanban dispatcher (max_concurrent={self.max_concurrent})")
+        
+        while True:
+            try:
+                await self.dispatch_once()
+                await asyncio.sleep(self.poll_interval)
+                
+            except asyncio.CancelledError:
+                logger.info("Kanban dispatcher cancelled")
+                break
+            except Exception as e:
+                logger.error(f"Dispatcher error: {e}")
+                await asyncio.sleep(self.poll_interval)
+        
+        # Cleanup on exit
+        await self._shutdown()
+    
+    async def _shutdown(self):
+        """Clean shutdown - wait for running tasks."""
+        if self.running_tasks:
+            logger.info(f"Waiting for {len(self.running_tasks)} running tasks to complete...")
+            
+            # Give tasks 30 seconds to complete gracefully
+            timeout = 30
+            start_time = time.time()
+            
+            while self.running_tasks and (time.time() - start_time) < timeout:
+                store = self._get_kanban_store()
+                if store:
+                    self._cleanup_completed_tasks(store)
+                await asyncio.sleep(1)
+            
+            # Force terminate remaining tasks
+            for task_id, process in self.running_tasks.items():
+                try:
+                    logger.warning(f"Force terminating task {task_id}")
+                    process.terminate()
+                    process.wait(timeout=5)
+                except:
+                    process.kill()
+            
+            self.running_tasks.clear()
+        
+        logger.info("Kanban dispatcher shutdown complete")
+
+
+# Global dispatcher management functions
+
+async def start_kanban_dispatcher(max_concurrent: int = 3, poll_interval: float = 5.0):
+    """
+    Start the global kanban dispatcher.
+    
+    Args:
+        max_concurrent: Maximum concurrent task executions
+        poll_interval: Seconds between polling cycles
+    """
+    global _dispatcher_running, _dispatcher_task
+    
+    if _dispatcher_running:
+        logger.warning("Kanban dispatcher already running")
+        return
+    
+    # Check if disabled by environment
+    if not _is_dispatcher_enabled():
+        logger.info("Kanban dispatcher disabled by PRAISONAI_KANBAN_DISPATCH=0")
+        return
+    
+    dispatcher = KanbanDispatcher(max_concurrent, poll_interval)
+    
+    _dispatcher_task = asyncio.create_task(dispatcher.run_forever())
+    _dispatcher_running = True
+    
+    logger.info("Kanban dispatcher started")
+
+
+async def stop_kanban_dispatcher():
+    """Stop the global kanban dispatcher."""
+    global _dispatcher_running, _dispatcher_task
+    
+    if not _dispatcher_running or _dispatcher_task is None:
+        return
+    
+    _dispatcher_task.cancel()
+    
+    try:
+        await _dispatcher_task
+    except asyncio.CancelledError:
+        pass
+    
+    _dispatcher_running = False
+    _dispatcher_task = None
+    
+    logger.info("Kanban dispatcher stopped")
+
+
+def is_dispatcher_running() -> bool:
+    """Check if the kanban dispatcher is running."""
+    return _dispatcher_running
+
+
+def _is_dispatcher_enabled() -> bool:
+    """Check if kanban dispatcher is enabled via environment variable."""
+    return os.environ.get('PRAISONAI_KANBAN_DISPATCH', '1').strip().lower() not in ('0', 'false', 'no')
+
+
+# Manual dispatch function for CLI/testing
+
+async def dispatch_once(max_spawn: int = 3) -> int:
+    """
+    Manually trigger a single dispatch cycle.
+    
+    Args:
+        max_spawn: Maximum tasks to spawn in this cycle
+        
+    Returns:
+        Number of tasks spawned
+    """
+    dispatcher = KanbanDispatcher(max_spawn)
+    return await dispatcher.dispatch_once()

src/praisonai/praisonai/integration/bridges/kanban_bridge.py

@@ -0,0 +1,113 @@
+"""Wire kanban store and jobs components for PraisonAIUI integration."""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+
+def get_kanban_store_factory():
+    """Return kanban store factory for PraisonAIUI injection."""
+    try:
+        from praisonai.kanban.sqlite_store import SQLiteKanbanStore
+        
+        def _kanban_store_factory():
+            """Create SQLite kanban store instance."""
+            # Reads PRAISONAI_KANBAN_BOARD env var for board selection
+            return SQLiteKanbanStore()
+        
+        return _kanban_store_factory
+    except ImportError:
+        return None
+
+
+def get_jobs_store():
+    """Return jobs store for PraisonAIUI injection."""
+    try:
+        from praisonai.jobs.store import get_store
+        return get_store
+    except ImportError:
+        return None
+
+
+def get_jobs_executor():
+    """Return jobs executor for PraisonAIUI injection."""
+    try:
+        from praisonai.jobs.executor import get_executor
+        return get_executor
+    except ImportError:
+        return None
+
+
+def register_kanban_backends():
+    """Register kanban-related backends with PraisonAIUI."""
+    import logging
+    
+    log = logging.getLogger(__name__)
+    
+    try:
+        import praisonaiui.backends as backends
+        
+        # Kanban store factory
+        kanban_factory = get_kanban_store_factory()
+        if kanban_factory is not None:
+            backends.set_backend("kanban_store", kanban_factory)
+            log.debug("kanban_store backend registered")
+        
+        # Jobs store
+        jobs_store = get_jobs_store()
+        if jobs_store is not None:
+            backends.set_backend("jobs_store", jobs_store)
+            log.debug("jobs_store backend registered")
+        
+        # Jobs executor
+        jobs_executor = get_jobs_executor()
+        if jobs_executor is not None:
+            backends.set_backend("jobs_executor", jobs_executor)
+            log.debug("jobs_executor backend registered")
+        
+        # Register kanban dispatcher startup handler
+        register_kanban_startup_handler()
+            
+        return True
+        
+    except Exception as exc:
+        log.debug("kanban backend registration failed: %s", exc)
+        return False
+
+
+def register_kanban_startup_handler():
+    """Register kanban dispatcher startup handler with the ASGI app."""
+    import logging
+    
+    log = logging.getLogger(__name__)
+    
+    try:
+        import praisonaiui.backends as backends
+        from praisonai.gateway.kanban_dispatcher import start_kanban_dispatcher, is_dispatcher_running
+        
+        async def _startup_handler():
+            """Startup handler for kanban dispatcher."""
+            try:
+                if not is_dispatcher_running():
+                    await start_kanban_dispatcher()
+                    log.info("kanban dispatcher started")
+            except Exception as exc:
+                log.warning("failed to start kanban dispatcher: %s", exc)
+        
+        # Register startup handler if the backend supports it
+        if hasattr(backends, 'register_startup_handler'):
+            backends.register_startup_handler(_startup_handler)
+            log.debug("kanban dispatcher startup handler registered")
+        else:
+            # Fallback: try immediate start if event loop is running
+            try:
+                import asyncio
+                loop = asyncio.get_running_loop()
+                if loop and not is_dispatcher_running():
+                    asyncio.create_task(start_kanban_dispatcher())
+                    log.debug("kanban dispatcher started via task")
+            except RuntimeError:
+                log.debug("kanban dispatcher deferred (no running event loop)")
+                
+    except Exception as exc:
+        log.debug("kanban startup handler registration failed: %s", exc)

src/praisonai/praisonai/integration/host_app.py

@@ -165,7 +165,12 @@ def setup_bridges() -> None:
 
         backends.set_backend("approvals_pending", list_pending_approvals)
         backends.set_backend("approvals_policies", get_approval_policies)
-    except ImportError as exc:
+        
+        # Kanban and jobs backends
+        from praisonai.integration.bridges.kanban_bridge import register_kanban_backends
+        register_kanban_backends()
+        
+    except Exception as exc:
         log.debug("aiui backend injection failed: %s", exc)
     except Exception as exc:
         log.warning("aiui backend injection failed: %s", exc)

src/praisonai/praisonai/integrations/managed_local.py

@@ -599,9 +599,11 @@ class LocalManagedAgent:
                 subprocess.run(cmd, check=True, capture_output=True, timeout=120)
                 logger.info("[local_managed] host pip install completed")
             except subprocess.CalledProcessError as e:
-                logger.warning("[local_managed] pip install failed: %s", e.stderr)
-            except subprocess.TimeoutExpired:
-                logger.warning("[local_managed] pip install timed out")
+                raise RuntimeError(
+                    f"pip install failed for {pip_pkgs}: {e.stderr.decode(errors='replace')}"
+                ) from e
+            except subprocess.TimeoutExpired as e:
+                raise RuntimeError(f"pip install timed out after 120s for {pip_pkgs}") from e
         else:
             # Sandbox installation via compute provider
             self._install_packages_in_compute(pip_pkgs)
@@ -629,10 +631,14 @@ class LocalManagedAgent:
             if result.get("exit_code", 0) == 0:
                 logger.info("[local_managed] compute pip install completed")
             else:
-                logger.warning("[local_managed] compute pip install failed: %s", result.get("stderr", ""))
+                raise RuntimeError(
+                    f"pip install failed in compute for {pip_pkgs}: {result.get('stderr', '')}"
+                )
                 
         except Exception as e:
-            logger.warning("[local_managed] compute pip install error: %s", e)
+            if "pip install failed in compute" in str(e):
+                raise  # Re-raise RuntimeError from above
+            raise RuntimeError(f"pip install error in compute for {pip_pkgs}: {e}") from e
 
     def _ensure_agent(self) -> Any:
         """Create or return the inner PraisonAI Agent."""

src/praisonai/praisonai/kanban/__init__.py

@@ -0,0 +1,12 @@
+# Kanban module for PraisonAI - persistent task management
+from .sqlite_store import SQLiteKanbanStore
+from .models import TaskStatus, Task, TaskEvent
+from .paths import get_kanban_db_path
+
+__all__ = [
+    'SQLiteKanbanStore',
+    'TaskStatus', 
+    'Task',
+    'TaskEvent',
+    'get_kanban_db_path'
+]

src/praisonai/praisonai/kanban/models.py

@@ -0,0 +1,198 @@
+"""Task dataclasses and models for kanban system."""
+import json
+from dataclasses import dataclass, asdict
+from datetime import datetime
+from enum import Enum
+from typing import Optional, Dict, Any, List
+
+
+class TaskStatus(str, Enum):
+    """Task status enum matching Hermes v1."""
+    TRIAGE = "triage"
+    TODO = "todo"
+    SCHEDULED = "scheduled"
+    READY = "ready"
+    RUNNING = "running"
+    BLOCKED = "blocked"
+    REVIEW = "review"
+    DONE = "done"
+    ARCHIVED = "archived"
+
+
+@dataclass
+class Task:
+    """Task model for kanban system."""
+    id: str
+    title: str
+    body: str = ""
+    status: TaskStatus = TaskStatus.TODO
+    assignee: str = ""
+    priority: int = 0  # Higher number = higher priority
+    tenant: str = "default"
+    board: str = "default"
+    workspace_kind: str = "default"
+    claim_lock: Optional[str] = None
+    created_at: Optional[datetime] = None
+    updated_at: Optional[datetime] = None
+    metadata: Optional[Dict[str, Any]] = None
+
+    def __post_init__(self):
+        from datetime import timezone
+        if self.created_at is None:
+            self.created_at = datetime.now(timezone.utc)
+        if self.updated_at is None:
+            self.updated_at = datetime.now(timezone.utc)
+        if self.metadata is None:
+            self.metadata = {}
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary for API/JSON serialization."""
+        result = asdict(self)
+        result['created_at'] = self.created_at.isoformat() if self.created_at else None
+        result['updated_at'] = self.updated_at.isoformat() if self.updated_at else None
+        result['status'] = self.status.value
+        result['metadata'] = json.dumps(self.metadata) if self.metadata else None
+        return result
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> 'Task':
+        """Create from dictionary."""
+        # Remove fields that don't belong to Task model
+        filtered_data = data.copy()
+        filtered_data.pop('version', None)  # Remove version field used for optimistic locking
+        
+        # Handle datetime fields
+        if 'created_at' in filtered_data and filtered_data['created_at']:
+            filtered_data['created_at'] = datetime.fromisoformat(filtered_data['created_at'].replace('Z', '+00:00'))
+        if 'updated_at' in filtered_data and filtered_data['updated_at']:
+            filtered_data['updated_at'] = datetime.fromisoformat(filtered_data['updated_at'].replace('Z', '+00:00'))
+        
+        # Handle status enum
+        if 'status' in filtered_data:
+            filtered_data['status'] = TaskStatus(filtered_data['status'])
+        
+        # Handle metadata JSON
+        if 'metadata' in filtered_data and filtered_data['metadata']:
+            try:
+                filtered_data['metadata'] = json.loads(filtered_data['metadata']) if isinstance(filtered_data['metadata'], str) else filtered_data['metadata']
+            except (json.JSONDecodeError, TypeError):
+                filtered_data['metadata'] = {}
+        
+        return cls(**filtered_data)
+
+
+@dataclass
+class TaskLink:
+    """Task dependency link (DAG)."""
+    parent_id: str
+    child_id: str
+    created_at: Optional[datetime] = None
+
+    def __post_init__(self):
+        if self.created_at is None:
+            self.created_at = datetime.utcnow()
+
+
+@dataclass
+class TaskComment:
+    """Task comment model."""
+    id: str
+    task_id: str
+    author: str
+    text: str
+    created_at: Optional[datetime] = None
+
+    def __post_init__(self):
+        if self.created_at is None:
+            self.created_at = datetime.utcnow()
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary."""
+        return {
+            'id': self.id,
+            'task_id': self.task_id,
+            'author': self.author,
+            'text': self.text,
+            'created_at': self.created_at.isoformat() if self.created_at else None
+        }
+
+
+@dataclass
+class TaskEvent:
+    """Task audit event for SSE/WebSocket."""
+    id: str
+    task_id: str
+    event_type: str  # created, updated, moved, deleted, commented
+    data: Dict[str, Any]
+    created_at: Optional[datetime] = None
+
+    def __post_init__(self):
+        if self.created_at is None:
+            self.created_at = datetime.utcnow()
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary."""
+        return {
+            'id': self.id,
+            'task_id': self.task_id,
+            'event_type': self.event_type,
+            'data': self.data,
+            'created_at': self.created_at.isoformat() if self.created_at else None
+        }
+
+
+# Protocol placeholder - will import from praisonaiagents when available
+class KanbanStoreProtocol:
+    """Minimal kanban store protocol - matches praisonaiagents when available."""
+    
+    def create_task(self, task_data: Dict[str, Any]) -> Task:
+        """Create a new task."""
+        ...
+    
+    def get_task(self, task_id: str) -> Optional[Task]:
+        """Get task by ID."""
+        ...
+    
+    def update_task(self, task_id: str, updates: Dict[str, Any]) -> Task:
+        """Update task."""
+        ...
+    
+    def delete_task(self, task_id: str) -> bool:
+        """Delete task."""
+        ...
+    
+    def list_tasks(self, filters: Optional[Dict[str, Any]] = None) -> List[Task]:
+        """List tasks with optional filters."""
+        ...
+    
+    def move_task(self, task_id: str, status: str) -> Task:
+        """Move task to new status."""
+        ...
+    
+    def get_board(self, board: str = "default") -> Dict[str, Any]:
+        """Get board layout for UI."""
+        ...
+    
+    def bulk_update(self, operations: List[Dict[str, Any]]) -> List[Task]:
+        """Bulk update operations."""
+        ...
+    
+    def add_comment(self, task_id: str, author: str, text: str) -> TaskComment:
+        """Add comment to task."""
+        ...
+    
+    def get_comments(self, task_id: str) -> List[TaskComment]:
+        """Get task comments."""
+        ...
+    
+    def add_link(self, parent_id: str, child_id: str) -> TaskLink:
+        """Create task dependency."""
+        ...
+    
+    def remove_link(self, parent_id: str, child_id: str) -> bool:
+        """Remove task dependency."""
+        ...
+    
+    def list_events(self, since: Optional[datetime] = None) -> List[TaskEvent]:
+        """List events for SSE."""
+        ...

src/praisonai/praisonai/kanban/paths.py

@@ -0,0 +1,71 @@
+"""Path utilities for kanban data storage."""
+import os
+from pathlib import Path
+from typing import Optional
+
+
+def get_praisonai_home() -> Path:
+    """Get PraisonAI home directory."""
+    return Path.home() / ".praisonai"
+
+
+def get_kanban_db_path(board: Optional[str] = None) -> Path:
+    """Get kanban database path.
+    
+    Args:
+        board: Board slug for multi-board support. None for default board.
+        
+    Returns:
+        Path to SQLite database file.
+    """
+    home = get_praisonai_home()
+    
+    # Read from environment variable for override
+    if board is None:
+        board = os.environ.get('PRAISONAI_KANBAN_BOARD')
+        if not board:
+            # Try to read from persisted config file
+            try:
+                import json
+                config_file = home / "kanban_config.json"
+                if config_file.exists():
+                    with open(config_file) as f:
+                        config = json.load(f)
+                        board = config.get('active_board', 'default')
+                else:
+                    board = 'default'
+            except Exception:
+                board = 'default'
+    
+    # Legacy override for single DB file
+    if db_override := os.environ.get('PRAISONAI_KANBAN_DB'):
+        return Path(db_override)
+    
+    # Default board at root
+    if board == 'default':
+        return home / "kanban.db"
+    
+    # Multi-board at boards/<slug>/
+    boards_dir = home / "kanban" / "boards" / board
+    boards_dir.mkdir(parents=True, exist_ok=True)
+    return boards_dir / "kanban.db"
+
+
+def get_kanban_boards_dir() -> Path:
+    """Get kanban boards directory."""
+    return get_praisonai_home() / "kanban" / "boards"
+
+
+def list_available_boards() -> list[str]:
+    """List all available board slugs."""
+    boards_dir = get_kanban_boards_dir()
+    if not boards_dir.exists():
+        return ["default"]
+    
+    boards = ["default"]  # Always include default
+    
+    for board_dir in boards_dir.iterdir():
+        if board_dir.is_dir() and (board_dir / "kanban.db").exists():
+            boards.append(board_dir.name)
+    
+    return sorted(set(boards))

src/praisonai/praisonai/kanban/sqlite_store.py

@@ -0,0 +1,565 @@
+"""SQLite kanban store implementation."""
+import sqlite3
+import json
+import uuid
+from contextlib import contextmanager
+from datetime import datetime
+from typing import Optional, Dict, Any, List
+from pathlib import Path
+
+from .models import Task, TaskStatus, TaskComment, TaskLink, TaskEvent, KanbanStoreProtocol
+from .paths import get_kanban_db_path
+
+
+class SQLiteKanbanStore:
+    """SQLite implementation of kanban store with WAL mode and CAS operations."""
+    
+    def __init__(self, board: Optional[str] = None):
+        """Initialize store.
+        
+        Args:
+            board: Board slug. Uses PRAISONAI_KANBAN_BOARD env var or 'default'.
+        """
+        self.board = board
+        self.db_path = get_kanban_db_path(board)
+        
+        # Ensure db_path is a Path object
+        if not isinstance(self.db_path, Path):
+            self.db_path = Path(self.db_path)
+            
+        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._init_db()
+
+    def _init_db(self):
+        """Initialize database with schema."""
+        with self._get_connection() as conn:
+            # Enable WAL mode for better concurrency
+            conn.execute("PRAGMA journal_mode=WAL")
+            conn.execute("PRAGMA foreign_keys=ON")
+            
+            # Tasks table
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS tasks (
+                    id TEXT PRIMARY KEY,
+                    title TEXT NOT NULL,
+                    body TEXT DEFAULT '',
+                    status TEXT DEFAULT 'todo',
+                    assignee TEXT DEFAULT '',
+                    priority INTEGER DEFAULT 0,
+                    tenant TEXT DEFAULT 'default',
+                    board TEXT DEFAULT 'default',
+                    workspace_kind TEXT DEFAULT 'default',
+                    claim_lock TEXT,
+                    metadata TEXT,  -- JSON blob
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    version INTEGER DEFAULT 1  -- For optimistic locking
+                )
+            """)
+            
+            # Task links table (DAG)
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS task_links (
+                    parent_id TEXT NOT NULL,
+                    child_id TEXT NOT NULL,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    PRIMARY KEY (parent_id, child_id),
+                    FOREIGN KEY (parent_id) REFERENCES tasks(id) ON DELETE CASCADE,
+                    FOREIGN KEY (child_id) REFERENCES tasks(id) ON DELETE CASCADE
+                )
+            """)
+            
+            # Task comments table
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS task_comments (
+                    id TEXT PRIMARY KEY,
+                    task_id TEXT NOT NULL,
+                    author TEXT NOT NULL,
+                    text TEXT NOT NULL,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    FOREIGN KEY (task_id) REFERENCES tasks(id) ON DELETE CASCADE
+                )
+            """)
+            
+            # Task events table (audit log)
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS task_events (
+                    id TEXT PRIMARY KEY,
+                    task_id TEXT NOT NULL,
+                    event_type TEXT NOT NULL,
+                    data TEXT NOT NULL,  -- JSON blob
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                )
+            """)
+            
+            # Indexes for performance
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status)")
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_tasks_assignee ON tasks(assignee)")
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_tasks_board ON tasks(board)")
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_tasks_priority ON tasks(priority DESC)")
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_task_comments_task ON task_comments(task_id)")
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_task_events_task ON task_events(task_id)")
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_task_events_created ON task_events(created_at)")
+
+    @contextmanager
+    def _get_connection(self):
+        """Get database connection with proper cleanup."""
+        conn = sqlite3.connect(self.db_path, timeout=30.0)
+        conn.row_factory = sqlite3.Row
+        # Enable foreign key constraints for this connection
+        conn.execute("PRAGMA foreign_keys=ON")
+        try:
+            yield conn
+            conn.commit()
+        except Exception:
+            conn.rollback()
+            raise
+        finally:
+            conn.close()
+
+    def _generate_id(self) -> str:
+        """Generate unique task ID."""
+        return f"task_{uuid.uuid4().hex[:12]}"
+
+    def _log_event(self, conn: sqlite3.Connection, task_id: str, event_type: str, data: Dict[str, Any]):
+        """Log audit event."""
+        event_id = f"event_{uuid.uuid4().hex[:12]}"
+        conn.execute(
+            "INSERT INTO task_events (id, task_id, event_type, data) VALUES (?, ?, ?, ?)",
+            (event_id, task_id, event_type, json.dumps(data))
+        )
+
+    def create_task(self, task_data: Dict[str, Any]) -> Task:
+        """Create a new task."""
+        task_id = task_data.get('id', self._generate_id())
+        now = datetime.utcnow()
+        
+        task = Task(
+            id=task_id,
+            title=task_data['title'],
+            body=task_data.get('body', ''),
+            status=TaskStatus(task_data.get('status', 'todo')),
+            assignee=task_data.get('assignee', ''),
+            priority=task_data.get('priority', 0),
+            tenant=task_data.get('tenant', 'default'),
+            board=task_data.get('board', self.board or 'default'),
+            workspace_kind=task_data.get('workspace_kind', 'default'),
+            metadata=task_data.get('metadata', {}),
+            created_at=now,
+            updated_at=now
+        )
+        
+        with self._get_connection() as conn:
+            conn.execute("""
+                INSERT INTO tasks (
+                    id, title, body, status, assignee, priority, 
+                    tenant, board, workspace_kind, metadata, 
+                    created_at, updated_at
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """, (
+                task.id, task.title, task.body, task.status.value,
+                task.assignee, task.priority, task.tenant, task.board,
+                task.workspace_kind, json.dumps(task.metadata),
+                task.created_at.isoformat(), task.updated_at.isoformat()
+            ))
+            
+            self._log_event(conn, task_id, 'created', task.to_dict())
+        
+        return task
+
+    def get_task(self, task_id: str) -> Optional[Task]:
+        """Get task by ID."""
+        with self._get_connection() as conn:
+            cursor = conn.execute("SELECT * FROM tasks WHERE id = ?", (task_id,))
+            row = cursor.fetchone()
+            
+            if not row:
+                return None
+            
+            return Task.from_dict(dict(row))
+
+    def update_task(self, task_id: str, updates: Dict[str, Any]) -> Task:
+        """Update task with optimistic locking."""
+        with self._get_connection() as conn:
+            # Get current version for CAS
+            cursor = conn.execute("SELECT version FROM tasks WHERE id = ?", (task_id,))
+            row = cursor.fetchone()
+            if not row:
+                raise ValueError(f"Task {task_id} not found")
+            
+            current_version = row['version']
+            new_version = current_version + 1
+            now = datetime.utcnow()
+            
+            # Build update query dynamically
+            update_fields = []
+            values = []
+            
+            for field, value in updates.items():
+                if field in ['status', 'title', 'body', 'assignee', 'priority', 'claim_lock', 'metadata']:
+                    if field == 'status' and isinstance(value, str):
+                        value = TaskStatus(value).value
+                    elif field == 'metadata':
+                        value = json.dumps(value) if value else None
+                    update_fields.append(f"{field} = ?")
+                    values.append(value)
+            
+            if not update_fields:
+                # No valid updates, return current task
+                return self.get_task(task_id)
+            
+            update_fields.append("updated_at = ?")
+            update_fields.append("version = ?")
+            values.extend([now.isoformat(), new_version, task_id, current_version])
+            
+            # CAS update with version check
+            result = conn.execute(f"""
+                UPDATE tasks 
+                SET {', '.join(update_fields)}
+                WHERE id = ? AND version = ?
+            """, values)
+            
+            if result.rowcount == 0:
+                raise ValueError(f"Task {task_id} was modified by another process")
+            
+            # Log event
+            self._log_event(conn, task_id, 'updated', updates)
+            
+            return self.get_task(task_id)
+
+    def delete_task(self, task_id: str) -> bool:
+        """Delete task."""
+        with self._get_connection() as conn:
+            # Log before deletion
+            self._log_event(conn, task_id, 'deleted', {'task_id': task_id})
+            
+            result = conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
+            return result.rowcount > 0
+
+    def list_tasks(self, filters: Optional[Dict[str, Any]] = None) -> List[Task]:
+        """List tasks with optional filters."""
+        query = "SELECT * FROM tasks WHERE 1=1"
+        values = []
+        
+        if filters:
+            if 'status' in filters:
+                if isinstance(filters['status'], list):
+                    placeholders = ','.join(['?'] * len(filters['status']))
+                    query += f" AND status IN ({placeholders})"
+                    values.extend(filters['status'])
+                else:
+                    query += " AND status = ?"
+                    values.append(filters['status'])
+            
+            if 'assignee' in filters:
+                query += " AND assignee = ?"
+                values.append(filters['assignee'])
+            
+            if 'board' in filters:
+                query += " AND board = ?"
+                values.append(filters['board'])
+            
+            if 'tenant' in filters:
+                query += " AND tenant = ?"
+                values.append(filters['tenant'])
+        
+        query += " ORDER BY priority DESC, created_at ASC"
+        
+        with self._get_connection() as conn:
+            cursor = conn.execute(query, values)
+            return [Task.from_dict(dict(row)) for row in cursor.fetchall()]
+
+    def _get_task_with_conn(self, task_id: str, conn: sqlite3.Connection) -> Optional[Task]:
+        """Get task using existing connection to avoid nesting."""
+        cursor = conn.execute(
+            "SELECT * FROM tasks WHERE id = ?", (task_id,)
+        )
+        row = cursor.fetchone()
+        if row:
+            return Task.from_dict(dict(row))
+        return None
+    
+    def _update_task_with_conn(self, task_id: str, updates: Dict[str, Any], conn: sqlite3.Connection) -> Task:
+        """Update task using existing connection to avoid nesting."""
+        from datetime import timezone
+        
+        if not updates:
+            return self._get_task_with_conn(task_id, conn)
+        
+        # Build update query
+        set_clauses = []
+        params = []
+        
+        for key, value in updates.items():
+            if key in ['id', 'created_at']:  # Don't allow updating immutable fields
+                continue
+            set_clauses.append(f"{key} = ?")
+            params.append(value)
+        
+        if not set_clauses:
+            return self._get_task_with_conn(task_id, conn)
+        
+        # Always update timestamp and version
+        set_clauses.append("updated_at = ?")
+        set_clauses.append("version = version + 1")
+        params.append(datetime.now(timezone.utc).isoformat())
+        params.append(task_id)
+        
+        query = f"UPDATE tasks SET {', '.join(set_clauses)} WHERE id = ?"
+        cursor = conn.execute(query, params)
+        
+        if cursor.rowcount == 0:
+            raise ValueError(f"Task {task_id} not found")
+        
+        # Log the update
+        self._log_event(conn, task_id, 'updated', updates)
+        
+        return self._get_task_with_conn(task_id, conn)
+
+    def move_task(self, task_id: str, status: str) -> Task:
+        """Move task to new status with parent/child promotion logic."""
+        with self._get_connection() as conn:
+            # Check if task exists
+            task = self._get_task_with_conn(task_id, conn)
+            if not task:
+                raise ValueError(f"Task {task_id} not found")
+            
+            new_status = TaskStatus(status)
+            
+            # Check for blocking parents if moving to ready
+            if new_status == TaskStatus.READY:
+                cursor = conn.execute("""
+                    SELECT COUNT(*) as incomplete_parents
+                    FROM task_links tl
+                    JOIN tasks t ON t.id = tl.parent_id
+                    WHERE tl.child_id = ? AND t.status NOT IN ('done', 'archived')
+                """, (task_id,))
+                
+                if cursor.fetchone()['incomplete_parents'] > 0:
+                    raise ValueError("Cannot move to ready: incomplete parent tasks")
+            
+            # Update task status
+            updated_task = self._update_task_with_conn(task_id, {'status': status}, conn)
+            
+            # Promote children if moving to done
+            if new_status == TaskStatus.DONE:
+                cursor = conn.execute("""
+                    SELECT child_id FROM task_links WHERE parent_id = ?
+                """, (task_id,))
+                
+                for row in cursor.fetchall():
+                    child_id = row['child_id']
+                    child_task = self._get_task_with_conn(child_id, conn)
+                    
+                    if child_task and child_task.status == TaskStatus.TODO:
+                        # Check if all other parents are done
+                        parent_check = conn.execute("""
+                            SELECT COUNT(*) as incomplete_parents
+                            FROM task_links tl
+                            JOIN tasks t ON t.id = tl.parent_id
+                            WHERE tl.child_id = ? AND t.status NOT IN ('done', 'archived')
+                        """, (child_id,))
+                        
+                        if parent_check.fetchone()['incomplete_parents'] == 0:
+                            self._update_task_with_conn(child_id, {'status': TaskStatus.READY.value}, conn)
+            
+            self._log_event(conn, task_id, 'moved', {
+                'old_status': task.status.value,
+                'new_status': status
+            })
+            
+            return updated_task
+
+    def get_board(self, board: str = "default") -> Dict[str, Any]:
+        """Get board layout for UI."""
+        tasks = self.list_tasks({'board': board})
+        
+        # Group by status for kanban columns
+        columns = {}
+        for status in TaskStatus:
+            columns[status.value] = []
+        
+        for task in tasks:
+            if task.status.value in columns:
+                columns[task.status.value].append(task.to_dict())
+        
+        return {
+            'board': board,
+            'columns': columns,
+            'task_count': len(tasks)
+        }
+
+    def bulk_update(self, operations: List[Dict[str, Any]]) -> List[Task]:
+        """Bulk update operations."""
+        results = []
+        
+        with self._get_connection() as conn:
+            for op in operations:
+                op_type = op.get('operation')
+                
+                if op_type == 'update':
+                    task = self.update_task(op['task_id'], op['updates'])
+                    results.append(task)
+                elif op_type == 'move':
+                    task = self.move_task(op['task_id'], op['status'])
+                    results.append(task)
+                elif op_type == 'delete':
+                    if self.delete_task(op['task_id']):
+                        results.append({'deleted': op['task_id']})
+        
+        return results
+
+    def add_comment(self, task_id: str, author: str, text: str) -> TaskComment:
+        """Add comment to task."""
+        comment_id = f"comment_{uuid.uuid4().hex[:12]}"
+        comment = TaskComment(
+            id=comment_id,
+            task_id=task_id,
+            author=author,
+            text=text
+        )
+        
+        with self._get_connection() as conn:
+            conn.execute("""
+                INSERT INTO task_comments (id, task_id, author, text, created_at)
+                VALUES (?, ?, ?, ?, ?)
+            """, (
+                comment.id, comment.task_id, comment.author, 
+                comment.text, comment.created_at.isoformat()
+            ))
+            
+            self._log_event(conn, task_id, 'commented', {
+                'author': author,
+                'text': text[:100] + '...' if len(text) > 100 else text
+            })
+        
+        return comment
+
+    def get_comments(self, task_id: str) -> List[TaskComment]:
+        """Get task comments."""
+        with self._get_connection() as conn:
+            cursor = conn.execute("""
+                SELECT * FROM task_comments 
+                WHERE task_id = ? 
+                ORDER BY created_at ASC
+            """, (task_id,))
+            
+            comments = []
+            for row in cursor.fetchall():
+                comment_data = dict(row)
+                comment_data['created_at'] = datetime.fromisoformat(comment_data['created_at'])
+                comments.append(TaskComment(**comment_data))
+            
+            return comments
+
+    def add_link(self, parent_id: str, child_id: str) -> TaskLink:
+        """Create task dependency."""
+        link = TaskLink(parent_id=parent_id, child_id=child_id)
+        
+        with self._get_connection() as conn:
+            # Check for cycles (simplified cycle detection)
+            cursor = conn.execute("""
+                WITH RECURSIVE cycle_check AS (
+                    SELECT parent_id, child_id, 1 as depth
+                    FROM task_links 
+                    WHERE parent_id = ?
+                    
+                    UNION ALL
+                    
+                    SELECT tl.parent_id, tl.child_id, cc.depth + 1
+                    FROM task_links tl
+                    JOIN cycle_check cc ON cc.child_id = tl.parent_id
+                    WHERE cc.depth < 10
+                )
+                SELECT COUNT(*) as cycle_found
+                FROM cycle_check
+                WHERE child_id = ?
+            """, (child_id, parent_id))
+            
+            if cursor.fetchone()['cycle_found'] > 0:
+                raise ValueError("Would create dependency cycle")
+            
+            conn.execute("""
+                INSERT OR IGNORE INTO task_links (parent_id, child_id, created_at)
+                VALUES (?, ?, ?)
+            """, (parent_id, child_id, link.created_at.isoformat()))
+            
+            self._log_event(conn, child_id, 'linked', {
+                'parent_id': parent_id,
+                'child_id': child_id
+            })
+        
+        return link
+
+    def remove_link(self, parent_id: str, child_id: str) -> bool:
+        """Remove task dependency."""
+        with self._get_connection() as conn:
+            result = conn.execute("""
+                DELETE FROM task_links 
+                WHERE parent_id = ? AND child_id = ?
+            """, (parent_id, child_id))
+            
+            if result.rowcount > 0:
+                self._log_event(conn, child_id, 'unlinked', {
+                    'parent_id': parent_id,
+                    'child_id': child_id
+                })
+            
+            return result.rowcount > 0
+
+    def list_events(self, since: Optional[datetime] = None) -> List[TaskEvent]:
+        """List events for SSE."""
+        query = "SELECT * FROM task_events"
+        values = []
+        
+        if since:
+            query += " WHERE created_at > ?"
+            values.append(since.isoformat())
+        
+        query += " ORDER BY created_at DESC LIMIT 100"
+        
+        with self._get_connection() as conn:
+            cursor = conn.execute(query, values)
+            events = []
+            
+            for row in cursor.fetchall():
+                event_data = dict(row)
+                event_data['created_at'] = datetime.fromisoformat(event_data['created_at'])
+                event_data['data'] = json.loads(event_data['data'])
+                events.append(TaskEvent(**event_data))
+            
+            return events
+
+    def claim_task(self, task_id: str, worker_id: str) -> bool:
+        """Claim a ready task for execution (CAS operation)."""
+        with self._get_connection() as conn:
+            # Atomic claim with CAS on status and claim_lock, increment version for optimistic locking
+            from datetime import timezone
+            result = conn.execute("""
+                UPDATE tasks 
+                SET claim_lock = ?, updated_at = ?, status = 'running', version = version + 1
+                WHERE id = ? AND status = 'ready' AND (claim_lock IS NULL OR claim_lock = '')
+            """, (worker_id, datetime.now(timezone.utc).isoformat(), task_id))
+            
+            if result.rowcount > 0:
+                self._log_event(conn, task_id, 'claimed', {'worker_id': worker_id})
+                return True
+            
+            return False
+
+    def release_claim(self, task_id: str, worker_id: str) -> bool:
+        """Release claim on task."""
+        with self._get_connection() as conn:
+            # Release claim and increment version for optimistic locking
+            from datetime import timezone
+            result = conn.execute("""
+                UPDATE tasks 
+                SET claim_lock = NULL, updated_at = ?, status = 'ready', version = version + 1
+                WHERE id = ? AND claim_lock = ?
+            """, (datetime.now(timezone.utc).isoformat(), task_id, worker_id))
+            
+            if result.rowcount > 0:
+                self._log_event(conn, task_id, 'released', {'worker_id': worker_id})
+                return True
+            
+            return False

src/praisonai/praisonai/persistence/_valkey_client.py

@@ -41,6 +41,7 @@ def create_valkey_client(
     config = GlideClientConfiguration(
         addresses=addresses,
         credentials=creds,
+        client_name="praisonai_valkey_client",
         database_id=db,
         client_name=_CLIENT_NAME,
     )

src/praisonai/praisonai/persistence/hooks/agent_hooks.py

@@ -5,6 +5,7 @@ Provides wrapper functions to add persistence capabilities to PraisonAI agents
 without modifying the core SDK.
 """
 
+import asyncio
 import time
 import uuid
 import logging
@@ -60,15 +61,13 @@ def wrap_agent_with_persistence(
     session_id = session_id or f"session-{uuid.uuid4().hex[:8]}"
     user_id = user_id or getattr(agent, 'user_id', 'default')
     
-    # Store original chat method
+    # Store original methods
     original_chat = agent.chat
+    original_achat = getattr(agent, 'achat', None)
     _session_initialized = [False]  # Use list to allow mutation in closure
     
-    @wraps(original_chat)
-    def chat_with_persistence(prompt, *args, **kwargs):
-        nonlocal _session_initialized
-        
-        # Initialize session on first call
+    def _ensure_session_sync():
+        """Ensure session is initialized (sync version)."""
         if not _session_initialized[0]:
             history = orchestrator.on_agent_start(
                 agent, 
@@ -86,6 +85,31 @@ def wrap_agent_with_persistence(
                 logger.info(f"Resumed session {session_id} with {len(history)} messages")
             
             _session_initialized[0] = True
+    
+    async def _ensure_session_async():
+        """Ensure session is initialized (async version)."""
+        if not _session_initialized[0]:
+            history = await orchestrator.aon_agent_start(
+                agent, 
+                session_id=session_id, 
+                user_id=user_id,
+                resume=auto_resume
+            )
+            
+            # Inject history into agent's chat_history if resuming
+            if auto_resume and history:
+                agent.chat_history = [
+                    {"role": msg.role, "content": msg.content}
+                    for msg in history
+                ]
+                logger.info(f"Resumed session {session_id} with {len(history)} messages")
+            
+            _session_initialized[0] = True
+    
+    @wraps(original_chat)
+    def chat_with_persistence(prompt, *args, **kwargs):
+        # Initialize session on first call
+        _ensure_session_sync()
         
         # Persist user message
         orchestrator.on_message(session_id, "user", prompt if isinstance(prompt, str) else str(prompt))
@@ -102,6 +126,27 @@ def wrap_agent_with_persistence(
     # Replace chat method
     agent.chat = chat_with_persistence
     
+    # Replace achat method if it exists
+    if original_achat is not None:
+        @wraps(original_achat)
+        async def achat_with_persistence(prompt, *args, **kwargs):
+            # Initialize session on first call
+            await _ensure_session_async()
+            
+            # Persist user message
+            await orchestrator.aon_message(session_id, "user", prompt if isinstance(prompt, str) else str(prompt))
+            
+            # Call original achat
+            response = await original_achat(prompt, *args, **kwargs)
+            
+            # Persist assistant response
+            if response:
+                await orchestrator.aon_message(session_id, "assistant", response)
+            
+            return response
+        
+        agent.achat = achat_with_persistence
+    
     # Add persistence-related methods to agent
     agent._persistence_orchestrator = orchestrator
     agent._persistence_session_id = session_id
@@ -118,9 +163,14 @@ def wrap_agent_with_persistence(
         """End the current session."""
         orchestrator.on_agent_end(agent, session_id)
     
+    async def aend_session():
+        """End the current session (async version)."""
+        await orchestrator.aon_agent_end(agent, session_id)
+    
     agent.get_session = get_session
     agent.get_messages = get_messages
     agent.end_session = end_session
+    agent.aend_session = aend_session
     
     return agent
 
@@ -164,7 +214,7 @@ class PersistentAgent:
         return getattr(self._agent, name)
     
     def _ensure_session(self):
-        """Ensure session is initialized."""
+        """Ensure session is initialized (sync version)."""
         if not self._session_initialized:
             history = self._orchestrator.on_agent_start(
                 self._agent,
@@ -182,6 +232,25 @@ class PersistentAgent:
             
             self._session_initialized = True
     
+    async def _ensure_session_async(self):
+        """Ensure session is initialized (async version)."""
+        if not self._session_initialized:
+            history = await self._orchestrator.aon_agent_start(
+                self._agent,
+                session_id=self._session_id,
+                user_id=self._user_id,
+                resume=self._auto_resume
+            )
+            
+            if self._auto_resume and history:
+                self._agent.chat_history = [
+                    {"role": msg.role, "content": msg.content}
+                    for msg in history
+                ]
+                logger.info(f"Resumed session {self._session_id} with {len(history)} messages")
+            
+            self._session_initialized = True
+    
     def chat(self, prompt, *args, **kwargs):
         """Chat with automatic persistence."""
         self._ensure_session()
@@ -201,6 +270,29 @@ class PersistentAgent:
         
         return response
     
+    async def achat(self, prompt, *args, **kwargs):
+        """Async chat with automatic persistence."""
+        await self._ensure_session_async()
+        
+        # Persist user message
+        await self._orchestrator.aon_message(
+            self._session_id, "user", 
+            prompt if isinstance(prompt, str) else str(prompt)
+        )
+        
+        # Call agent achat
+        if hasattr(self._agent, 'achat'):
+            response = await self._agent.achat(prompt, *args, **kwargs)
+        else:
+            # Fallback to sync chat if achat is not available - use asyncio.to_thread to avoid blocking
+            response = await asyncio.to_thread(self._agent.chat, prompt, *args, **kwargs)
+        
+        # Persist assistant response
+        if response:
+            await self._orchestrator.aon_message(self._session_id, "assistant", response)
+        
+        return response
+    
     def get_session(self) -> Optional[ConversationSession]:
         """Get the current session."""
         return self._orchestrator.get_session(self._session_id)
@@ -213,6 +305,10 @@ class PersistentAgent:
         """End the current session."""
         self._orchestrator.on_agent_end(self._agent, self._session_id)
     
+    async def aend_session(self):
+        """End the current session (async version)."""
+        await self._orchestrator.aon_agent_end(self._agent, self._session_id)
+    
     @property
     def session_id(self) -> str:
         """Get the session ID."""

src/praisonai/praisonai/persistence/orchestrator.py

@@ -9,6 +9,7 @@ import logging
 import time
 import threading
 import uuid
+import inspect
 from copy import deepcopy
 from typing import Any, Dict, List, Optional, TYPE_CHECKING
 
@@ -244,6 +245,164 @@ class PersistenceOrchestrator:
             self._cache_put(session)
             logger.debug(f"Updated session metadata: {session_id}")
     
+    # =========================================================================
+    # Async Agent Lifecycle Hooks
+    # =========================================================================
+    
+    async def aon_agent_start(
+        self,
+        agent: Any,
+        session_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+        resume: bool = True,
+    ) -> List[ConversationMessage]:
+        """
+        Async version of on_agent_start.
+        
+        Args:
+            agent: The agent instance
+            session_id: Session ID (generated if not provided)
+            user_id: User ID for session
+            resume: Whether to load existing session history
+        
+        Returns:
+            List of previous messages if resuming, empty list otherwise
+        """
+        if not session_id:
+            session_id = str(uuid.uuid4())
+        
+        if not self.conversation:
+            logger.debug("No conversation store configured, skipping session load")
+            return []
+        
+        # Try to load existing session
+        session = None
+        if resume:
+            getter = getattr(self.conversation, 'async_get_session', self.conversation.get_session)
+            if inspect.iscoroutinefunction(getter):
+                session = await getter(session_id)
+            else:
+                session = getter(session_id)
+        
+        if session:
+            logger.info(f"Resuming session: {session_id}")
+            self._current_session = session
+            self._cache_put(session)
+            
+            # Load previous messages
+            msg_getter = getattr(self.conversation, 'async_get_messages', self.conversation.get_messages)
+            if inspect.iscoroutinefunction(msg_getter):
+                messages = await msg_getter(session_id)
+            else:
+                messages = msg_getter(session_id)
+            return messages
+        else:
+            # Create new session
+            agent_id = getattr(agent, "name", None) or getattr(agent, "agent_id", None)
+            session = ConversationSession(
+                session_id=session_id,
+                user_id=user_id,
+                agent_id=agent_id,
+                name=f"Session {session_id[:8]}",
+                metadata={"agent_type": type(agent).__name__},
+            )
+            
+            creator = getattr(self.conversation, 'async_create_session', self.conversation.create_session)
+            if inspect.iscoroutinefunction(creator):
+                await creator(session)
+            else:
+                creator(session)
+                
+            logger.info(f"Created new session: {session_id}")
+            self._current_session = session
+            self._cache_put(session)
+            return []
+    
+    async def aon_message(
+        self,
+        session_id: str,
+        role: str,
+        content: str,
+        tool_calls: Optional[List[Dict]] = None,
+        tool_call_id: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> Optional[ConversationMessage]:
+        """
+        Async version of on_message.
+        
+        Args:
+            session_id: Session ID
+            role: Message role (user, assistant, system, tool)
+            content: Message content
+            tool_calls: Tool calls (for assistant messages)
+            tool_call_id: Tool call ID (for tool response messages)
+            metadata: Additional metadata
+        
+        Returns:
+            The persisted message, or None if no store configured
+        """
+        if not self.conversation:
+            return None
+        
+        message = ConversationMessage(
+            id=str(uuid.uuid4()),
+            session_id=session_id,
+            role=role,
+            content=content,
+            tool_calls=tool_calls,
+            tool_call_id=tool_call_id,
+            metadata=metadata,
+        )
+        
+        adder = getattr(self.conversation, 'async_add_message', self.conversation.add_message)
+        if inspect.iscoroutinefunction(adder):
+            await adder(session_id, message)
+        else:
+            adder(session_id, message)
+            
+        logger.debug(f"Persisted {role} message to session {session_id}")
+        return message
+    
+    async def aon_agent_end(
+        self,
+        agent: Any,
+        session_id: str,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """
+        Async version of on_agent_end.
+        
+        Args:
+            agent: The agent instance
+            session_id: Session ID
+            metadata: Additional metadata to store
+        """
+        if not self.conversation:
+            return
+        
+        session = self._cache_get(session_id)
+        if not session:
+            getter = getattr(self.conversation, 'async_get_session', self.conversation.get_session)
+            if inspect.iscoroutinefunction(getter):
+                session = await getter(session_id)
+            else:
+                session = getter(session_id)
+                
+        if session:
+            session.updated_at = time.time()
+            if metadata:
+                session.metadata = {**(session.metadata or {}), **metadata}
+                
+            updater = getattr(self.conversation, 'async_update_session', self.conversation.update_session)
+            if inspect.iscoroutinefunction(updater):
+                await updater(session)
+            else:
+                updater(session)
+                
+            # Update cache with the modified session
+            self._cache_put(session)
+            logger.debug(f"Updated session metadata: {session_id}")
+    
     # =========================================================================
     # Knowledge Retrieval
     # =========================================================================

src/praisonai/praisonai/scheduler/async_agent_scheduler.py

@@ -71,9 +71,9 @@ class AsyncAgentScheduler:
     - Cancellation support  
     - No global state pollution
     - Native async coordination
-    - Timeout support (TODO: needs porting from sync version)
-    - Budget tracking (TODO: needs porting from sync version)
-    - YAML/recipe constructors (TODO: needs porting from sync version)
+    - Timeout support
+    - Budget tracking
+    - YAML/recipe constructors
     
     Example:
         scheduler = AsyncAgentScheduler(agent, task="Check news")
@@ -89,9 +89,8 @@ class AsyncAgentScheduler:
         config: Optional[Dict[str, Any]] = None,
         on_success: Optional[Callable] = None,
         on_failure: Optional[Callable] = None,
-        # TODO: Add these missing features from sync version:
-        # timeout: Optional[int] = None,
-        # max_cost: Optional[float] = 1.00
+        timeout: Optional[int] = None,
+        max_cost: Optional[float] = 1.00
     ):
         """
         Initialize async agent scheduler.
@@ -102,18 +101,17 @@ class AsyncAgentScheduler:
             config: Optional configuration dict
             on_success: Callback function on successful execution
             on_failure: Callback function on failed execution
-            # TODO: Add timeout and max_cost parameters
+            timeout: Maximum execution time per run in seconds (None = no limit)
+            max_cost: Maximum total cost in USD (default: $1.00 for safety)
         """
         self.agent = agent
         self.task = task
         self.config = config or {}
         self.on_success = on_success
         self.on_failure = on_failure
-        
-        # TODO: Add these missing features from sync version:
-        # self.timeout = timeout
-        # self.max_cost = max_cost
-        # self._total_cost = 0.0
+        self.timeout = timeout
+        self.max_cost = max_cost
+        self._total_cost = 0.0
         
         self.is_running = False
         self._task: Optional[asyncio.Task] = None
@@ -172,7 +170,10 @@ class AsyncAgentScheduler:
             logger.info(f"Starting async agent scheduler: {getattr(self.agent, 'name', 'Agent')}")
             logger.info(f"Task: {self.task}")
             logger.info(f"Schedule: {schedule_expr} ({interval}s interval)")
-            # TODO: Add timeout and budget logging from sync version
+            if self.timeout:
+                logger.info(f"Timeout per execution: {self.timeout}s")
+            if self.max_cost is not None:
+                logger.info(f"Budget limit: ${self.max_cost}")
             
             # Run immediately if requested
             if run_immediately:
@@ -265,26 +266,12 @@ class AsyncAgentScheduler:
     
     async def get_stats(self) -> Dict[str, Any]:
         """
-        Get current execution statistics (best-effort synchronous access).
-        
-        Warning: This method provides a best-effort view of stats without
-        guaranteeing atomicity. For consistent snapshots in async context,
-        use get_stats_async() instead.
+        Get current execution statistics (async, atomic snapshot).
         
         Returns:
             Dictionary with execution stats
         """
-        # Best-effort read without lock for backward compatibility
-        return {
-            "is_running": self.is_running,
-            "total_executions": self._execution_count,
-            "successful_executions": self._success_count,
-            "failed_executions": self._failure_count,
-            "success_rate": (self._success_count / self._execution_count * 100) if self._execution_count > 0 else 0,
-            # TODO: Add cost tracking from sync version:
-            # "total_cost": self._total_cost,
-            # "remaining_budget": (self.max_cost - self._total_cost) if self.max_cost else None,
-        }
+        return await self.get_stats_async()
     
     async def get_stats_async(self) -> Dict[str, Any]:
         """
@@ -295,13 +282,14 @@ class AsyncAgentScheduler:
         """
         if self._stats_lock is None:
             # Not yet started: stats are all zero, no lock needed
-            execs, success, failed = 0, 0, 0
+            execs, success, failed, total_cost = 0, 0, 0, 0.0
         else:
             # Take atomic snapshot of all counters
             async with self._stats_lock:
                 execs = self._execution_count
                 success = self._success_count
                 failed = self._failure_count
+                total_cost = self._total_cost
         
         return {
             "is_running": self.is_running,
@@ -309,9 +297,8 @@ class AsyncAgentScheduler:
             "successful_executions": success,
             "failed_executions": failed,
             "success_rate": (success / execs * 100) if execs > 0 else 0,
-            # TODO: Add cost tracking from sync version:
-            # "total_cost": self._total_cost,
-            # "remaining_budget": (self.max_cost - self._total_cost) if self.max_cost else None,
+            "total_cost_usd": round(total_cost, 4),
+            "remaining_budget": round(self.max_cost - total_cost, 4) if self.max_cost is not None else None,
         }
     
     def get_stats_sync(self) -> Dict[str, Any]:
@@ -328,6 +315,8 @@ class AsyncAgentScheduler:
             "successful_executions": self._success_count,
             "failed_executions": self._failure_count,
             "success_rate": (self._success_count / self._execution_count * 100) if self._execution_count > 0 else 0,
+            "total_cost_usd": round(self._total_cost, 4),
+            "remaining_budget": round(self.max_cost - self._total_cost, 4) if self.max_cost is not None else None,
         }
     
     async def _run_schedule(self, interval: int, max_retries: int):
@@ -362,33 +351,37 @@ class AsyncAgentScheduler:
         async with self._stats_lock:
             self._execution_count += 1
         
-        # TODO: Add budget check from sync version:
-        # if self.max_cost and self._total_cost >= self.max_cost:
-        #     logger.warning(f"Budget limit reached: ${self._total_cost:.4f} >= ${self.max_cost}")
-        #     return
+        # Check budget limit before execution
+        if self.max_cost is not None and self._total_cost >= self.max_cost:
+            logger.warning(f"Budget limit reached: ${self._total_cost:.4f} >= ${self.max_cost}")
+            if self._stop_event is not None:
+                self._stop_event.set()
+            self.is_running = False
+            return
         
         last_exc: Optional[Exception] = None
         for attempt in range(max_retries):
             try:
                 logger.info(f"Async attempt {attempt + 1}/{max_retries}")
                 
-                # TODO: Add timeout support from sync version:
-                # if self.timeout:
-                #     result = await asyncio.wait_for(
-                #         self._executor.execute(self.task), 
-                #         timeout=self.timeout
-                #     )
-                # else:
-                result = await self._executor.execute(self.task)
+                # Execute with timeout if specified
+                if self.timeout:
+                    result = await asyncio.wait_for(
+                        self._executor.execute(self.task), 
+                        timeout=self.timeout
+                    )
+                else:
+                    result = await self._executor.execute(self.task)
                 
                 logger.info(f"Async agent execution successful on attempt {attempt + 1}")
                 logger.info(f"Result: {result}")
                 
+                # Estimate cost (rough: ~$0.0001 per execution for gpt-4o-mini)
+                estimated_cost = 0.0001  # Base cost estimate
                 async with self._stats_lock:
                     self._success_count += 1
-                    # TODO: Add cost tracking from sync version:
-                    # estimated_cost = self._estimate_cost(result)
-                    # self._total_cost += estimated_cost
+                    self._total_cost += estimated_cost
+                logger.info(f"Estimated cost this run: ${estimated_cost:.4f}, Total: ${self._total_cost:.4f}")
                 
                 safe_call(self.on_success, result)
                 # TODO: Add daemon state update from sync version:
@@ -438,16 +431,197 @@ class AsyncAgentScheduler:
             logger.error(f"One-time async execution failed: {e}")
             raise
 
-    # TODO: Add these missing factory methods from sync version:
-    # @classmethod
-    # async def from_yaml(cls, yaml_path: str, ...) -> "AsyncAgentScheduler":
-    #     """Create scheduler from YAML configuration."""
-    #     pass
-    #
-    # @classmethod  
-    # async def from_recipe(cls, recipe_name: str, ...) -> "AsyncAgentScheduler":
-    #     """Create scheduler from recipe configuration."""
-    #     pass
+    @classmethod
+    def from_yaml(
+        cls,
+        yaml_path: str = "agents.yaml",
+        interval_override: Optional[str] = None,
+        max_retries_override: Optional[int] = None,
+        timeout_override: Optional[int] = None,
+        max_cost_override: Optional[float] = None,
+        on_success: Optional[Callable] = None,
+        on_failure: Optional[Callable] = None
+    ) -> 'AsyncAgentScheduler':
+        """
+        Create AsyncAgentScheduler from agents.yaml file.
+        
+        Args:
+            yaml_path: Path to agents.yaml file
+            interval_override: Override schedule interval from YAML
+            max_retries_override: Override max_retries from YAML
+            timeout_override: Override timeout from YAML
+            max_cost_override: Override max_cost from YAML
+            on_success: Callback function on successful execution
+            on_failure: Callback function on failed execution
+            
+        Returns:
+            Configured AsyncAgentScheduler instance
+            
+        Example:
+            scheduler = AsyncAgentScheduler.from_yaml("agents.yaml")
+            await scheduler.start("hourly")
+        """
+        from .yaml_loader import load_agent_yaml_with_schedule, create_agent_from_config
+        
+        # Load configuration from YAML
+        agent_config, schedule_config = load_agent_yaml_with_schedule(yaml_path)
+        
+        # Create agent from config
+        agent = create_agent_from_config(agent_config)
+        
+        # Get task
+        task = agent_config.get('task', '')
+        if not task:
+            raise ValueError("No task specified in YAML file")
+        
+        # Apply overrides to schedule config
+        if interval_override:
+            schedule_config['interval'] = interval_override
+        if max_retries_override is not None:
+            schedule_config['max_retries'] = max_retries_override
+        if timeout_override is not None:
+            schedule_config['timeout'] = timeout_override
+        if max_cost_override is not None:
+            schedule_config['max_cost'] = max_cost_override
+        
+        # Create scheduler instance with timeout and cost limits
+        scheduler = cls(
+            agent=agent,
+            task=task,
+            config=agent_config,
+            timeout=schedule_config.get('timeout'),
+            max_cost=schedule_config.get('max_cost'),
+            on_success=on_success,
+            on_failure=on_failure
+        )
+        
+        # Store schedule config for later use
+        scheduler._yaml_schedule_config = schedule_config
+        
+        return scheduler
+
+    async def start_from_yaml_config(self) -> bool:
+        """
+        Start scheduler using configuration from YAML file.
+        
+        Must be called after from_yaml() class method.
+        
+        Returns:
+            True if started successfully
+        """
+        if not hasattr(self, '_yaml_schedule_config'):
+            raise ValueError("No YAML configuration found. Use from_yaml() first.")
+        
+        schedule_config = self._yaml_schedule_config
+        interval = schedule_config.get('interval', 'hourly')
+        max_retries = schedule_config.get('max_retries', 3)
+        run_immediately = schedule_config.get('run_immediately', False)
+        
+        return await self.start(interval, max_retries, run_immediately)
+
+    @classmethod
+    def from_recipe(
+        cls,
+        recipe_name: str,
+        *,
+        input_data: Any = None,
+        config: Optional[Dict[str, Any]] = None,
+        interval_override: Optional[str] = None,
+        max_retries_override: Optional[int] = None,
+        timeout_override: Optional[int] = None,
+        max_cost_override: Optional[float] = None,
+        on_success: Optional[Callable] = None,
+        on_failure: Optional[Callable] = None
+    ) -> 'AsyncAgentScheduler':
+        """
+        Create AsyncAgentScheduler from a recipe name.
+        
+        Args:
+            recipe_name: Name of the recipe to schedule
+            input_data: Input data for the recipe
+            config: Configuration overrides for the recipe
+            interval_override: Override schedule interval from recipe runtime config
+            max_retries_override: Override max_retries from recipe runtime config
+            timeout_override: Override timeout from recipe runtime config
+            max_cost_override: Override max_cost from recipe runtime config
+            on_success: Callback function on successful execution
+            on_failure: Callback function on failed execution
+            
+        Returns:
+            Configured AsyncAgentScheduler instance
+            
+        Example:
+            scheduler = AsyncAgentScheduler.from_recipe("news-monitor")
+            await scheduler.start("hourly")
+        """
+        from praisonai.recipe.bridge import resolve, execute_resolved_recipe, get_recipe_task_description
+        
+        # Resolve the recipe
+        resolved = resolve(
+            recipe_name,
+            input_data=input_data,
+            config=config or {},
+            options={'timeout_sec': timeout_override or 300},
+        )
+        
+        # Get runtime config defaults from recipe
+        interval = interval_override or "hourly"
+        max_retries = max_retries_override if max_retries_override is not None else 3
+        timeout = timeout_override or 300
+        max_cost = max_cost_override if max_cost_override is not None else 1.00
+        
+        runtime = resolved.runtime_config
+        if runtime and hasattr(runtime, 'schedule'):
+            sched_config = runtime.schedule
+            interval = interval_override or sched_config.interval
+            max_retries = max_retries_override if max_retries_override is not None else sched_config.max_retries
+            timeout = timeout_override or sched_config.timeout_sec
+            max_cost = max_cost_override if max_cost_override is not None else sched_config.max_cost_usd
+        
+        # Create a recipe executor agent wrapper that supports async
+        class AsyncRecipeExecutorAgent:
+            """Wrapper that makes a recipe look like an agent for the async scheduler."""
+            def __init__(self, resolved_recipe):
+                self.resolved = resolved_recipe
+                self.name = f"AsyncRecipeAgent:{resolved_recipe.name}"
+            
+            async def astart(self, task: str) -> Any:
+                # Run recipe execution in thread to avoid blocking async loop
+                loop = asyncio.get_running_loop()
+                return await loop.run_in_executor(
+                    None, lambda: execute_resolved_recipe(self.resolved)
+                )
+            
+            def start(self, task: str) -> Any:
+                # Fallback sync method
+                return execute_resolved_recipe(self.resolved)
+        
+        # Create the agent wrapper
+        agent = AsyncRecipeExecutorAgent(resolved)
+        task = get_recipe_task_description(resolved)
+        
+        # Create scheduler instance
+        scheduler = cls(
+            agent=agent,
+            task=task,
+            timeout=timeout,
+            max_cost=max_cost,
+            on_success=on_success,
+            on_failure=on_failure,
+        )
+        
+        # Store recipe metadata and schedule config
+        scheduler._recipe_name = recipe_name
+        scheduler._recipe_resolved = resolved
+        scheduler._yaml_schedule_config = {
+            'interval': interval,
+            'max_retries': max_retries,
+            'run_immediately': False,
+            'timeout': timeout,
+            'max_cost': max_cost,
+        }
+        
+        return scheduler
 
 
 def create_async_agent_scheduler(

src/praisonai/praisonai/tool_registry.py

@@ -19,8 +19,8 @@ class ToolRegistry:
     
     def __init__(self):
         self._functions: Dict[str, Callable] = {}
-        self._autogen_adapters: Dict[str, Callable] = {}
         self._lock = threading.Lock()
+        # Note: AutoGen-specific adapters moved to framework_adapters.autogen
         
     def register_function(self, name: str, func: Callable) -> None:
         """Register a function tool."""
@@ -30,13 +30,24 @@ class ToolRegistry:
             self._functions[name] = func
         logger.debug(f"Registered function tool: {name}")
     
-    def register_autogen_adapter(self, tool_type_name: str, adapter: Callable) -> None:
-        """Register an AutoGen-specific tool adapter."""
+    def register_autogen_adapter(self, tool_type_name: str, adapter: Callable, _suppress_deprecation_warning: bool = False) -> None:
+        """Deprecated: AutoGen adapters moved to framework_adapters.autogen module."""
+        if not _suppress_deprecation_warning:
+            import warnings
+            warnings.warn(
+                "ToolRegistry.register_autogen_adapter is deprecated. "
+                "AutoGen-specific logic has been moved to framework_adapters.autogen module.",
+                DeprecationWarning,
+                stacklevel=2
+            )
+        # For backward compatibility, still store but warn
         if not callable(adapter):
             raise ValueError(f"AutoGen adapter for {tool_type_name} must be callable")
         with self._lock:
+            if not hasattr(self, '_autogen_adapters'):
+                self._autogen_adapters: Dict[str, Callable] = {}
             self._autogen_adapters[tool_type_name] = adapter
-        logger.debug(f"Registered AutoGen adapter: {tool_type_name}")
+        logger.debug(f"Registered AutoGen adapter: {tool_type_name} (deprecated)")
     
     def get_function(self, name: str) -> Optional[Callable]:
         """Get a function tool by name."""
@@ -44,9 +55,19 @@ class ToolRegistry:
             return self._functions.get(name)
     
     def get_autogen_adapter(self, tool_type_name: str) -> Optional[Callable]:
-        """Get an AutoGen adapter by tool type name."""
-        with self._lock:
-            return self._autogen_adapters.get(tool_type_name)
+        """Deprecated: AutoGen adapters moved to framework_adapters.autogen module."""
+        import warnings
+        warnings.warn(
+            "ToolRegistry.get_autogen_adapter is deprecated. "
+            "AutoGen-specific logic has been moved to framework_adapters.autogen module.",
+            DeprecationWarning,
+            stacklevel=2
+        )
+        # For backward compatibility
+        if hasattr(self, '_autogen_adapters'):
+            with self._lock:
+                return self._autogen_adapters.get(tool_type_name)
+        return None
     
     def list_functions(self) -> List[str]:
         """List all registered function tool names."""
@@ -54,9 +75,19 @@ class ToolRegistry:
             return list(self._functions.keys())
     
     def list_autogen_adapters(self) -> List[str]:
-        """List all registered AutoGen adapter names."""
-        with self._lock:
-            return list(self._autogen_adapters.keys())
+        """Deprecated: AutoGen adapters moved to framework_adapters.autogen module."""
+        import warnings
+        warnings.warn(
+            "ToolRegistry.list_autogen_adapters is deprecated. "
+            "AutoGen-specific logic has been moved to framework_adapters.autogen module.",
+            DeprecationWarning,
+            stacklevel=2
+        )
+        # For backward compatibility
+        if hasattr(self, '_autogen_adapters'):
+            with self._lock:
+                return list(self._autogen_adapters.keys())
+        return []
     
     def get_functions_dict(self) -> Dict[str, Callable]:
         """Get a copy of all registered functions."""
@@ -67,7 +98,8 @@ class ToolRegistry:
         """Clear all registered tools."""
         with self._lock:
             self._functions.clear()
-            self._autogen_adapters.clear()
+            if hasattr(self, '_autogen_adapters'):
+                self._autogen_adapters.clear()
         logger.debug("Cleared tool registry")
     
     def register_from_module(self, module: Any) -> List[str]:
@@ -91,25 +123,27 @@ class ToolRegistry:
         logger.debug(f"Registered {len(registered)} functions from module: {registered}")
         return registered
     
-    def register_builtin_autogen_adapters(self) -> None:
-        """Register builtin AutoGen adapters from inbuilt_tools."""
+    def register_builtin_autogen_adapters(self, _suppress_deprecation_warning: bool = False) -> None:
+        """Deprecated: AutoGen adapters moved to framework_adapters.autogen module."""
+        if not _suppress_deprecation_warning:
+            import warnings
+            warnings.warn(
+                "ToolRegistry.register_builtin_autogen_adapters is deprecated. "
+                "AutoGen-specific logic has been moved to framework_adapters.autogen module.",
+                DeprecationWarning,
+                stacklevel=2
+            )
+        # For backward compatibility, attempt old logic but warn
         try:
-            # Lazy import to avoid circular dependencies
             from .inbuilt_tools import _get_autogen_tools
-            
-            # Get the autogen_tools module
             tools_module = _get_autogen_tools()
             if tools_module:
-                # Register adapters based on the pattern in the original code
                 for attr_name in dir(tools_module):
                     if attr_name.startswith('autogen_') and not attr_name.startswith('__'):
                         adapter = getattr(tools_module, attr_name)
                         if callable(adapter):
-                            # Extract tool type name from adapter function name
-                            # e.g., 'autogen_CodeDocsSearchTool' -> 'CodeDocsSearchTool'
                             tool_type_name = attr_name.replace('autogen_', '')
-                            self.register_autogen_adapter(tool_type_name, adapter)
-                            
+                            self.register_autogen_adapter(tool_type_name, adapter, _suppress_deprecation_warning=True)
         except ImportError as e:
             logger.warning(f"Could not register builtin AutoGen adapters: {e}")
         except Exception as e:
@@ -118,7 +152,8 @@ class ToolRegistry:
     def __len__(self) -> int:
         """Return total number of registered tools."""
         with self._lock:
-            return len(self._functions) + len(self._autogen_adapters)
+            autogen_count = len(self._autogen_adapters) if hasattr(self, '_autogen_adapters') else 0
+            return len(self._functions) + autogen_count
     
     def __contains__(self, name: str) -> bool:
         """Check if a tool function is registered."""

src/praisonai/praisonai/tool_resolver.py

@@ -621,15 +621,3 @@ def validate_yaml_tools(yaml_config: Dict[str, Any], resolver: Optional[ToolReso
     return (resolver or _get_default_resolver()).validate_yaml_tools(yaml_config)
 
 
-def reset_default_resolver() -> None:
-    """Clear the process-default resolver.
-    
-    Call this between tenants, on CWD change, or in test setup to ensure
-    that local tools.py resolution is not affected by previous calls.
-    
-    This follows the same pattern as _framework_availability.invalidate()
-    for resetting cached state.
-    """
-    global _default_resolver
-    with _default_resolver_lock:
-        _default_resolver = None

src/praisonai/praisonai/tools/__init__.py

@@ -13,6 +13,8 @@ __all__ = [
     "stt_tool",
     "create_tts_tool",
     "create_stt_tool",
+    "get_kanban_tools",
+    "KANBAN_TOOLS",
 ]
 
 _lazy_cache = {}
@@ -58,4 +60,14 @@ def __getattr__(name: str):
         _lazy_cache[name] = create_stt_tool
         return create_stt_tool
     
+    if name == "get_kanban_tools":
+        from .kanban_tools import get_kanban_tools
+        _lazy_cache[name] = get_kanban_tools
+        return get_kanban_tools
+    
+    if name == "KANBAN_TOOLS":
+        from .kanban_tools import KANBAN_TOOLS
+        _lazy_cache[name] = KANBAN_TOOLS
+        return KANBAN_TOOLS
+    
     raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

src/praisonai/praisonai/tools/kanban_tools.py

@@ -0,0 +1,399 @@
+"""
+Kanban Tools for PraisonAI Agents.
+
+Provides kanban task management tools that allow agents to create, update,
+and manage kanban tasks for multi-agent coordination and workflow management.
+
+All tools connect to the same SQLite kanban store as the PraisonAIUI dashboard.
+"""
+
+import logging
+import os
+from typing import Dict, Any, List, Optional
+
+logger = logging.getLogger(__name__)
+
+# Lazy-loaded kanban store instance 
+_kanban_store = None
+
+
+def _get_kanban_store():
+    """Lazy load kanban store."""
+    global _kanban_store
+    if _kanban_store is None:
+        try:
+            from praisonai.kanban.sqlite_store import SQLiteKanbanStore
+            _kanban_store = SQLiteKanbanStore()
+        except ImportError as e:
+            logger.error(f"Failed to import kanban store: {e}")
+            raise ImportError("Kanban functionality requires praisonai kanban module") from e
+    return _kanban_store
+
+
+def kanban_create(
+    title: str,
+    body: str = "",
+    assignee: str = "",
+    status: str = "todo",
+    priority: int = 0,
+    board: str = "default"
+) -> Dict[str, Any]:
+    """
+    Create a kanban task for multi-agent coordination.
+    
+    Args:
+        title: Task title (required)
+        body: Task description/details
+        assignee: Username or agent name to assign to
+        status: Task status (todo, ready, running, blocked, done)
+        priority: Task priority (higher number = higher priority)
+        board: Board name to create task on
+        
+    Returns:
+        Dict containing the created task details including task ID
+    
+    Example:
+        >>> kanban_create("Implement user authentication", assignee="coder")
+        {'id': 'task_abc123', 'title': 'Implement user authentication', ...}
+    """
+    try:
+        store = _get_kanban_store()
+        
+        task_data = {
+            'title': title,
+            'body': body,
+            'assignee': assignee,
+            'status': status,
+            'priority': priority,
+            'board': board
+        }
+        
+        task = store.create_task(task_data)
+        return task.to_dict()
+        
+    except Exception as e:
+        logger.error(f"Failed to create kanban task: {e}")
+        return {'error': str(e)}
+
+
+def kanban_list(
+    status: Optional[str] = None,
+    assignee: Optional[str] = None,
+    board: str = "default",
+    limit: int = 50
+) -> Dict[str, Any]:
+    """
+    List kanban tasks with optional filtering.
+    
+    Args:
+        status: Filter by status (todo, ready, running, blocked, done)
+        assignee: Filter by assignee username
+        board: Board name to list from
+        limit: Maximum number of tasks to return
+        
+    Returns:
+        Dict containing list of tasks and metadata
+    
+    Example:
+        >>> kanban_list(status="ready", assignee="coder")
+        {'tasks': [...], 'count': 3, 'filters': {...}}
+    """
+    try:
+        store = _get_kanban_store()
+        
+        filters = {'board': board}
+        if status:
+            filters['status'] = status
+        if assignee:
+            filters['assignee'] = assignee
+        
+        tasks = store.list_tasks(filters)
+        
+        # Limit results
+        if len(tasks) > limit:
+            tasks = tasks[:limit]
+        
+        return {
+            'tasks': [task.to_dict() for task in tasks],
+            'count': len(tasks),
+            'filters': filters,
+            'limited': len(tasks) == limit
+        }
+        
+    except Exception as e:
+        logger.error(f"Failed to list kanban tasks: {e}")
+        return {'error': str(e)}
+
+
+def kanban_show(task_id: str) -> Dict[str, Any]:
+    """
+    Show detailed information about a specific task.
+    
+    Args:
+        task_id: The task ID to retrieve
+        
+    Returns:
+        Dict containing task details, comments, and dependencies
+    
+    Example:
+        >>> kanban_show("task_abc123")
+        {'task': {...}, 'comments': [...], 'dependencies': [...]}
+    """
+    try:
+        store = _get_kanban_store()
+        
+        task = store.get_task(task_id)
+        if not task:
+            return {'error': f'Task {task_id} not found'}
+        
+        # Get comments
+        comments = store.get_comments(task_id)
+        
+        # Get parent/child links (simplified - would need more complex query for full DAG)
+        result = {
+            'task': task.to_dict(),
+            'comments': [comment.to_dict() for comment in comments],
+            'comment_count': len(comments)
+        }
+        
+        return result
+        
+    except Exception as e:
+        logger.error(f"Failed to show kanban task {task_id}: {e}")
+        return {'error': str(e)}
+
+
+def kanban_complete(task_id: str, comment: str = "") -> Dict[str, Any]:
+    """
+    Mark a task as completed.
+    
+    Args:
+        task_id: The task ID to complete
+        comment: Optional completion comment
+        
+    Returns:
+        Dict containing the updated task
+    
+    Example:
+        >>> kanban_complete("task_abc123", "Authentication system working")
+        {'task': {...}, 'status': 'done'}
+    """
+    try:
+        store = _get_kanban_store()
+        
+        # Move to done status
+        task = store.move_task(task_id, 'done')
+        
+        # Add completion comment if provided
+        if comment:
+            store.add_comment(task_id, 'agent', f"Completed: {comment}")
+        
+        return {
+            'task': task.to_dict(),
+            'status': 'done',
+            'completed': True
+        }
+        
+    except Exception as e:
+        logger.error(f"Failed to complete kanban task {task_id}: {e}")
+        return {'error': str(e)}
+
+
+def kanban_block(task_id: str, reason: str) -> Dict[str, Any]:
+    """
+    Mark a task as blocked with a reason.
+    
+    Args:
+        task_id: The task ID to block
+        reason: Reason why the task is blocked
+        
+    Returns:
+        Dict containing the updated task
+    
+    Example:
+        >>> kanban_block("task_abc123", "Waiting for API credentials")
+        {'task': {...}, 'status': 'blocked'}
+    """
+    try:
+        store = _get_kanban_store()
+        
+        # Move to blocked status
+        task = store.move_task(task_id, 'blocked')
+        
+        # Add blocking reason as comment
+        store.add_comment(task_id, 'agent', f"Blocked: {reason}")
+        
+        return {
+            'task': task.to_dict(),
+            'status': 'blocked',
+            'reason': reason
+        }
+        
+    except Exception as e:
+        logger.error(f"Failed to block kanban task {task_id}: {e}")
+        return {'error': str(e)}
+
+
+def kanban_comment(task_id: str, text: str, author: str = "agent") -> Dict[str, Any]:
+    """
+    Add a comment to a kanban task.
+    
+    Args:
+        task_id: The task ID to comment on
+        text: Comment text
+        author: Comment author (defaults to 'agent')
+        
+    Returns:
+        Dict containing the created comment
+    
+    Example:
+        >>> kanban_comment("task_abc123", "Made progress on auth module")
+        {'comment': {...}, 'task_id': 'task_abc123'}
+    """
+    try:
+        store = _get_kanban_store()
+        
+        comment = store.add_comment(task_id, author, text)
+        
+        return {
+            'comment': comment.to_dict(),
+            'task_id': task_id
+        }
+        
+    except Exception as e:
+        logger.error(f"Failed to add comment to kanban task {task_id}: {e}")
+        return {'error': str(e)}
+
+
+def kanban_link(parent_id: str, child_id: str) -> Dict[str, Any]:
+    """
+    Create a dependency link between two tasks.
+    
+    Args:
+        parent_id: The parent task ID (must be completed before child)
+        child_id: The child task ID (depends on parent)
+        
+    Returns:
+        Dict containing the created link
+    
+    Example:
+        >>> kanban_link("task_design", "task_implement")
+        {'link': {...}, 'parent_id': 'task_design', 'child_id': 'task_implement'}
+    """
+    try:
+        store = _get_kanban_store()
+        
+        link = store.add_link(parent_id, child_id)
+        
+        return {
+            'link': {
+                'parent_id': link.parent_id,
+                'child_id': link.child_id,
+                'created_at': link.created_at.isoformat()
+            },
+            'parent_id': parent_id,
+            'child_id': child_id
+        }
+        
+    except Exception as e:
+        logger.error(f"Failed to link kanban tasks {parent_id} -> {child_id}: {e}")
+        return {'error': str(e)}
+
+
+def kanban_heartbeat(task_id: str, status: str = "working") -> Dict[str, Any]:
+    """
+    Send a heartbeat signal while working on a task.
+    
+    Args:
+        task_id: The task ID being worked on
+        status: Status message (e.g., "working", "testing", "debugging")
+        
+    Returns:
+        Dict containing heartbeat confirmation
+    
+    Example:
+        >>> kanban_heartbeat("task_abc123", "testing authentication flow")
+        {'heartbeat': True, 'task_id': 'task_abc123', 'status': 'testing...'}
+    """
+    try:
+        store = _get_kanban_store()
+        
+        # Verify task exists and is claimed
+        task = store.get_task(task_id)
+        if not task:
+            return {'error': f'Task {task_id} not found'}
+        
+        # Add heartbeat comment
+        store.add_comment(
+            task_id, 
+            'agent', 
+            f"Heartbeat: {status} at {task.updated_at.strftime('%H:%M:%S')}"
+        )
+        
+        return {
+            'heartbeat': True,
+            'task_id': task_id,
+            'status': status,
+            'timestamp': task.updated_at.isoformat()
+        }
+        
+    except Exception as e:
+        logger.error(f"Failed to send heartbeat for kanban task {task_id}: {e}")
+        return {'error': str(e)}
+
+
+# Export all kanban tools for agent registration
+KANBAN_TOOLS = [
+    kanban_create,
+    kanban_list,
+    kanban_show,
+    kanban_complete,
+    kanban_block,
+    kanban_comment,
+    kanban_link,
+    kanban_heartbeat,
+]
+
+
+def get_kanban_tools() -> List[callable]:
+    """
+    Get all kanban tools for agent registration.
+    
+    Returns:
+        List of kanban tool functions
+    
+    Example:
+        >>> from praisonai.tools.kanban_tools import get_kanban_tools
+        >>> agent = Agent(name="coordinator", tools=get_kanban_tools())
+    """
+    return KANBAN_TOOLS
+
+
+def create_kanban_agent_tools(board: str = "default") -> List[callable]:
+    """
+    Create kanban tools bound to a specific board.
+    
+    Args:
+        board: Board name to bind tools to
+        
+    Returns:
+        List of board-specific kanban tools
+    """
+    # Create board-specific versions of tools with board parameter pre-filled
+    def _make_board_tool(tool_func, board_name):
+        def wrapper(*args, **kwargs):
+            if 'board' not in kwargs:
+                kwargs['board'] = board_name
+            return tool_func(*args, **kwargs)
+        wrapper.__name__ = f"{tool_func.__name__}_{board_name}"
+        wrapper.__doc__ = tool_func.__doc__
+        return wrapper
+    
+    board_tools = []
+    for tool in KANBAN_TOOLS:
+        if 'board' in tool.__code__.co_varnames:
+            board_tools.append(_make_board_tool(tool, board))
+        else:
+            board_tools.append(tool)
+    
+    return board_tools

src/praisonai/praisonai/version.py

@@ -1 +1 @@
-__version__ = "4.6.48"
+__version__ = "4.6.50"

src/praisonai/pyproject.toml

@@ -12,7 +12,7 @@ dependencies = [
     "rich>=13.7",
     "markdown>=3.5",
     "pyparsing>=3.0.0",
-    "praisonaiagents>=1.6.48",
+    "praisonaiagents>=1.6.50",
     "python-dotenv>=0.19.0",
     "litellm>=1.83.14,<2",
     "PyYAML>=6.0",

src/praisonai/tests/integration/test_kanban_host_injection.py

@@ -0,0 +1,127 @@
+"""Integration tests for kanban host injection."""
+
+import pytest
+import tempfile
+import shutil
+from pathlib import Path
+from unittest.mock import patch, MagicMock
+
+
+@pytest.fixture
+def temp_kanban_dir():
+    """Create temporary directory for kanban testing."""
+    temp_dir = Path(tempfile.mkdtemp())
+    yield temp_dir
+    if temp_dir.exists():
+        shutil.rmtree(temp_dir)
+
+
+def test_kanban_bridge_registration():
+    """Test that kanban bridge can register backends."""
+    # Mock praisonaiui backends
+    mock_backends = MagicMock()
+    
+    with patch.dict('sys.modules', {'praisonaiui.backends': mock_backends}):
+        from praisonai.integration.bridges.kanban_bridge import register_kanban_backends
+        
+        result = register_kanban_backends()
+        
+        # Should return True indicating successful registration
+        assert result is True
+        
+        # Verify backends were set
+        assert mock_backends.set_backend.call_count >= 1
+        
+        # Check specific backend calls
+        backend_calls = [call[0] for call in mock_backends.set_backend.call_args_list]
+        assert any('kanban_store' in str(call) for call in backend_calls)
+
+
+def test_kanban_store_factory():
+    """Test kanban store factory creation."""
+    from praisonai.integration.bridges.kanban_bridge import get_kanban_store_factory
+    
+    factory = get_kanban_store_factory()
+    
+    assert factory is not None
+    
+    # Test factory creates store
+    with patch('praisonai.integration.bridges.kanban_bridge.SQLiteKanbanStore') as mock_store:
+        store = factory()
+        mock_store.assert_called_once()
+
+
+def test_setup_bridges_includes_kanban(temp_kanban_dir):
+    """Test that setup_bridges includes kanban registration."""
+    # Mock all the dependencies
+    mock_backends = MagicMock()
+    mock_praisonaiui = MagicMock()
+    mock_praisonaiui.backends = mock_backends
+    
+    # Mock the kanban path to use temp directory
+    def mock_get_path(board=None):
+        return temp_kanban_dir / "test_kanban.db"
+    
+    with patch.dict('sys.modules', {
+        'praisonaiui': mock_praisonaiui,
+        'praisonaiui.backends': mock_backends
+    }), patch('praisonai.kanban.paths.get_kanban_db_path', mock_get_path):
+        
+        from praisonai.integration.host_app import setup_bridges
+        
+        # This should not raise and should call kanban registration
+        setup_bridges()
+        
+        # Verify some backend was registered (exact calls depend on available components)
+        assert mock_backends.set_backend.called
+
+
+def test_kanban_store_environment_variable_support(temp_kanban_dir):
+    """Test that kanban store respects environment variables."""
+    import os
+    
+    test_board = "test_board"
+    
+    with patch.dict(os.environ, {'PRAISONAI_KANBAN_BOARD': test_board}):
+        with patch('praisonai.kanban.paths.get_kanban_db_path') as mock_path:
+            mock_path.return_value = temp_kanban_dir / f"{test_board}_kanban.db"
+            
+            from praisonai.integration.bridges.kanban_bridge import get_kanban_store_factory
+            
+            factory = get_kanban_store_factory()
+            
+            # This should use the environment variable for board selection
+            with patch('praisonai.integration.bridges.kanban_bridge.SQLiteKanbanStore') as mock_store:
+                store = factory()
+                mock_store.assert_called_once()
+
+
+def test_missing_dependencies_graceful_degradation():
+    """Test that missing dependencies don't break setup."""
+    # Simulate missing praisonaiui
+    with patch.dict('sys.modules', {}, clear=True):
+        # Remove praisonaiui from modules to simulate missing dependency
+        if 'praisonaiui' in __import__('sys').modules:
+            del __import__('sys').modules['praisonaiui']
+        if 'praisonaiui.backends' in __import__('sys').modules:
+            del __import__('sys').modules['praisonaiui.backends']
+        
+        from praisonai.integration.bridges.kanban_bridge import register_kanban_backends
+        
+        # Should not raise, just return False
+        result = register_kanban_backends()
+        assert result is False
+
+
+def test_jobs_store_bridge():
+    """Test jobs store bridge functionality."""
+    from praisonai.integration.bridges.kanban_bridge import get_jobs_store, get_jobs_executor
+    
+    # These may return None if jobs module isn't available, which is fine
+    jobs_store = get_jobs_store()
+    jobs_executor = get_jobs_executor()
+    
+    # Just test that the functions don't raise exceptions
+    # Actual functionality depends on jobs module implementation
+    assert jobs_store is None or callable(jobs_store)
+    assert jobs_executor is None or callable(jobs_executor)

src/praisonai/tests/unit/scheduler/test_async_agent_scheduler.py

@@ -239,6 +239,22 @@ class TestRunScheduleFinally:
 
         assert scheduler.is_running is False
 
+    @pytest.mark.asyncio
+    async def test_execute_with_retry_zero_budget_sets_stop_event_without_stop_call(self):
+        scheduler = _make_scheduler()
+        scheduler.is_running = True
+        scheduler.max_cost = 0.0
+        scheduler._total_cost = 0.0
+        scheduler._ensure_async_primitives()
+
+        with patch.object(scheduler, "stop", new=AsyncMock()) as stop_mock:
+            await scheduler._execute_with_retry(max_retries=1)
+
+        assert scheduler._stop_event.is_set()
+        assert scheduler.is_running is False
+        stop_mock.assert_not_called()
+        scheduler.agent.astart.assert_not_called()
+
 
 # ---------------------------------------------------------------------------
 # Full lifecycle
@@ -280,6 +296,12 @@ class TestAsyncAgentSchedulerLifecycle:
         assert stats["failed_executions"] == 0
         assert stats["success_rate"] == 0
 
+    def test_get_stats_sync_includes_cost_fields(self):
+        scheduler = _make_scheduler()
+        stats = scheduler.get_stats_sync()
+        assert "total_cost_usd" in stats
+        assert "remaining_budget" in stats
+
     @pytest.mark.asyncio
     async def test_execute_once_success(self):
         scheduler = _make_scheduler()

src/praisonai/tests/unit/test_bot_gap_features.py

@@ -289,6 +289,28 @@ class TestConfigSchema:
         with pytest.raises(ValueError, match="Unknown channel"):
             validate_bot_config({"channels": {"signal": {"token": "x"}}})
 
+    def test_multi_channel_custom_keys_with_platform(self, monkeypatch):
+        """Onboard multi-telegram YAML uses telegram_<role> keys plus platform field."""
+        from praisonai.bots._config_schema import validate_bot_config
+
+        monkeypatch.setenv("_TEST_CFO_TOKEN", "cfo-token")
+        raw = {
+            "channels": {
+                "telegram_cfo": {
+                    "platform": "telegram",
+                    "token": "${_TEST_CFO_TOKEN}",
+                    "group_policy": "mention_only",
+                },
+                "telegram_ops": {
+                    "platform": "telegram",
+                    "token": "${_TEST_CFO_TOKEN}",
+                },
+            }
+        }
+        result = validate_bot_config(raw)
+        assert result.channels["telegram_cfo"].platform == "telegram"
+        assert result.channels["telegram_cfo"].token == "cfo-token"
+
     def test_invalid_group_policy(self):
         from praisonai.bots._config_schema import validate_bot_config
         with pytest.raises(ValueError, match="group_policy"):

src/praisonai/tests/unit/test_kanban_sqlite_store.py

@@ -0,0 +1,315 @@
+"""Unit tests for SQLite kanban store."""
+
+import pytest
+import tempfile
+import shutil
+from pathlib import Path
+from datetime import datetime, timedelta
+
+from praisonai.kanban.sqlite_store import SQLiteKanbanStore
+from praisonai.kanban.models import Task, TaskStatus
+
+
+@pytest.fixture
+def temp_db():
+    """Create a temporary database for testing."""
+    temp_dir = Path(tempfile.mkdtemp())
+    db_path = temp_dir / "test_kanban.db"
+    
+    yield db_path
+    
+    # Cleanup
+    if temp_dir.exists():
+        shutil.rmtree(temp_dir)
+
+
+@pytest.fixture
+def store(temp_db):
+    """Create a kanban store with temp database."""
+    # Patch the db path to use temp
+    original_path = None
+    
+    try:
+        import praisonai.kanban.sqlite_store
+        original_method = praisonai.kanban.sqlite_store.get_kanban_db_path
+        praisonai.kanban.sqlite_store.get_kanban_db_path = lambda board=None: temp_db
+        
+        store = SQLiteKanbanStore()
+        yield store
+    finally:
+        # Restore original method
+        if original_method:
+            praisonai.kanban.sqlite_store.get_kanban_db_path = original_method
+
+
+class TestSQLiteKanbanStore:
+    """Test kanban store operations."""
+
+    def test_create_task(self, store):
+        """Test task creation."""
+        task_data = {
+            'title': 'Test Task',
+            'body': 'Test description',
+            'status': 'todo',
+            'assignee': 'test_user'
+        }
+        
+        task = store.create_task(task_data)
+        
+        assert task.title == 'Test Task'
+        assert task.body == 'Test description'
+        assert task.status == TaskStatus.TODO
+        assert task.assignee == 'test_user'
+        assert task.id.startswith('task_')
+        assert task.created_at is not None
+        assert task.updated_at is not None
+
+    def test_get_task(self, store):
+        """Test task retrieval."""
+        task_data = {'title': 'Test Task', 'assignee': 'user1'}
+        created_task = store.create_task(task_data)
+        
+        retrieved_task = store.get_task(created_task.id)
+        
+        assert retrieved_task is not None
+        assert retrieved_task.id == created_task.id
+        assert retrieved_task.title == 'Test Task'
+        assert retrieved_task.assignee == 'user1'
+
+    def test_get_nonexistent_task(self, store):
+        """Test getting non-existent task returns None."""
+        task = store.get_task('nonexistent_id')
+        assert task is None
+
+    def test_update_task(self, store):
+        """Test task updates."""
+        task_data = {'title': 'Original Title', 'status': 'todo'}
+        created_task = store.create_task(task_data)
+        
+        updates = {
+            'title': 'Updated Title',
+            'status': 'ready',
+            'assignee': 'new_user'
+        }
+        
+        updated_task = store.update_task(created_task.id, updates)
+        
+        assert updated_task.title == 'Updated Title'
+        assert updated_task.status == TaskStatus.READY
+        assert updated_task.assignee == 'new_user'
+        assert updated_task.updated_at > created_task.updated_at
+
+    def test_delete_task(self, store):
+        """Test task deletion."""
+        task_data = {'title': 'To Delete'}
+        created_task = store.create_task(task_data)
+        
+        deleted = store.delete_task(created_task.id)
+        assert deleted is True
+        
+        # Verify task is gone
+        task = store.get_task(created_task.id)
+        assert task is None
+
+    def test_list_tasks(self, store):
+        """Test task listing."""
+        # Create multiple tasks
+        tasks_data = [
+            {'title': 'Task 1', 'status': 'todo', 'assignee': 'user1'},
+            {'title': 'Task 2', 'status': 'ready', 'assignee': 'user1'},
+            {'title': 'Task 3', 'status': 'todo', 'assignee': 'user2'},
+        ]
+        
+        created_tasks = []
+        for data in tasks_data:
+            created_tasks.append(store.create_task(data))
+        
+        # List all tasks
+        all_tasks = store.list_tasks()
+        assert len(all_tasks) == 3
+        
+        # Filter by status
+        todo_tasks = store.list_tasks({'status': 'todo'})
+        assert len(todo_tasks) == 2
+        
+        # Filter by assignee
+        user1_tasks = store.list_tasks({'assignee': 'user1'})
+        assert len(user1_tasks) == 2
+        
+        # Filter by status list
+        ready_tasks = store.list_tasks({'status': ['ready', 'running']})
+        assert len(ready_tasks) == 1
+
+    def test_move_task(self, store):
+        """Test task status changes."""
+        task_data = {'title': 'Move Test', 'status': 'todo'}
+        created_task = store.create_task(task_data)
+        
+        moved_task = store.move_task(created_task.id, 'ready')
+        
+        assert moved_task.status == TaskStatus.READY
+        assert moved_task.updated_at > created_task.updated_at
+
+    def test_get_board(self, store):
+        """Test board layout generation."""
+        # Create tasks in different statuses
+        tasks_data = [
+            {'title': 'Todo Task', 'status': 'todo'},
+            {'title': 'Ready Task', 'status': 'ready'},
+            {'title': 'Done Task', 'status': 'done'},
+        ]
+        
+        for data in tasks_data:
+            store.create_task(data)
+        
+        board = store.get_board()
+        
+        assert 'board' in board
+        assert 'columns' in board
+        assert 'task_count' in board
+        
+        assert len(board['columns']['todo']) == 1
+        assert len(board['columns']['ready']) == 1
+        assert len(board['columns']['done']) == 1
+        assert board['task_count'] == 3
+
+    def test_bulk_update(self, store):
+        """Test bulk operations."""
+        # Create tasks
+        task1 = store.create_task({'title': 'Task 1', 'status': 'todo'})
+        task2 = store.create_task({'title': 'Task 2', 'status': 'todo'})
+        
+        # Bulk operations
+        operations = [
+            {
+                'operation': 'move',
+                'task_id': task1.id,
+                'status': 'ready'
+            },
+            {
+                'operation': 'update',
+                'task_id': task2.id,
+                'updates': {'title': 'Updated Task 2'}
+            }
+        ]
+        
+        results = store.bulk_update(operations)
+        
+        assert len(results) == 2
+        assert results[0].status == TaskStatus.READY
+        assert results[1].title == 'Updated Task 2'
+
+    def test_comments(self, store):
+        """Test task comments."""
+        task = store.create_task({'title': 'Comment Test'})
+        
+        # Add comment
+        comment = store.add_comment(task.id, 'test_user', 'This is a test comment')
+        
+        assert comment.task_id == task.id
+        assert comment.author == 'test_user'
+        assert comment.text == 'This is a test comment'
+        assert comment.id.startswith('comment_')
+        
+        # Get comments
+        comments = store.get_comments(task.id)
+        assert len(comments) == 1
+        assert comments[0].text == 'This is a test comment'
+
+    def test_task_links(self, store):
+        """Test task dependencies."""
+        parent = store.create_task({'title': 'Parent Task'})
+        child = store.create_task({'title': 'Child Task'})
+        
+        # Add dependency
+        link = store.add_link(parent.id, child.id)
+        
+        assert link.parent_id == parent.id
+        assert link.child_id == child.id
+        
+        # Test cycle detection
+        with pytest.raises(ValueError, match="cycle"):
+            store.add_link(child.id, parent.id)  # Would create cycle
+        
+        # Remove link
+        removed = store.remove_link(parent.id, child.id)
+        assert removed is True
+
+    def test_claim_and_release(self, store):
+        """Test task claiming for execution."""
+        task = store.create_task({'title': 'Claimable Task', 'status': 'ready'})
+        
+        # Claim task
+        claimed = store.claim_task(task.id, 'worker_1')
+        assert claimed is True
+        
+        # Verify task is running and claimed
+        updated_task = store.get_task(task.id)
+        assert updated_task.status == TaskStatus.RUNNING
+        assert updated_task.claim_lock == 'worker_1'
+        
+        # Another worker cannot claim
+        claimed_again = store.claim_task(task.id, 'worker_2')
+        assert claimed_again is False
+        
+        # Release claim
+        released = store.release_claim(task.id, 'worker_1')
+        assert released is True
+        
+        # Verify task is ready again
+        released_task = store.get_task(task.id)
+        assert released_task.status == TaskStatus.READY
+        assert released_task.claim_lock is None
+
+    def test_parent_child_promotion(self, store):
+        """Test automatic promotion of child tasks when parent completes."""
+        # Create parent and child
+        parent = store.create_task({'title': 'Parent', 'status': 'running'})
+        child = store.create_task({'title': 'Child', 'status': 'todo'})
+        
+        # Link them
+        store.add_link(parent.id, child.id)
+        
+        # Complete parent
+        store.move_task(parent.id, 'done')
+        
+        # Child should be promoted to ready
+        updated_child = store.get_task(child.id)
+        assert updated_child.status == TaskStatus.READY
+
+    def test_blocked_by_parents(self, store):
+        """Test that child cannot move to ready if parents are incomplete."""
+        parent = store.create_task({'title': 'Incomplete Parent', 'status': 'todo'})
+        child = store.create_task({'title': 'Child', 'status': 'todo'})
+        
+        # Link them
+        store.add_link(parent.id, child.id)
+        
+        # Try to move child to ready - should fail
+        with pytest.raises(ValueError, match="incomplete parent"):
+            store.move_task(child.id, 'ready')
+
+    def test_list_events(self, store):
+        """Test event listing for audit trail."""
+        task = store.create_task({'title': 'Event Test'})
+        store.update_task(task.id, {'title': 'Updated Title'})
+        
+        events = store.list_events()
+        
+        # Should have created and updated events
+        assert len(events) >= 2
+        event_types = [e.event_type for e in events]
+        assert 'created' in event_types
+        assert 'updated' in event_types
+
+    def test_concurrent_access_simulation(self, store):
+        """Test optimistic locking prevents concurrent modifications."""
+        task = store.create_task({'title': 'Concurrent Test'})
+        
+        # Simulate concurrent update by manually manipulating version
+        with store._get_connection() as conn:
+            conn.execute("UPDATE tasks SET version = 2 WHERE id = ?", (task.id,))
+        
+        # This update should fail due to version mismatch
+        with pytest.raises(ValueError, match="modified by another process"):
+            store.update_task(task.id, {'title': 'Concurrent Update'})

src/praisonai/uv.lock

@@ -5491,7 +5491,7 @@ wheels = [
 
 [[package]]
 name = "praisonaiagents"
-version = "1.6.48"
+version = "1.6.50"
 source = { directory = "../praisonai-agents" }
 dependencies = [
     { name = "aiohttp" },