Release v2.9.0

api_server.py

@@ -0,0 +1,71 @@
+"""
+Auto-generated API server for PraisonAI agents.
+"""
+from flask import Flask, request, jsonify
+from flask_cors import CORS
+from praisonai import PraisonAI
+import os
+
+app = Flask(__name__)
+
+# CORS configuration
+CORS(app)
+
+# Authentication
+AUTH_ENABLED = False
+AUTH_TOKEN = None
+
+def check_auth():
+    """Check authentication if enabled."""
+    if not AUTH_ENABLED:
+        return True
+    
+    token = request.headers.get('Authorization', '').replace('Bearer ', '')
+    return token == AUTH_TOKEN
+
+@app.route('/health', methods=['GET'])
+def health():
+    """Health check endpoint."""
+    return jsonify({"status": "ok", "service": "praisonai-api"})
+
+@app.route('/chat', methods=['POST'])
+def chat():
+    """Chat endpoint for agent interaction."""
+    if not check_auth():
+        return jsonify({"error": "Unauthorized"}), 401
+    
+    data = request.get_json()
+    if not data or 'message' not in data:
+        return jsonify({"error": "Message required"}), 400
+    
+    try:
+        praisonai = PraisonAI(agent_file="agents.yaml")
+        result = praisonai.run()
+        
+        return jsonify({
+            "response": result,
+            "status": "success"
+        })
+    except Exception as e:
+        return jsonify({
+            "error": str(e),
+            "status": "error"
+        }), 500
+
+@app.route('/agents', methods=['GET'])
+def list_agents():
+    """List available agents."""
+    if not check_auth():
+        return jsonify({"error": "Unauthorized"}), 401
+    
+    return jsonify({
+        "agents": ["default"],
+        "agent_file": "agents.yaml"
+    })
+
+if __name__ == '__main__':
+    app.run(
+        host='0.0.0.0',
+        port=8080,
+        debug=False
+    )

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>=2.8.9" \
+    "praisonai>=2.9.0" \
     "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>=2.8.9" \
+    "praisonai>=2.9.0" \
     "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>=2.8.9" \
+    "praisonai>=2.9.0" \
     "praisonai[ui]" \
     "praisonai[crewai]"
 

examples/debug/debug_interactive_example.py

@@ -0,0 +1,228 @@
+"""
+Debug Interactive Mode Example
+
+This example demonstrates how to use the debug CLI commands to test
+the Interactive Coding Assistant without entering TUI mode.
+
+Features demonstrated:
+- LSP symbol listing
+- LSP definition lookup
+- ACP action planning
+- ACP action execution
+- Trace capture and JSON output
+"""
+
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+# Add the src directory to path for development
+sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src" / "praisonai"))
+
+from praisonai.cli.features.interactive_runtime import create_runtime
+from praisonai.cli.features.code_intelligence import CodeIntelligenceRouter
+from praisonai.cli.features.action_orchestrator import ActionOrchestrator
+
+
+async def demo_lsp_queries(workspace: str):
+    """Demonstrate LSP-based code queries."""
+    print("\n" + "=" * 60)
+    print("LSP CODE QUERIES DEMO")
+    print("=" * 60)
+    
+    runtime = create_runtime(
+        workspace=workspace,
+        lsp=True,
+        acp=False,
+        trace=True
+    )
+    
+    try:
+        await runtime.start()
+        router = CodeIntelligenceRouter(runtime)
+        
+        # Query 1: List symbols
+        print("\n[Query 1] List all symbols in app.py")
+        result = await router.handle_query("list all functions in app.py", "app.py")
+        print(f"  Success: {result.success}")
+        print(f"  LSP Used: {result.lsp_used}")
+        print(f"  Fallback Used: {result.fallback_used}")
+        if result.data:
+            print(f"  Symbols found: {len(result.data)}")
+            for symbol in result.data[:5]:
+                print(f"    - {symbol['name']} ({symbol['kind']}) at line {symbol['line']}")
+        
+        # Query 2: Find definition
+        print("\n[Query 2] Find definition of Calculator")
+        result = await router.handle_query("where is Calculator defined", "app.py")
+        print(f"  Success: {result.success}")
+        if result.citations:
+            print(f"  Found at: {result.citations[0]}")
+        
+        # Get trace
+        trace = runtime.get_trace()
+        print(f"\n[Trace] {len(trace.entries)} entries captured")
+        
+    finally:
+        await runtime.stop()
+
+
+async def demo_acp_actions(workspace: str):
+    """Demonstrate ACP-based code modifications."""
+    print("\n" + "=" * 60)
+    print("ACP CODE MODIFICATIONS DEMO")
+    print("=" * 60)
+    
+    runtime = create_runtime(
+        workspace=workspace,
+        lsp=False,
+        acp=True,
+        approval="auto",
+        trace=True
+    )
+    
+    try:
+        await runtime.start()
+        print(f"\n  ACP Ready: {runtime.acp_ready}")
+        print(f"  Read Only: {runtime.read_only}")
+        
+        orchestrator = ActionOrchestrator(runtime)
+        
+        # Action 1: Create a plan
+        print("\n[Action 1] Create plan for new file")
+        result = await orchestrator.create_plan("create a new file called demo_output.py")
+        print(f"  Success: {result.success}")
+        if result.plan:
+            print(f"  Plan ID: {result.plan.id}")
+            print(f"  Steps: {len(result.plan.steps)}")
+            for step in result.plan.steps:
+                print(f"    - {step.action_type.value}: {step.target}")
+        
+        # Action 2: Execute with auto-approve
+        print("\n[Action 2] Execute file creation")
+        exec_result = await orchestrator.execute(
+            "create a new file called demo_output.py with a hello function",
+            auto_approve=True
+        )
+        print(f"  Success: {exec_result.success}")
+        print(f"  Applied Steps: {exec_result.applied_steps}")
+        
+        # Verify file was created
+        output_file = Path(workspace) / "demo_output.py"
+        if output_file.exists():
+            print(f"  File created: {output_file}")
+            print(f"  Content preview: {output_file.read_text()[:100]}...")
+        
+        # Get trace
+        trace = runtime.get_trace()
+        print(f"\n[Trace] {len(trace.entries)} entries captured")
+        for entry in trace.entries:
+            print(f"    [{entry.category}] {entry.action}")
+        
+    finally:
+        await runtime.stop()
+
+
+async def demo_full_interactive_turn(workspace: str):
+    """Demonstrate full interactive turn simulation."""
+    print("\n" + "=" * 60)
+    print("FULL INTERACTIVE TURN SIMULATION")
+    print("=" * 60)
+    
+    runtime = create_runtime(
+        workspace=workspace,
+        lsp=True,
+        acp=True,
+        approval="auto",
+        trace=True
+    )
+    
+    try:
+        status = await runtime.start()
+        print("\n  Runtime Status:")
+        print(f"    LSP: {'ready' if status['lsp']['ready'] else 'fallback'}")
+        print(f"    ACP: {'ready' if status['acp']['ready'] else 'unavailable'}")
+        
+        router = CodeIntelligenceRouter(runtime)
+        orchestrator = ActionOrchestrator(runtime)
+        
+        # Simulate user prompt: code query
+        prompt1 = "What functions are in app.py?"
+        print(f"\n[Turn 1] User: {prompt1}")
+        
+        # Classify and route
+        result = await router.handle_query(prompt1, "app.py")
+        print(f"  Intent: {result.intent.value}")
+        print(f"  Response: Found {len(result.data) if result.data else 0} symbols")
+        
+        # Simulate user prompt: code modification
+        prompt2 = "Create a new test file"
+        print(f"\n[Turn 2] User: {prompt2}")
+        
+        exec_result = await orchestrator.execute(
+            "create a new file called test_demo.py",
+            auto_approve=True
+        )
+        print("  Action: file_create")
+        print(f"  Result: {'success' if exec_result.success else 'failed'}")
+        
+        # Export trace as JSON
+        trace = runtime.get_trace()
+        trace_json = trace.to_dict()
+        print("[Trace Export]")
+        print(json.dumps(trace_json, indent=2, default=str)[:500] + "...")
+        
+    finally:
+        await runtime.stop()
+
+
+def main():
+    """Run the debug demo."""
+    # Create a temporary workspace
+    workspace = Path(__file__).parent / "_demo_workspace"
+    workspace.mkdir(exist_ok=True)
+    
+    # Create a sample app.py
+    app_file = workspace / "app.py"
+    app_file.write_text('''"""Sample application."""
+
+def greet(name: str) -> str:
+    """Greet someone."""
+    return f"Hello, {name}!"
+
+def add(a: int, b: int) -> int:
+    """Add two numbers."""
+    return a + b
+
+class Calculator:
+    """Simple calculator."""
+    
+    def __init__(self):
+        self.value = 0
+    
+    def add(self, x):
+        self.value += x
+        return self.value
+
+if __name__ == "__main__":
+    print(greet("World"))
+''')
+    
+    print("=" * 60)
+    print("PRAISONAI DEBUG MODE DEMO")
+    print("=" * 60)
+    print(f"Workspace: {workspace}")
+    
+    # Run demos
+    asyncio.run(demo_lsp_queries(str(workspace)))
+    asyncio.run(demo_acp_actions(str(workspace)))
+    asyncio.run(demo_full_interactive_turn(str(workspace)))
+    
+    print("\n" + "=" * 60)
+    print("DEMO COMPLETE")
+    print("=" * 60)
+
+
+if __name__ == "__main__":
+    main()

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/v2.8.9.tar.gz"
-    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.8.9.tar.gz | shasum -a 256`.split.first
+    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.9.0.tar.gz"
+    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.9.0.tar.gz | shasum -a 256`.split.first
     license "MIT"
   
     depends_on "[email protected]"

src/praisonai/praisonai/cli/features/action_orchestrator.py

