enhanced-mcp-tools/examples/bulk_operations_safety_guide.py

"""
BulkToolCaller Safety Guide and Best Practices

Comprehensive guide for safely using BulkToolCaller with Enhanced MCP Tools.
Demonstrates SACRED TRUST safety framework integration and best practices.
"""

import asyncio
import sys
from pathlib import Path

# Add src to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))

from enhanced_mcp.bulk_operations import BulkToolCaller, BulkOperationMode
from enhanced_mcp.security_manager import SecurityManager
from enhanced_mcp.base import SecurityLevel


class SafetyGuide:
    """Comprehensive safety guide for BulkToolCaller usage"""

    @staticmethod
    def print_safety_principles():
        """Print the core safety principles for bulk operations"""
        print("🛡️ SACRED TRUST Safety Principles for Bulk Operations")
        print("=" * 60)
        print()
        print("1. 🧪 ALWAYS DRY RUN FIRST")
        print("   • Every workflow must be validated with dry_run_bulk_workflow")
        print("   • Review dry run results before live execution")
        print("   • Understand what each operation will do")
        print()
        print("2. 🔒 PROGRESSIVE CONFIRMATION")
        print("   • Safe operations: No confirmation needed")
        print("   • Caution operations: Standard confirmation")
        print("   • Destructive operations: Explicit confirmation required")
        print()
        print("3. 🔄 BACKUP BEFORE MODIFICATION")
        print("   • Create backups for any destructive operations")
        print("   • Use versioned backup names")
        print("   • Test backup restoration procedures")
        print()
        print("4. 📊 MONITOR AND VALIDATE")
        print("   • Watch execution progress in real-time")
        print("   • Validate results after completion")
        print("   • Check for unexpected side effects")
        print()
        print("5. 🚨 FAIL FAST AND SAFE")
        print("   • Stop execution on first critical error")
        print("   • Preserve system state when possible")
        print("   • Provide clear error messages and recovery steps")

    @staticmethod
    def demonstrate_security_levels():
        """Demonstrate the three security levels and their implications"""
        print("\n🔐 Security Levels in Bulk Operations")
        print("=" * 45)
        print()
        print("🟢 SAFE LEVEL (Always Available)")
        print("   • Read-only operations")
        print("   • Information gathering")
        print("   • Status checks and monitoring")
        print("   • No risk of data loss or system changes")
        print("   Examples:")
        print("     - git_status")
        print("     - file analysis")
        print("     - security scans")
        print("     - dependency analysis")
        print()
        print("🟡 CAUTION LEVEL (Normal Mode)")
        print("   • Create/modify operations that are reversible")
        print("   • Operations with built-in safeguards")
        print("   • Require standard confirmation")
        print("   Examples:")
        print("     - file backups")
        print("     - code formatting")
        print("     - test execution")
        print("     - log analysis")
        print()
        print("🔴 DESTRUCTIVE LEVEL (Requires Explicit Enablement)")
        print("   • Operations that can cause data loss")
        print("   • Irreversible system changes")
        print("   • Require explicit confirmation")
        print("   Examples:")
        print("     - file deletion")
        print("     - database modifications")
        print("     - system configuration changes")
        print("     - bulk file modifications")

    @staticmethod
    async def demonstrate_safety_workflow():
        """Demonstrate a complete safety workflow"""
        print("\n🎯 Complete Safety Workflow Demonstration")
        print("=" * 50)

        # Create instances
        security_manager = SecurityManager()
        bulk_operations = BulkToolCaller()
        bulk_operations.set_security_manager(security_manager)

        print("\n1. 🔍 Initial Security Assessment")
        # Check initial security status
        security_status = await security_manager.security_status()
        print(f"   Initial protection level: {security_status.get('safety_summary', {}).get('protection_level', 'UNKNOWN')}")

        # List tools by security level
        tools_by_security = await security_manager.list_tools_by_security()
        safe_tools = len(tools_by_security.get('security_levels', {}).get('safe', {}).get('tools', []))
        print(f"   Safe tools available: {safe_tools}")

        print("\n2. 📋 Creating Safe Workflow")
        # Create a workflow with only safe operations
        safe_operations = [
            {
                "id": "check_status",
                "tool_name": "git_status",
                "arguments": {"path": "/example/project"},
                "description": "Check project status",
                "security_level": SecurityLevel.SAFE
            },
            {
                "id": "analyze_files",
                "tool_name": "file_ops_analyze_file_complexity",
                "arguments": {"path": "/example/project"},
                "description": "Analyze code complexity",
                "security_level": SecurityLevel.SAFE,
                "depends_on": ["check_status"]
            }
        ]

        create_result = await bulk_operations.create_bulk_workflow(
            name="Safe Analysis Workflow",
            description="Read-only analysis workflow",
            operations=safe_operations,
            mode="staged"
        )

        if create_result.get("success"):
            workflow_id = create_result["workflow_id"]
            print(f"   ✅ Safe workflow created: {workflow_id}")

            print("\n3. 🧪 Mandatory Dry Run")
            dry_run_result = await bulk_operations.dry_run_bulk_workflow(workflow_id)

            if dry_run_result.get("ready_for_execution"):
                print("   ✅ Dry run passed - workflow is safe to execute")

                print("\n4. 🚀 Safe Execution")
                execution_result = await bulk_operations.execute_bulk_workflow(
                    workflow_id=workflow_id,
                    dry_run=False,  # This is safe because all operations are SAFE level
                    confirm_destructive=False
                )

                if execution_result.get("success"):
                    print("   ✅ Safe workflow executed successfully")
                else:
                    print(f"   ❌ Execution failed: {execution_result.get('error')}")
            else:
                print("   ❌ Dry run failed - workflow has safety issues")
        else:
            print(f"   ❌ Workflow creation failed: {create_result.get('error')}")

    @staticmethod
    async def demonstrate_destructive_workflow_safety():
        """Demonstrate handling destructive operations safely"""
        print("\n⚠️ Destructive Operations Safety Demonstration")
        print("=" * 55)

        security_manager = SecurityManager()
        bulk_operations = BulkToolCaller()
        bulk_operations.set_security_manager(security_manager)

        print("\n1. 🚫 Attempting Destructive Operation (Should Fail)")
        # Create workflow with destructive operations
        destructive_operations = [
            {
                "id": "backup_first",
                "tool_name": "archive_create_backup",
                "arguments": {"source_paths": ["/example/file.txt"]},
                "description": "Create backup before deletion",
                "security_level": SecurityLevel.CAUTION
            },
            {
                "id": "delete_file",
                "tool_name": "file_ops_delete_file",
                "arguments": {"file_path": "/example/file.txt"},
                "description": "Delete the file",
                "security_level": SecurityLevel.DESTRUCTIVE,
                "depends_on": ["backup_first"]
            }
        ]

        create_result = await bulk_operations.create_bulk_workflow(
            name="Destructive Workflow",
            description="Workflow with destructive operations",
            operations=destructive_operations,
            mode="staged"
        )

        if create_result.get("success"):
            workflow_id = create_result["workflow_id"]
            print(f"   ⚠️ Destructive workflow created: {workflow_id}")

            # Try to execute without enabling destructive tools
            print("\n2. ❌ Execution Without Proper Enablement")
            execution_result = await bulk_operations.execute_bulk_workflow(
                workflow_id=workflow_id,
                dry_run=False,
                confirm_destructive=True  # Even with confirmation, should fail
            )

            if "error" in execution_result:
                print(f"   ✅ Properly blocked: {execution_result['error']}")

            print("\n3. 🔓 Enabling Destructive Tools (Requires Confirmation)")
            enable_result = await security_manager.enable_destructive_tools(
                enabled=True,
                confirm_destructive=True
            )

            if enable_result.get("success"):
                print("   ⚠️ Destructive tools now enabled")

                print("\n4. 🧪 Mandatory Dry Run for Destructive Operations")
                dry_run_result = await bulk_operations.dry_run_bulk_workflow(workflow_id)

                if dry_run_result.get("ready_for_execution"):
                    print("   ✅ Dry run validation passed")
                    print("   🛡️ In real scenario, would proceed with extreme caution")
                    print("   📋 Would execute with:")
                    print("     • Full confirmation")
                    print("     • Real-time monitoring")
                    print("     • Rollback plan ready")
                else:
                    print("   ❌ Dry run failed - unsafe to proceed")

            print("\n5. 🔒 Re-securing System")
            disable_result = await security_manager.enable_destructive_tools(
                enabled=False,
                confirm_destructive=False
            )
            print("   🛡️ Destructive tools disabled for safety")

    @staticmethod
    def print_error_handling_guide():
        """Print comprehensive error handling guide"""
        print("\n🚨 Error Handling and Recovery Guide")
        print("=" * 42)
        print()
        print("Common Error Scenarios and Solutions:")
        print()
        print("1. 🔧 Tool Not Found Error")
        print("   Error: 'Tool not found in registry'")
        print("   Solution:")
        print("     • Check tool name spelling")
        print("     • Verify tool is registered with BulkToolCaller")
        print("     • Use list_tools to see available tools")
        print()
        print("2. 🔒 Security Validation Failed")
        print("   Error: 'Destructive tools are not enabled'")
        print("   Solution:")
        print("     • Use security_manager_enable_destructive_tools")
        print("     • Set confirm_destructive=True")
        print("     • Review security implications carefully")
        print()
        print("3. 🔄 Dependency Resolution Failed")
        print("   Error: 'Cannot resolve dependencies'")
        print("   Solution:")
        print("     • Check for circular dependencies")
        print("     • Verify all dependency IDs exist")
        print("     • Use sequential mode if dependencies are complex")
        print()
        print("4. ⏰ Operation Timeout")
        print("   Error: 'Operation timed out'")
        print("   Solution:")
        print("     • Break large operations into smaller chunks")
        print("     • Use parallel mode for independent operations")
        print("     • Increase timeout values if needed")
        print()
        print("5. 🔄 Rollback Required")
        print("   Error: 'Operation completed but results unexpected'")
        print("   Solution:")
        print("     • Use rollback_workflow if available")
        print("     • Restore from backups")
        print("     • Manually reverse changes if necessary")

    @staticmethod
    def print_best_practices():
        """Print comprehensive best practices"""
        print("\n💡 Best Practices for Bulk Operations")
        print("=" * 40)
        print()
        print("📋 Planning Phase:")
        print("   • Start with small, safe operations")
        print("   • Test workflows on non-critical data first")
        print("   • Document dependencies clearly")
        print("   • Plan rollback strategies")
        print()
        print("🧪 Testing Phase:")
        print("   • Always run dry_run_bulk_workflow first")
        print("   • Validate all operation arguments")
        print("   • Check for sufficient permissions")
        print("   • Verify backup creation works")
        print()
        print("🚀 Execution Phase:")
        print("   • Monitor progress in real-time")
        print("   • Be ready to cancel if needed")
        print("   • Validate results after completion")
        print("   • Document any issues encountered")
        print()
        print("🔄 Post-Execution:")
        print("   • Verify all operations completed successfully")
        print("   • Check for any side effects")
        print("   • Update documentation")
        print("   • Share lessons learned")
        print()
        print("🛡️ Security Considerations:")
        print("   • Use least privilege principle")
        print("   • Enable destructive tools only when necessary")
        print("   • Disable destructive tools after use")
        print("   • Audit bulk operations regularly")


