Initial commit: scheduler service extraction from portainer-core

Extracted standalone scheduler service with:
- FastAPI REST API for task management
- APScheduler-based task execution
- PostgreSQL persistence
- Docker container support
- Gitea Actions CI/CD workflow

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

src/config.py

@@ -0,0 +1,73 @@
+"""
+Configuration management for The Scheduler.
+Uses Pydantic BaseSettings for type-safe environment variable loading.
+"""
+from functools import lru_cache
+from pydantic_settings import BaseSettings
+from pydantic import Field
+
+
+class Settings(BaseSettings):
+    """Application settings loaded from environment variables."""
+
+    # Application
+    app_name: str = Field(default="The Scheduler", alias="APP_NAME")
+    app_version: str = Field(default="1.0.0", alias="APP_VERSION")
+    debug: bool = Field(default=False, alias="DEBUG")
+    host: str = Field(default="0.0.0.0", alias="HOST")
+    port: int = Field(default=8090, alias="PORT")
+    log_level: str = Field(default="INFO", alias="LOG_LEVEL")
+
+    # Security
+    scheduler_api_key: str = Field(default="dev-key-change-me", alias="SCHEDULER_API_KEY")
+
+    # PostgreSQL
+    postgres_host: str = Field(default="postgres-shared", alias="POSTGRES_HOST")
+    postgres_port: int = Field(default=5432, alias="POSTGRES_PORT")
+    postgres_db: str = Field(default="library_scheduler", alias="POSTGRES_DB")
+    postgres_user: str = Field(default="library_scheduler_user", alias="POSTGRES_USER")
+    postgres_password: str = Field(default="", alias="POSTGRES_PASSWORD")
+
+    # Redis
+    redis_host: str = Field(default="redis-shared", alias="REDIS_HOST")
+    redis_port: int = Field(default=6379, alias="REDIS_PORT")
+    redis_db: int = Field(default=3, alias="REDIS_DB")
+
+    # Gitea
+    gitea_url: str = Field(default="http://gitea:3000", alias="GITEA_URL")
+    gitea_user: str = Field(default="library", alias="GITEA_USER")
+    gitea_password: str = Field(default="", alias="GITEA_PASSWORD")
+    gitea_ssh_host: str = Field(default="gitea", alias="GITEA_SSH_HOST")
+    gitea_ssh_port: int = Field(default=22, alias="GITEA_SSH_PORT")
+
+    # Backup Configuration
+    backup_retention_daily: int = Field(default=7, alias="BACKUP_RETENTION_DAILY")
+    backup_retention_weekly: int = Field(default=4, alias="BACKUP_RETENTION_WEEKLY")
+    backup_retention_monthly: int = Field(default=12, alias="BACKUP_RETENTION_MONTHLY")
+
+    # Documentation Mirroring
+    docs_mirror_path: str = Field(default="/docs-mirror", alias="DOCS_MIRROR_PATH")
+    docs_check_interval: int = Field(default=21600, alias="DOCS_CHECK_INTERVAL")  # 6 hours
+
+    @property
+    def database_url(self) -> str:
+        """PostgreSQL connection URL for APScheduler."""
+        return f"postgresql://{self.postgres_user}:{self.postgres_password}@{self.postgres_host}:{self.postgres_port}/{self.postgres_db}"
+
+    @property
+    def redis_url(self) -> str:
+        """Redis connection URL."""
+        return f"redis://{self.redis_host}:{self.redis_port}/{self.redis_db}"
+
+    class Config:
+        env_file = ".env"
+        case_sensitive = False
+
+
+@lru_cache
+def get_settings() -> Settings:
+    """
+    Get cached settings instance.
+    Using lru_cache ensures we only create one Settings instance.
+    """
+    return Settings()

src/executors/__init__.py

@@ -0,0 +1,23 @@
+"""
+Executor modules for The Scheduler.
+
+Each executor must implement an async execute(config, settings) function.
+
+Example:
+    async def execute(config: dict, settings: Settings) -> str:
+        '''
+        Perform the task.
+
+        Args:
+            config: Task-specific configuration from scheduled_tasks.config
+            settings: Global scheduler settings
+
+        Returns:
+            Output message (success)
+
+        Raises:
+            Exception: On failure (will be logged as error)
+        '''
+        # Your task logic here
+        return "Task completed successfully"
+"""

src/executors/config_backup_executor.py

@@ -0,0 +1,159 @@
+"""
+Config Backup Executor
+Backs up Docker container configs and host-based service configs.
+Replicates functionality of maintenance container's backup-configs.sh
+"""
+import asyncio
+import logging
+import tarfile
+import tempfile
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import List, Dict, Any
+
+from src.config import Settings
+
+logger = logging.getLogger(__name__)
+
+
+async def execute(config: dict, settings: Settings) -> str:
+    """
+    Execute config backup task.
+
+    Config schema:
+        {
+            "sources": [
+                {
+                    "path": "/data/docker-data",
+                    "name": "docker-data",
+                    "excludes": ["*/cache/*", "*/temp/*", "*.log"]
+                }
+            ],
+            "backup_dir": "/backups/docker-configs",
+            "retention_days": 30,
+            "compress": true
+        }
+
+    Args:
+        config: Backup configuration
+        settings: Global scheduler settings
+
+    Returns:
+        Summary of backup operation
+
+    Raises:
+        Exception: On backup failure
+    """
+    sources = config.get('sources', [])
+    backup_dir = Path(config.get('backup_dir', '/backups/docker-configs'))
+    retention_days = config.get('retention_days', 30)
+    compress = config.get('compress', True)
+
+    if not sources:
+        raise ValueError("No backup sources configured")
+
+    timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')
+    backup_filename = f"docker-configs-{timestamp}.tar.gz"
+    backup_file = backup_dir / backup_filename
+
+    logger.info(f"Starting Docker configs backup: {backup_filename}")
+
+    # Create backup directory
+    backup_dir.mkdir(parents=True, exist_ok=True)
+
+    # Create temporary directory for staging
+    with tempfile.TemporaryDirectory(prefix='backup-') as temp_dir:
+        temp_path = Path(temp_dir)
+        results = []
+
+        # Backup each source
+        for source in sources:
+            source_path = Path(source['path'])
+            source_name = source['name']
+            excludes = source.get('excludes', [])
+
+            if not source_path.exists():
+                logger.warning(f"Source path does not exist: {source_path}")
+                continue
+
+            logger.info(f"Backing up {source_name} from {source_path}")
+
+            # Create tar for this source
+            source_tar = temp_path / f"{source_name}.tar.gz"
+
+            def tar_filter(tarinfo):
+                """Filter function to exclude patterns."""
+                for pattern in excludes:
+                    # Simple pattern matching (could be enhanced with fnmatch)
+                    if pattern.replace('*/', '').replace('/*', '') in tarinfo.name:
+                        logger.debug(f"Excluding: {tarinfo.name}")
+                        return None
+                return tarinfo
+
+            with tarfile.open(source_tar, 'w:gz') as tar:
+                tar.add(
+                    source_path,
+                    arcname=source_name,
+                    filter=tar_filter,
+                    recursive=True
+                )
+
+            source_size = source_tar.stat().st_size / (1024 * 1024)  # MB
+            results.append(f"{source_name}: {source_size:.2f}MB")
+            logger.info(f"Backed up {source_name}: {source_size:.2f}MB")
+
+        # Combine all source backups into final archive
+        logger.info("Creating combined backup archive...")
+        with tarfile.open(backup_file, 'w:gz') as final_tar:
+            for item in temp_path.glob('*.tar.gz'):
+                final_tar.add(item, arcname=item.name)
+
+    # Verify backup created
+    if not backup_file.exists():
+        raise Exception("Backup file was not created")
+
+    backup_size = backup_file.stat().st_size / (1024 * 1024)  # MB
+    logger.info(f"Backup created successfully: {backup_size:.2f}MB")
+
+    # Clean up old backups
+    await cleanup_old_backups(backup_dir, retention_days)
+
+    # Count remaining backups
+    backup_count = len(list(backup_dir.glob('docker-configs-*.tar.gz')))
+    total_size = sum(f.stat().st_size for f in backup_dir.glob('docker-configs-*.tar.gz'))
+    total_size_mb = total_size / (1024 * 1024)
+
+    output = (
+        f"Backup completed: {backup_filename} ({backup_size:.2f}MB). "
+        f"Sources: {', '.join(results)}. "
+        f"Retention: {backup_count} backups, {total_size_mb:.2f}MB total."
+    )
+
+    logger.info(output)
+    return output
+
+
+async def cleanup_old_backups(backup_dir: Path, retention_days: int):
+    """Remove backups older than retention period."""
+    cutoff_date = datetime.now() - timedelta(days=retention_days)
+    removed_count = 0
+    removed_size = 0
+
+    logger.info(f"Cleaning up backups older than {retention_days} days...")
+
+    for backup_file in backup_dir.glob('docker-configs-*.tar.gz'):
+        # Get file modification time
+        file_mtime = datetime.fromtimestamp(backup_file.stat().st_mtime)
+
+        if file_mtime < cutoff_date:
+            file_size = backup_file.stat().st_size
+            logger.info(f"Removing old backup: {backup_file.name} (from {file_mtime:%Y-%m-%d})")
+            backup_file.unlink()
+            removed_count += 1
+            removed_size += file_size
+
+    if removed_count > 0:
+        removed_size_mb = removed_size / (1024 * 1024)
+        logger.info(f"Removed {removed_count} old backups, freed {removed_size_mb:.2f}MB")
+    else:
+        logger.info("No old backups to remove")