@@ -0,0 +1,546 @@
+"""
+Action Orchestrator for PraisonAI.
+
+Orchestrates code modifications through ACP with plan → approval → apply → verify steps.
+Enforces read-only mode when ACP is unavailable.
+"""
+
+import logging
+import time
+from dataclasses import dataclass, field
+from enum import Enum
+from pathlib import Path
+from typing import Any, Dict, List, Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .interactive_runtime import InteractiveRuntime
+
+logger = logging.getLogger(__name__)
+
+
+class ActionType(Enum):
+    """Types of actions that can be orchestrated."""
+    FILE_CREATE = "file_create"
+    FILE_EDIT = "file_edit"
+    FILE_DELETE = "file_delete"
+    FILE_RENAME = "file_rename"
+    SHELL_COMMAND = "shell_command"
+    REFACTOR = "refactor"
+    UNKNOWN = "unknown"
+
+
+class ActionStatus(Enum):
+    """Status of an action."""
+    PENDING = "pending"
+    APPROVED = "approved"
+    REJECTED = "rejected"
+    APPLIED = "applied"
+    FAILED = "failed"
+    VERIFIED = "verified"
+
+
+@dataclass
+class ActionStep:
+    """A single step in an action plan."""
+    id: str
+    action_type: ActionType
+    description: str
+    target: str  # File path or command
+    params: Dict[str, Any] = field(default_factory=dict)
+    status: ActionStatus = ActionStatus.PENDING
+    result: Optional[Any] = None
+    error: Optional[str] = None
+
+
+@dataclass
+class ActionPlan:
+    """A plan of actions to execute."""
+    id: str
+    prompt: str
+    steps: List[ActionStep] = field(default_factory=list)
+    created_at: float = field(default_factory=time.time)
+    status: ActionStatus = ActionStatus.PENDING
+    approval_mode: str = "manual"
+    
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "id": self.id,
+            "prompt": self.prompt,
+            "steps": [
+                {
+                    "id": s.id,
+                    "action_type": s.action_type.value,
+                    "description": s.description,
+                    "target": s.target,
+                    "params": s.params,
+                    "status": s.status.value,
+                    "result": s.result,
+                    "error": s.error
+                }
+                for s in self.steps
+            ],
+            "created_at": self.created_at,
+            "status": self.status.value,
+            "approval_mode": self.approval_mode
+        }
+
+
+@dataclass
+class ActionResult:
+    """Result of action orchestration."""
+    success: bool
+    plan: Optional[ActionPlan] = None
+    applied_steps: int = 0
+    failed_steps: int = 0
+    error: Optional[str] = None
+    read_only_blocked: bool = False
+    diff_summary: Optional[str] = None
+    
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "success": self.success,
+            "plan": self.plan.to_dict() if self.plan else None,
+            "applied_steps": self.applied_steps,
+            "failed_steps": self.failed_steps,
+            "error": self.error,
+            "read_only_blocked": self.read_only_blocked,
+            "diff_summary": self.diff_summary
+        }
+
+
+class ActionOrchestrator:
+    """
+    Orchestrates code modifications through ACP.
+    
+    All edits must go through:
+    1. Plan creation
+    2. Approval (manual, auto, or scoped)
+    3. Application
+    4. Verification
+    
+    If ACP is unavailable, runtime is read-only and edits are blocked.
+    """
+    
+    def __init__(self, runtime: "InteractiveRuntime"):
+        """Initialize with runtime reference."""
+        self.runtime = runtime
+        self._plan_counter = 0
+    
+    def _generate_plan_id(self) -> str:
+        """Generate a unique plan ID."""
+        self._plan_counter += 1
+        return f"plan_{int(time.time())}_{self._plan_counter}"
+    
+    def _generate_step_id(self, plan_id: str, index: int) -> str:
+        """Generate a unique step ID."""
+        return f"{plan_id}_step_{index}"
+    
+    async def create_plan(self, prompt: str) -> ActionResult:
+        """
+        Create an action plan for a modification request.
+        
+        Args:
+            prompt: The user's modification request
+            
+        Returns:
+            ActionResult with the plan
+        """
+        if self.runtime.read_only:
+            return ActionResult(
+                success=False,
+                error="Runtime is in read-only mode. ACP is required for modifications.",
+                read_only_blocked=True
+            )
+        
+        plan_id = self._generate_plan_id()
+        
+        # Always use fallback plan creation (ACP session tracks but doesn't create plans)
+        # In production, this would integrate with an LLM for intelligent planning
+        return await self._create_fallback_plan(prompt, plan_id)
+    
+    async def _create_fallback_plan(self, prompt: str, plan_id: str) -> ActionResult:
+        """Create a fallback plan when ACP is not available for planning."""
+        # This is a simplified fallback - in production, this would use LLM
+        steps = []
+        
+        # Analyze prompt for common patterns
+        prompt_lower = prompt.lower()
+        
+        if "create" in prompt_lower or "new file" in prompt_lower:
+            # Extract file path if mentioned
+            import re
+            file_match = re.search(r'(?:create|new)\s+(?:a\s+)?(?:file\s+)?(?:called\s+)?([^\s]+\.\w+)', prompt_lower)
+            target = file_match.group(1) if file_match else "new_file.py"
+            
+            # Generate default content based on file type
+            if target.endswith('.py'):
+                content = f'"""\n{target} - Auto-generated file\n"""\n\ndef main():\n    pass\n\n\nif __name__ == "__main__":\n    main()\n'
+            else:
+                content = f"// {target} - Auto-generated file\n"
+            
+            steps.append(ActionStep(
+                id=self._generate_step_id(plan_id, 0),
+                action_type=ActionType.FILE_CREATE,
+                description=f"Create file: {target}",
+                target=target,
+                params={"content": content},
+                status=ActionStatus.APPROVED  # Auto-approve for fallback
+            ))
+        
+        elif "edit" in prompt_lower or "modify" in prompt_lower or "change" in prompt_lower:
+            file_match = re.search(r'(?:edit|modify|change)\s+([^\s]+\.\w+)', prompt_lower)
+            target = file_match.group(1) if file_match else ""
+            
+            if target:
+                steps.append(ActionStep(
+                    id=self._generate_step_id(plan_id, 0),
+                    action_type=ActionType.FILE_EDIT,
+                    description=f"Edit file: {target}",
+                    target=target,
+                    params={"changes": prompt}
+                ))
+        
+        elif "delete" in prompt_lower or "remove" in prompt_lower:
+            file_match = re.search(r'(?:delete|remove)\s+([^\s]+\.\w+)', prompt_lower)
+            target = file_match.group(1) if file_match else ""
+            
+            if target:
+                steps.append(ActionStep(
+                    id=self._generate_step_id(plan_id, 0),
+                    action_type=ActionType.FILE_DELETE,
+                    description=f"Delete file: {target}",
+                    target=target
+                ))
+        
+        elif "rename" in prompt_lower:
+            steps.append(ActionStep(
+                id=self._generate_step_id(plan_id, 0),
+                action_type=ActionType.REFACTOR,
+                description="Rename operation",
+                target="",
+                params={"prompt": prompt}
+            ))
+        
+        elif "run" in prompt_lower or "execute" in prompt_lower:
+            cmd_match = re.search(r'(?:run|execute)\s+[`"\']?([^`"\']+)[`"\']?', prompt_lower)
+            command = cmd_match.group(1) if cmd_match else ""
+            
+            if command:
+                steps.append(ActionStep(
+                    id=self._generate_step_id(plan_id, 0),
+                    action_type=ActionType.SHELL_COMMAND,
+                    description=f"Execute: {command}",
+                    target=command
+                ))
+        
+        if not steps:
+            return ActionResult(
+                success=False,
+                error="Could not determine action from prompt. Please be more specific."
+            )
+        
+        plan = ActionPlan(
+            id=plan_id,
+            prompt=prompt,
+            steps=steps,
+            approval_mode=self.runtime.config.approval_mode
+        )
+        
+        return ActionResult(
+            success=True,
+            plan=plan
+        )
+    
+    def _classify_action_type(self, step_data: Dict[str, Any]) -> ActionType:
+        """Classify the action type from step data."""
+        action = step_data.get("action", "").lower()
+        
+        if "create" in action:
+            return ActionType.FILE_CREATE
+        elif "edit" in action or "modify" in action:
+            return ActionType.FILE_EDIT
+        elif "delete" in action or "remove" in action:
+            return ActionType.FILE_DELETE
+        elif "rename" in action:
+            return ActionType.FILE_RENAME
+        elif "shell" in action or "command" in action or "run" in action:
+            return ActionType.SHELL_COMMAND
+        elif "refactor" in action:
+            return ActionType.REFACTOR
+        
+        return ActionType.UNKNOWN
+    
+    async def approve_plan(self, plan: ActionPlan, auto: bool = False) -> bool:
+        """
+        Request approval for a plan.
+        
+        Args:
+            plan: The plan to approve
+            auto: Whether to auto-approve
+            
+        Returns:
+            True if approved, False if rejected
+        """
+        if self.runtime.read_only:
+            logger.warning("Cannot approve plan in read-only mode")
+            return False
+        
+        approval_mode = self.runtime.config.approval_mode
+        
+        if approval_mode == "auto" or auto:
+            plan.status = ActionStatus.APPROVED
+            for step in plan.steps:
+                step.status = ActionStatus.APPROVED
+            return True
+        
+        elif approval_mode == "scoped":
+            # Auto-approve safe actions, require manual for dangerous ones
+            dangerous_types = {ActionType.FILE_DELETE, ActionType.SHELL_COMMAND}
+            
+            for step in plan.steps:
+                if step.action_type in dangerous_types:
+                    step.status = ActionStatus.PENDING  # Needs manual approval
+                else:
+                    step.status = ActionStatus.APPROVED
+            
+            # Plan is approved if all steps are approved
+            if all(s.status == ActionStatus.APPROVED for s in plan.steps):
+                plan.status = ActionStatus.APPROVED
+                return True
+            
+            return False  # Some steps need manual approval
+        
+        else:  # manual
+            # In manual mode, we return False to indicate approval is needed
+            return False
+    
+    async def apply_plan(self, plan: ActionPlan, force: bool = False) -> ActionResult:
+        """
+        Apply an approved plan.
+        
+        Args:
+            plan: The plan to apply
+            force: Force apply even if not fully approved
+            
+        Returns:
+            ActionResult with application status
+        """
+        if self.runtime.read_only:
+            return ActionResult(
+                success=False,
+                plan=plan,
+                error="Runtime is in read-only mode",
+                read_only_blocked=True
+            )
+        
+        if plan.status != ActionStatus.APPROVED and not force:
+            return ActionResult(
+                success=False,
+                plan=plan,
+                error="Plan is not approved"
+            )
+        
+        applied = 0
+        failed = 0
+        diff_parts = []
+        
+        # Track in ACP session if available (but always do manual application)
+        if self.runtime.acp_ready:
+            try:
+                auto_approve = self.runtime.config.approval_mode == "auto"
+                await self.runtime.acp_apply_plan(
+                    plan.to_dict(),
+                    auto_approve=auto_approve
+                )
+            except Exception as e:
+                logger.warning(f"ACP tracking failed: {e}")
+        
+        # Manual application (always execute)
+        for step in plan.steps:
+            if step.status not in [ActionStatus.APPROVED, ActionStatus.PENDING]:
+                continue
+            
+            try:
+                result = await self._apply_step(step)
+                if result:
+                    step.status = ActionStatus.APPLIED
+                    step.result = result
+                    applied += 1
+                    if isinstance(result, dict) and "diff" in result:
+                        diff_parts.append(result["diff"])
+                else:
+                    step.status = ActionStatus.FAILED
+                    failed += 1
+            except Exception as e:
+                step.status = ActionStatus.FAILED
+                step.error = str(e)
+                failed += 1
+        
+        plan.status = ActionStatus.APPLIED if failed == 0 else ActionStatus.FAILED
+        
+        return ActionResult(
+            success=failed == 0,
+            plan=plan,
+            applied_steps=applied,
+            failed_steps=failed,
+            diff_summary="\n".join(diff_parts) if diff_parts else None
+        )
+    
+    async def _apply_step(self, step: ActionStep) -> Optional[Dict[str, Any]]:
+        """Apply a single step."""
+        workspace = Path(self.runtime.config.workspace)
+        
+        if step.action_type == ActionType.FILE_CREATE:
+            target = workspace / step.target
+            target.parent.mkdir(parents=True, exist_ok=True)
+            content = step.params.get("content", "")
+            target.write_text(content)
+            return {"created": str(target), "diff": f"+++ {target}\n+ (new file)"}
+        
+        elif step.action_type == ActionType.FILE_EDIT:
+            target = workspace / step.target
+            if not target.exists():
+                return None
+            
+            old_content = target.read_text()
+            new_content = step.params.get("new_content", old_content)
+            target.write_text(new_content)
+            
+            return {
+                "edited": str(target),
+                "diff": f"--- {target}\n+++ {target}\n(content changed)"
+            }
+        
+        elif step.action_type == ActionType.FILE_DELETE:
+            target = workspace / step.target
+            if target.exists():
+                target.unlink()
+                return {"deleted": str(target), "diff": f"--- {target}\n- (deleted)"}
+            return None
+        
+        elif step.action_type == ActionType.FILE_RENAME:
+            source = workspace / step.params.get("source", "")
+            dest = workspace / step.params.get("dest", "")
+            if source.exists():
+                source.rename(dest)
+                return {"renamed": f"{source} -> {dest}"}
+            return None
+        
+        elif step.action_type == ActionType.SHELL_COMMAND:
+            import subprocess
+            result = subprocess.run(
+                step.target,
+                shell=True,
+                capture_output=True,
+                text=True,
+                cwd=str(workspace),
+                timeout=30
+            )
+            return {
+                "command": step.target,
+                "stdout": result.stdout,
+                "stderr": result.stderr,
+                "returncode": result.returncode
+            }
+        
+        return None
+    
+    async def verify_plan(self, plan: ActionPlan) -> ActionResult:
+        """
+        Verify that a plan was applied correctly.
+        
+        Args:
+            plan: The plan to verify
+            
+        Returns:
+            ActionResult with verification status
+        """
+        if plan.status != ActionStatus.APPLIED:
+            return ActionResult(
+                success=False,
+                plan=plan,
+                error="Plan was not applied"
+            )
+        
+        workspace = Path(self.runtime.config.workspace)
+        verified = 0
+        failed = 0
+        
+        for step in plan.steps:
+            if step.status != ActionStatus.APPLIED:
+                continue
+            
+            try:
+                if step.action_type == ActionType.FILE_CREATE:
+                    target = workspace / step.target
+                    if target.exists():
+                        step.status = ActionStatus.VERIFIED
+                        verified += 1
+                    else:
+                        step.status = ActionStatus.FAILED
+                        step.error = "File was not created"
+                        failed += 1
+                
+                elif step.action_type == ActionType.FILE_DELETE:
+                    target = workspace / step.target
+                    if not target.exists():
+                        step.status = ActionStatus.VERIFIED
+                        verified += 1
+                    else:
+                        step.status = ActionStatus.FAILED
+                        step.error = "File was not deleted"
+                        failed += 1
+                
+                else:
+                    # For other types, assume verified if applied
+                    step.status = ActionStatus.VERIFIED
+                    verified += 1
+                    
+            except Exception as e:
+                step.status = ActionStatus.FAILED
+                step.error = str(e)
+                failed += 1
+        
+        plan.status = ActionStatus.VERIFIED if failed == 0 else ActionStatus.FAILED
+        
+        return ActionResult(
+            success=failed == 0,
+            plan=plan,
+            applied_steps=verified,
+            failed_steps=failed
+        )
+    
+    async def execute(self, prompt: str, auto_approve: bool = False) -> ActionResult:
+        """
+        Execute a full plan → approve → apply → verify cycle.
+        
+        Args:
+            prompt: The modification request
+            auto_approve: Whether to auto-approve
+            
+        Returns:
+            ActionResult with final status
+        """
+        # Create plan
+        result = await self.create_plan(prompt)
+        if not result.success:
+            return result
+        
+        plan = result.plan
+        
+        # Approve
+        approved = await self.approve_plan(plan, auto=auto_approve)
+        if not approved and self.runtime.config.approval_mode == "manual":
+            return ActionResult(
+                success=False,
+                plan=plan,
+                error="Plan requires manual approval"
+            )
+        
+        # Apply
+        result = await self.apply_plan(plan)
+        if not result.success:
+            return result
+        
+        # Verify
+        return await self.verify_plan(plan)

src/praisonai/praisonai/cli/features/code_intelligence.py

