Merge pull request #45 from BrainDriveAI/feature/plugin-services-runtime

# Add Plugin Service Runtime Management + JWT UTC Fix

## Summary

This PR introduces **Plugin Service Runtime Management**—a framework for discovering, installing, starting, health-checking, and stopping plugin-scoped backend services (e.g., docker-compose stacks or Python microservices). It also updates JWT token issuance to use **UTC-aware timestamps** to prevent “signature expired” issues on machines with timezone drift.

## Why

* Plugins increasingly depend on **companion services** (retrievers, workers, vector stores, etc.). We need a first-class way to declare and manage those.
* JWTs created with `datetime.utcnow()` can cause **intermittent auth failures** under clock skew or timezone quirks. Moving to `datetime.now(UTC)` makes tokens robust and standards-aligned.

---

## What’s Changed

### Ignored paths

* `.gitignore`

  * `backend/services_runtime/` – local service checkouts, venvs, compose files
  * `backend/persistent/` – optional local data/state

### Security / Auth

* `backend/app/core/security.py`

  * Replace `datetime.utcnow()` with `datetime.now(UTC)` for `exp` and `iat`.

### Data Model

* `backend/app/models/plugin.py`

  * Add `required_services_runtime` (JSON string) to `Plugin`
  * Add `Plugin.service_runtimes` relationship
  * New model **`PluginServiceRuntime`** with fields:

    * ids, `plugin_id`, `plugin_slug`, `name`, `source_url`, `type`
    * `install_command`, `start_command`, `healthcheck_url`
    * `required_env_vars` (JSON), `status`
    * `created_at`, `updated_at` (UTC), `user_id`
* **Migration**: `backend/migrations/versions/5d073fe444c9_*.py`

  * Creates `plugin_service_runtime` table + indices
  * Adds `plugin.required_services_runtime` column

### DTOs

* `backend/app/dto/plugin.py`

  * Add `PluginServiceRuntimeDTO` (Pydantic) and conversion helpers

### Plugin Installer & Repository

* `backend/app/plugins/remote_installer.py`

  * Extract `required_services_runtime` from plugin sources
  * After install, **automatically install & start required services**
* `backend/app/plugins/repository.py`

  * Add `get_all_service_runtimes()` returning DTOs (for startup orchestration)

### Runtime Orchestration

* New package: `backend/app/plugins/service_installler/` (sic)

  * `service_runtime_extractor.py` – robust parser for `required_services_runtime` in plugin code
  * `plugin_service_manager.py` – download/extract sources, env prechecks, type-based runners
  * `docker_manager.py` – docker checks, `docker compose` start/stop, health waits
  * `python_manager.py` – venv create, install, start, health wait
  * `service_health_checker.py` – async health polling
  * `prerequisites.py` – .env loading and required vars validation
  * `service_debugger.py` – developer utility for parsing verification
  * `start_stop_plugin_services.py` – **startup**: start all persisted runtimes; **shutdown**: stop all

### App lifecycle hooks

* `backend/app/main.py` and `backend/main.py`

  * On **startup**: `start_plugin_services_on_startup()`
  * On **shutdown**: `stop_all_plugin_services_on_shutdown()`

---

## Database Migration

* **Alembic Revision:** `5d073fe444c9`
* **Upgrade:**

  ```bash
  cd backend
  alembic upgrade head
  ```
* **Downgrade:**

  ```bash
  cd backend
  alembic downgrade -1
  ```
* **Backfill:** none required.
* **Compatibility:** preserves existing plugin rows; new column is nullable; new table is additive.

---

## How It Works (High-Level)

1. **Plugin declares services** in its lifecycle manager as `required_services_runtime = [...]`.
2. **Remote installer** extracts and validates the list during install.
3. **Manager** downloads sources (ZIP/TAR from Git host), prepares environment (.env projection), and starts services:

   * `type=="docker-compose"` → `docker compose up --build -d` + health check
   * `type=="python"` → venv create → install → start + health check
4. **On app startup**, any persisted `plugin_service_runtime` with status in `pending/stopped/running` is resumed.
5. **On shutdown**, all services are stopped gracefully.