async def demonstrate_complete_safety_workflow():
    """Demonstrate a complete end-to-end safety workflow"""
    print("\n🎯 Complete End-to-End Safety Demonstration")
    print("=" * 55)

    guide = SafetyGuide()

    # Print all safety information
    guide.print_safety_principles()
    guide.demonstrate_security_levels()

    # Demonstrate safe workflow
    await guide.demonstrate_safety_workflow()

    # Demonstrate destructive operation safety
    await guide.demonstrate_destructive_workflow_safety()

    # Print guides
    guide.print_error_handling_guide()
    guide.print_best_practices()

    print("\n✅ Safety Demonstration Complete!")
    print("\n🛡️ Remember: SACRED TRUST means:")
    print("   • Security first, always")
    print("   • Always validate before execution")
    print("   • Create backups for destructive operations")
    print("   • Review and understand every operation")
    print("   • Emergency stops and rollbacks ready")
    print("   • Document and learn from every workflow")
    print("   • Trust is earned through consistent safety practices")


def main():
    """Main safety guide demonstration"""
    print("🛡️ BulkToolCaller Safety Guide")
    print("=" * 35)
    print()
    print("This guide demonstrates safe usage of BulkToolCaller")
    print("with Enhanced MCP Tools security framework.")
    print()

    # Run the complete demonstration
    asyncio.run(demonstrate_complete_safety_workflow())


if __name__ == "__main__":
    main()