src/executors/doc_sync_executor.py

@@ -0,0 +1,259 @@
+"""
+Documentation Sync Executor
+Syncs documentation from upstream Git repositories to Gitea.
+Clones source repos, extracts docs directories, pushes to Gitea.
+"""
+import asyncio
+import logging
+import shutil
+from datetime import datetime
+from pathlib import Path
+from typing import List, Optional
+
+from src.config import Settings
+
+logger = logging.getLogger(__name__)
+
+
+async def execute(config: dict, settings: Settings) -> str:
+    """
+    Sync documentation from upstream Git repo to Gitea.
+
+    Config schema:
+        {
+            "project": "fastapi",
+            "upstream_repo": "https://github.com/tiangolo/fastapi.git",
+            "docs_paths": ["/docs", "/docs_src"],
+            "gitea_repo": "library/docs-fastapi",
+            "gitea_url": "http://gitea:3000",
+            "branch": "main"
+        }
+
+    Args:
+        config: Documentation sync configuration
+        settings: Global scheduler settings
+
+    Returns:
+        Summary of sync operation
+
+    Raises:
+        Exception: On sync failure
+    """
+    project = config.get('project')
+    upstream_repo = config.get('upstream_repo')
+    docs_paths = config.get('docs_paths', ['/docs'])
+    gitea_repo = config.get('gitea_repo')
+    gitea_url = config.get('gitea_url', settings.gitea_url)
+    branch = config.get('branch', 'main')
+
+    if not all([project, upstream_repo, gitea_repo]):
+        raise ValueError("Missing required config: project, upstream_repo, or gitea_repo")
+
+    logger.info(f"Starting doc sync for {project}")
+    logger.info(f"Upstream: {upstream_repo}")
+    logger.info(f"Gitea: {gitea_repo}")
+
+    # Work directory
+    work_dir = Path(f"/app/task-data/doc-sync/{project}")
+    upstream_dir = work_dir / "upstream"
+    gitea_dir = work_dir / "gitea"
+
+    try:
+        # Clean work directory
+        if work_dir.exists():
+            logger.info(f"Cleaning work directory: {work_dir}")
+            shutil.rmtree(work_dir)
+        work_dir.mkdir(parents=True)
+
+        # Clone upstream repo (shallow clone for speed)
+        logger.info(f"Cloning upstream repo...")
+        await _run_command([
+            'git', 'clone',
+            '--depth', '1',
+            '--branch', branch,
+            upstream_repo,
+            str(upstream_dir)
+        ])
+
+        # Get upstream version/commit
+        upstream_commit = await _get_git_commit(upstream_dir)
+        upstream_date = datetime.now().strftime('%Y-%m-%d')
+        logger.info(f"Upstream commit: {upstream_commit[:8]}")
+
+        # Clone Gitea repo (or create if doesn't exist)
+        # Build authenticated URL for Gitea
+        gitea_url_clean = gitea_url.replace('http://', '').replace('https://', '')
+        gitea_clone_url = f"http://{settings.gitea_user}:{settings.gitea_password}@{gitea_url_clean}/{gitea_repo}.git"
+        # Log without credentials
+        logger.info(f"Cloning Gitea repo: {gitea_url}/{gitea_repo}.git")
+
+        # Try to clone, if fails create new repo
+        try:
+            await _run_command([
+                'git', 'clone',
+                gitea_clone_url,
+                str(gitea_dir)
+            ])
+        except Exception as e:
+            logger.warning(f"Gitea repo doesn't exist, will create: {e}")
+            gitea_dir.mkdir(parents=True)
+            await _run_command(['git', 'init'], cwd=gitea_dir)
+            await _run_command(['git', 'checkout', '-b', branch], cwd=gitea_dir)
+
+            # Set remote
+            await _run_command([
+                'git', 'remote', 'add', 'origin',
+                gitea_clone_url
+            ], cwd=gitea_dir)
+
+        # Clear existing content in Gitea repo (except .git)
+        for item in gitea_dir.iterdir():
+            if item.name != '.git':
+                if item.is_dir():
+                    shutil.rmtree(item)
+                else:
+                    item.unlink()
+
+        # Determine what to copy based on docs_paths
+        copied_paths = []
+
+        # If docs_paths is empty or contains "." or "/", sync entire repo
+        if not docs_paths or any(p in [".", "/", ""] for p in docs_paths):
+            logger.info(f"Copying entire repository...")
+            for item in upstream_dir.iterdir():
+                if item.name != '.git':
+                    dest = gitea_dir / item.name
+                    if item.is_dir():
+                        shutil.copytree(item, dest)
+                    else:
+                        shutil.copy2(item, dest)
+            copied_paths = ["entire repository"]
+        else:
+            # Copy only specified paths
+            logger.info(f"Copying specific paths: {docs_paths}")
+            for doc_path in docs_paths:
+                source = upstream_dir / doc_path.lstrip('/')
+                if source.exists():
+                    dest = gitea_dir / source.name
+                    logger.info(f"Copying {source.name}...")
+                    if source.is_dir():
+                        shutil.copytree(source, dest)
+                    else:
+                        shutil.copy2(source, dest)
+                    copied_paths.append(source.name)
+                else:
+                    logger.warning(f"Path not found in upstream: {doc_path}")
+
+            if not copied_paths:
+                raise Exception("No documentation paths were copied")
+
+        # Create .SYNC_INFO.md with metadata (don't overwrite README.md from upstream)
+        sync_info_path = gitea_dir / ".SYNC_INFO.md"
+        content_desc = "Complete repository mirror" if "entire repository" in copied_paths else f"Paths: {', '.join(copied_paths)}"
+        sync_info_content = f"""# Sync Information
+
+This is a mirror of the {project.title()} repository.
+
+**Synced from:** {upstream_repo}
+**Branch:** {branch}
+**Commit:** {upstream_commit}
+**Sync Date:** {upstream_date}
+**Content:** {content_desc}
+
+---
+
+This repository is automatically synced monthly by The Scheduler.
+For the latest updates, visit the official repository.
+"""
+        sync_info_path.write_text(sync_info_content)
+
+        # Git add, commit, push
+        await _run_command(['git', 'add', '.'], cwd=gitea_dir)
+
+        # Check if there are changes
+        status = await _run_command(
+            ['git', 'status', '--porcelain'],
+            cwd=gitea_dir,
+            capture=True
+        )
+
+        if not status.strip():
+            logger.info("No changes detected, skipping commit")
+            return f"Documentation already up to date (commit: {upstream_commit[:8]})"
+
+        # Commit changes
+        commit_msg = f"Sync {project} docs from {upstream_commit[:8]} on {upstream_date}"
+        await _run_command([
+            'git', 'commit',
+            '-m', commit_msg
+        ], cwd=gitea_dir)
+
+        # Tag with date
+        tag = f"sync-{upstream_date}"
+        await _run_command([
+            'git', 'tag', '-f', tag,
+            '-m', f"Documentation snapshot {upstream_date}"
+        ], cwd=gitea_dir)
+
+        # Push to Gitea
+        logger.info("Pushing to Gitea...")
+        await _run_command([
+            'git', 'push', 'origin', branch, '--force'
+        ], cwd=gitea_dir)
+
+        await _run_command([
+            'git', 'push', 'origin', tag, '--force'
+        ], cwd=gitea_dir)
+
+        # Cleanup
+        logger.info("Cleaning up work directory...")
+        shutil.rmtree(work_dir)
+
+        result = (
+            f"Successfully synced {project} documentation. "
+            f"Copied: {', '.join(copied_paths)}. "
+            f"Upstream commit: {upstream_commit[:8]}. "
+            f"Tagged: {tag}"
+        )
+        logger.info(result)
+        return result
+
+    except Exception as e:
+        # Cleanup on error
+        if work_dir.exists():
+            shutil.rmtree(work_dir)
+        raise
+
+
+async def _run_command(
+    cmd: List[str],
+    cwd: Optional[Path] = None,
+    capture: bool = False
+) -> str:
+    """Run shell command asynchronously."""
+    logger.debug(f"Running: {' '.join(cmd)} (cwd: {cwd})")
+
+    proc = await asyncio.create_subprocess_exec(
+        *cmd,
+        cwd=cwd,
+        stdout=asyncio.subprocess.PIPE if capture else asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE
+    )
+
+    stdout, stderr = await proc.communicate()
+
+    if proc.returncode != 0:
+        error_msg = stderr.decode() if stderr else "Unknown error"
+        raise Exception(f"Command failed: {' '.join(cmd)}\n{error_msg}")
+
+    return stdout.decode() if capture else ""
+
+
+async def _get_git_commit(repo_dir: Path) -> str:
+    """Get current git commit hash."""
+    output = await _run_command(
+        ['git', 'rev-parse', 'HEAD'],
+        cwd=repo_dir,
+        capture=True
+    )
+    return output.strip()