---

## Operational Notes

### Requirements

* Docker & Docker Compose available and daemon running (for dockerized services).
* Root backend `.env` must define **all required env vars** referenced by services. The installer writes a **service-local .env** with those keys.

### New Local Folders

* `backend/services_runtime/` – service code, venvs, compose files (ignored)
* `backend/persistent/` – optional runtime data (ignored)

---

## Security & Compliance

* **JWT**: Use UTC (`datetime.now(UTC)`) for `exp`/`iat` → reduces clock-skew issues.
* **Process execution**: service runners use explicit commands from plugin metadata; health checks guard readiness.
* **Env handling**: only copies whitelisted keys from root `.env` into the service scope.

Author: DJJones66

.gitignore

@@ -47,6 +47,9 @@ backend/alembic/versions/
 *.log
 backend/logs/
 
+backend/services_runtime/
+backend/persistent/
+
 # Environment variables
 .env
 .env.local

ROADMAP.md

@@ -59,11 +59,12 @@ Our roadmap is broken into clearly defined versions, each building toward a stab
 ## Version 0.6.0 – Open Beta
 > Goal: AI System with core functionality for developers
 
-- [ ] All plugins moved to the Life Cycle Manager
-- [ ] Ollama plugin updated to include server manager
-- [ ] User initializer - Plugin install from remote
-- [ ] User initializer - Restructure navigation 
-- [ ] Prompt Library
+- [x] All plugins moved to the Life Cycle Manager
+- [x] Ollama plugin updated to include server manager
+- [x] User initializer - Plugin install from remote
+- [x] User initializer - Restructure navigation 
+- [ ] Improve Registration
+- [ ] Ollama AI Provider
 - [ ] One-Click Installer - (Windows first)
 
 ---
@@ -73,6 +74,7 @@ Our roadmap is broken into clearly defined versions, each building toward a stab
 
 - [ ] Unified Dynamic Page Renderer - Bounce
 - [ ] Unified Dynamic Page Renderer - Finetune
+- [ ] Prompt Library
 
 ---
 

backend/app/core/security.py

@@ -1,5 +1,5 @@
 from sqlalchemy.ext.asyncio import AsyncSession
-from datetime import datetime, timedelta as datetime_timedelta
+from datetime import datetime, timedelta as datetime_timedelta, UTC
 from typing import Optional
 from jose import jwt, JWTError
 import logging
