fix: address 6 critical issues in kanban system implementation

- Fix subprocess deadlock by redirecting output to temp files instead of PIPE
- Enable foreign key constraints on all SQLite connections for proper cascades
- Replace immediate dispatcher startup with proper event handler registration
- Add persistent CLI board switching via ~/.praisonai/kanban_config.json
- Prevent database locking with connection-reusing internal methods
- Replace deprecated datetime.utcnow() with datetime.now(timezone.utc)
- Fix optimistic locking by incrementing version in claim/release operations

Fixes all issues identified in code review by Gemini.

Co-authored-by: Mervin Praison <MervinPraison@users.noreply.github.com>

Author: praisonai-triage-agent[bot]

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

@@ -282,10 +282,25 @@ def kanban_boards_cmd(
                 raise typer.Exit(1)
 
             import os
+            from pathlib import Path
+            import json
+            
+            # Set for current session
             os.environ['PRAISONAI_KANBAN_BOARD'] = board
 
-            output.print_success(f"Switched to board '{board}'")
-            output.print_info("Note: This only affects the current CLI session")
+            # 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}")

src/praisonai/praisonai/gateway/kanban_dispatcher.py

@@ -144,15 +144,24 @@ async def _spawn_worker(self, task: Any, store: Any) -> bool:
 
             logger.info(f"Spawning worker for task {task.id}: {' '.join(cmd)}")
 
-            # Start process
+            # 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=subprocess.PIPE,
+                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
 
@@ -208,8 +217,17 @@ def _cleanup_completed_tasks(self, store: Any):
                     # Get return code
                     return_code = process.returncode
 
-                    # Read output
-                    stdout_data, _ = process.communicate(timeout=1)
+                    # 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:
@@ -260,9 +278,19 @@ def _cleanup_completed_tasks(self, store: Any):
                     except:
                         pass
 
-        # Remove completed tasks from tracking
+        # 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):
         """

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

@@ -65,8 +65,8 @@ def register_kanban_backends():
             backends.set_backend("jobs_executor", jobs_executor)
             log.debug("jobs_executor backend registered")
 
-        # Start kanban dispatcher
-        ensure_kanban_dispatcher()
+        # Register kanban dispatcher startup handler
+        register_kanban_startup_handler()
 
         return True
 
@@ -75,34 +75,39 @@ def register_kanban_backends():
         return False
 
 
-def ensure_kanban_dispatcher():
-    """Ensure kanban dispatcher is running in background."""
-    import asyncio
+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
 
-        if not is_dispatcher_running():
-            # Start dispatcher in background
-            loop = None
+        async def _startup_handler():
+            """Startup handler for kanban dispatcher."""
             try:
-                loop = asyncio.get_event_loop()
-            except RuntimeError:
-                # No event loop in this thread, create one
-                loop = asyncio.new_event_loop()
-                asyncio.set_event_loop(loop)
-            
-            if loop and loop.is_running():
-                # Loop is running, schedule the task
-                asyncio.create_task(start_kanban_dispatcher())
-                log.debug("kanban dispatcher scheduled to start")
-            else:
-                # No loop or loop not running, we can't start dispatcher here
-                # This is expected in CLI/sync contexts
-                log.debug("kanban dispatcher cannot start (no async context)")
+                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 dispatcher startup failed: %s", exc)
+        log.debug("kanban startup handler registration failed: %s", exc)

src/praisonai/praisonai/kanban/models.py

@@ -37,10 +37,11 @@ class Task:
     metadata: Optional[Dict[str, Any]] = None
 
     def __post_init__(self):
+        from datetime import timezone
         if self.created_at is None:
-            self.created_at = datetime.utcnow()
+            self.created_at = datetime.now(timezone.utc)
         if self.updated_at is None:
-            self.updated_at = datetime.utcnow()
+            self.updated_at = datetime.now(timezone.utc)
         if self.metadata is None:
             self.metadata = {}
 

src/praisonai/praisonai/kanban/paths.py

@@ -22,7 +22,20 @@ def get_kanban_db_path(board: Optional[str] = None) -> Path:
 
     # Read from environment variable for override
     if board is None:
-        board = os.environ.get('PRAISONAI_KANBAN_BOARD', 'default')
+        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'):

src/praisonai/praisonai/kanban/sqlite_store.py

@@ -106,6 +106,8 @@ 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()
@@ -267,11 +269,58 @@ def list_tasks(self, filters: Optional[Dict[str, Any]] = None) -> List[Task]:
             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(task_id)
+            task = self._get_task_with_conn(task_id, conn)
             if not task:
                 raise ValueError(f"Task {task_id} not found")
 
@@ -290,7 +339,7 @@ def move_task(self, task_id: str, status: str) -> Task:
                     raise ValueError("Cannot move to ready: incomplete parent tasks")
 
             # Update task status
-            updated_task = self.update_task(task_id, {'status': status})
+            updated_task = self._update_task_with_conn(task_id, {'status': status}, conn)
 
             # Promote children if moving to done
             if new_status == TaskStatus.DONE:
@@ -300,7 +349,7 @@ def move_task(self, task_id: str, status: str) -> Task:
 
                 for row in cursor.fetchall():
                     child_id = row['child_id']
-                    child_task = self.get_task(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
@@ -312,7 +361,7 @@ def move_task(self, task_id: str, status: str) -> Task:
                         """, (child_id,))
 
                         if parent_check.fetchone()['incomplete_parents'] == 0:
-                            self.update_task(child_id, {'status': TaskStatus.READY.value})
+                            self._update_task_with_conn(child_id, {'status': TaskStatus.READY.value}, conn)
 
             self._log_event(conn, task_id, 'moved', {
                 'old_status': task.status.value,
@@ -484,12 +533,13 @@ def list_events(self, since: Optional[datetime] = None) -> List[TaskEvent]:
     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
+            # 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'
+                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.utcnow().isoformat(), task_id))
+            """, (worker_id, datetime.now(timezone.utc).isoformat(), task_id))
 
             if result.rowcount > 0:
                 self._log_event(conn, task_id, 'claimed', {'worker_id': worker_id})
@@ -500,11 +550,13 @@ def claim_task(self, task_id: str, worker_id: str) -> bool:
     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'
+                SET claim_lock = NULL, updated_at = ?, status = 'ready', version = version + 1
                 WHERE id = ? AND claim_lock = ?
-            """, (datetime.utcnow().isoformat(), task_id, worker_id))
+            """, (datetime.now(timezone.utc).isoformat(), task_id, worker_id))
 
             if result.rowcount > 0:
                 self._log_event(conn, task_id, 'released', {'worker_id': worker_id})