src/executors/example_executor.py

@@ -0,0 +1,42 @@
+"""
+Example executor demonstrating the pattern.
+Shows how to write task executors for The Scheduler.
+"""
+import asyncio
+import logging
+from src.config import Settings
+
+logger = logging.getLogger(__name__)
+
+
+async def execute(config: dict, settings: Settings) -> str:
+    """
+    Example task executor.
+
+    Args:
+        config: Task configuration from scheduled_tasks.config JSONB field
+                Example: {"message": "Hello", "delay_seconds": 2}
+        settings: Global scheduler settings (database, API keys, etc.)
+
+    Returns:
+        Output message describing what was done
+
+    Raises:
+        Exception: On failure (will be caught and logged by executor framework)
+    """
+    message = config.get('message', 'No message configured')
+    delay = config.get('delay_seconds', 1)
+
+    logger.info(f"Example executor starting: {message}")
+
+    # Simulate some work
+    await asyncio.sleep(delay)
+
+    # You can access settings
+    logger.info(f"Using database: {settings.postgres_db}")
+
+    # Return success message
+    output = f"Executed example task: {message} (took {delay}s)"
+    logger.info(output)
+
+    return output

src/executors/rest_api_executor.py

@@ -0,0 +1,266 @@
+"""
+Generic REST API Executor
+
+Universal executor for calling any REST API endpoint across the system.
+Supports GET, POST, PUT, DELETE with configurable payloads, headers, and authentication.
+
+This executor can be used to trigger any service endpoint:
+- Library Desk knowledge consolidation
+- Core API operations
+- External webhooks
+- Any HTTP-based task
+
+Config schema:
+    {
+        "url": "http://service:port/endpoint",
+        "method": "POST",  # GET, POST, PUT, DELETE, PATCH
+        "payload": {...},  # Request body (for POST/PUT/PATCH)
+        "headers": {...},  # Additional headers
+        "auth": {
+            "type": "bearer",  # bearer, basic, api_key
+            "token": "${ENV_VAR}",  # Use ${VAR} for env vars
+            "header": "Authorization"  # Optional: header name for API key
+        },
+        "timeout": 300,  # Timeout in seconds (default: 300)
+        "verify_ssl": true,  # SSL verification (default: true)
+        "success_codes": [200, 201, 202],  # Expected success codes
+        "response_path": "result.message"  # JSONPath to extract from response
+    }
+
+Example configs:
+
+1. Library Desk Knowledge Consolidation:
+   {
+       "url": "http://library-desk:8089/consolidate/knowledge",
+       "method": "POST",
+       "payload": {"process_limit": 10, "lookback_days": 7, "dry_run": false},
+       "auth": {"type": "bearer", "token": "${LIBRARY_DESK_API_KEY}"}
+   }
+
+2. Core API Container Restart:
+   {
+       "url": "http://core-api:8088/v1/infrastructure/containers/nginx/restart",
+       "method": "POST",
+       "auth": {"type": "bearer", "token": "${CORE_API_KEY}"}
+   }
+
+3. External Webhook:
+   {
+       "url": "https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
+       "method": "POST",
+       "payload": {"text": "Scheduled task completed"},
+       "verify_ssl": true
+   }
+"""
+import logging
+import os
+import re
+import httpx
+from typing import Any, Dict, Optional
+
+from src.config import Settings
+
+logger = logging.getLogger(__name__)
+
+
+async def execute(config: dict, settings: Settings) -> str:
+    """
+    Execute REST API call with configured parameters.
+
+    Args:
+        config: REST API call configuration (see module docstring)
+        settings: Global scheduler settings
+
+    Returns:
+        Response summary or extracted result
+
+    Raises:
+        ValueError: On configuration error
+        Exception: On API call failure
+    """
+    # Required configuration
+    url = config.get('url')
+    if not url:
+        raise ValueError("Missing required config: 'url'")
+
+    method = config.get('method', 'POST').upper()
+    if method not in ['GET', 'POST', 'PUT', 'DELETE', 'PATCH']:
+        raise ValueError(f"Invalid HTTP method: {method}")
+
+    # Optional configuration
+    payload = config.get('payload', {})
+    headers = config.get('headers', {})
+    timeout = config.get('timeout', 300)
+    verify_ssl = config.get('verify_ssl', True)
+    success_codes = config.get('success_codes', [200, 201, 202, 204])
+    response_path = config.get('response_path')
+
+    # Handle authentication
+    auth_config = config.get('auth', {})
+    if auth_config:
+        auth_header = _build_auth_header(auth_config, settings)
+        if auth_header:
+            headers.update(auth_header)
+
+    # Substitute environment variables in URL and payload
+    url = _substitute_env_vars(url)
+    payload = _substitute_env_vars_recursive(payload)
+
+    logger.info(f"Executing REST API call: {method} {url}")
+    if payload:
+        logger.debug(f"Payload: {_redact_sensitive(payload)}")
+
+    # Make HTTP request
+    try:
+        async with httpx.AsyncClient(timeout=timeout, verify=verify_ssl) as client:
+            if method == 'GET':
+                response = await client.get(url, headers=headers)
+            elif method == 'POST':
+                response = await client.post(url, json=payload, headers=headers)
+            elif method == 'PUT':
+                response = await client.put(url, json=payload, headers=headers)
+            elif method == 'DELETE':
+                response = await client.delete(url, headers=headers)
+            elif method == 'PATCH':
+                response = await client.patch(url, json=payload, headers=headers)
+
+            # Check status code
+            if response.status_code not in success_codes:
+                error_msg = (
+                    f"API call failed with status {response.status_code}: "
+                    f"{response.text[:500]}"
+                )
+                logger.error(error_msg)
+                raise Exception(error_msg)
+
+            # Parse response
+            try:
+                response_data = response.json()
+            except:
+                response_data = {"text": response.text}
+
+            # Extract specific field if response_path provided
+            result_text = None
+            if response_path and isinstance(response_data, dict):
+                result_text = _extract_json_path(response_data, response_path)
+
+            if not result_text:
+                # Build summary from response
+                if isinstance(response_data, dict):
+                    # Look for common result fields
+                    result_text = (
+                        response_data.get('message') or
+                        response_data.get('result') or
+                        response_data.get('summary') or
+                        f"Success ({response.status_code})"
+                    )
+                else:
+                    result_text = f"Success ({response.status_code})"
+
+            logger.info(f"API call succeeded: {result_text}")
+            return str(result_text)
+
+    except httpx.HTTPStatusError as e:
+        error_msg = f"HTTP {e.response.status_code}: {e.response.text[:500]}"
+        logger.error(error_msg)
+        raise Exception(error_msg)
+    except httpx.RequestError as e:
+        error_msg = f"Request failed: {str(e)}"
+        logger.error(error_msg)
+        raise Exception(error_msg)
+    except Exception as e:
+        logger.error(f"REST API call failed: {e}", exc_info=True)
+        raise
+
+
+def _build_auth_header(auth_config: dict, settings: Settings) -> Optional[Dict[str, str]]:
+    """Build authentication header from config."""
+    auth_type = auth_config.get('type', '').lower()
+
+    if auth_type == 'bearer':
+        token = auth_config.get('token', '')
+        token = _substitute_env_vars(token)
+        if token:
+            return {"Authorization": f"Bearer {token}"}
+
+    elif auth_type == 'basic':
+        username = _substitute_env_vars(auth_config.get('username', ''))
+        password = _substitute_env_vars(auth_config.get('password', ''))
+        if username and password:
+            import base64
+            credentials = base64.b64encode(f"{username}:{password}".encode()).decode()
+            return {"Authorization": f"Basic {credentials}"}
+
+    elif auth_type == 'api_key':
+        key = _substitute_env_vars(auth_config.get('key', ''))
+        header_name = auth_config.get('header', 'X-API-Key')
+        if key:
+            return {header_name: key}
+
+    return None
+
+
+def _substitute_env_vars(text: str) -> str:
+    """Substitute ${ENV_VAR} placeholders with environment variables."""
+    if not isinstance(text, str):
+        return text
+
+    # Find all ${VAR} patterns
+    pattern = r'\$\{([A-Z_][A-Z0-9_]*)\}'
+    matches = re.findall(pattern, text)
+
+    for var_name in matches:
+        env_value = os.getenv(var_name, '')
+        if not env_value:
+            logger.warning(f"Environment variable not found: {var_name}")
+        text = text.replace(f"${{{var_name}}}", env_value)
+
+    return text
+
+
+def _substitute_env_vars_recursive(data: Any) -> Any:
+    """Recursively substitute environment variables in nested structures."""
+    if isinstance(data, dict):
+        return {k: _substitute_env_vars_recursive(v) for k, v in data.items()}
+    elif isinstance(data, list):
+        return [_substitute_env_vars_recursive(item) for item in data]
+    elif isinstance(data, str):
+        return _substitute_env_vars(data)
+    else:
+        return data
+
+
+def _extract_json_path(data: dict, path: str) -> Optional[str]:
+    """
+    Extract value from nested dict using dot notation.
+
+    Example: "result.message" -> data["result"]["message"]
+    """
+    try:
+        keys = path.split('.')
+        value = data
+        for key in keys:
+            if isinstance(value, dict):
+                value = value.get(key)
+            else:
+                return None
+        return str(value) if value is not None else None
+    except:
+        return None
+
+
+def _redact_sensitive(data: Any) -> Any:
+    """Redact sensitive fields from logs."""
+    if isinstance(data, dict):
+        redacted = {}
+        sensitive_keys = ['password', 'token', 'api_key', 'secret', 'auth']
+        for k, v in data.items():
+            if any(s in k.lower() for s in sensitive_keys):
+                redacted[k] = '***REDACTED***'
+            else:
+                redacted[k] = _redact_sensitive(v)
+        return redacted
+    elif isinstance(data, list):
+        return [_redact_sensitive(item) for item in data]
+    else:
+        return data