@@ -41,16 +41,16 @@ def create_access_token(data: dict, expires_delta: Optional[datetime_timedelta]
     """Create a new access token."""
     to_encode = data.copy()
     if expires_delta:
-        expire = datetime.utcnow() + expires_delta
+        expire = datetime.now(UTC) + expires_delta
     else:
-        expire = datetime.utcnow() + datetime_timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+        expire = datetime.now(UTC) + datetime_timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
 
     # Convert datetime to Unix timestamp for JWT
     to_encode.update({"exp": expire.timestamp()})
 
     # Let JWT library handle iat automatically if not provided
     if "iat" not in to_encode:
-        to_encode.update({"iat": datetime.utcnow().timestamp()})
+        to_encode.update({"iat": datetime.now(UTC).timestamp()})
 
     encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
     return encoded_jwt

backend/app/dto/__init__.py

@@ -0,0 +1,65 @@
+from pydantic import BaseModel
+from typing import List, Optional, Union
+from datetime import datetime
+import uuid
+
+# This schema is used for returning data from the repository.
+# It ensures that JSON fields are correctly converted to Python types.
+class PluginServiceRuntimeDTO(BaseModel):
+    """
+    A Pydantic model to represent a PluginServiceRuntime object,
+    with required_env_vars as a list of strings.
+    """
+    id: str
+    plugin_id: str
+    plugin_slug: str
+    name: str
+    source_url: Optional[str] = None
+    type: Optional[str] = None
+    install_command: Optional[str] = None
+    start_command: Optional[str] = None
+    healthcheck_url: Optional[str] = None
+    required_env_vars: List[str] = []
+    status: Optional[str] = None
+    user_id: str
+    created_at: Optional[datetime] = None
+    updated_at: Optional[datetime] = None
+
+    @classmethod
+    def from_github_data(cls, service_dict: dict, plugin_id: str, plugin_slug: str, user_id: str) -> 'PluginServiceRuntimeDTO':
+        """
+        Create a PluginServiceRuntimeDTO from raw GitHub service data (dict).
+        This handles first-time installation where database fields don't exist yet.
+        """
+        return cls(
+            id=str(uuid.uuid4()),  # Generate new UUID for first install
+            plugin_id=plugin_id,
+            plugin_slug=plugin_slug,
+            user_id=user_id,
+            name=service_dict.get('name'),
+            source_url=service_dict.get('source_url'),
+            type=service_dict.get('type', 'python'),
+            install_command=service_dict.get('install_command'),
+            start_command=service_dict.get('start_command'),
+            healthcheck_url=service_dict.get('healthcheck_url'),
+            required_env_vars=service_dict.get('required_env_vars', []),
+            status='installing',
+            created_at=datetime.now(),
+            updated_at=datetime.now()
+        )
+
+    @classmethod
+    def from_dict_or_dto(cls, data: Union[dict, 'PluginServiceRuntimeDTO'], plugin_id: str = None, plugin_slug: str = None, user_id: str = None) -> 'PluginServiceRuntimeDTO':
+        """
+        Flexible factory method that handles both dict (GitHub) and DTO (database) inputs.
+        """
+        if isinstance(data, cls):
+            return data  # Already a DTO, return as-is
+        elif isinstance(data, dict):
+            # Dict from GitHub, convert using factory method
+            if not all([plugin_id, plugin_slug, user_id]):
+                raise ValueError("plugin_id, plugin_slug, and user_id are required when converting from dict")
+            return cls.from_github_data(data, plugin_id, plugin_slug, user_id)
+        else:
+            raise TypeError(f"Expected dict or {cls.__name__}, got {type(data)}")
+

backend/app/dto/plugin.py

@@ -5,6 +5,7 @@
 from app.api.v1.api import api_router
 from app.core.config import settings
 from app.routers.plugins import plugin_manager
+from app.plugins.service_installler.start_stop_plugin_services import start_plugin_services_on_startup, stop_all_plugin_services_on_shutdown
 import logging
 import time
 import structlog
@@ -29,8 +30,21 @@ async def startup_event():
     logger.info("Initializing application settings...")
     from app.init_settings import init_ollama_settings
     await init_ollama_settings()
+    # Start plugin services
+    await start_plugin_services_on_startup()
     logger.info("Settings initialization completed")
 
+
+# Add shutdown event to gracefully stop services
+@app.on_event("shutdown")
+async def shutdown_event():
+    """Gracefully stop all plugin services on application shutdown."""
+    logger.info("Shutting down application and stopping plugin services...")
+    # Stop all plugin services gracefully
+    await stop_all_plugin_services_on_shutdown()
+    logger.info("Application shutdown completed.")
+
+
 # Add middleware to log all requests
 logger = structlog.get_logger()
 
@@ -94,5 +108,3 @@ async def validation_exception_handler(request: Request, exc: RequestValidationE
 
 # Include API routers
 app.include_router(api_router)
-
-

backend/app/main.py

@@ -1,10 +1,13 @@
-from sqlalchemy import Column, String, Integer, Boolean, ForeignKey, Text, JSON, UniqueConstraint, TIMESTAMP
+from sqlalchemy import Column, String, Integer, Boolean, ForeignKey, Text, JSON, UniqueConstraint, TIMESTAMP, DateTime
 import sqlalchemy
 from sqlalchemy.orm import relationship
 from sqlalchemy.sql import func
+from datetime import datetime, UTC
+import json
 
 from app.models.base import Base
 
+
 class Plugin(Base):
     """SQLAlchemy model for plugins."""
 
@@ -44,6 +47,7 @@ class Plugin(Base):
     config_fields = Column(Text)  # Stored as JSON string
     messages = Column(Text)       # Stored as JSON string
     dependencies = Column(Text)   # Stored as JSON string
+    required_services_runtime = Column(Text, nullable=True)
 
     # Timestamps
     created_at = Column(String, default=func.now())
@@ -60,6 +64,7 @@ class Plugin(Base):
 
     # Relationships
     modules = relationship("Module", back_populates="plugin", cascade="all, delete-orphan")
+    service_runtimes = relationship("PluginServiceRuntime", back_populates="plugin", cascade="all, delete-orphan")
 
     def to_dict(self):
         """Convert model to dictionary."""
@@ -118,6 +123,11 @@ def to_dict(self):
         else:
             result["permissions"] = []
 
+        if self.required_services_runtime:
+            result["requiredServicesRuntime"] = json.loads(self.required_services_runtime)
+        else:
+            result["requiredServicesRuntime"] = []
+
         return result
 
     @classmethod
@@ -162,10 +172,81 @@ def from_dict(cls, data):
         # Remove modules from data as they are handled separately
         if "modules" in db_data:
             db_data.pop("modules")
+
+        # Handle service runtimes (only store names in plugin table)
+        if "requiredServicesRuntime" in db_data and db_data["requiredServicesRuntime"] is not None:
+            db_data["required_services_runtime"] = json.dumps([
+                r["name"] for r in db_data["requiredServicesRuntime"]
+            ])
+            db_data.pop("requiredServicesRuntime")
 
         return cls(**db_data)
 
 
+class PluginServiceRuntime(Base):
+    """SQLAlchemy model for plugin service runtimes."""
+
+    __tablename__ = "plugin_service_runtime"
+
+    id = Column(String, primary_key=True, index=True)
+    plugin_id = Column(String, ForeignKey("plugin.id"), nullable=False, index=True)
+    plugin_slug = Column(String, nullable=False, index=True)
+
+    name = Column(String, nullable=False)
+    source_url = Column(String)
+    type = Column(String)
+    install_command = Column(Text)
+    start_command = Column(Text)
+    healthcheck_url = Column(String)
+    required_env_vars = Column(Text)  # store as JSON string
+    status = Column(String, default="pending")
+
+    created_at = Column(DateTime, default=datetime.now(UTC))
+    updated_at = Column(DateTime, default=datetime.now(UTC), onupdate=datetime.now(UTC))
+
+    user_id = Column(String(32), ForeignKey("users.id", name="fk_plugin_service_runtime_user_id"), nullable=False)
+    user = relationship("User")
+
+    # Relationship back to plugin
+    plugin = relationship("Plugin", back_populates="service_runtimes")
+
+    def to_dict(self):
+        """
+        Convert the model instance to a dictionary, handling JSON fields.
+        """
+        return {
+            "id": self.id,
+            "plugin_id": self.plugin_id,
+            "plugin_slug": self.plugin_slug,
+            "name": self.name,
+            "source_url": self.source_url,
+            "type": self.type,
+            "install_command": self.install_command,
+            "start_command": self.start_command,
+            "healthcheck_url": self.healthcheck_url,
+            "required_env_vars": json.loads(self.required_env_vars) if self.required_env_vars else [],
+            "status": self.status,
+            "user_id": self.user_id,
+            "created_at": self.created_at.isoformat() if self.created_at else None,
+            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict):
+        """
+        Create a new instance from a dictionary, serializing JSON fields.
+        """
+        db_data = data.copy()
+        
+        if "required_env_vars" in db_data and db_data["required_env_vars"] is not None:
+            db_data["required_env_vars"] = json.dumps(db_data["required_env_vars"])
+            
+        # Handle conversion from camelCase to snake_case if necessary
+        # For simplicity, we are assuming keys in the incoming dict match model attributes
+        
+        return cls(**db_data)
+
+
 class Module(Base):
     """SQLAlchemy model for plugin modules."""
 

backend/app/models/plugin.py

@@ -20,6 +20,8 @@
 from typing import Dict, Any, Optional, List, Tuple
 from urllib.parse import urlparse
 import structlog
+from .service_installler.plugin_service_manager import install_and_run_required_services
+from .service_installler.service_runtime_extractor import extract_required_services_runtime
 
 logger = structlog.get_logger()
 
@@ -237,6 +239,17 @@ async def install_from_url(self, repo_url: str, user_id: str, version: str = "la
             if install_result['success']:
                 logger.info(f"Plugin installation successful: {install_result}")
 
+                service_runtime: list = validation_result.get("service_runtime", [])
+                logger.info(f"\n\n>>>>>>>>SERVICE RUNTIME\n\n: {service_runtime}\n\n>>>>>>>>>>")
+                if service_runtime:
+                    plugin_slug = validation_result["plugin_info"].get("plugin_slug")
+                    await install_and_run_required_services(
+                        service_runtime,
+                        plugin_slug,
+                        install_result['plugin_id'],
+                        user_id
+                    )
+
                 # Store installation metadata
                 try:
                     await self._store_installation_metadata(
@@ -718,6 +731,7 @@ async def _validate_plugin_structure(self, plugin_dir: Path) -> Dict[str, Any]:
 
             # Try to load plugin metadata
             plugin_info = {}
+            service_runtime = []
 
             # Check package.json
             package_json_path = plugin_dir / 'package.json'
@@ -828,6 +842,12 @@ async def _validate_plugin_structure(self, plugin_dir: Path) -> Dict[str, Any]:
                                     extracted_slug = slug_match.group(1)
                                     plugin_info['plugin_slug'] = extracted_slug
                                     logger.info(f"Extracted plugin_slug from source: {extracted_slug}")
+
+                                # Extract services using the dedicated function
+                                services = extract_required_services_runtime(content, plugin_info.get('plugin_slug'))
+                                if services:
+                                    plugin_info['required_services_runtime'] = services
+                                    service_runtime.extend(services)
                     except Exception as extract_error:
                         logger.warning(f"Could not extract plugin_slug from source: {extract_error}")
 
@@ -839,7 +859,8 @@ async def _validate_plugin_structure(self, plugin_dir: Path) -> Dict[str, Any]:
 
             return {
                 'valid': True,
-                'plugin_info': plugin_info
+                'plugin_info': plugin_info,
+                'service_runtime': service_runtime
             }
 
         except Exception as e:

backend/app/plugins/remote_installer.py

@@ -6,7 +6,8 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 import structlog
 
-from app.models.plugin import Plugin, Module
+from app.models.plugin import Plugin, Module, PluginServiceRuntime
+from app.dto.plugin import PluginServiceRuntimeDTO
 
 logger = structlog.get_logger()
 
@@ -85,7 +86,7 @@ async def get_all_plugins_with_modules(self, user_id: str = None) -> List[Dict[s
             plugin_dicts = []
             for plugin in plugins:
                 plugin_dict = plugin.to_dict()
-                
+
                 # Get modules for this plugin
                 modules_query = select(Module).where(Module.plugin_id == plugin.id)
 
@@ -104,6 +105,25 @@ async def get_all_plugins_with_modules(self, user_id: str = None) -> List[Dict[s
         except Exception as e:
             logger.error("Error getting plugins with modules", error=str(e))
             raise
+
+    async def get_all_service_runtimes(self) -> List[PluginServiceRuntimeDTO]:
+        """
+        Get all plugin service runtimes for startup and return them as DTOs.
+        """
+        try:
+            query = select(PluginServiceRuntime).where(
+                PluginServiceRuntime.status.in_(["pending", "stopped", "running"])
+            )
+            
+            result = await self.db.execute(query)
+            services = result.scalars().all()
+            
+            # Convert SQLAlchemy models to Pydantic DTOs for a typed return
+            return [PluginServiceRuntimeDTO(**service.to_dict()) for service in services]
+            
+        except Exception as e:
+            logger.error("Error getting service runtimes", error=str(e))
+            raise
 
     async def get_plugin(self, plugin_id: str) -> Optional[Dict[str, Any]]:
         """Get a specific plugin by ID."""