@@ -0,0 +1,652 @@
+"""
+Code Intelligence Router for PraisonAI.
+
+Routes code-related queries to LSP when available, with fallback to grep/file search.
+"""
+
+import logging
+import re
+from dataclasses import dataclass
+from enum import Enum
+from pathlib import Path
+from typing import Any, Dict, List, Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .interactive_runtime import InteractiveRuntime
+
+logger = logging.getLogger(__name__)
+
+
+class CodeIntent(Enum):
+    """Classification of code-related intents."""
+    LIST_SYMBOLS = "list_symbols"
+    GO_TO_DEFINITION = "go_to_definition"
+    FIND_REFERENCES = "find_references"
+    GET_DIAGNOSTICS = "get_diagnostics"
+    EXPLAIN_CODE = "explain_code"
+    SEARCH_CODE = "search_code"
+    UNKNOWN = "unknown"
+
+
+@dataclass
+class CodeQueryResult:
+    """Result of a code intelligence query."""
+    intent: CodeIntent
+    success: bool
+    lsp_used: bool
+    data: Any = None
+    citations: List[Dict[str, Any]] = None
+    error: Optional[str] = None
+    fallback_used: bool = False
+    
+    def __post_init__(self):
+        if self.citations is None:
+            self.citations = []
+    
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "intent": self.intent.value,
+            "success": self.success,
+            "lsp_used": self.lsp_used,
+            "data": self.data,
+            "citations": self.citations,
+            "error": self.error,
+            "fallback_used": self.fallback_used
+        }
+
+
+class CodeIntelligenceRouter:
+    """
+    Routes code intelligence queries to LSP or fallback mechanisms.
+    
+    LSP-first approach:
+    - If LSP is available, use it for semantic queries
+    - If LSP fails or unavailable, fall back to grep/file search
+    - Always provide citations with file:line references
+    """
+    
+    # Patterns for intent classification
+    INTENT_PATTERNS = {
+        CodeIntent.LIST_SYMBOLS: [
+            r"list\s+(all\s+)?(functions?|classes?|methods?|symbols?)",
+            r"what\s+(functions?|classes?|methods?)\s+(are|exist)",
+            r"show\s+(me\s+)?(all\s+)?(functions?|classes?|symbols?)",
+            r"find\s+(all\s+)?(functions?|classes?|methods?)",
+        ],
+        CodeIntent.GO_TO_DEFINITION: [
+            r"(go\s+to|find|show|where\s+is)\s+(the\s+)?definition",
+            r"where\s+is\s+(\w+)\s+defined",
+            r"definition\s+of\s+(\w+)",
+            r"(\w+)\s+is\s+defined\s+where",
+        ],
+        CodeIntent.FIND_REFERENCES: [
+            r"(find|show|list)\s+(all\s+)?references",
+            r"where\s+is\s+(\w+)\s+used",
+            r"usages?\s+of\s+(\w+)",
+            r"who\s+(calls?|uses?)\s+(\w+)",
+        ],
+        CodeIntent.GET_DIAGNOSTICS: [
+            r"(show|list|get)\s+(all\s+)?(errors?|warnings?|diagnostics?|problems?)",
+            r"what('s|\s+is)\s+wrong",
+            r"any\s+(errors?|issues?|problems?)",
+            r"check\s+(for\s+)?(errors?|issues?)",
+        ],
+        CodeIntent.EXPLAIN_CODE: [
+            r"explain\s+(this|the)\s+code",
+            r"what\s+does\s+(this|the)\s+code\s+do",
+            r"how\s+does\s+(\w+)\s+work",
+        ],
+        CodeIntent.SEARCH_CODE: [
+            r"search\s+(for\s+)?",
+            r"find\s+(the\s+)?string",
+            r"grep\s+",
+        ],
+    }
+    
+    def __init__(self, runtime: "InteractiveRuntime"):
+        """Initialize with runtime reference."""
+        self.runtime = runtime
+    
+    def classify_intent(self, query: str) -> CodeIntent:
+        """Classify the intent of a code query."""
+        query_lower = query.lower()
+        
+        for intent, patterns in self.INTENT_PATTERNS.items():
+            for pattern in patterns:
+                if re.search(pattern, query_lower):
+                    return intent
+        
+        return CodeIntent.UNKNOWN
+    
+    async def handle_query(self, query: str, file_path: str = None) -> CodeQueryResult:
+        """
+        Handle a code intelligence query.
+        
+        Args:
+            query: The user's query
+            file_path: Optional file path context
+            
+        Returns:
+            CodeQueryResult with data and citations
+        """
+        intent = self.classify_intent(query)
+        
+        if intent == CodeIntent.LIST_SYMBOLS:
+            return await self._handle_list_symbols(query, file_path)
+        elif intent == CodeIntent.GO_TO_DEFINITION:
+            return await self._handle_go_to_definition(query, file_path)
+        elif intent == CodeIntent.FIND_REFERENCES:
+            return await self._handle_find_references(query, file_path)
+        elif intent == CodeIntent.GET_DIAGNOSTICS:
+            return await self._handle_get_diagnostics(query, file_path)
+        elif intent == CodeIntent.SEARCH_CODE:
+            return await self._handle_search_code(query, file_path)
+        else:
+            return CodeQueryResult(
+                intent=intent,
+                success=False,
+                lsp_used=False,
+                error="Could not determine code query intent"
+            )
+    
+    async def _handle_list_symbols(self, query: str, file_path: str = None) -> CodeQueryResult:
+        """Handle list symbols query."""
+        if not file_path:
+            # Try to find a file from the query
+            file_path = self._extract_file_from_query(query)
+        
+        if not file_path:
+            return CodeQueryResult(
+                intent=CodeIntent.LIST_SYMBOLS,
+                success=False,
+                lsp_used=False,
+                error="No file specified. Please specify a file path."
+            )
+        
+        # Try LSP first
+        if self.runtime.lsp_ready:
+            try:
+                symbols = await self.runtime.lsp_get_symbols(file_path)
+                if symbols:
+                    citations = [
+                        {"file": file_path, "type": "symbols", "count": len(symbols)}
+                    ]
+                    return CodeQueryResult(
+                        intent=CodeIntent.LIST_SYMBOLS,
+                        success=True,
+                        lsp_used=True,
+                        data=symbols,
+                        citations=citations
+                    )
+            except Exception as e:
+                logger.warning(f"LSP symbols failed: {e}")
+        
+        # Fallback to regex-based symbol extraction
+        return await self._fallback_list_symbols(file_path)
+    
+    async def _fallback_list_symbols(self, file_path: str) -> CodeQueryResult:
+        """Fallback symbol listing using regex."""
+        try:
+            path = Path(file_path)
+            if not path.exists():
+                path = Path(self.runtime.config.workspace) / file_path
+            
+            if not path.exists():
+                return CodeQueryResult(
+                    intent=CodeIntent.LIST_SYMBOLS,
+                    success=False,
+                    lsp_used=False,
+                    fallback_used=True,
+                    error=f"File not found: {file_path}"
+                )
+            
+            content = path.read_text()
+            symbols = []
+            
+            # Python patterns
+            if path.suffix == ".py":
+                # Functions
+                for match in re.finditer(r'^(async\s+)?def\s+(\w+)\s*\(', content, re.MULTILINE):
+                    line_num = content[:match.start()].count('\n') + 1
+                    symbols.append({
+                        "name": match.group(2),
+                        "kind": "function",
+                        "line": line_num
+                    })
+                # Classes
+                for match in re.finditer(r'^class\s+(\w+)\s*[:\(]', content, re.MULTILINE):
+                    line_num = content[:match.start()].count('\n') + 1
+                    symbols.append({
+                        "name": match.group(1),
+                        "kind": "class",
+                        "line": line_num
+                    })
+            
+            # JavaScript/TypeScript patterns
+            elif path.suffix in [".js", ".ts", ".jsx", ".tsx"]:
+                # Functions
+                for match in re.finditer(r'(?:async\s+)?function\s+(\w+)\s*\(', content):
+                    line_num = content[:match.start()].count('\n') + 1
+                    symbols.append({
+                        "name": match.group(1),
+                        "kind": "function",
+                        "line": line_num
+                    })
+                # Arrow functions assigned to const/let/var
+                for match in re.finditer(r'(?:const|let|var)\s+(\w+)\s*=\s*(?:async\s+)?\(', content):
+                    line_num = content[:match.start()].count('\n') + 1
+                    symbols.append({
+                        "name": match.group(1),
+                        "kind": "function",
+                        "line": line_num
+                    })
+                # Classes
+                for match in re.finditer(r'class\s+(\w+)\s*(?:extends|implements|{)', content):
+                    line_num = content[:match.start()].count('\n') + 1
+                    symbols.append({
+                        "name": match.group(1),
+                        "kind": "class",
+                        "line": line_num
+                    })
+            
+            citations = [
+                {"file": str(path), "type": "symbols", "count": len(symbols)}
+            ]
+            
+            return CodeQueryResult(
+                intent=CodeIntent.LIST_SYMBOLS,
+                success=True,
+                lsp_used=False,
+                fallback_used=True,
+                data=symbols,
+                citations=citations
+            )
+            
+        except Exception as e:
+            return CodeQueryResult(
+                intent=CodeIntent.LIST_SYMBOLS,
+                success=False,
+                lsp_used=False,
+                fallback_used=True,
+                error=str(e)
+            )
+    
+    async def _handle_go_to_definition(self, query: str, file_path: str = None) -> CodeQueryResult:
+        """Handle go to definition query."""
+        # Extract symbol name from query
+        symbol = self._extract_symbol_from_query(query)
+        
+        if not symbol:
+            return CodeQueryResult(
+                intent=CodeIntent.GO_TO_DEFINITION,
+                success=False,
+                lsp_used=False,
+                error="Could not identify symbol to find definition for"
+            )
+        
+        if not file_path:
+            file_path = self._extract_file_from_query(query)
+        
+        # Try LSP first
+        if self.runtime.lsp_ready and file_path:
+            try:
+                # Need to find the symbol position first
+                line, col = await self._find_symbol_position(file_path, symbol)
+                if line is not None:
+                    definitions = await self.runtime.lsp_get_definition(file_path, line, col)
+                    if definitions:
+                        citations = [
+                            {"file": d.get("uri", "").replace("file://", ""), 
+                             "line": d.get("range", {}).get("start", {}).get("line", 0) + 1,
+                             "type": "definition"}
+                            for d in definitions
+                        ]
+                        return CodeQueryResult(
+                            intent=CodeIntent.GO_TO_DEFINITION,
+                            success=True,
+                            lsp_used=True,
+                            data={"symbol": symbol, "definitions": definitions},
+                            citations=citations
+                        )
+            except Exception as e:
+                logger.warning(f"LSP definition failed: {e}")
+        
+        # Fallback to grep
+        return await self._fallback_find_definition(symbol)
+    
+    async def _fallback_find_definition(self, symbol: str) -> CodeQueryResult:
+        """Fallback definition finding using grep."""
+        try:
+            import subprocess
+            workspace = self.runtime.config.workspace
+            
+            # Search for definition patterns
+            patterns = [
+                f"def {symbol}\\s*\\(",  # Python function
+                f"class {symbol}\\s*[:\\(]",  # Python class
+                f"function {symbol}\\s*\\(",  # JS function
+                f"const {symbol}\\s*=",  # JS const
+                f"let {symbol}\\s*=",  # JS let
+            ]
+            
+            results = []
+            for pattern in patterns:
+                try:
+                    result = subprocess.run(
+                        ["grep", "-rn", "-E", pattern, workspace],
+                        capture_output=True,
+                        text=True,
+                        timeout=5
+                    )
+                    if result.stdout:
+                        for line in result.stdout.strip().split('\n'):
+                            if line:
+                                parts = line.split(':', 2)
+                                if len(parts) >= 2:
+                                    results.append({
+                                        "file": parts[0],
+                                        "line": int(parts[1]),
+                                        "content": parts[2] if len(parts) > 2 else ""
+                                    })
+                except subprocess.TimeoutExpired:
+                    pass
+            
+            if results:
+                citations = [
+                    {"file": r["file"], "line": r["line"], "type": "definition"}
+                    for r in results
+                ]
+                return CodeQueryResult(
+                    intent=CodeIntent.GO_TO_DEFINITION,
+                    success=True,
+                    lsp_used=False,
+                    fallback_used=True,
+                    data={"symbol": symbol, "definitions": results},
+                    citations=citations
+                )
+            
+            return CodeQueryResult(
+                intent=CodeIntent.GO_TO_DEFINITION,
+                success=False,
+                lsp_used=False,
+                fallback_used=True,
+                error=f"Definition for '{symbol}' not found"
+            )
+            
+        except Exception as e:
+            return CodeQueryResult(
+                intent=CodeIntent.GO_TO_DEFINITION,
+                success=False,
+                lsp_used=False,
+                fallback_used=True,
+                error=str(e)
+            )
+    
+    async def _handle_find_references(self, query: str, file_path: str = None) -> CodeQueryResult:
+        """Handle find references query."""
+        symbol = self._extract_symbol_from_query(query)
+        
+        if not symbol:
+            return CodeQueryResult(
+                intent=CodeIntent.FIND_REFERENCES,
+                success=False,
+                lsp_used=False,
+                error="Could not identify symbol to find references for"
+            )
+        
+        if not file_path:
+            file_path = self._extract_file_from_query(query)
+        
+        # Try LSP first
+        if self.runtime.lsp_ready and file_path:
+            try:
+                line, col = await self._find_symbol_position(file_path, symbol)
+                if line is not None:
+                    references = await self.runtime.lsp_get_references(file_path, line, col)
+                    if references:
+                        citations = [
+                            {"file": r.get("uri", "").replace("file://", ""),
+                             "line": r.get("range", {}).get("start", {}).get("line", 0) + 1,
+                             "type": "reference"}
+                            for r in references
+                        ]
+                        return CodeQueryResult(
+                            intent=CodeIntent.FIND_REFERENCES,
+                            success=True,
+                            lsp_used=True,
+                            data={"symbol": symbol, "references": references},
+                            citations=citations
+                        )
+            except Exception as e:
+                logger.warning(f"LSP references failed: {e}")
+        
+        # Fallback to grep
+        return await self._fallback_find_references(symbol)
+    
+    async def _fallback_find_references(self, symbol: str) -> CodeQueryResult:
+        """Fallback reference finding using grep."""
+        try:
+            import subprocess
+            workspace = self.runtime.config.workspace
+            
+            result = subprocess.run(
+                ["grep", "-rn", "-w", symbol, workspace],
+                capture_output=True,
+                text=True,
+                timeout=10
+            )
+            
+            results = []
+            if result.stdout:
+                for line in result.stdout.strip().split('\n'):
+                    if line:
+                        parts = line.split(':', 2)
+                        if len(parts) >= 2:
+                            results.append({
+                                "file": parts[0],
+                                "line": int(parts[1]),
+                                "content": parts[2] if len(parts) > 2 else ""
+                            })
+            
+            if results:
+                citations = [
+                    {"file": r["file"], "line": r["line"], "type": "reference"}
+                    for r in results
+                ]
+                return CodeQueryResult(
+                    intent=CodeIntent.FIND_REFERENCES,
+                    success=True,
+                    lsp_used=False,
+                    fallback_used=True,
+                    data={"symbol": symbol, "references": results},
+                    citations=citations
+                )
+            
+            return CodeQueryResult(
+                intent=CodeIntent.FIND_REFERENCES,
+                success=False,
+                lsp_used=False,
+                fallback_used=True,
+                error=f"No references found for '{symbol}'"
+            )
+            
+        except Exception as e:
+            return CodeQueryResult(
+                intent=CodeIntent.FIND_REFERENCES,
+                success=False,
+                lsp_used=False,
+                fallback_used=True,
+                error=str(e)
+            )
+    
+    async def _handle_get_diagnostics(self, query: str, file_path: str = None) -> CodeQueryResult:
+        """Handle get diagnostics query."""
+        if not file_path:
+            file_path = self._extract_file_from_query(query)
+        
+        # Try LSP
+        if self.runtime.lsp_ready:
+            try:
+                diagnostics = await self.runtime.lsp_get_diagnostics(file_path)
+                citations = []
+                if diagnostics:
+                    for d in diagnostics:
+                        if isinstance(d, dict):
+                            citations.append({
+                                "file": file_path or "workspace",
+                                "line": d.get("range", {}).get("start", {}).get("line", 0) + 1,
+                                "type": "diagnostic",
+                                "severity": d.get("severity", "unknown")
+                            })
+                
+                return CodeQueryResult(
+                    intent=CodeIntent.GET_DIAGNOSTICS,
+                    success=True,
+                    lsp_used=True,
+                    data=diagnostics,
+                    citations=citations
+                )
+            except Exception as e:
+                logger.warning(f"LSP diagnostics failed: {e}")
+        
+        return CodeQueryResult(
+            intent=CodeIntent.GET_DIAGNOSTICS,
+            success=False,
+            lsp_used=False,
+            error="LSP not available for diagnostics"
+        )
+    
+    async def _handle_search_code(self, query: str, file_path: str = None) -> CodeQueryResult:
+        """Handle code search query."""
+        # Extract search term
+        search_term = self._extract_search_term(query)
+        
+        if not search_term:
+            return CodeQueryResult(
+                intent=CodeIntent.SEARCH_CODE,
+                success=False,
+                lsp_used=False,
+                error="Could not identify search term"
+            )
+        
+        try:
+            import subprocess
+            workspace = self.runtime.config.workspace
+            
+            cmd = ["grep", "-rn", search_term]
+            if file_path:
+                cmd.append(file_path)
+            else:
+                cmd.append(workspace)
+            
+            result = subprocess.run(
+                cmd,
+                capture_output=True,
+                text=True,
+                timeout=10
+            )
+            
+            results = []
+            if result.stdout:
+                for line in result.stdout.strip().split('\n')[:50]:  # Limit results
+                    if line:
+                        parts = line.split(':', 2)
+                        if len(parts) >= 2:
+                            results.append({
+                                "file": parts[0],
+                                "line": int(parts[1]),
+                                "content": parts[2] if len(parts) > 2 else ""
+                            })
+            
+            citations = [
+                {"file": r["file"], "line": r["line"], "type": "search_result"}
+                for r in results
+            ]
+            
+            return CodeQueryResult(
+                intent=CodeIntent.SEARCH_CODE,
+                success=True,
+                lsp_used=False,
+                data={"search_term": search_term, "results": results},
+                citations=citations
+            )
+            
+        except Exception as e:
+            return CodeQueryResult(
+                intent=CodeIntent.SEARCH_CODE,
+                success=False,
+                lsp_used=False,
+                error=str(e)
+            )
+    
+    def _extract_file_from_query(self, query: str) -> Optional[str]:
+        """Extract file path from query."""
+        # Look for file patterns
+        patterns = [
+            r'(?:in|from|file)\s+["\']?([^\s"\']+\.\w+)["\']?',
+            r'([^\s]+\.(?:py|js|ts|tsx|jsx|go|rs|java|cpp|c|h))',
+        ]
+        
+        for pattern in patterns:
+            match = re.search(pattern, query, re.IGNORECASE)
+            if match:
+                return match.group(1)
+        
+        return None
+    
+    def _extract_symbol_from_query(self, query: str) -> Optional[str]:
+        """Extract symbol name from query."""
+        # Common patterns for symbol extraction
+        patterns = [
+            r'(?:definition\s+of|where\s+is|find|show)\s+["\']?(\w+)["\']?',
+            r'["\'](\w+)["\']',
+            r'`(\w+)`',
+            r'(\w+)\s+(?:is\s+)?(?:defined|used)',
+        ]
+        
+        for pattern in patterns:
+            match = re.search(pattern, query, re.IGNORECASE)
+            if match:
+                symbol = match.group(1)
+                # Filter out common words
+                if symbol.lower() not in ['the', 'a', 'an', 'is', 'are', 'in', 'of', 'to', 'for']:
+                    return symbol
+        
+        return None
+    
+    def _extract_search_term(self, query: str) -> Optional[str]:
+        """Extract search term from query."""
+        patterns = [
+            r'search\s+(?:for\s+)?["\']([^"\']+)["\']',
+            r'find\s+["\']([^"\']+)["\']',
+            r'grep\s+["\']?([^\s"\']+)["\']?',
+        ]
+        
+        for pattern in patterns:
+            match = re.search(pattern, query, re.IGNORECASE)
+            if match:
+                return match.group(1)
+        
+        return None
+    
+    async def _find_symbol_position(self, file_path: str, symbol: str) -> tuple:
+        """Find the position of a symbol in a file."""
+        try:
+            path = Path(file_path)
+            if not path.exists():
+                path = Path(self.runtime.config.workspace) / file_path
+            
+            if not path.exists():
+                return None, None
+            
+            content = path.read_text()
+            lines = content.split('\n')
+            
+            for i, line in enumerate(lines):
+                col = line.find(symbol)
+                if col >= 0:
+                    return i, col
+            
+            return None, None
+        except Exception:
+            return None, None

src/praisonai/praisonai/cli/features/debug.py