src/main.py

@@ -0,0 +1,530 @@
+"""
+The Scheduler - System-wide maintenance orchestration.
+Handles backups, documentation mirroring, cleanup, and automated tasks.
+
+Architecture: Hybrid APScheduler + DB-based priority system
+- APScheduler runs a single job every minute
+- Job queries DB for tasks scheduled in that minute
+- Executes up to 5 tasks concurrently based on priority
+"""
+from fastapi import FastAPI, HTTPException, Depends, Header
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
+from apscheduler.triggers.cron import CronTrigger
+from contextlib import asynccontextmanager
+import logging
+
+from src.config import get_settings, Settings
+from src.tasks.executor import TaskExecutor
+from src.models import TaskCreate, TaskUpdate, TaskResponse
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+# Global instances
+scheduler: AsyncIOScheduler | None = None
+task_executor: TaskExecutor | None = None
+
+
+def get_scheduler() -> AsyncIOScheduler:
+    """Dependency to get scheduler instance."""
+    if scheduler is None:
+        raise HTTPException(500, "Scheduler not initialized")
+    return scheduler
+
+
+def get_task_executor() -> TaskExecutor:
+    """Dependency to get task executor instance."""
+    if task_executor is None:
+        raise HTTPException(500, "Task executor not initialized")
+    return task_executor
+
+# Lifespan manager for startup/shutdown
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Manage application lifecycle - startup and shutdown."""
+    global scheduler, task_executor
+
+    settings = get_settings()
+
+    # Startup
+    logger.info("=" * 60)
+    logger.info("The Scheduler - System-wide Maintenance Orchestration")
+    logger.info("=" * 60)
+    logger.info(f"Architecture: Hybrid APScheduler + DB-based priority system")
+    logger.info(f"Database: {settings.postgres_host}:{settings.postgres_port}/{settings.postgres_db}")
+    logger.info(f"API: http://{settings.host}:{settings.port}")
+    logger.info(f"Docs: http://{settings.host}:{settings.port}/docs")
+    logger.info("=" * 60)
+
+    # Initialize task executor
+    task_executor = TaskExecutor(settings)
+    logger.info("Task executor initialized (max 5 concurrent tasks)")
+
+    # Initialize APScheduler with minimal configuration
+    # No jobstore needed - we only have one in-memory job
+    scheduler = AsyncIOScheduler(
+        job_defaults={
+            'coalesce': True,        # Combine missed runs
+            'max_instances': 1,      # Only one instance running
+            'misfire_grace_time': 30  # 30s grace period for minute-based execution
+        }
+    )
+
+    # Add the single minute-based task processor
+    scheduler.add_job(
+        func=task_executor.process_minute,
+        trigger=CronTrigger(minute='*'),  # Run every minute
+        id='process_tasks',
+        name='Process scheduled tasks',
+        replace_existing=True
+    )
+
+    try:
+        scheduler.start()
+        logger.info("Scheduler started - processing tasks every minute")
+        logger.info("Priority system: 1-5 (emergency/system), 10-30 (user), 40-70+ (maintenance)")
+    except Exception as e:
+        logger.error(f"Failed to start scheduler: {e}")
+        raise
+
+    yield
+
+    # Shutdown
+    logger.info("Stopping The Scheduler...")
+    if scheduler:
+        scheduler.shutdown(wait=True)
+    logger.info("Scheduler stopped")
+
+
+# FastAPI app
+app = FastAPI(
+    title="The Scheduler",
+    version="1.0.0",
+    description="System-wide maintenance orchestration - backups, doc mirroring, cleanup, task automation",
+    lifespan=lifespan
+)
+
+# Dependencies
+async def verify_api_key(
+    authorization: str = Header(None),
+    settings: Settings = Depends(get_settings)
+):
+    """Verify API key from Authorization header."""
+    if not authorization or not authorization.startswith("Bearer "):
+        raise HTTPException(401, "Missing API key")
+    key = authorization.replace("Bearer ", "")
+    if key != settings.scheduler_api_key:
+        raise HTTPException(403, "Invalid API key")
+    return key
+
+# ============================================================================
+# Public Endpoints (no auth required)
+# ============================================================================
+
+@app.get("/health")
+async def health(
+    settings: Settings = Depends(get_settings),
+    sched: AsyncIOScheduler = Depends(get_scheduler)
+):
+    """Health check endpoint."""
+    return {
+        "status": "healthy",
+        "scheduler_running": sched.running,
+        "jobs_count": len(sched.get_jobs()),
+        "database": settings.postgres_db
+    }
+
+# ============================================================================
+# Protected Endpoints (require API key)
+# ============================================================================
+
+@app.get("/tasks")
+async def list_tasks(
+    enabled: bool = None,
+    service: str = None,
+    api_key: str = Depends(verify_api_key),
+    executor: TaskExecutor = Depends(get_task_executor)
+):
+    """List all scheduled tasks from database."""
+    import psycopg2.extras
+
+    query = "SELECT * FROM scheduled_tasks WHERE 1=1"
+    params = []
+
+    if enabled is not None:
+        query += " AND enabled = %s"
+        params.append(enabled)
+
+    if service:
+        query += " AND service = %s"
+        params.append(service)
+
+    query += " ORDER BY priority ASC, task_name ASC"
+
+    with executor.get_db_connection() as conn:
+        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            cur.execute(query, params)
+            tasks = [dict(task) for task in cur.fetchall()]
+
+    return {
+        "tasks": tasks,
+        "count": len(tasks)
+    }
+
+@app.get("/tasks/{task_name}")
+async def get_task_details(
+    task_name: str,
+    api_key: str = Depends(verify_api_key),
+    executor: TaskExecutor = Depends(get_task_executor)
+):
+    """Get details for a specific task."""
+    import psycopg2.extras
+
+    with executor.get_db_connection() as conn:
+        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            cur.execute("SELECT * FROM scheduled_tasks WHERE task_name = %s", (task_name,))
+            task = cur.fetchone()
+
+    if not task:
+        raise HTTPException(404, f"Task '{task_name}' not found")
+
+    return dict(task)
+
+@app.post("/tasks", response_model=TaskResponse, tags=["Task Management"])
+async def create_task(
+    task: TaskCreate,
+    api_key: str = Depends(verify_api_key),
+    executor: TaskExecutor = Depends(get_task_executor)
+):
+    """
+    Create a new scheduled task.
+
+    ## Schedule Pattern
+
+    Use cron-style fields where `-1` means "every":
+    - `minute: -1, hour: -1` → Runs every minute
+    - `minute: 0, hour: -1` → Runs at minute 0 of every hour (hourly)
+    - `minute: 0, hour: 2` → Runs at 2:00 AM every day
+    - `minute: 0, hour: 2, day_of_week: 0` → Runs at 2:00 AM every Monday
+
+    ## Priority Levels
+
+    - **1-5**: Emergency/critical system tasks
+    - **10-30**: User-initiated tasks
+    - **40-70**: Background maintenance
+    - **70+**: Low priority cleanup
+
+    ## Executor Types
+
+    ### `rest_api` Executor
+
+    Calls HTTP endpoints. Config format:
+    ```json
+    {
+      "method": "POST",
+      "url": "http://service:port/endpoint",
+      "headers": {
+        "Authorization": "Bearer ${ENV_VAR}",
+        "Content-Type": "application/json"
+      },
+      "body": {"key": "value"}
+    }
+    ```
+
+    Environment variables can be referenced with `${VAR_NAME}` syntax.
+
+    ## Example: Librarian Consolidation Task
+
+    ```json
+    {
+      "task_name": "librarian_consolidation",
+      "service": "library-desk",
+      "executor": "rest_api",
+      "priority": 25,
+      "description": "Processes search queries and consolidates knowledge",
+      "minute": 0,
+      "hour": -1,
+      "config": {
+        "method": "POST",
+        "url": "http://library-desk:8089/consolidation/run",
+        "headers": {
+          "Authorization": "Bearer ${LIBRARY_DESK_API_KEY}",
+          "Content-Type": "application/json"
+        },
+        "body": {
+          "dry_run": false,
+          "process_limit": 10
+        }
+      }
+    }
+    ```
+
+    This creates an hourly task that calls the consolidation endpoint.
+    """
+    import psycopg2.extras
+    import json
+
+    # Convert task model to dict
+    task_data = task.model_dump()
+
+    # Convert config dict to JSON string if present
+    if 'config' in task_data and isinstance(task_data['config'], dict):
+        task_data['config'] = json.dumps(task_data['config'])
+
+    with executor.get_db_connection() as conn:
+        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            cur.execute("""
+                INSERT INTO scheduled_tasks
+                    (task_name, service, executor, priority, minute, hour,
+                     day_of_month, month, day_of_week, enabled, description,
+                     config, max_retries, timeout_seconds, created_by)
+                VALUES
+                    (%(task_name)s, %(service)s, %(executor)s, %(priority)s,
+                     %(minute)s, %(hour)s, %(day_of_month)s, %(month)s,
+                     %(day_of_week)s, %(enabled)s, %(description)s,
+                     %(config)s::jsonb, %(max_retries)s, %(timeout_seconds)s,
+                     %(created_by)s)
+                RETURNING *
+            """, task_data)
+            new_task = dict(cur.fetchone())
+            conn.commit()
+
+    logger.info(f"Created task: {new_task['task_name']}")
+    return new_task
+
+@app.put("/tasks/{task_name}")
+async def update_task(
+    task_name: str,
+    task_data: dict,
+    api_key: str = Depends(verify_api_key),
+    executor: TaskExecutor = Depends(get_task_executor)
+):
+    """Update an existing scheduled task."""
+    import psycopg2.extras
+    import json
+
+    # Build update query dynamically
+    allowed_fields = ['service', 'executor', 'priority', 'minute', 'hour',
+                     'day_of_month', 'month', 'day_of_week', 'enabled',
+                     'description', 'config', 'max_retries', 'timeout_seconds']
+
+    updates = {k: v for k, v in task_data.items() if k in allowed_fields}
+    if not updates:
+        raise HTTPException(400, "No valid fields to update")
+
+    # Convert config dict to JSON string if present
+    if 'config' in updates and isinstance(updates['config'], dict):
+        updates['config'] = json.dumps(updates['config'])
+
+    set_clause = ', '.join([f"{k} = %({k})s" for k in updates.keys()])
+
+    with executor.get_db_connection() as conn:
+        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            cur.execute(f"""
+                UPDATE scheduled_tasks
+                SET {set_clause}, updated_at = NOW()
+                WHERE task_name = %(task_name)s
+                RETURNING *
+            """, {**updates, 'task_name': task_name})
+
+            updated_task = cur.fetchone()
+            if not updated_task:
+                raise HTTPException(404, f"Task '{task_name}' not found")
+
+            conn.commit()
+
+    logger.info(f"Updated task: {task_name}")
+    return dict(updated_task)
+
+@app.delete("/tasks/{task_name}")
+async def delete_task(
+    task_name: str,
+    api_key: str = Depends(verify_api_key),
+    executor: TaskExecutor = Depends(get_task_executor)
+):
+    """Delete a scheduled task."""
+    with executor.get_db_connection() as conn:
+        with conn.cursor() as cur:
+            cur.execute("""
+                DELETE FROM scheduled_tasks
+                WHERE task_name = %s
+                RETURNING task_name
+            """, (task_name,))
+
+            deleted = cur.fetchone()
+            if not deleted:
+                raise HTTPException(404, f"Task '{task_name}' not found")
+
+            conn.commit()
+
+    logger.info(f"Deleted task: {task_name}")
+    return {"message": f"Task '{task_name}' deleted successfully"}
+
+@app.post("/tasks/{task_name}/trigger")
+async def trigger_task(
+    task_name: str,
+    api_key: str = Depends(verify_api_key),
+    executor: TaskExecutor = Depends(get_task_executor)
+):
+    """Manually trigger a task to run immediately."""
+    import psycopg2.extras
+
+    # Get task details
+    with executor.get_db_connection() as conn:
+        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            cur.execute("SELECT * FROM scheduled_tasks WHERE task_name = %s", (task_name,))
+            task = cur.fetchone()
+
+    if not task:
+        raise HTTPException(404, f"Task '{task_name}' not found")
+
+    if not task['enabled']:
+        raise HTTPException(400, f"Task '{task_name}' is disabled")
+
+    # Execute task immediately in background
+    import asyncio
+    asyncio.create_task(executor.execute_task(dict(task)))
+
+    logger.info(f"Manually triggered task: {task_name}")
+    return {
+        "message": f"Task '{task_name}' triggered successfully",
+        "task_name": task_name,
+        "priority": task['priority'],
+        "executor": task['executor']
+    }
+
+# Legacy endpoints (deprecated)
+@app.post("/tasks/backup")
+async def trigger_backup(api_key: str = Depends(verify_api_key)):
+    """Trigger backup tasks manually (deprecated - use POST /tasks/{name}/trigger)"""
+    # TODO: Implement backup executor
+    logger.info("Manual backup triggered")
+    return {
+        "message": "Backup task triggered",
+        "status": "not_implemented",
+        "note": "Backup executor needs to be implemented"
+    }
+
+@app.post("/tasks/docs/update")
+async def trigger_docs_update(
+    project: str = None,
+    api_key: str = Depends(verify_api_key)
+):
+    """Trigger documentation mirror update"""
+    # TODO: Implement doc mirror executor
+    logger.info(f"Doc mirror update triggered for project: {project or 'all'}")
+    return {
+        "message": f"Documentation update triggered for {project or 'all projects'}",
+        "status": "not_implemented",
+        "note": "Doc mirror executor needs to be implemented"
+    }
+
+@app.post("/tasks/docs/check-versions")
+async def check_doc_versions(api_key: str = Depends(verify_api_key)):
+    """Check for new documentation versions"""
+    # TODO: Implement version check executor
+    logger.info("Version check triggered")
+    return {
+        "message": "Version check triggered",
+        "status": "not_implemented",
+        "note": "Version check executor needs to be implemented"
+    }
+
+@app.post("/tasks/cleanup")
+async def trigger_cleanup(api_key: str = Depends(verify_api_key)):
+    """Run cleanup tasks"""
+    # TODO: Implement cleanup executor
+    logger.info("Cleanup task triggered")
+    return {
+        "message": "Cleanup task triggered",
+        "status": "not_implemented",
+        "note": "Cleanup executor needs to be implemented"
+    }
+
+@app.get("/executions")
+async def task_history(
+    limit: int = 20,
+    task_name: str = None,
+    service: str = None,
+    status: str = None,
+    api_key: str = Depends(verify_api_key),
+    executor: TaskExecutor = Depends(get_task_executor)
+):
+    """View task execution history."""
+    import psycopg2.extras
+
+    query = "SELECT * FROM task_executions WHERE 1=1"
+    params = []
+
+    if task_name:
+        query += " AND task_name = %s"
+        params.append(task_name)
+
+    if service:
+        query += " AND service = %s"
+        params.append(service)
+
+    if status:
+        query += " AND status = %s"
+        params.append(status)
+
+    query += " ORDER BY triggered_at DESC LIMIT %s"
+    params.append(limit)
+
+    with executor.get_db_connection() as conn:
+        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            cur.execute(query, params)
+            executions = [dict(ex) for ex in cur.fetchall()]
+
+    return {
+        "executions": executions,
+        "count": len(executions),
+        "limit": limit
+    }
+
+@app.get("/stats")
+async def stats(
+    api_key: str = Depends(verify_api_key),
+    settings: Settings = Depends(get_settings),
+    sched: AsyncIOScheduler = Depends(get_scheduler),
+    task_exec: TaskExecutor = Depends(get_task_executor)
+):
+    """Get system statistics."""
+    import psycopg2.extras
+
+    # Query task stats from database
+    with task_exec.get_db_connection() as conn:
+        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            # Count enabled tasks
+            cur.execute("SELECT COUNT(*) as count FROM scheduled_tasks WHERE enabled = true")
+            enabled_tasks = cur.fetchone()['count']
+
+            # Count running tasks
+            cur.execute("SELECT COUNT(*) as count FROM task_executions WHERE status = 'running'")
+            running_tasks = cur.fetchone()['count']
+
+            # Recent execution stats (last 24 hours)
+            cur.execute("""
+                SELECT status, COUNT(*) as count
+                FROM task_executions
+                WHERE triggered_at > NOW() - INTERVAL '24 hours'
+                GROUP BY status
+            """)
+            execution_stats = {row['status']: row['count'] for row in cur.fetchall()}
+
+    return {
+        "scheduler_running": sched.running,
+        "minute_processor_active": True,  # If we got here, it's running
+        "database": settings.postgres_db,
+        "tasks_enabled": enabled_tasks,
+        "tasks_currently_running": running_tasks,
+        "concurrent_limit": 5,
+        "execution_stats_24h": execution_stats,
+        "priority_system": "1-5 (emergency/system), 10-30 (user), 40-70+ (maintenance)"
+    }
+

src/models.py

@@ -0,0 +1,210 @@
+"""
+Pydantic models for The Scheduler API.
+"""
+from pydantic import BaseModel, Field
+from typing import Optional, Dict, Any
+from enum import Enum
+
+
+class SchedulePattern(str, Enum):
+    """Common schedule patterns."""
+    EVERY_MINUTE = "every_minute"
+    HOURLY = "hourly"
+    DAILY = "daily"
+    WEEKLY = "weekly"
+    MONTHLY = "monthly"
+
+
+class TaskExecutor(str, Enum):
+    """Available task executors."""
+    REST_API = "rest_api"
+    SHELL = "shell"
+    PYTHON = "python"
+
+
+class TaskPriorityLevel(str, Enum):
+    """Task priority levels."""
+    EMERGENCY = "emergency"  # 1-5: Critical system tasks
+    SYSTEM = "system"        # 5-10: System maintenance
+    USER = "user"            # 10-30: User-initiated tasks
+    MAINTENANCE = "maintenance"  # 40-70: Background maintenance
+    LOW = "low"             # 70+: Low priority cleanup
+
+
+class TaskBase(BaseModel):
+    """Base task fields."""
+    task_name: str = Field(
+        ...,
+        description="Unique task identifier (e.g., 'librarian_consolidation')",
+        example="librarian_consolidation"
+    )
+    service: str = Field(
+        ...,
+        description="Service that owns this task (e.g., 'library-desk', 'core-api')",
+        example="library-desk"
+    )
+    executor: str = Field(
+        ...,
+        description="Executor type: 'rest_api', 'shell', or 'python'",
+        example="rest_api"
+    )
+    priority: int = Field(
+        ...,
+        ge=1,
+        le=100,
+        description="Priority level (1-5: emergency, 10-30: user, 40-70: maintenance, 70+: low)",
+        example=25
+    )
+    description: Optional[str] = Field(
+        None,
+        description="Human-readable task description",
+        example="Processes unprocessed search queries and consolidates knowledge into wiki pages"
+    )
+
+
+class TaskSchedule(BaseModel):
+    """Cron-style schedule fields."""
+    minute: int = Field(
+        -1,
+        ge=-1,
+        le=59,
+        description="Minute to run (-1 = every minute, 0-59 = specific minute)",
+        example=-1
+    )
+    hour: int = Field(
+        -1,
+        ge=-1,
+        le=23,
+        description="Hour to run (-1 = every hour, 0-23 = specific hour)",
+        example=-1
+    )
+    day_of_month: int = Field(
+        -1,
+        ge=-1,
+        le=31,
+        description="Day of month to run (-1 = every day, 1-31 = specific day)",
+        example=-1
+    )
+    month: int = Field(
+        -1,
+        ge=-1,
+        le=12,
+        description="Month to run (-1 = every month, 1-12 = specific month)",
+        example=-1
+    )
+    day_of_week: int = Field(
+        -1,
+        ge=-1,
+        le=6,
+        description="Day of week to run (-1 = every day, 0-6 = Monday-Sunday)",
+        example=-1
+    )
+
+
+class TaskConfig(BaseModel):
+    """Task execution configuration."""
+    enabled: bool = Field(
+        True,
+        description="Whether task is enabled",
+        example=True
+    )
+    max_retries: int = Field(
+        3,
+        ge=0,
+        le=10,
+        description="Maximum retry attempts on failure",
+        example=3
+    )
+    timeout_seconds: int = Field(
+        3600,
+        ge=1,
+        description="Execution timeout in seconds",
+        example=3600
+    )
+    config: Optional[Dict[str, Any]] = Field(
+        None,
+        description="Executor-specific configuration (varies by executor type)",
+        example={
+            "method": "POST",
+            "url": "http://library-desk:8089/consolidation/run",
+            "headers": {
+                "Authorization": "Bearer ${LIBRARY_DESK_API_KEY}",
+                "Content-Type": "application/json"
+            },
+            "body": {
+                "dry_run": False,
+                "process_limit": 10,
+                "lookback_days": 7,
+                "min_web_results": 2
+            }
+        }
+    )
+
+
+class TaskCreate(TaskBase, TaskSchedule, TaskConfig):
+    """Request model for creating a new scheduled task."""
+    created_by: Optional[str] = Field(
+        "api",
+        description="User or system that created this task",
+        example="api"
+    )
+
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "task_name": "librarian_consolidation",
+                "service": "library-desk",
+                "executor": "rest_api",
+                "priority": 25,
+                "description": "Processes unprocessed search queries and consolidates knowledge",
+                "minute": 0,
+                "hour": -1,
+                "day_of_month": -1,
+                "month": -1,
+                "day_of_week": -1,
+                "enabled": True,
+                "max_retries": 3,
+                "timeout_seconds": 3600,
+                "config": {
+                    "method": "POST",
+                    "url": "http://library-desk:8089/consolidation/run",
+                    "headers": {
+                        "Authorization": "Bearer ${LIBRARY_DESK_API_KEY}",
+                        "Content-Type": "application/json"
+                    },
+                    "body": {
+                        "dry_run": False,
+                        "process_limit": 10,
+                        "lookback_days": 7,
+                        "min_web_results": 2
+                    }
+                },
+                "created_by": "api"
+            }
+        }
+
+
+class TaskUpdate(BaseModel):
+    """Request model for updating a scheduled task."""
+    service: Optional[str] = None
+    executor: Optional[str] = None
+    priority: Optional[int] = Field(None, ge=1, le=100)
+    minute: Optional[int] = Field(None, ge=-1, le=59)
+    hour: Optional[int] = Field(None, ge=-1, le=23)
+    day_of_month: Optional[int] = Field(None, ge=-1, le=31)
+    month: Optional[int] = Field(None, ge=-1, le=12)
+    day_of_week: Optional[int] = Field(None, ge=-1, le=6)
+    enabled: Optional[bool] = None
+    description: Optional[str] = None
+    config: Optional[Dict[str, Any]] = None
+    max_retries: Optional[int] = Field(None, ge=0, le=10)
+    timeout_seconds: Optional[int] = Field(None, ge=1)
+
+
+class TaskResponse(TaskCreate):
+    """Response model for task operations."""
+    created_at: str
+    updated_at: Optional[str] = None
+
+    class Config:
+        from_attributes = True

src/tasks/__init__.py

@@ -0,0 +1 @@
+"""Task execution system for The Scheduler."""

src/tasks/executor.py

@@ -0,0 +1,278 @@
+"""
+Task execution engine for The Scheduler.
+Implements minute-based polling with priority-based concurrent execution.
+"""
+import asyncio
+import logging
+from datetime import datetime, timezone
+from typing import List, Dict, Any, Optional
+import psycopg2
+from psycopg2.extras import RealDictCursor
+import traceback
+
+from src.config import Settings
+
+logger = logging.getLogger(__name__)
+
+# Priority ranges (for reference)
+# 1:  Emergency/recovery tasks
+# 2:  Primary system tasks
+# 3:  Secondary system tasks
+# 5:  Urgent user-triggered tasks
+# 10: High-priority user tasks
+# 15: Reserved
+# 20: Backup tasks
+# 25: Reserved
+# 30: Low-priority user tasks
+# 40: Cleanup tasks
+# 50: Documentation version checks
+# 60: Documentation mirroring
+# 70+: Future/experimental tasks
+
+MAX_CONCURRENT_TASKS = 5
+
+
+class TaskExecutor:
+    """Executes scheduled tasks based on priority with concurrency control."""
+
+    def __init__(self, settings: Settings):
+        self.settings = settings
+        self.running_tasks: Dict[int, asyncio.Task] = {}  # task_id -> asyncio.Task
+        self.semaphore = asyncio.Semaphore(MAX_CONCURRENT_TASKS)
+
+    def get_db_connection(self):
+        """Create database connection."""
+        return psycopg2.connect(
+            host=self.settings.postgres_host,
+            port=self.settings.postgres_port,
+            database=self.settings.postgres_db,
+            user=self.settings.postgres_user,
+            password=self.settings.postgres_password
+        )
+
+    def get_tasks_for_minute(self, now: datetime) -> List[Dict[str, Any]]:
+        """
+        Query database for tasks scheduled for this minute.
+        Supports wildcards (-1 = any value).
+        """
+        minute = now.minute
+        hour = now.hour
+        day = now.day
+        month = now.month
+        # Python: Monday=0, Sunday=6; PostgreSQL: Monday=0, Sunday=6 (same)
+        weekday = now.weekday()
+
+        query = """
+            SELECT
+                id, task_name, service, executor, priority,
+                config, timeout_seconds, max_retries, retry_count,
+                last_run, last_status
+            FROM scheduled_tasks
+            WHERE enabled = true
+              AND (minute = -1 OR minute = %s)
+              AND (hour = -1 OR hour = %s)
+              AND (day_of_month = -1 OR day_of_month = %s)
+              AND (month = -1 OR month = %s)
+              AND (day_of_week = -1 OR day_of_week = %s)
+              AND id NOT IN (
+                  SELECT task_id
+                  FROM task_executions
+                  WHERE status = 'running'
+              )
+            ORDER BY priority ASC, task_name ASC
+        """
+
+        with self.get_db_connection() as conn:
+            with conn.cursor(cursor_factory=RealDictCursor) as cur:
+                cur.execute(query, (minute, hour, day, month, weekday))
+                tasks = cur.fetchall()
+
+        logger.info(f"Found {len(tasks)} tasks scheduled for {now.strftime('%Y-%m-%d %H:%M')}")
+        return [dict(task) for task in tasks]
+
+    def should_run_task(self, task: Dict[str, Any], now: datetime) -> bool:
+        """
+        Determine if task should run based on last execution.
+        Prevents running the same task multiple times in the same minute.
+        """
+        if not task['last_run']:
+            return True
+
+        last_run = task['last_run']
+        if last_run.tzinfo is None:
+            last_run = last_run.replace(tzinfo=timezone.utc)
+        if now.tzinfo is None:
+            now = now.replace(tzinfo=timezone.utc)
+
+        # Don't run if already executed this minute
+        if (last_run.year == now.year and
+            last_run.month == now.month and
+            last_run.day == now.day and
+            last_run.hour == now.hour and
+            last_run.minute == now.minute):
+            logger.debug(f"Task {task['task_name']} already ran this minute")
+            return False
+
+        return True
+
+    async def execute_task(self, task: Dict[str, Any]):
+        """
+        Execute a single task with timeout and error handling.
+        Updates task_executions table with results.
+        """
+        task_id = task['id']
+        task_name = task['task_name']
+        executor_name = task['executor']
+        timeout = task.get('timeout_seconds', 3600)
+
+        execution_id = None
+        started_at = datetime.now(timezone.utc)
+
+        logger.info(f"[Priority {task['priority']}] Starting task: {task_name} (executor: {executor_name})")
+
+        try:
+            # Create execution record
+            with self.get_db_connection() as conn:
+                with conn.cursor() as cur:
+                    cur.execute("""
+                        INSERT INTO task_executions
+                            (task_id, task_name, service, executor, priority,
+                             status, triggered_by, started_at)
+                        VALUES (%s, %s, %s, %s, %s, 'running', 'scheduler', %s)
+                        RETURNING id
+                    """, (task_id, task_name, task['service'], executor_name,
+                          task['priority'], started_at))
+                    execution_id = cur.fetchone()[0]
+                    conn.commit()
+
+            # Load and execute the task
+            output, error = await self._run_executor(executor_name, task, timeout)
+
+            completed_at = datetime.now(timezone.utc)
+            duration = int((completed_at - started_at).total_seconds())
+            status = 'success' if error is None else 'failed'
+
+            # Update execution record
+            with self.get_db_connection() as conn:
+                with conn.cursor() as cur:
+                    cur.execute("""
+                        UPDATE task_executions
+                        SET status = %s, completed_at = %s, duration_seconds = %s,
+                            output = %s, error = %s
+                        WHERE id = %s
+                    """, (status, completed_at, duration, output, error, execution_id))
+
+                    # Update scheduled_tasks
+                    cur.execute("""
+                        UPDATE scheduled_tasks
+                        SET last_run = %s, last_status = %s, last_duration_seconds = %s,
+                            retry_count = 0, updated_at = %s
+                        WHERE id = %s
+                    """, (completed_at, status, duration, completed_at, task_id))
+                    conn.commit()
+
+            if error:
+                logger.error(f"Task {task_name} failed: {error}")
+            else:
+                logger.info(f"Task {task_name} completed successfully in {duration}s")
+
+        except asyncio.TimeoutError:
+            logger.error(f"Task {task_name} timed out after {timeout}s")
+            self._update_execution_status(execution_id, 'timeout',
+                                         error=f"Task exceeded timeout of {timeout}s")
+
+        except Exception as e:
+            logger.error(f"Task {task_name} failed with exception: {e}")
+            logger.error(traceback.format_exc())
+            self._update_execution_status(execution_id, 'failed',
+                                         error=f"{str(e)}\n{traceback.format_exc()}")
+
+        finally:
+            # Remove from running tasks
+            if task_id in self.running_tasks:
+                del self.running_tasks[task_id]
+
+    async def _run_executor(self, executor_name: str, task: Dict[str, Any], timeout: int) -> tuple[Optional[str], Optional[str]]:
+        """
+        Dynamically load and run the executor module.
+        Returns (output, error) tuple.
+        """
+        try:
+            # Import executor dynamically
+            module_path = f"src.executors.{executor_name}"
+            module = __import__(module_path, fromlist=['execute'])
+
+            if not hasattr(module, 'execute'):
+                return None, f"Executor {executor_name} missing execute() function"
+
+            # Run with timeout
+            config = task.get('config', {})
+            result = await asyncio.wait_for(
+                module.execute(config, self.settings),
+                timeout=timeout
+            )
+
+            return result, None
+
+        except ModuleNotFoundError:
+            return None, f"Executor module not found: {executor_name}"
+        except Exception as e:
+            return None, f"Executor error: {str(e)}\n{traceback.format_exc()}"
+
+    def _update_execution_status(self, execution_id: int, status: str, error: str = None):
+        """Update execution record with final status."""
+        if execution_id is None:
+            return
+
+        try:
+            with self.get_db_connection() as conn:
+                with conn.cursor() as cur:
+                    cur.execute("""
+                        UPDATE task_executions
+                        SET status = %s, completed_at = %s, error = %s
+                        WHERE id = %s
+                    """, (status, datetime.now(timezone.utc), error, execution_id))
+                    conn.commit()
+        except Exception as e:
+            logger.error(f"Failed to update execution status: {e}")
+
+    async def process_minute(self):
+        """
+        Main entry point: Process all tasks scheduled for the current minute.
+        Executes up to MAX_CONCURRENT_TASKS in parallel, prioritized by priority field.
+        """
+        now = datetime.now(timezone.utc)
+        logger.info(f"Processing tasks for {now.strftime('%Y-%m-%d %H:%M')}")
+
+        # Get tasks scheduled for this minute
+        tasks = self.get_tasks_for_minute(now)
+
+        # Filter out tasks that already ran this minute
+        tasks_to_run = [task for task in tasks if self.should_run_task(task, now)]
+
+        if not tasks_to_run:
+            logger.debug("No tasks to run this minute")
+            return
+
+        logger.info(f"Will execute {len(tasks_to_run)} tasks (max {MAX_CONCURRENT_TASKS} concurrent)")
+
+        # Process tasks in priority order, respecting concurrency limit
+        for task in tasks_to_run:
+            # Wait for available slot
+            await self.semaphore.acquire()
+
+            # Start task
+            task_coro = self._execute_with_semaphore(task)
+            asyncio_task = asyncio.create_task(task_coro)
+            self.running_tasks[task['id']] = asyncio_task
+
+        # Wait for all tasks in this batch to complete
+        if self.running_tasks:
+            await asyncio.gather(*self.running_tasks.values(), return_exceptions=True)
+
+    async def _execute_with_semaphore(self, task: Dict[str, Any]):
+        """Execute task and release semaphore when done."""
+        try:
+            await self.execute_task(task)
+        finally:
+            self.semaphore.release()