@@ -0,0 +1,810 @@
+"""
+Debug CLI handler for PraisonAI.
+
+Provides non-interactive debug commands for testing interactive flows:
+- debug interactive: Run single interactive turn
+- debug lsp: Direct LSP probes
+- debug acp: Direct ACP probes
+- debug trace: Trace record/replay/diff
+"""
+
+import argparse
+import asyncio
+import json
+import logging
+import sys
+import time
+from typing import Any, Dict, List
+
+logger = logging.getLogger(__name__)
+
+
+class DebugHandler:
+    """Handler for debug CLI commands."""
+    
+    @staticmethod
+    def setup_subparser(subparsers) -> None:
+        """Set up the debug subcommand parser."""
+        debug_parser = subparsers.add_parser(
+            "debug",
+            help="Debug and test interactive flows",
+            description="Debug commands for testing interactive coding assistant flows"
+        )
+        
+        debug_subparsers = debug_parser.add_subparsers(dest="debug_command", help="Debug subcommands")
+        
+        # debug interactive
+        interactive_parser = debug_subparsers.add_parser(
+            "interactive",
+            help="Run single interactive turn non-interactively",
+            description="Execute one interactive turn with full LSP/ACP pipeline"
+        )
+        interactive_parser.add_argument(
+            "-p", "--prompt",
+            type=str,
+            required=True,
+            help="Prompt to execute"
+        )
+        interactive_parser.add_argument(
+            "--lsp",
+            action="store_true",
+            default=True,
+            help="Enable LSP (default: enabled)"
+        )
+        interactive_parser.add_argument(
+            "--no-lsp",
+            action="store_true",
+            help="Disable LSP"
+        )
+        interactive_parser.add_argument(
+            "--acp",
+            action="store_true",
+            default=True,
+            help="Enable ACP (default: enabled)"
+        )
+        interactive_parser.add_argument(
+            "--no-acp",
+            action="store_true",
+            help="Disable ACP"
+        )
+        interactive_parser.add_argument(
+            "--approval",
+            type=str,
+            choices=["manual", "auto", "scoped"],
+            default="auto",
+            help="Approval mode (default: auto for debug)"
+        )
+        interactive_parser.add_argument(
+            "--workspace", "-w",
+            type=str,
+            default=".",
+            help="Workspace root directory"
+        )
+        interactive_parser.add_argument(
+            "--json",
+            action="store_true",
+            help="Output JSON trace"
+        )
+        interactive_parser.add_argument(
+            "--trace-file",
+            type=str,
+            help="Save trace to file"
+        )
+        interactive_parser.add_argument(
+            "--timeout",
+            type=float,
+            default=60.0,
+            help="Timeout in seconds"
+        )
+        interactive_parser.add_argument(
+            "--model",
+            type=str,
+            help="LLM model to use"
+        )
+        interactive_parser.set_defaults(func=DebugHandler.handle_interactive)
+        
+        # debug lsp
+        lsp_parser = debug_subparsers.add_parser(
+            "lsp",
+            help="Direct LSP probes",
+            description="Probe LSP server directly"
+        )
+        lsp_subparsers = lsp_parser.add_subparsers(dest="lsp_command", help="LSP subcommands")
+        
+        # debug lsp status
+        lsp_status = lsp_subparsers.add_parser("status", help="Show LSP status")
+        lsp_status.add_argument("--workspace", "-w", type=str, default=".")
+        lsp_status.add_argument("--json", action="store_true")
+        lsp_status.set_defaults(func=DebugHandler.handle_lsp_status)
+        
+        # debug lsp symbols
+        lsp_symbols = lsp_subparsers.add_parser("symbols", help="List symbols in file")
+        lsp_symbols.add_argument("file", type=str, help="File path")
+        lsp_symbols.add_argument("--workspace", "-w", type=str, default=".")
+        lsp_symbols.add_argument("--json", action="store_true")
+        lsp_symbols.set_defaults(func=DebugHandler.handle_lsp_symbols)
+        
+        # debug lsp definition
+        lsp_def = lsp_subparsers.add_parser("definition", help="Get definition location")
+        lsp_def.add_argument("location", type=str, help="file:line:col")
+        lsp_def.add_argument("--workspace", "-w", type=str, default=".")
+        lsp_def.add_argument("--json", action="store_true")
+        lsp_def.set_defaults(func=DebugHandler.handle_lsp_definition)
+        
+        # debug lsp references
+        lsp_refs = lsp_subparsers.add_parser("references", help="Get references")
+        lsp_refs.add_argument("location", type=str, help="file:line:col")
+        lsp_refs.add_argument("--workspace", "-w", type=str, default=".")
+        lsp_refs.add_argument("--json", action="store_true")
+        lsp_refs.set_defaults(func=DebugHandler.handle_lsp_references)
+        
+        # debug lsp diagnostics
+        lsp_diag = lsp_subparsers.add_parser("diagnostics", help="Get diagnostics")
+        lsp_diag.add_argument("file", type=str, nargs="?", help="File path (optional)")
+        lsp_diag.add_argument("--all", action="store_true", help="Get all diagnostics")
+        lsp_diag.add_argument("--workspace", "-w", type=str, default=".")
+        lsp_diag.add_argument("--json", action="store_true")
+        lsp_diag.set_defaults(func=DebugHandler.handle_lsp_diagnostics)
+        
+        # debug acp
+        acp_parser = debug_subparsers.add_parser(
+            "acp",
+            help="Direct ACP probes",
+            description="Probe ACP server directly"
+        )
+        acp_subparsers = acp_parser.add_subparsers(dest="acp_command", help="ACP subcommands")
+        
+        # debug acp status
+        acp_status = acp_subparsers.add_parser("status", help="Show ACP status")
+        acp_status.add_argument("--workspace", "-w", type=str, default=".")
+        acp_status.add_argument("--json", action="store_true")
+        acp_status.set_defaults(func=DebugHandler.handle_acp_status)
+        
+        # debug acp plan
+        acp_plan = acp_subparsers.add_parser("plan", help="Create action plan")
+        acp_plan.add_argument("-p", "--prompt", type=str, required=True)
+        acp_plan.add_argument("--workspace", "-w", type=str, default=".")
+        acp_plan.add_argument("--json", action="store_true")
+        acp_plan.set_defaults(func=DebugHandler.handle_acp_plan)
+        
+        # debug acp apply
+        acp_apply = acp_subparsers.add_parser("apply", help="Apply action plan")
+        acp_apply.add_argument("-p", "--prompt", type=str, required=True)
+        acp_apply.add_argument("--approval", type=str, choices=["manual", "auto", "scoped"], default="auto")
+        acp_apply.add_argument("--workspace", "-w", type=str, default=".")
+        acp_apply.add_argument("--json", action="store_true")
+        acp_apply.set_defaults(func=DebugHandler.handle_acp_apply)
+        
+        # debug trace
+        trace_parser = debug_subparsers.add_parser(
+            "trace",
+            help="Trace record/replay/diff",
+            description="Manage trace recordings"
+        )
+        trace_subparsers = trace_parser.add_subparsers(dest="trace_command", help="Trace subcommands")
+        
+        # debug trace record
+        trace_record = trace_subparsers.add_parser("record", help="Record session trace")
+        trace_record.add_argument("-o", "--output", type=str, required=True, help="Output file")
+        trace_record.set_defaults(func=DebugHandler.handle_trace_record)
+        
+        # debug trace replay
+        trace_replay = trace_subparsers.add_parser("replay", help="Replay recorded trace")
+        trace_replay.add_argument("file", type=str, help="Trace file to replay")
+        trace_replay.add_argument("--json", action="store_true")
+        trace_replay.set_defaults(func=DebugHandler.handle_trace_replay)
+        
+        # debug trace diff
+        trace_diff = trace_subparsers.add_parser("diff", help="Compare two traces")
+        trace_diff.add_argument("file1", type=str, help="First trace file")
+        trace_diff.add_argument("file2", type=str, help="Second trace file")
+        trace_diff.add_argument("--json", action="store_true")
+        trace_diff.set_defaults(func=DebugHandler.handle_trace_diff)
+    
+    @staticmethod
+    def handle_interactive(args: argparse.Namespace) -> int:
+        """Handle debug interactive command."""
+        return asyncio.run(DebugHandler._run_interactive(args))
+    
+    @staticmethod
+    async def _run_interactive(args: argparse.Namespace) -> int:
+        """Run interactive turn asynchronously."""
+        from .interactive_runtime import create_runtime
+        from .code_intelligence import CodeIntelligenceRouter
+        from .action_orchestrator import ActionOrchestrator
+        
+        start_time = time.time()
+        
+        # Create runtime
+        runtime = create_runtime(
+            workspace=args.workspace,
+            lsp=args.lsp and not args.no_lsp,
+            acp=args.acp and not args.no_acp,
+            approval=args.approval,
+            trace=True,
+            trace_file=args.trace_file,
+            json_output=args.json,
+            timeout=args.timeout,
+            model=getattr(args, 'model', None)
+        )
+        
+        result = {
+            "prompt": args.prompt,
+            "workspace": args.workspace,
+            "start_time": start_time
+        }
+        
+        try:
+            # Start runtime
+            status = await runtime.start()
+            result["runtime_status"] = status
+            
+            if not status.get("started"):
+                result["error"] = "Failed to start runtime"
+                DebugHandler._output_result(result, args.json)
+                return 1
+            
+            # Classify intent
+            code_router = CodeIntelligenceRouter(runtime)
+            action_orchestrator = ActionOrchestrator(runtime)
+            
+            prompt = args.prompt.lower()
+            
+            # Determine if this is a code query or action
+            is_code_query = any(kw in prompt for kw in [
+                "list", "show", "find", "where", "definition", "reference",
+                "symbol", "function", "class", "diagnostic", "error"
+            ])
+            
+            is_action = any(kw in prompt for kw in [
+                "create", "edit", "modify", "delete", "rename", "add", "remove",
+                "change", "update", "refactor"
+            ])
+            
+            if is_code_query:
+                # Route to code intelligence
+                query_result = await code_router.handle_query(args.prompt)
+                result["type"] = "code_query"
+                result["query_result"] = query_result.to_dict()
+                result["lsp_used"] = query_result.lsp_used
+                result["citations"] = query_result.citations
+                
+            elif is_action:
+                # Route to action orchestrator
+                action_result = await action_orchestrator.execute(
+                    args.prompt,
+                    auto_approve=(args.approval == "auto")
+                )
+                result["type"] = "action"
+                result["action_result"] = action_result.to_dict()
+                result["read_only_blocked"] = action_result.read_only_blocked
+                
+            else:
+                # General query - just return status
+                result["type"] = "general"
+                result["message"] = "Query processed"
+            
+            # Get trace
+            trace = runtime.get_trace()
+            if trace:
+                result["trace"] = trace.to_dict()
+            
+            result["success"] = True
+            
+        except Exception as e:
+            result["error"] = str(e)
+            result["success"] = False
+            logger.exception("Debug interactive failed")
+            
+        finally:
+            await runtime.stop()
+            result["end_time"] = time.time()
+            result["duration_ms"] = (result["end_time"] - start_time) * 1000
+        
+        # Save trace if requested
+        if args.trace_file and "trace" in result:
+            with open(args.trace_file, "w") as f:
+                json.dump(result["trace"], f, indent=2, default=str)
+            if not args.json:
+                print(f"Trace saved to: {args.trace_file}")
+        
+        DebugHandler._output_result(result, args.json)
+        return 0 if result.get("success") else 1
+    
+    @staticmethod
+    def handle_lsp_status(args: argparse.Namespace) -> int:
+        """Handle debug lsp status command."""
+        return asyncio.run(DebugHandler._run_lsp_status(args))
+    
+    @staticmethod
+    async def _run_lsp_status(args: argparse.Namespace) -> int:
+        """Run LSP status check."""
+        from .interactive_runtime import create_runtime
+        
+        runtime = create_runtime(
+            workspace=args.workspace,
+            lsp=True,
+            acp=False
+        )
+        
+        try:
+            await runtime.start()
+            result = {
+                "lsp_enabled": True,
+                "lsp_ready": runtime.lsp_ready,
+                "lsp_status": runtime._lsp_state.status.value,
+                "lsp_error": runtime._lsp_state.error,
+                "workspace": args.workspace
+            }
+            
+            DebugHandler._output_result(result, args.json)
+            return 0 if runtime.lsp_ready else 1
+            
+        finally:
+            await runtime.stop()
+    
+    @staticmethod
+    def handle_lsp_symbols(args: argparse.Namespace) -> int:
+        """Handle debug lsp symbols command."""
+        return asyncio.run(DebugHandler._run_lsp_symbols(args))
+    
+    @staticmethod
+    async def _run_lsp_symbols(args: argparse.Namespace) -> int:
+        """Run LSP symbols query."""
+        from .interactive_runtime import create_runtime
+        from .code_intelligence import CodeIntelligenceRouter
+        
+        runtime = create_runtime(
+            workspace=args.workspace,
+            lsp=True,
+            acp=False
+        )
+        
+        try:
+            await runtime.start()
+            
+            if not runtime.lsp_ready:
+                # Use fallback
+                router = CodeIntelligenceRouter(runtime)
+                result = await router._fallback_list_symbols(args.file)
+            else:
+                symbols = await runtime.lsp_get_symbols(args.file)
+                result = {
+                    "file": args.file,
+                    "symbols": symbols,
+                    "count": len(symbols) if symbols else 0,
+                    "lsp_used": True
+                }
+            
+            if hasattr(result, 'to_dict'):
+                result = result.to_dict()
+            
+            DebugHandler._output_result(result, args.json)
+            return 0
+            
+        finally:
+            await runtime.stop()
+    
+    @staticmethod
+    def handle_lsp_definition(args: argparse.Namespace) -> int:
+        """Handle debug lsp definition command."""
+        return asyncio.run(DebugHandler._run_lsp_definition(args))
+    
+    @staticmethod
+    async def _run_lsp_definition(args: argparse.Namespace) -> int:
+        """Run LSP definition query."""
+        from .interactive_runtime import create_runtime
+        
+        # Parse location
+        parts = args.location.split(":")
+        if len(parts) != 3:
+            print("Error: Location must be in format file:line:col", file=sys.stderr)
+            return 1
+        
+        file_path, line, col = parts[0], int(parts[1]), int(parts[2])
+        
+        runtime = create_runtime(
+            workspace=args.workspace,
+            lsp=True,
+            acp=False
+        )
+        
+        try:
+            await runtime.start()
+            
+            definitions = await runtime.lsp_get_definition(file_path, line, col)
+            result = {
+                "file": file_path,
+                "line": line,
+                "col": col,
+                "definitions": definitions,
+                "count": len(definitions) if definitions else 0,
+                "lsp_used": runtime.lsp_ready
+            }
+            
+            DebugHandler._output_result(result, args.json)
+            return 0
+            
+        finally:
+            await runtime.stop()
+    
+    @staticmethod
+    def handle_lsp_references(args: argparse.Namespace) -> int:
+        """Handle debug lsp references command."""
+        return asyncio.run(DebugHandler._run_lsp_references(args))
+    
+    @staticmethod
+    async def _run_lsp_references(args: argparse.Namespace) -> int:
+        """Run LSP references query."""
+        from .interactive_runtime import create_runtime
+        
+        # Parse location
+        parts = args.location.split(":")
+        if len(parts) != 3:
+            print("Error: Location must be in format file:line:col", file=sys.stderr)
+            return 1
+        
+        file_path, line, col = parts[0], int(parts[1]), int(parts[2])
+        
+        runtime = create_runtime(
+            workspace=args.workspace,
+            lsp=True,
+            acp=False
+        )
+        
+        try:
+            await runtime.start()
+            
+            references = await runtime.lsp_get_references(file_path, line, col)
+            result = {
+                "file": file_path,
+                "line": line,
+                "col": col,
+                "references": references,
+                "count": len(references) if references else 0,
+                "lsp_used": runtime.lsp_ready
+            }
+            
+            DebugHandler._output_result(result, args.json)
+            return 0
+            
+        finally:
+            await runtime.stop()
+    
+    @staticmethod
+    def handle_lsp_diagnostics(args: argparse.Namespace) -> int:
+        """Handle debug lsp diagnostics command."""
+        return asyncio.run(DebugHandler._run_lsp_diagnostics(args))
+    
+    @staticmethod
+    async def _run_lsp_diagnostics(args: argparse.Namespace) -> int:
+        """Run LSP diagnostics query."""
+        from .interactive_runtime import create_runtime
+        
+        runtime = create_runtime(
+            workspace=args.workspace,
+            lsp=True,
+            acp=False
+        )
+        
+        try:
+            await runtime.start()
+            
+            file_path = args.file if not args.all else None
+            diagnostics = await runtime.lsp_get_diagnostics(file_path)
+            
+            result = {
+                "file": file_path or "all",
+                "diagnostics": diagnostics,
+                "count": len(diagnostics) if diagnostics else 0,
+                "lsp_used": runtime.lsp_ready
+            }
+            
+            DebugHandler._output_result(result, args.json)
+            return 0
+            
+        finally:
+            await runtime.stop()
+    
+    @staticmethod
+    def handle_acp_status(args: argparse.Namespace) -> int:
+        """Handle debug acp status command."""
+        return asyncio.run(DebugHandler._run_acp_status(args))
+    
+    @staticmethod
+    async def _run_acp_status(args: argparse.Namespace) -> int:
+        """Run ACP status check."""
+        from .interactive_runtime import create_runtime
+        
+        runtime = create_runtime(
+            workspace=args.workspace,
+            lsp=False,
+            acp=True
+        )
+        
+        try:
+            await runtime.start()
+            result = {
+                "acp_enabled": True,
+                "acp_ready": runtime.acp_ready,
+                "acp_status": runtime._acp_state.status.value,
+                "acp_error": runtime._acp_state.error,
+                "read_only": runtime.read_only,
+                "workspace": args.workspace
+            }
+            
+            DebugHandler._output_result(result, args.json)
+            return 0 if runtime.acp_ready else 1
+            
+        finally:
+            await runtime.stop()
+    
+    @staticmethod
+    def handle_acp_plan(args: argparse.Namespace) -> int:
+        """Handle debug acp plan command."""
+        return asyncio.run(DebugHandler._run_acp_plan(args))
+    
+    @staticmethod
+    async def _run_acp_plan(args: argparse.Namespace) -> int:
+        """Run ACP plan creation."""
+        from .interactive_runtime import create_runtime
+        from .action_orchestrator import ActionOrchestrator
+        
+        runtime = create_runtime(
+            workspace=args.workspace,
+            lsp=False,
+            acp=True
+        )
+        
+        try:
+            await runtime.start()
+            
+            orchestrator = ActionOrchestrator(runtime)
+            result = await orchestrator.create_plan(args.prompt)
+            
+            DebugHandler._output_result(result.to_dict(), args.json)
+            return 0 if result.success else 1
+            
+        finally:
+            await runtime.stop()
+    
+    @staticmethod
+    def handle_acp_apply(args: argparse.Namespace) -> int:
+        """Handle debug acp apply command."""
+        return asyncio.run(DebugHandler._run_acp_apply(args))
+    
+    @staticmethod
+    async def _run_acp_apply(args: argparse.Namespace) -> int:
+        """Run ACP plan and apply."""
+        from .interactive_runtime import create_runtime
+        from .action_orchestrator import ActionOrchestrator
+        
+        runtime = create_runtime(
+            workspace=args.workspace,
+            lsp=False,
+            acp=True,
+            approval=args.approval
+        )
+        
+        try:
+            await runtime.start()
+            
+            orchestrator = ActionOrchestrator(runtime)
+            result = await orchestrator.execute(
+                args.prompt,
+                auto_approve=(args.approval == "auto")
+            )
+            
+            DebugHandler._output_result(result.to_dict(), args.json)
+            return 0 if result.success else 1
+            
+        finally:
+            await runtime.stop()
+    
+    @staticmethod
+    def handle_trace_record(args: argparse.Namespace) -> int:
+        """Handle debug trace record command."""
+        print(f"Trace recording will be saved to: {args.output}")
+        print("Start an interactive session with --trace-file to record.")
+        return 0
+    
+    @staticmethod
+    def handle_trace_replay(args: argparse.Namespace) -> int:
+        """Handle debug trace replay command."""
+        try:
+            with open(args.file) as f:
+                trace = json.load(f)
+            
+            if args.json:
+                print(json.dumps(trace, indent=2))
+            else:
+                print(f"Trace from: {args.file}")
+                print(f"Start time: {trace.get('start_time')}")
+                print(f"End time: {trace.get('end_time')}")
+                print(f"Entries: {len(trace.get('entries', []))}")
+                print("\nEntries:")
+                for entry in trace.get("entries", []):
+                    print(f"  [{entry.get('category')}] {entry.get('action')}")
+                    if entry.get('duration_ms'):
+                        print(f"    Duration: {entry.get('duration_ms'):.2f}ms")
+                    if entry.get('error'):
+                        print(f"    Error: {entry.get('error')}")
+            
+            return 0
+        except Exception as e:
+            print(f"Error: {e}", file=sys.stderr)
+            return 1
+    
+    @staticmethod
+    def handle_trace_diff(args: argparse.Namespace) -> int:
+        """Handle debug trace diff command."""
+        try:
+            with open(args.file1) as f:
+                trace1 = json.load(f)
+            with open(args.file2) as f:
+                trace2 = json.load(f)
+            
+            diff = {
+                "file1": args.file1,
+                "file2": args.file2,
+                "entries_diff": {
+                    "file1_count": len(trace1.get("entries", [])),
+                    "file2_count": len(trace2.get("entries", []))
+                },
+                "duration_diff": {
+                    "file1": trace1.get("end_time", 0) - trace1.get("start_time", 0),
+                    "file2": trace2.get("end_time", 0) - trace2.get("start_time", 0)
+                }
+            }
+            
+            DebugHandler._output_result(diff, args.json)
+            return 0
+        except Exception as e:
+            print(f"Error: {e}", file=sys.stderr)
+            return 1
+    
+    @staticmethod
+    def _output_result(result: Dict[str, Any], as_json: bool) -> None:
+        """Output result in appropriate format."""
+        if as_json:
+            print(json.dumps(result, indent=2, default=str))
+        else:
+            for key, value in result.items():
+                if isinstance(value, dict):
+                    print(f"{key}:")
+                    for k, v in value.items():
+                        print(f"  {k}: {v}")
+                elif isinstance(value, list):
+                    print(f"{key}: ({len(value)} items)")
+                    for item in value[:5]:  # Show first 5
+                        print(f"  - {item}")
+                    if len(value) > 5:
+                        print(f"  ... and {len(value) - 5} more")
+                else:
+                    print(f"{key}: {value}")
+
+
+def setup_debug_subparser(subparsers) -> None:
+    """Set up the debug subcommand parser."""
+    DebugHandler.setup_subparser(subparsers)
+
+
+def run_debug_command(args: List[str]) -> int:
+    """Run debug command from CLI."""
+    parser = argparse.ArgumentParser(
+        prog="praisonai debug",
+        description="Debug and test interactive coding assistant flows"
+    )
+    subparsers = parser.add_subparsers(dest="debug_command", help="Debug subcommands")
+    
+    # debug interactive
+    interactive_parser = subparsers.add_parser(
+        "interactive",
+        help="Run single interactive turn non-interactively",
+        description="Execute one interactive turn with full LSP/ACP pipeline"
+    )
+    interactive_parser.add_argument("-p", "--prompt", type=str, required=True, help="Prompt to execute")
+    interactive_parser.add_argument("--lsp", action="store_true", default=True, help="Enable LSP (default: enabled)")
+    interactive_parser.add_argument("--no-lsp", action="store_true", help="Disable LSP")
+    interactive_parser.add_argument("--acp", action="store_true", default=True, help="Enable ACP (default: enabled)")
+    interactive_parser.add_argument("--no-acp", action="store_true", help="Disable ACP")
+    interactive_parser.add_argument("--approval", type=str, choices=["manual", "auto", "scoped"], default="auto", help="Approval mode")
+    interactive_parser.add_argument("--workspace", "-w", type=str, default=".", help="Workspace root directory")
+    interactive_parser.add_argument("--json", action="store_true", help="Output JSON trace")
+    interactive_parser.add_argument("--trace-file", type=str, help="Save trace to file")
+    interactive_parser.add_argument("--timeout", type=float, default=60.0, help="Timeout in seconds")
+    interactive_parser.add_argument("--model", type=str, help="LLM model to use")
+    interactive_parser.set_defaults(func=DebugHandler.handle_interactive)
+    
+    # debug lsp
+    lsp_parser = subparsers.add_parser("lsp", help="Direct LSP probes")
+    lsp_subparsers = lsp_parser.add_subparsers(dest="lsp_command", help="LSP subcommands")
+    
+    # debug lsp status
+    lsp_status = lsp_subparsers.add_parser("status", help="Show LSP status")
+    lsp_status.add_argument("--workspace", "-w", type=str, default=".")
+    lsp_status.add_argument("--json", action="store_true")
+    lsp_status.set_defaults(func=DebugHandler.handle_lsp_status)
+    
+    # debug lsp symbols
+    lsp_symbols = lsp_subparsers.add_parser("symbols", help="List symbols in file")
+    lsp_symbols.add_argument("file", type=str, help="File path")
+    lsp_symbols.add_argument("--workspace", "-w", type=str, default=".")
+    lsp_symbols.add_argument("--json", action="store_true")
+    lsp_symbols.set_defaults(func=DebugHandler.handle_lsp_symbols)
+    
+    # debug lsp definition
+    lsp_def = lsp_subparsers.add_parser("definition", help="Get definition location")
+    lsp_def.add_argument("location", type=str, help="file:line:col")
+    lsp_def.add_argument("--workspace", "-w", type=str, default=".")
+    lsp_def.add_argument("--json", action="store_true")
+    lsp_def.set_defaults(func=DebugHandler.handle_lsp_definition)
+    
+    # debug lsp references
+    lsp_refs = lsp_subparsers.add_parser("references", help="Get references")
+    lsp_refs.add_argument("location", type=str, help="file:line:col")
+    lsp_refs.add_argument("--workspace", "-w", type=str, default=".")
+    lsp_refs.add_argument("--json", action="store_true")
+    lsp_refs.set_defaults(func=DebugHandler.handle_lsp_references)
+    
+    # debug lsp diagnostics
+    lsp_diag = lsp_subparsers.add_parser("diagnostics", help="Get diagnostics")
+    lsp_diag.add_argument("file", type=str, nargs="?", help="File path (optional)")
+    lsp_diag.add_argument("--all", action="store_true", help="Get all diagnostics")
+    lsp_diag.add_argument("--workspace", "-w", type=str, default=".")
+    lsp_diag.add_argument("--json", action="store_true")
+    lsp_diag.set_defaults(func=DebugHandler.handle_lsp_diagnostics)
+    
+    # debug acp
+    acp_parser = subparsers.add_parser("acp", help="Direct ACP probes")
+    acp_subparsers = acp_parser.add_subparsers(dest="acp_command", help="ACP subcommands")
+    
+    # debug acp status
+    acp_status = acp_subparsers.add_parser("status", help="Show ACP status")
+    acp_status.add_argument("--workspace", "-w", type=str, default=".")
+    acp_status.add_argument("--json", action="store_true")
+    acp_status.set_defaults(func=DebugHandler.handle_acp_status)
+    
+    # debug acp plan
+    acp_plan = acp_subparsers.add_parser("plan", help="Create action plan")
+    acp_plan.add_argument("-p", "--prompt", type=str, required=True)
+    acp_plan.add_argument("--workspace", "-w", type=str, default=".")
+    acp_plan.add_argument("--json", action="store_true")
+    acp_plan.set_defaults(func=DebugHandler.handle_acp_plan)
+    
+    # debug acp apply
+    acp_apply = acp_subparsers.add_parser("apply", help="Apply action plan")
+    acp_apply.add_argument("-p", "--prompt", type=str, required=True)
+    acp_apply.add_argument("--approval", type=str, choices=["manual", "auto", "scoped"], default="auto")
+    acp_apply.add_argument("--workspace", "-w", type=str, default=".")
+    acp_apply.add_argument("--json", action="store_true")
+    acp_apply.set_defaults(func=DebugHandler.handle_acp_apply)
+    
+    # debug trace
+    trace_parser = subparsers.add_parser("trace", help="Trace record/replay/diff")
+    trace_subparsers = trace_parser.add_subparsers(dest="trace_command", help="Trace subcommands")
+    
+    # debug trace record
+    trace_record = trace_subparsers.add_parser("record", help="Record session trace")
+    trace_record.add_argument("-o", "--output", type=str, required=True, help="Output file")
+    trace_record.set_defaults(func=DebugHandler.handle_trace_record)
+    
+    # debug trace replay
+    trace_replay = trace_subparsers.add_parser("replay", help="Replay recorded trace")
+    trace_replay.add_argument("file", type=str, help="Trace file to replay")
+    trace_replay.add_argument("--json", action="store_true")
+    trace_replay.set_defaults(func=DebugHandler.handle_trace_replay)
+    
+    # debug trace diff
+    trace_diff = trace_subparsers.add_parser("diff", help="Compare two traces")
+    trace_diff.add_argument("file1", type=str, help="First trace file")
+    trace_diff.add_argument("file2", type=str, help="Second trace file")
+    trace_diff.add_argument("--json", action="store_true")
+    trace_diff.set_defaults(func=DebugHandler.handle_trace_diff)
+    
+    parsed = parser.parse_args(args)
+    if hasattr(parsed, 'func'):
+        return parsed.func(parsed)
+    else:
+        parser.print_help()
+        return 0

src/praisonai/praisonai/cli/features/diag.py

@@ -0,0 +1,289 @@
+"""
+Diagnostics export CLI handler for PraisonAI.
+
+Provides diagnostic export functionality:
+- diag export: Bundle logs, traces, config for bug reports
+"""
+
+import argparse
+import json
+import logging
+import os
+import shutil
+import tempfile
+import time
+import zipfile
+from pathlib import Path
+from typing import List
+
+logger = logging.getLogger(__name__)
+
+
+class DiagHandler:
+    """Handler for diag CLI commands."""
+    
+    @staticmethod
+    def setup_subparser(subparsers) -> None:
+        """Set up the diag subcommand parser."""
+        diag_parser = subparsers.add_parser(
+            "diag",
+            help="Diagnostics export",
+            description="Export diagnostic information for bug reports"
+        )
+        
+        diag_subparsers = diag_parser.add_subparsers(dest="diag_command", help="Diag subcommands")
+        
+        # diag export
+        export_parser = diag_subparsers.add_parser("export", help="Export diagnostic bundle")
+        export_parser.add_argument(
+            "-o", "--output",
+            type=str,
+            help="Output path (default: praisonai-diag-{timestamp}.zip)"
+        )
+        export_parser.add_argument(
+            "--include-logs",
+            action="store_true",
+            default=True,
+            help="Include log files (default: true)"
+        )
+        export_parser.add_argument(
+            "--include-config",
+            action="store_true",
+            default=True,
+            help="Include configuration (default: true)"
+        )
+        export_parser.add_argument(
+            "--include-trace",
+            action="store_true",
+            default=True,
+            help="Include recent traces (default: true)"
+        )
+        export_parser.add_argument(
+            "--workspace", "-w",
+            type=str,
+            default=".",
+            help="Workspace root directory"
+        )
+        export_parser.add_argument("--json", action="store_true", help="Output JSON")
+        export_parser.set_defaults(func=DiagHandler.handle_export)
+    
+    @staticmethod
+    def handle_export(args: argparse.Namespace) -> int:
+        """Handle diag export command."""
+        timestamp = int(time.time())
+        output_path = args.output or f"praisonai-diag-{timestamp}.zip"
+        
+        collected_files = []
+        
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tmppath = Path(tmpdir)
+            
+            # Collect system info
+            system_info = DiagHandler._collect_system_info()
+            system_file = tmppath / "system_info.json"
+            with open(system_file, "w") as f:
+                json.dump(system_info, f, indent=2, default=str)
+            collected_files.append("system_info.json")
+            
+            # Collect config
+            if args.include_config:
+                config_files = DiagHandler._collect_config(args.workspace, tmppath)
+                collected_files.extend(config_files)
+            
+            # Collect logs
+            if args.include_logs:
+                log_files = DiagHandler._collect_logs(tmppath)
+                collected_files.extend(log_files)
+            
+            # Collect traces
+            if args.include_trace:
+                trace_files = DiagHandler._collect_traces(args.workspace, tmppath)
+                collected_files.extend(trace_files)
+            
+            # Create zip
+            with zipfile.ZipFile(output_path, 'w', zipfile.ZIP_DEFLATED) as zf:
+                for file in tmppath.rglob("*"):
+                    if file.is_file():
+                        arcname = file.relative_to(tmppath)
+                        zf.write(file, arcname)
+        
+        result = {
+            "action": "export",
+            "output": output_path,
+            "files_collected": collected_files,
+            "count": len(collected_files)
+        }
+        
+        DiagHandler._output_result(result, args.json)
+        return 0
+    
+    @staticmethod
+    def _collect_system_info() -> dict:
+        """Collect system information."""
+        import platform
+        import sys
+        
+        info = {
+            "timestamp": time.time(),
+            "platform": {
+                "system": platform.system(),
+                "release": platform.release(),
+                "version": platform.version(),
+                "machine": platform.machine(),
+                "python_version": sys.version
+            },
+            "praisonai": {}
+        }
+        
+        # Get PraisonAI version
+        try:
+            from praisonai.version import __version__
+            info["praisonai"]["version"] = __version__
+        except ImportError:
+            info["praisonai"]["version"] = "unknown"
+        
+        # Get praisonaiagents version
+        try:
+            import praisonaiagents
+            info["praisonai"]["agents_version"] = getattr(praisonaiagents, '__version__', 'unknown')
+        except ImportError:
+            info["praisonai"]["agents_version"] = "not installed"
+        
+        # Check for optional dependencies
+        info["dependencies"] = {}
+        for dep in ["acp", "litellm", "openai", "anthropic", "google.generativeai"]:
+            try:
+                __import__(dep.split(".")[0])
+                info["dependencies"][dep] = "installed"
+            except ImportError:
+                info["dependencies"][dep] = "not installed"
+        
+        # Environment (redacted)
+        env_keys = [
+            "OPENAI_API_KEY", "ANTHROPIC_API_KEY", "GOOGLE_API_KEY",
+            "PRAISONAI_APPROVAL_MODE", "LOGLEVEL"
+        ]
+        info["environment"] = {}
+        for key in env_keys:
+            value = os.environ.get(key)
+            if value:
+                if "KEY" in key or "SECRET" in key:
+                    info["environment"][key] = f"{value[:4]}...{value[-4:]}" if len(value) > 8 else "***"
+                else:
+                    info["environment"][key] = value
+            else:
+                info["environment"][key] = None
+        
+        return info
+    
+    @staticmethod
+    def _collect_config(workspace: str, tmppath: Path) -> List[str]:
+        """Collect configuration files."""
+        collected = []
+        workspace_path = Path(workspace)
+        config_dir = tmppath / "config"
+        config_dir.mkdir(exist_ok=True)
+        
+        # Look for config files
+        config_patterns = [
+            "agents.yaml",
+            "workflow.yaml",
+            ".praisonai.yaml",
+            ".env.example",  # Not .env to avoid secrets
+            "pyproject.toml",
+        ]
+        
+        for pattern in config_patterns:
+            for file in workspace_path.glob(pattern):
+                if file.is_file():
+                    dest = config_dir / file.name
+                    shutil.copy(file, dest)
+                    collected.append(f"config/{file.name}")
+        
+        return collected
+    
+    @staticmethod
+    def _collect_logs(tmppath: Path) -> List[str]:
+        """Collect log files."""
+        collected = []
+        logs_dir = tmppath / "logs"
+        logs_dir.mkdir(exist_ok=True)
+        
+        # Look for log files in common locations
+        log_locations = [
+            Path.home() / ".praison" / "logs",
+            Path.home() / ".praisonai" / "logs",
+            Path("/var/log/praisonai"),
+        ]
+        
+        for log_dir in log_locations:
+            if log_dir.exists():
+                for log_file in log_dir.glob("*.log"):
+                    if log_file.is_file():
+                        dest = logs_dir / log_file.name
+                        # Only copy last 1000 lines
+                        try:
+                            with open(log_file) as f:
+                                lines = f.readlines()[-1000:]
+                            with open(dest, "w") as f:
+                                f.writelines(lines)
+                            collected.append(f"logs/{log_file.name}")
+                        except Exception:
+                            pass
+        
+        return collected
+    
+    @staticmethod
+    def _collect_traces(workspace: str, tmppath: Path) -> List[str]:
+        """Collect trace files."""
+        collected = []
+        traces_dir = tmppath / "traces"
+        traces_dir.mkdir(exist_ok=True)
+        
+        workspace_path = Path(workspace)
+        
+        # Look for trace files
+        trace_patterns = ["*_trace.json", "praisonai_trace*.json", "*.trace"]
+        
+        for pattern in trace_patterns:
+            for trace_file in workspace_path.glob(pattern):
+                if trace_file.is_file():
+                    dest = traces_dir / trace_file.name
+                    shutil.copy(trace_file, dest)
+                    collected.append(f"traces/{trace_file.name}")
+        
+        return collected
+    
+    @staticmethod
+    def _output_result(result: dict, as_json: bool) -> None:
+        """Output result in appropriate format."""
+        if as_json:
+            print(json.dumps(result, indent=2, default=str))
+        else:
+            print(f"Diagnostic bundle created: {result['output']}")
+            print(f"Files collected: {result['count']}")
+            for f in result['files_collected']:
+                print(f"  - {f}")
+
+
+def run_diag_command(args: List[str]) -> int:
+    """Run diag command from CLI."""
+    parser = argparse.ArgumentParser(prog="praisonai diag")
+    subparsers = parser.add_subparsers(dest="diag_command", help="Diag subcommands")
+    
+    # diag export
+    export_parser = subparsers.add_parser("export", help="Export diagnostic bundle")
+    export_parser.add_argument("-o", "--output", type=str)
+    export_parser.add_argument("--include-logs", action="store_true", default=True)
+    export_parser.add_argument("--include-config", action="store_true", default=True)
+    export_parser.add_argument("--include-trace", action="store_true", default=True)
+    export_parser.add_argument("--workspace", "-w", type=str, default=".")
+    export_parser.add_argument("--json", action="store_true")
+    export_parser.set_defaults(func=DiagHandler.handle_export)
+    
+    parsed = parser.parse_args(args)
+    if hasattr(parsed, 'func'):
+        return parsed.func(parsed)
+    else:
+        parser.print_help()
+        return 0

src/praisonai/praisonai/cli/features/doctor/checks/__init__.py

@@ -20,3 +20,5 @@ def register_all_checks():
     from . import performance_checks
     from . import selftest_checks
     from . import serve_checks
+    from . import lsp_checks
+    from . import acp_checks

src/praisonai/praisonai/cli/features/doctor/checks/acp_checks.py

@@ -0,0 +1,240 @@
+"""
+ACP health checks for PraisonAI Doctor.
+"""
+
+import asyncio
+import logging
+
+from ..models import CheckCategory, CheckResult, CheckStatus
+from ..registry import register_check
+
+logger = logging.getLogger(__name__)
+
+
+@register_check(
+    id="acp_module",
+    title="ACP Module Available",
+    category=CheckCategory.TOOLS,
+    description="Check if ACP module is installed"
+)
+def check_acp_module() -> CheckResult:
+    """Check if ACP module is available."""
+    try:
+        import importlib.util
+        spec = importlib.util.find_spec("praisonai.acp")
+        
+        if spec is not None:
+            return CheckResult(
+                status=CheckStatus.PASS,
+                message="ACP module is available",
+                details={"module": "praisonai.acp"}
+            )
+        else:
+            return CheckResult(
+                status=CheckStatus.WARN,
+                message="ACP module not found",
+                details={"hint": "ACP module should be part of praisonai package"}
+            )
+    except Exception as e:
+        return CheckResult(
+            status=CheckStatus.FAIL,
+            message=f"Error checking ACP module: {e}",
+            details={"error": str(e)}
+        )
+
+
+@register_check(
+    id="acp_sdk",
+    title="ACP SDK Available",
+    category=CheckCategory.TOOLS,
+    description="Check if agent-client-protocol SDK is installed"
+)
+def check_acp_sdk() -> CheckResult:
+    """Check if ACP SDK is available."""
+    try:
+        import importlib.util
+        spec = importlib.util.find_spec("acp")
+        
+        if spec is not None:
+            return CheckResult(
+                status=CheckStatus.PASS,
+                message="ACP SDK (agent-client-protocol) is installed",
+                details={"package": "agent-client-protocol"}
+            )
+        else:
+            return CheckResult(
+                status=CheckStatus.INFO,
+                message="ACP SDK not installed (optional)",
+                details={"hint": "Install with: pip install agent-client-protocol"}
+            )
+    except Exception as e:
+        return CheckResult(
+            status=CheckStatus.FAIL,
+            message=f"Error checking ACP SDK: {e}",
+            details={"error": str(e)}
+        )
+
+
+@register_check(
+    id="acp_server",
+    title="ACP Server Import",
+    category=CheckCategory.TOOLS,
+    description="Check if ACP server can be imported"
+)
+def check_acp_server() -> CheckResult:
+    """Check if ACP server can be imported."""
+    try:
+        from praisonai.acp import ACPServer  # noqa: F401
+        
+        return CheckResult(
+            status=CheckStatus.PASS,
+            message="ACP server can be imported",
+            details={"class": "ACPServer"}
+        )
+    except ImportError as e:
+        return CheckResult(
+            status=CheckStatus.WARN,
+            message=f"ACP server import failed: {e}",
+            details={"error": str(e)}
+        )
+    except Exception as e:
+        return CheckResult(
+            status=CheckStatus.FAIL,
+            message=f"Error importing ACP server: {e}",
+            details={"error": str(e)}
+        )
+
+
+@register_check(
+    id="acp_config",
+    title="ACP Configuration",
+    category=CheckCategory.CONFIG,
+    description="Check ACP configuration"
+)
+def check_acp_config() -> CheckResult:
+    """Check ACP configuration."""
+    import os
+    
+    config = {
+        "approval_mode": os.environ.get("PRAISONAI_APPROVAL_MODE", "manual"),
+        "allow_write": os.environ.get("PRAISONAI_ALLOW_WRITE", "false"),
+        "allow_shell": os.environ.get("PRAISONAI_ALLOW_SHELL", "false"),
+    }
+    
+    return CheckResult(
+        status=CheckStatus.PASS,
+        message=f"ACP configured with approval_mode={config['approval_mode']}",
+        details=config
+    )
+
+
+@register_check(
+    id="acp_runtime",
+    title="ACP Runtime Check",
+    category=CheckCategory.TOOLS,
+    description="Check if ACP runtime can start",
+    requires_deep=True
+)
+def check_acp_runtime() -> CheckResult:
+    """Check if ACP runtime can start."""
+    try:
+        from praisonai.cli.features.interactive_runtime import create_runtime
+        
+        async def _check():
+            runtime = create_runtime(
+                workspace=".",
+                lsp=False,
+                acp=True
+            )
+            try:
+                await runtime.start()
+                ready = runtime.acp_ready
+                error = runtime._acp_state.error
+                read_only = runtime.read_only
+                return ready, error, read_only
+            finally:
+                await runtime.stop()
+        
+        ready, error, read_only = asyncio.run(_check())
+        
+        if ready:
+            return CheckResult(
+                status=CheckStatus.PASS,
+                message="ACP runtime started successfully",
+                details={"ready": True, "read_only": read_only}
+            )
+        else:
+            return CheckResult(
+                status=CheckStatus.WARN,
+                message=f"ACP runtime failed to start: {error}",
+                details={"ready": False, "error": error, "read_only": read_only}
+            )
+            
+    except Exception as e:
+        return CheckResult(
+            status=CheckStatus.FAIL,
+            message=f"ACP runtime check failed: {e}",
+            details={"error": str(e)}
+        )
+
+
+@register_check(
+    id="interactive_runtime",
+    title="Interactive Runtime Check",
+    category=CheckCategory.TOOLS,
+    description="Check if full interactive runtime can start",
+    requires_deep=True
+)
+def check_interactive_runtime() -> CheckResult:
+    """Check if full interactive runtime can start."""
+    try:
+        from praisonai.cli.features.interactive_runtime import create_runtime
+        
+        async def _check():
+            runtime = create_runtime(
+                workspace=".",
+                lsp=True,
+                acp=True
+            )
+            try:
+                await runtime.start()
+                status = runtime.get_status()
+                return status
+            finally:
+                await runtime.stop()
+        
+        status = asyncio.run(_check())
+        
+        lsp_ready = status.get("lsp", {}).get("ready", False)
+        acp_ready = status.get("acp", {}).get("ready", False)
+        
+        if lsp_ready and acp_ready:
+            return CheckResult(
+                status=CheckStatus.PASS,
+                message="Interactive runtime fully operational",
+                details=status
+            )
+        elif lsp_ready or acp_ready:
+            return CheckResult(
+                status=CheckStatus.WARN,
+                message="Interactive runtime partially operational",
+                details=status
+            )
+        else:
+            return CheckResult(
+                status=CheckStatus.FAIL,
+                message="Interactive runtime failed to start",
+                details=status
+            )
+            
+    except Exception as e:
+        return CheckResult(
+            status=CheckStatus.FAIL,
+            message=f"Interactive runtime check failed: {e}",
+            details={"error": str(e)}
+        )
+
+
+def register_acp_checks():
+    """Register all ACP checks."""
+    pass

src/praisonai/praisonai/cli/features/doctor/checks/lsp_checks.py

@@ -0,0 +1,199 @@
+"""
+LSP health checks for PraisonAI Doctor.
+"""
+
+import asyncio
+import logging
+
+from ..models import CheckCategory, CheckResult, CheckStatus
+from ..registry import register_check
+
+logger = logging.getLogger(__name__)
+
+
+@register_check(
+    id="lsp_module",
+    title="LSP Module Available",
+    category=CheckCategory.TOOLS,
+    description="Check if LSP module is installed"
+)
+def check_lsp_module() -> CheckResult:
+    """Check if LSP module is available."""
+    try:
+        import importlib.util
+        spec = importlib.util.find_spec("praisonaiagents.lsp")
+        
+        if spec is not None:
+            return CheckResult(
+                status=CheckStatus.PASS,
+                message="LSP module is available",
+                details={"module": "praisonaiagents.lsp"}
+            )
+        else:
+            return CheckResult(
+                status=CheckStatus.WARN,
+                message="LSP module not found",
+                details={"hint": "Install praisonaiagents with LSP support"}
+            )
+    except Exception as e:
+        return CheckResult(
+            status=CheckStatus.FAIL,
+            message=f"Error checking LSP module: {e}",
+            details={"error": str(e)}
+        )
+
+
+@register_check(
+    id="lsp_client",
+    title="LSP Client Import",
+    category=CheckCategory.TOOLS,
+    description="Check if LSP client can be imported"
+)
+def check_lsp_client() -> CheckResult:
+    """Check if LSP client can be imported."""
+    try:
+        from praisonaiagents.lsp import LSPClient  # noqa: F401
+        
+        return CheckResult(
+            status=CheckStatus.PASS,
+            message="LSP client can be imported",
+            details={"class": "LSPClient"}
+        )
+    except ImportError as e:
+        return CheckResult(
+            status=CheckStatus.WARN,
+            message=f"LSP client import failed: {e}",
+            details={"error": str(e)}
+        )
+    except Exception as e:
+        return CheckResult(
+            status=CheckStatus.FAIL,
+            message=f"Error importing LSP client: {e}",
+            details={"error": str(e)}
+        )
+
+
+@register_check(
+    id="lsp_server_python",
+    title="Python LSP Server",
+    category=CheckCategory.TOOLS,
+    description="Check if Python language server is available"
+)
+def check_lsp_server_python() -> CheckResult:
+    """Check if Python language server is available."""
+    import shutil
+    
+    # Check for common Python language servers
+    servers = [
+        ("pyright-langserver", "pyright"),
+        ("pylsp", "python-lsp-server"),
+        ("pyls", "python-language-server"),
+    ]
+    
+    found = []
+    for cmd, package in servers:
+        if shutil.which(cmd):
+            found.append({"command": cmd, "package": package})
+    
+    if found:
+        return CheckResult(
+            status=CheckStatus.PASS,
+            message=f"Python LSP server available: {found[0]['command']}",
+            details={"servers": found}
+        )
+    else:
+        return CheckResult(
+            status=CheckStatus.WARN,
+            message="No Python LSP server found",
+            details={
+                "hint": "Install with: npm install -g pyright OR pip install python-lsp-server",
+                "checked": [s[0] for s in servers]
+            }
+        )
+
+
+@register_check(
+    id="lsp_server_typescript",
+    title="TypeScript LSP Server",
+    category=CheckCategory.TOOLS,
+    description="Check if TypeScript language server is available"
+)
+def check_lsp_server_typescript() -> CheckResult:
+    """Check if TypeScript language server is available."""
+    import shutil
+    
+    # Check for TypeScript language server
+    if shutil.which("typescript-language-server"):
+        return CheckResult(
+            status=CheckStatus.PASS,
+            message="TypeScript LSP server available",
+            details={"command": "typescript-language-server"}
+        )
+    elif shutil.which("tsserver"):
+        return CheckResult(
+            status=CheckStatus.PASS,
+            message="TypeScript server available (tsserver)",
+            details={"command": "tsserver"}
+        )
+    else:
+        return CheckResult(
+            status=CheckStatus.INFO,
+            message="TypeScript LSP server not found (optional)",
+            details={
+                "hint": "Install with: npm install -g typescript-language-server typescript"
+            }
+        )
+
+
+@register_check(
+    id="lsp_runtime",
+    title="LSP Runtime Check",
+    category=CheckCategory.TOOLS,
+    description="Check if LSP runtime can start",
+    requires_deep=True
+)
+def check_lsp_runtime() -> CheckResult:
+    """Check if LSP runtime can start."""
+    try:
+        from praisonai.cli.features.interactive_runtime import create_runtime
+        
+        async def _check():
+            runtime = create_runtime(
+                workspace=".",
+                lsp=True,
+                acp=False
+            )
+            try:
+                await runtime.start()
+                ready = runtime.lsp_ready
+                error = runtime._lsp_state.error
+                return ready, error
+            finally:
+                await runtime.stop()
+        
+        ready, error = asyncio.run(_check())
+        
+        if ready:
+            return CheckResult(
+                status=CheckStatus.PASS,
+                message="LSP runtime started successfully",
+                details={"ready": True}
+            )
+        else:
+            return CheckResult(
+                status=CheckStatus.WARN,
+                message=f"LSP runtime failed to start: {error}",
+                details={"ready": False, "error": error}
+            )
+            
+    except Exception as e:
+        return CheckResult(
+            status=CheckStatus.FAIL,
+            message=f"LSP runtime check failed: {e}",
+            details={"error": str(e)}
+        )
+
+
+def register_lsp_checks():
+    """Register all LSP checks."""
+    pass

src/praisonai/praisonai/cli/features/interactive_runtime.py

@@ -0,0 +1,585 @@
+"""
+Interactive Runtime for PraisonAI.
+
+Provides the core runtime that powers both TUI interactive mode and debug non-interactive mode.
+Manages LSP and ACP subsystem lifecycle with graceful degradation.
+"""
+
+import asyncio
+import logging
+import time
+from dataclasses import dataclass, field
+from enum import Enum
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class SubsystemStatus(Enum):
+    """Status of a subsystem."""
+    NOT_STARTED = "not_started"
+    STARTING = "starting"
+    READY = "ready"
+    FAILED = "failed"
+    STOPPED = "stopped"
+
+
+@dataclass
+class SubsystemState:
+    """State of a subsystem."""
+    status: SubsystemStatus = SubsystemStatus.NOT_STARTED
+    error: Optional[str] = None
+    start_time: Optional[float] = None
+    ready_time: Optional[float] = None
+
+
+@dataclass
+class RuntimeConfig:
+    """Configuration for InteractiveRuntime."""
+    workspace: str = "."
+    lsp_enabled: bool = True
+    acp_enabled: bool = True
+    approval_mode: str = "manual"  # manual, auto, scoped
+    trace_enabled: bool = False
+    trace_file: Optional[str] = None
+    json_output: bool = False
+    timeout: float = 60.0
+    model: Optional[str] = None
+    verbose: bool = False
+
+
+@dataclass
+class TraceEntry:
+    """A single trace entry."""
+    timestamp: float
+    category: str  # lsp, acp, tool, llm, file
+    action: str
+    params: Dict[str, Any] = field(default_factory=dict)
+    result: Optional[Any] = None
+    duration_ms: Optional[float] = None
+    error: Optional[str] = None
+
+
+@dataclass
+class RuntimeTrace:
+    """Trace of a runtime session."""
+    version: str = "1.0"
+    start_time: float = field(default_factory=time.time)
+    end_time: Optional[float] = None
+    config: Optional[RuntimeConfig] = None
+    entries: List[TraceEntry] = field(default_factory=list)
+    
+    def add_entry(self, category: str, action: str, params: Dict[str, Any] = None,
+                  result: Any = None, duration_ms: float = None, error: str = None):
+        """Add a trace entry."""
+        self.entries.append(TraceEntry(
+            timestamp=time.time(),
+            category=category,
+            action=action,
+            params=params or {},
+            result=result,
+            duration_ms=duration_ms,
+            error=error
+        ))
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert trace to dictionary."""
+        return {
+            "version": self.version,
+            "start_time": self.start_time,
+            "end_time": self.end_time,
+            "config": {
+                "workspace": self.config.workspace if self.config else None,
+                "lsp_enabled": self.config.lsp_enabled if self.config else None,
+                "acp_enabled": self.config.acp_enabled if self.config else None,
+                "approval_mode": self.config.approval_mode if self.config else None,
+            } if self.config else None,
+            "entries": [
+                {
+                    "timestamp": e.timestamp,
+                    "category": e.category,
+                    "action": e.action,
+                    "params": e.params,
+                    "result": e.result,
+                    "duration_ms": e.duration_ms,
+                    "error": e.error
+                }
+                for e in self.entries
+            ]
+        }
+
+
+class InteractiveRuntime:
+    """
+    Core runtime for interactive coding assistant.
+    
+    Manages:
+    - LSP subsystem for code intelligence
+    - ACP subsystem for action orchestration
+    - Trace collection for debugging
+    - Graceful degradation when subsystems fail
+    """
+    
+    def __init__(self, config: RuntimeConfig = None):
+        """Initialize the runtime."""
+        self.config = config or RuntimeConfig()
+        self._lsp_state = SubsystemState()
+        self._acp_state = SubsystemState()
+        self._lsp_client = None
+        self._acp_session = None
+        self._trace = RuntimeTrace(config=self.config) if self.config.trace_enabled else None
+        self._started = False
+        self._read_only = False  # Enforced when ACP fails
+        
+    @property
+    def lsp_ready(self) -> bool:
+        """Check if LSP is ready."""
+        return self._lsp_state.status == SubsystemStatus.READY
+    
+    @property
+    def acp_ready(self) -> bool:
+        """Check if ACP is ready."""
+        return self._acp_state.status == SubsystemStatus.READY
+    
+    @property
+    def read_only(self) -> bool:
+        """Check if runtime is in read-only mode."""
+        return self._read_only or not self.acp_ready
+    
+    def get_status(self) -> Dict[str, Any]:
+        """Get runtime status."""
+        return {
+            "started": self._started,
+            "workspace": self.config.workspace,
+            "lsp": {
+                "enabled": self.config.lsp_enabled,
+                "status": self._lsp_state.status.value,
+                "ready": self.lsp_ready,
+                "error": self._lsp_state.error
+            },
+            "acp": {
+                "enabled": self.config.acp_enabled,
+                "status": self._acp_state.status.value,
+                "ready": self.acp_ready,
+                "error": self._acp_state.error
+            },
+            "read_only": self.read_only,
+            "approval_mode": self.config.approval_mode
+        }
+    
+    async def start(self) -> Dict[str, Any]:
+        """
+        Start the runtime and all enabled subsystems.
+        
+        Returns status dict with subsystem states.
+        """
+        if self._started:
+            return self.get_status()
+        
+        workspace = Path(self.config.workspace).resolve()
+        if not workspace.exists():
+            workspace.mkdir(parents=True, exist_ok=True)
+        
+        self.config.workspace = str(workspace)
+        
+        # Start subsystems in parallel
+        tasks = []
+        if self.config.lsp_enabled:
+            tasks.append(self._start_lsp())
+        if self.config.acp_enabled:
+            tasks.append(self._start_acp())
+        
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)
+        
+        # Enforce read-only if ACP failed
+        if self.config.acp_enabled and not self.acp_ready:
+            self._read_only = True
+            logger.warning("ACP unavailable - runtime is READ-ONLY")
+        
+        self._started = True
+        
+        if self._trace:
+            self._trace.add_entry(
+                category="runtime",
+                action="start",
+                params={"workspace": self.config.workspace},
+                result=self.get_status()
+            )
+        
+        return self.get_status()
+    
+    async def _start_lsp(self):
+        """Start LSP subsystem."""
+        self._lsp_state.status = SubsystemStatus.STARTING
+        self._lsp_state.start_time = time.time()
+        
+        try:
+            # Lazy import LSP client
+            from praisonaiagents.lsp import LSPClient
+            
+            # Detect language from workspace
+            language = self._detect_workspace_language()
+            
+            self._lsp_client = LSPClient(
+                language=language,
+                root_uri=f"file://{self.config.workspace}"
+            )
+            await self._lsp_client.start()
+            
+            # Wait for initialization
+            if self._lsp_client.is_running:
+                self._lsp_state.status = SubsystemStatus.READY
+                self._lsp_state.ready_time = time.time()
+                logger.info(f"LSP ready for {language}")
+            else:
+                raise RuntimeError("LSP client failed to start")
+                
+        except ImportError as e:
+            self._lsp_state.status = SubsystemStatus.FAILED
+            self._lsp_state.error = f"LSP module not available: {e}"
+            logger.warning(self._lsp_state.error)
+        except Exception as e:
+            self._lsp_state.status = SubsystemStatus.FAILED
+            self._lsp_state.error = str(e)
+            logger.warning(f"LSP failed to start: {e}")
+    
+    async def _start_acp(self):
+        """Start ACP subsystem (in-process session)."""
+        self._acp_state.status = SubsystemStatus.STARTING
+        self._acp_state.start_time = time.time()
+        
+        try:
+            # Lazy import ACP
+            from praisonai.acp.session import ACPSession
+            from pathlib import Path
+            
+            # Create a session for the workspace
+            self._acp_session = ACPSession.create(
+                workspace=Path(self.config.workspace)
+            )
+            self._acp_session.mode = self.config.approval_mode
+            
+            self._acp_state.status = SubsystemStatus.READY
+            self._acp_state.ready_time = time.time()
+            logger.info("ACP session ready")
+            
+        except ImportError as e:
+            self._acp_state.status = SubsystemStatus.FAILED
+            self._acp_state.error = f"ACP module not available: {e}"
+            logger.warning(self._acp_state.error)
+        except Exception as e:
+            self._acp_state.status = SubsystemStatus.FAILED
+            self._acp_state.error = str(e)
+            logger.warning(f"ACP failed to start: {e}")
+    
+    async def stop(self):
+        """Stop the runtime and all subsystems."""
+        if self._lsp_client:
+            try:
+                await self._lsp_client.stop()
+            except Exception as e:
+                logger.warning(f"Error stopping LSP: {e}")
+            self._lsp_state.status = SubsystemStatus.STOPPED
+        
+        if self._acp_session:
+            # ACPSession is a dataclass, no close method needed
+            self._acp_session = None
+            self._acp_state.status = SubsystemStatus.STOPPED
+        
+        if self._trace:
+            self._trace.end_time = time.time()
+            self._trace.add_entry(
+                category="runtime",
+                action="stop",
+                result=self.get_status()
+            )
+        
+        self._started = False
+    
+    def _detect_workspace_language(self) -> str:
+        """Detect primary language in workspace."""
+        workspace = Path(self.config.workspace)
+        
+        # Count files by extension
+        extensions = {}
+        for ext in [".py", ".js", ".ts", ".tsx", ".jsx", ".go", ".rs", ".java"]:
+            count = len(list(workspace.rglob(f"*{ext}")))
+            if count > 0:
+                extensions[ext] = count
+        
+        if not extensions:
+            return "python"  # Default
+        
+        # Map extensions to languages
+        ext_to_lang = {
+            ".py": "python",
+            ".js": "javascript",
+            ".ts": "typescript",
+            ".tsx": "typescript",
+            ".jsx": "javascript",
+            ".go": "go",
+            ".rs": "rust",
+            ".java": "java"
+        }
+        
+        # Return language with most files
+        top_ext = max(extensions, key=extensions.get)
+        return ext_to_lang.get(top_ext, "python")
+    
+    # LSP Operations
+    async def lsp_get_symbols(self, file_path: str) -> List[Dict[str, Any]]:
+        """Get symbols in a file via LSP."""
+        if not self.lsp_ready:
+            return []
+        
+        start = time.time()
+        try:
+            # Use workspace symbols or document symbols
+            result = await self._lsp_client.get_document_symbols(file_path)
+            duration = (time.time() - start) * 1000
+            
+            if self._trace:
+                self._trace.add_entry(
+                    category="lsp",
+                    action="get_symbols",
+                    params={"file": file_path},
+                    result={"count": len(result) if result else 0},
+                    duration_ms=duration
+                )
+            
+            return result or []
+        except Exception as e:
+            if self._trace:
+                self._trace.add_entry(
+                    category="lsp",
+                    action="get_symbols",
+                    params={"file": file_path},
+                    error=str(e)
+                )
+            return []
+    
+    async def lsp_get_definition(self, file_path: str, line: int, col: int) -> List[Dict[str, Any]]:
+        """Get definition location via LSP."""
+        if not self.lsp_ready:
+            return []
+        
+        start = time.time()
+        try:
+            result = await self._lsp_client.get_definition(file_path, line, col)
+            duration = (time.time() - start) * 1000
+            
+            if self._trace:
+                self._trace.add_entry(
+                    category="lsp",
+                    action="get_definition",
+                    params={"file": file_path, "line": line, "col": col},
+                    result={"count": len(result) if result else 0},
+                    duration_ms=duration
+                )
+            
+            return [loc.__dict__ if hasattr(loc, '__dict__') else loc for loc in (result or [])]
+        except Exception as e:
+            if self._trace:
+                self._trace.add_entry(
+                    category="lsp",
+                    action="get_definition",
+                    params={"file": file_path, "line": line, "col": col},
+                    error=str(e)
+                )
+            return []
+    
+    async def lsp_get_references(self, file_path: str, line: int, col: int) -> List[Dict[str, Any]]:
+        """Get references via LSP."""
+        if not self.lsp_ready:
+            return []
+        
+        start = time.time()
+        try:
+            result = await self._lsp_client.get_references(file_path, line, col)
+            duration = (time.time() - start) * 1000
+            
+            if self._trace:
+                self._trace.add_entry(
+                    category="lsp",
+                    action="get_references",
+                    params={"file": file_path, "line": line, "col": col},
+                    result={"count": len(result) if result else 0},
+                    duration_ms=duration
+                )
+            
+            return [loc.__dict__ if hasattr(loc, '__dict__') else loc for loc in (result or [])]
+        except Exception as e:
+            if self._trace:
+                self._trace.add_entry(
+                    category="lsp",
+                    action="get_references",
+                    params={"file": file_path, "line": line, "col": col},
+                    error=str(e)
+                )
+            return []
+    
+    async def lsp_get_diagnostics(self, file_path: str = None) -> List[Dict[str, Any]]:
+        """Get diagnostics via LSP."""
+        if not self.lsp_ready:
+            return []
+        
+        start = time.time()
+        try:
+            if file_path:
+                result = await self._lsp_client.get_diagnostics(file_path)
+            else:
+                result = self._lsp_client._diagnostics  # All cached diagnostics
+            duration = (time.time() - start) * 1000
+            
+            if self._trace:
+                self._trace.add_entry(
+                    category="lsp",
+                    action="get_diagnostics",
+                    params={"file": file_path},
+                    result={"count": len(result) if result else 0},
+                    duration_ms=duration
+                )
+            
+            return [d.__dict__ if hasattr(d, '__dict__') else d for d in (result or [])]
+        except Exception as e:
+            if self._trace:
+                self._trace.add_entry(
+                    category="lsp",
+                    action="get_diagnostics",
+                    params={"file": file_path},
+                    error=str(e)
+                )
+            return []
+    
+    # ACP Operations
+    async def acp_create_plan(self, prompt: str) -> Dict[str, Any]:
+        """Create an action plan via ACP session tracking."""
+        if not self.acp_ready:
+            return {"error": "ACP not available", "read_only": True}
+        
+        start = time.time()
+        try:
+            # ACP session tracks the plan but doesn't create it
+            # Return empty steps - ActionOrchestrator handles actual planning
+            result = {"steps": [], "prompt": prompt}
+            duration = (time.time() - start) * 1000
+            
+            if self._trace:
+                self._trace.add_entry(
+                    category="acp",
+                    action="create_plan",
+                    params={"prompt": prompt[:100]},
+                    result=result,
+                    duration_ms=duration
+                )
+            
+            return result
+        except Exception as e:
+            if self._trace:
+                self._trace.add_entry(
+                    category="acp",
+                    action="create_plan",
+                    params={"prompt": prompt[:100]},
+                    error=str(e)
+                )
+            return {"error": str(e)}
+    
+    async def acp_apply_plan(self, plan: Dict[str, Any], auto_approve: bool = False) -> Dict[str, Any]:
+        """Apply an action plan via ACP session tracking."""
+        if self.read_only:
+            return {"error": "Runtime is read-only", "applied": False}
+        
+        start = time.time()
+        try:
+            # Track the action in session
+            if self._acp_session:
+                self._acp_session.add_tool_call(
+                    tool_call_id=f"plan_{int(time.time())}",
+                    title="apply_plan",
+                    status="applied" if auto_approve else "pending"
+                )
+            
+            result = {"applied": True, "auto_approve": auto_approve}
+            duration = (time.time() - start) * 1000
+            
+            if self._trace:
+                self._trace.add_entry(
+                    category="acp",
+                    action="apply_plan",
+                    params={"plan_id": plan.get("id"), "auto_approve": auto_approve},
+                    result=result,
+                    duration_ms=duration
+                )
+            
+            return result
+        except Exception as e:
+            if self._trace:
+                self._trace.add_entry(
+                    category="acp",
+                    action="apply_plan",
+                    params={"plan_id": plan.get("id")},
+                    error=str(e)
+                )
+            return {"error": str(e), "applied": False}
+    
+    def get_trace(self) -> Optional[RuntimeTrace]:
+        """Get the runtime trace."""
+        return self._trace
+    
+    def save_trace(self, path: str = None):
+        """Save trace to file."""
+        if not self._trace:
+            return
+        
+        import json
+        path = path or self.config.trace_file or "praisonai_trace.json"
+        with open(path, "w") as f:
+            json.dump(self._trace.to_dict(), f, indent=2, default=str)
+        logger.info(f"Trace saved to {path}")
+
+
+def create_runtime(
+    workspace: str = ".",
+    lsp: bool = True,
+    acp: bool = True,
+    approval: str = "manual",
+    trace: bool = False,
+    trace_file: str = None,
+    json_output: bool = False,
+    timeout: float = 60.0,
+    model: str = None,
+    verbose: bool = False
+) -> InteractiveRuntime:
+    """
+    Factory function to create an InteractiveRuntime.
+    
+    Args:
+        workspace: Workspace root directory
+        lsp: Enable LSP subsystem
+        acp: Enable ACP subsystem
+        approval: Approval mode (manual, auto, scoped)
+        trace: Enable trace collection
+        trace_file: Path to save trace
+        json_output: Enable JSON output mode
+        timeout: Operation timeout
+        model: LLM model to use
+        verbose: Enable verbose output
+        
+    Returns:
+        Configured InteractiveRuntime instance
+    """
+    config = RuntimeConfig(
+        workspace=workspace,
+        lsp_enabled=lsp,
+        acp_enabled=acp,
+        approval_mode=approval,
+        trace_enabled=trace,
+        trace_file=trace_file,
+        json_output=json_output,
+        timeout=timeout,
+        model=model,
+        verbose=verbose
+    )
+    return InteractiveRuntime(config)

src/praisonai/praisonai/cli/features/lsp_cli.py

@@ -0,0 +1,225 @@
+"""
+LSP CLI handler for PraisonAI.
+
+Provides LSP service lifecycle commands:
+- lsp start: Start LSP server(s)
+- lsp stop: Stop LSP server(s)
+- lsp status: Show status
+- lsp logs: Show recent logs
+"""
+
+import argparse
+import asyncio
+import json
+import logging
+from typing import List
+
+logger = logging.getLogger(__name__)
+
+
+class LSPHandler:
+    """Handler for LSP CLI commands."""
+    
+    @staticmethod
+    def setup_subparser(subparsers) -> None:
+        """Set up the lsp subcommand parser."""
+        lsp_parser = subparsers.add_parser(
+            "lsp",
+            help="LSP service lifecycle management",
+            description="Manage Language Server Protocol servers"
+        )
+        
+        lsp_subparsers = lsp_parser.add_subparsers(dest="lsp_command", help="LSP subcommands")
+        
+        # lsp start
+        start_parser = lsp_subparsers.add_parser("start", help="Start LSP server(s)")
+        start_parser.add_argument(
+            "--language", "-l",
+            type=str,
+            help="Language to start (default: auto-detect)"
+        )
+        start_parser.add_argument(
+            "--workspace", "-w",
+            type=str,
+            default=".",
+            help="Workspace root directory"
+        )
+        start_parser.add_argument("--json", action="store_true", help="Output JSON")
+        start_parser.set_defaults(func=LSPHandler.handle_start)
+        
+        # lsp stop
+        stop_parser = lsp_subparsers.add_parser("stop", help="Stop LSP server(s)")
+        stop_parser.add_argument("--json", action="store_true", help="Output JSON")
+        stop_parser.set_defaults(func=LSPHandler.handle_stop)
+        
+        # lsp status
+        status_parser = lsp_subparsers.add_parser("status", help="Show LSP status")
+        status_parser.add_argument(
+            "--workspace", "-w",
+            type=str,
+            default=".",
+            help="Workspace root directory"
+        )
+        status_parser.add_argument("--json", action="store_true", help="Output JSON")
+        status_parser.set_defaults(func=LSPHandler.handle_status)
+        
+        # lsp logs
+        logs_parser = lsp_subparsers.add_parser("logs", help="Show recent logs")
+        logs_parser.add_argument(
+            "--tail", "-n",
+            type=int,
+            default=50,
+            help="Number of lines to show"
+        )
+        logs_parser.add_argument("--json", action="store_true", help="Output JSON")
+        logs_parser.set_defaults(func=LSPHandler.handle_logs)
+    
+    @staticmethod
+    def handle_start(args: argparse.Namespace) -> int:
+        """Handle lsp start command."""
+        return asyncio.run(LSPHandler._run_start(args))
+    
+    @staticmethod
+    async def _run_start(args: argparse.Namespace) -> int:
+        """Start LSP server."""
+        from .interactive_runtime import create_runtime
+        
+        runtime = create_runtime(
+            workspace=args.workspace,
+            lsp=True,
+            acp=False
+        )
+        
+        try:
+            await runtime.start()
+            
+            result = {
+                "action": "start",
+                "lsp_ready": runtime.lsp_ready,
+                "lsp_status": runtime._lsp_state.status.value,
+                "workspace": args.workspace,
+                "language": getattr(args, 'language', None) or "auto-detected"
+            }
+            
+            if runtime._lsp_state.error:
+                result["error"] = runtime._lsp_state.error
+            
+            LSPHandler._output_result(result, args.json)
+            
+            # Keep running if started successfully
+            if runtime.lsp_ready:
+                if not args.json:
+                    print("\nLSP server running. Press Ctrl+C to stop.")
+                try:
+                    while True:
+                        await asyncio.sleep(1)
+                except KeyboardInterrupt:
+                    pass
+            
+            return 0 if runtime.lsp_ready else 1
+            
+        finally:
+            await runtime.stop()
+    
+    @staticmethod
+    def handle_stop(args: argparse.Namespace) -> int:
+        """Handle lsp stop command."""
+        result = {
+            "action": "stop",
+            "message": "LSP servers stopped"
+        }
+        LSPHandler._output_result(result, args.json)
+        return 0
+    
+    @staticmethod
+    def handle_status(args: argparse.Namespace) -> int:
+        """Handle lsp status command."""
+        return asyncio.run(LSPHandler._run_status(args))
+    
+    @staticmethod
+    async def _run_status(args: argparse.Namespace) -> int:
+        """Check LSP status."""
+        from .interactive_runtime import create_runtime
+        
+        runtime = create_runtime(
+            workspace=args.workspace,
+            lsp=True,
+            acp=False
+        )
+        
+        try:
+            await runtime.start()
+            
+            result = {
+                "lsp_enabled": True,
+                "lsp_ready": runtime.lsp_ready,
+                "lsp_status": runtime._lsp_state.status.value,
+                "workspace": args.workspace
+            }
+            
+            if runtime._lsp_state.error:
+                result["error"] = runtime._lsp_state.error
+            
+            LSPHandler._output_result(result, args.json)
+            return 0 if runtime.lsp_ready else 1
+            
+        finally:
+            await runtime.stop()
+    
+    @staticmethod
+    def handle_logs(args: argparse.Namespace) -> int:
+        """Handle lsp logs command."""
+        result = {
+            "action": "logs",
+            "tail": args.tail,
+            "logs": [],
+            "message": "LSP logging not yet implemented"
+        }
+        LSPHandler._output_result(result, args.json)
+        return 0
+    
+    @staticmethod
+    def _output_result(result: dict, as_json: bool) -> None:
+        """Output result in appropriate format."""
+        if as_json:
+            print(json.dumps(result, indent=2, default=str))
+        else:
+            for key, value in result.items():
+                print(f"{key}: {value}")
+
+
+def run_lsp_command(args: List[str]) -> int:
+    """Run lsp command from CLI."""
+    parser = argparse.ArgumentParser(prog="praisonai lsp")
+    subparsers = parser.add_subparsers(dest="lsp_command", help="LSP subcommands")
+    
+    # lsp start
+    start_parser = subparsers.add_parser("start", help="Start LSP server(s)")
+    start_parser.add_argument("--language", "-l", type=str)
+    start_parser.add_argument("--workspace", "-w", type=str, default=".")
+    start_parser.add_argument("--json", action="store_true")
+    start_parser.set_defaults(func=LSPHandler.handle_start)
+    
+    # lsp stop
+    stop_parser = subparsers.add_parser("stop", help="Stop LSP server(s)")
+    stop_parser.add_argument("--json", action="store_true")
+    stop_parser.set_defaults(func=LSPHandler.handle_stop)
+    
+    # lsp status
+    status_parser = subparsers.add_parser("status", help="Show LSP status")
+    status_parser.add_argument("--workspace", "-w", type=str, default=".")
+    status_parser.add_argument("--json", action="store_true")
+    status_parser.set_defaults(func=LSPHandler.handle_status)
+    
+    # lsp logs
+    logs_parser = subparsers.add_parser("logs", help="Show recent logs")
+    logs_parser.add_argument("--tail", "-n", type=int, default=50)
+    logs_parser.add_argument("--json", action="store_true")
+    logs_parser.set_defaults(func=LSPHandler.handle_logs)
+    
+    parsed = parser.parse_args(args)
+    if hasattr(parsed, 'func'):
+        return parsed.func(parsed)
+    else:
+        parser.print_help()
+        return 0

src/praisonai/praisonai/cli/main.py

@@ -254,6 +254,11 @@ class PraisonAI:
         invocation_cmd = "praisonai"
         version_string = f"PraisonAI version {__version__}"
 
+        # Handle -p/--prompt flag - treat as direct prompt
+        if getattr(args, 'prompt_flag', None):
+            args.direct_prompt = args.prompt_flag
+            args.command = None
+        
         # Handle --interactive flag - start interactive TUI mode
         if getattr(args, 'interactive', False):
             self._start_interactive_mode(args)
@@ -261,7 +266,7 @@ class PraisonAI:
         
         # Handle --chat-mode flag - run single prompt in interactive style (non-interactive)
         if getattr(args, 'chat_mode', False):
-            prompt = args.command or getattr(args, 'direct_prompt', None)
+            prompt = args.command or getattr(args, 'direct_prompt', None) or getattr(args, 'prompt_flag', None)
             if prompt:
                 return self._run_chat_mode(prompt, args)
 
@@ -698,7 +703,7 @@ class PraisonAI:
             return default_args
         
         # Define special commands
-        special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research', 'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools', 'todo', 'docs', 'mcp', 'commit', 'serve', 'schedule', 'skills', 'profile', 'eval', 'agents', 'run', 'thinking', 'compaction', 'output', 'deploy', 'templates', 'recipe', 'endpoints', 'audio', 'embed', 'images', 'moderate', 'files', 'batches', 'vector-stores', 'rerank', 'ocr', 'assistants', 'fine-tuning', 'completions', 'messages', 'guardrails', 'rag', 'videos', 'a2a', 'containers', 'passthrough', 'responses', 'search', 'realtime-api', 'doctor', 'registry', 'package', 'install', 'uninstall', 'acp']
+        special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research', 'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools', 'todo', 'docs', 'mcp', 'commit', 'serve', 'schedule', 'skills', 'profile', 'eval', 'agents', 'run', 'thinking', 'compaction', 'output', 'deploy', 'templates', 'recipe', 'endpoints', 'audio', 'embed', 'images', 'moderate', 'files', 'batches', 'vector-stores', 'rerank', 'ocr', 'assistants', 'fine-tuning', 'completions', 'messages', 'guardrails', 'rag', 'videos', 'a2a', 'containers', 'passthrough', 'responses', 'search', 'realtime-api', 'doctor', 'registry', 'package', 'install', 'uninstall', 'acp', 'debug', 'lsp', 'diag']
         
         parser = argparse.ArgumentParser(prog="praisonai", description="praisonAI command-line interface")
         parser.add_argument("--framework", choices=["crewai", "autogen", "praisonai"], help="Specify the framework")
@@ -818,6 +823,9 @@ class PraisonAI:
         parser.add_argument("--interactive", "-i", action="store_true", help="Start interactive terminal mode with slash commands")
         parser.add_argument("--chat-mode", "--chat", action="store_true", dest="chat_mode", help="Run single prompt in interactive style (non-interactive, for testing)")
         
+        # Direct prompt flag - alternative to positional command
+        parser.add_argument("-p", "--prompt", type=str, dest="prompt_flag", help="Direct prompt to execute (alternative to positional argument)")
+        
         # Autonomy Mode - control AI action approval
         parser.add_argument("--autonomy", type=str, choices=["suggest", "auto_edit", "full_auto"], help="Set autonomy mode for AI actions")
         
@@ -903,6 +911,24 @@ class PraisonAI:
             from praisonai.cli.features.acp import run_acp_command
             exit_code = run_acp_command(unknown_args)
             sys.exit(exit_code)
+        
+        # Handle debug command - Debug and test interactive flows
+        if args.command == 'debug':
+            from praisonai.cli.features.debug import run_debug_command
+            exit_code = run_debug_command(unknown_args)
+            sys.exit(exit_code)
+        
+        # Handle lsp command - LSP service lifecycle
+        if args.command == 'lsp':
+            from praisonai.cli.features.lsp_cli import run_lsp_command
+            exit_code = run_lsp_command(unknown_args)
+            sys.exit(exit_code)
+        
+        # Handle diag command - Diagnostics export
+        if args.command == 'diag':
+            from praisonai.cli.features.diag import run_diag_command
+            exit_code = run_diag_command(unknown_args)
+            sys.exit(exit_code)
 
         # Handle both command and flag versions for call
         if args.command == 'call' or args.call:

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==2.8.9 gunicorn markdown\n")
+            file.write("RUN pip install flask praisonai==2.9.0 gunicorn markdown\n")
             file.write("EXPOSE 8080\n")
             file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
             

src/praisonai/praisonai/version.py

@@ -1 +1 @@
-__version__ = "2.8.9"
+__version__ = "2.9.0"