Merge origin/main into sid/persistent-backend

Resolve conflict in local.py: keep refactored _make_run_env helper
over inline _sanitize_subprocess_env logic.

tools/__init__.py

@@ -84,14 +84,13 @@ from .browser_tool import (
 
 # Cronjob management tools (CLI-only, hermes-cli toolset)
 from .cronjob_tools import (
+    cronjob,
     schedule_cronjob,
     list_cronjobs,
     remove_cronjob,
     check_cronjob_requirements,
     get_cronjob_tool_definitions,
-    SCHEDULE_CRONJOB_SCHEMA,
-    LIST_CRONJOBS_SCHEMA,
-    REMOVE_CRONJOB_SCHEMA
+    CRONJOB_SCHEMA,
 )
 
 # RL Training tools (Tinker-Atropos)
@@ -211,14 +210,13 @@ __all__ = [
     'check_browser_requirements',
     'BROWSER_TOOL_SCHEMAS',
     # Cronjob management tools (CLI-only)
+    'cronjob',
     'schedule_cronjob',
     'list_cronjobs',
     'remove_cronjob',
     'check_cronjob_requirements',
     'get_cronjob_tool_definitions',
-    'SCHEDULE_CRONJOB_SCHEMA',
-    'LIST_CRONJOBS_SCHEMA',
-    'REMOVE_CRONJOB_SCHEMA',
+    'CRONJOB_SCHEMA',
     # RL Training tools
     'rl_list_environments',
     'rl_select_environment',

tools/approval.py

@@ -38,7 +38,7 @@ DANGEROUS_PATTERNS = [
     (r'\bsystemctl\s+(stop|disable|mask)\b', "stop/disable system service"),
     (r'\bkill\s+-9\s+-1\b', "kill all processes"),
     (r'\bpkill\s+-9\b', "force kill processes"),
-    (r':()\s*{\s*:\s*\|\s*:&\s*}\s*;:', "fork bomb"),
+    (r':\(\)\s*\{\s*:\s*\|\s*:\s*&\s*\}\s*;\s*:', "fork bomb"),
     (r'\b(bash|sh|zsh)\s+-c\s+', "shell command via -c flag"),
     (r'\b(python[23]?|perl|ruby|node)\s+-[ec]\s+', "script execution via -e/-c flag"),
     (r'\b(curl|wget)\b.*\|\s*(ba)?sh\b', "pipe remote content to shell"),
@@ -50,6 +50,29 @@ DANGEROUS_PATTERNS = [
 ]
 
 
+def _legacy_pattern_key(pattern: str) -> str:
+    """Reproduce the old regex-derived approval key for backwards compatibility."""
+    return pattern.split(r'\b')[1] if r'\b' in pattern else pattern[:20]
+
+
+_PATTERN_KEY_ALIASES: dict[str, set[str]] = {}
+for _pattern, _description in DANGEROUS_PATTERNS:
+    _legacy_key = _legacy_pattern_key(_pattern)
+    _canonical_key = _description
+    _PATTERN_KEY_ALIASES.setdefault(_canonical_key, set()).update({_canonical_key, _legacy_key})
+    _PATTERN_KEY_ALIASES.setdefault(_legacy_key, set()).update({_legacy_key, _canonical_key})
+
+
+def _approval_key_aliases(pattern_key: str) -> set[str]:
+    """Return all approval keys that should match this pattern.
+
+    New approvals use the human-readable description string, but older
+    command_allowlist entries and session approvals may still contain the
+    historical regex-derived key.
+    """
+    return _PATTERN_KEY_ALIASES.get(pattern_key, {pattern_key})
+
+
 # =========================================================================
 # Detection
 # =========================================================================
@@ -63,7 +86,7 @@ def detect_dangerous_command(command: str) -> tuple:
     command_lower = command.lower()
     for pattern, description in DANGEROUS_PATTERNS:
         if re.search(pattern, command_lower, re.IGNORECASE | re.DOTALL):
-            pattern_key = pattern.split(r'\b')[1] if r'\b' in pattern else pattern[:20]
+            pattern_key = description
             return (True, pattern_key, description)
     return (False, None, None)
 
@@ -103,11 +126,17 @@ def approve_session(session_key: str, pattern_key: str):
 
 
 def is_approved(session_key: str, pattern_key: str) -> bool:
-    """Check if a pattern is approved (session-scoped or permanent)."""
+    """Check if a pattern is approved (session-scoped or permanent).
+
+    Accept both the current canonical key and the legacy regex-derived key so
+    existing command_allowlist entries continue to work after key migrations.
+    """
+    aliases = _approval_key_aliases(pattern_key)
     with _lock:
-        if pattern_key in _permanent_approved:
+        if any(alias in _permanent_approved for alias in aliases):
             return True
-        return pattern_key in _session_approved.get(session_key, set())
+        session_approvals = _session_approved.get(session_key, set())
+        return any(alias in session_approvals for alias in aliases)
 
 
 def approve_permanent(pattern_key: str):

tools/code_execution_tool.py

@@ -440,6 +440,11 @@ def execute_code(
                 child_env[k] = v
         child_env["HERMES_RPC_SOCKET"] = sock_path
         child_env["PYTHONDONTWRITEBYTECODE"] = "1"
+        # Ensure the hermes-agent root is importable in the sandbox so
+        # modules like minisweagent_path are available to child scripts.
+        _hermes_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+        _existing_pp = child_env.get("PYTHONPATH", "")
+        child_env["PYTHONPATH"] = _hermes_root + (os.pathsep + _existing_pp if _existing_pp else "")
         # Inject user's configured timezone so datetime.now() in sandboxed
         # code reflects the correct wall-clock time.
         _tz_name = os.getenv("HERMES_TIMEZONE", "").strip()

tools/cronjob_tools.py

@@ -1,24 +1,32 @@
 """
 Cron job management tools for Hermes Agent.
 
-These tools allow the agent to schedule, list, and remove automated tasks.
-Only available when running via CLI (hermes-cli toolset).
-
-IMPORTANT: Cronjobs run in isolated sessions with NO prior context.
-The prompt must contain ALL necessary information.
+Expose a single compressed action-oriented tool to avoid schema/context bloat.
+Compatibility wrappers remain for direct Python callers and legacy tests.
 """
 
 import json
 import os
 import re
-from typing import Optional
-
-# Import from cron module (will be available when properly installed)
+import shutil
 import sys
 from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+# Import from cron module (will be available when properly installed)
 sys.path.insert(0, str(Path(__file__).parent.parent))
 
-from cron.jobs import create_job, get_job, list_jobs, remove_job
+from cron.jobs import (
+    create_job,
+    get_job,
+    list_jobs,
+    parse_schedule,
+    pause_job,
+    remove_job,
+    resume_job,
+    trigger_job,
+    update_job,
+)
 
 
 # ---------------------------------------------------------------------------
@@ -56,9 +64,233 @@ def _scan_cron_prompt(prompt: str) -> str:
     return ""
 
 
-# =============================================================================
-# Tool: schedule_cronjob
-# =============================================================================
+def _origin_from_env() -> Optional[Dict[str, str]]:
+    origin_platform = os.getenv("HERMES_SESSION_PLATFORM")
+    origin_chat_id = os.getenv("HERMES_SESSION_CHAT_ID")
+    if origin_platform and origin_chat_id:
+        return {
+            "platform": origin_platform,
+            "chat_id": origin_chat_id,
+            "chat_name": os.getenv("HERMES_SESSION_CHAT_NAME"),
+            "thread_id": os.getenv("HERMES_SESSION_THREAD_ID"),
+        }
+    return None
+
+
+def _repeat_display(job: Dict[str, Any]) -> str:
+    times = (job.get("repeat") or {}).get("times")
+    completed = (job.get("repeat") or {}).get("completed", 0)
+    if times is None:
+        return "forever"
+    if times == 1:
+        return "once" if completed == 0 else "1/1"
+    return f"{completed}/{times}" if completed else f"{times} times"
+
+
+def _canonical_skills(skill: Optional[str] = None, skills: Optional[Any] = None) -> List[str]:
+    if skills is None:
+        raw_items = [skill] if skill else []
+    elif isinstance(skills, str):
+        raw_items = [skills]
+    else:
+        raw_items = list(skills)
+
+    normalized: List[str] = []
+    for item in raw_items:
+        text = str(item or "").strip()
+        if text and text not in normalized:
+            normalized.append(text)
+    return normalized
+
+
+
+def _normalize_optional_job_value(value: Optional[Any], *, strip_trailing_slash: bool = False) -> Optional[str]:
+    if value is None:
+        return None
+    text = str(value).strip()
+    if strip_trailing_slash:
+        text = text.rstrip("/")
+    return text or None
+
+
+
+def _format_job(job: Dict[str, Any]) -> Dict[str, Any]:
+    prompt = job.get("prompt", "")
+    skills = _canonical_skills(job.get("skill"), job.get("skills"))
+    return {
+        "job_id": job["id"],
+        "name": job["name"],
+        "skill": skills[0] if skills else None,
+        "skills": skills,
+        "prompt_preview": prompt[:100] + "..." if len(prompt) > 100 else prompt,
+        "model": job.get("model"),
+        "provider": job.get("provider"),
+        "base_url": job.get("base_url"),
+        "schedule": job.get("schedule_display"),
+        "repeat": _repeat_display(job),
+        "deliver": job.get("deliver", "local"),
+        "next_run_at": job.get("next_run_at"),
+        "last_run_at": job.get("last_run_at"),
+        "last_status": job.get("last_status"),
+        "enabled": job.get("enabled", True),
+        "state": job.get("state", "scheduled" if job.get("enabled", True) else "paused"),
+        "paused_at": job.get("paused_at"),
+        "paused_reason": job.get("paused_reason"),
+    }
+
+
+def cronjob(
+    action: str,
+    job_id: Optional[str] = None,
+    prompt: Optional[str] = None,
+    schedule: Optional[str] = None,
+    name: Optional[str] = None,
+    repeat: Optional[int] = None,
+    deliver: Optional[str] = None,
+    include_disabled: bool = False,
+    skill: Optional[str] = None,
+    skills: Optional[List[str]] = None,
+    model: Optional[str] = None,
+    provider: Optional[str] = None,
+    base_url: Optional[str] = None,
+    reason: Optional[str] = None,
+    task_id: str = None,
+) -> str:
+    """Unified cron job management tool."""
+    del task_id  # unused but kept for handler signature compatibility
+
+    try:
+        normalized = (action or "").strip().lower()
+
+        if normalized == "create":
+            if not schedule:
+                return json.dumps({"success": False, "error": "schedule is required for create"}, indent=2)
+            canonical_skills = _canonical_skills(skill, skills)
+            if not prompt and not canonical_skills:
+                return json.dumps({"success": False, "error": "create requires either prompt or at least one skill"}, indent=2)
+            if prompt:
+                scan_error = _scan_cron_prompt(prompt)
+                if scan_error:
+                    return json.dumps({"success": False, "error": scan_error}, indent=2)
+
+            job = create_job(
+                prompt=prompt or "",
+                schedule=schedule,
+                name=name,
+                repeat=repeat,
+                deliver=deliver,
+                origin=_origin_from_env(),
+                skills=canonical_skills,
+                model=_normalize_optional_job_value(model),
+                provider=_normalize_optional_job_value(provider),
+                base_url=_normalize_optional_job_value(base_url, strip_trailing_slash=True),
+            )
+            return json.dumps(
+                {
+                    "success": True,
+                    "job_id": job["id"],
+                    "name": job["name"],
+                    "skill": job.get("skill"),
+                    "skills": job.get("skills", []),
+                    "schedule": job["schedule_display"],
+                    "repeat": _repeat_display(job),
+                    "deliver": job.get("deliver", "local"),
+                    "next_run_at": job["next_run_at"],
+                    "job": _format_job(job),
+                    "message": f"Cron job '{job['name']}' created.",
+                },
+                indent=2,
+            )
+
+        if normalized == "list":
+            jobs = [_format_job(job) for job in list_jobs(include_disabled=include_disabled)]
+            return json.dumps({"success": True, "count": len(jobs), "jobs": jobs}, indent=2)
+
+        if not job_id:
+            return json.dumps({"success": False, "error": f"job_id is required for action '{normalized}'"}, indent=2)
+
+        job = get_job(job_id)
+        if not job:
+            return json.dumps(
+                {"success": False, "error": f"Job with ID '{job_id}' not found. Use cronjob(action='list') to inspect jobs."},
+                indent=2,
+            )
+
+        if normalized == "remove":
+            removed = remove_job(job_id)
+            if not removed:
+                return json.dumps({"success": False, "error": f"Failed to remove job '{job_id}'"}, indent=2)
+            return json.dumps(
+                {
+                    "success": True,
+                    "message": f"Cron job '{job['name']}' removed.",
+                    "removed_job": {
+                        "id": job_id,
+                        "name": job["name"],
+                        "schedule": job.get("schedule_display"),
+                    },
+                },
+                indent=2,
+            )
+
+        if normalized == "pause":
+            updated = pause_job(job_id, reason=reason)
+            return json.dumps({"success": True, "job": _format_job(updated)}, indent=2)
+
+        if normalized == "resume":
+            updated = resume_job(job_id)
+            return json.dumps({"success": True, "job": _format_job(updated)}, indent=2)
+
+        if normalized in {"run", "run_now", "trigger"}:
+            updated = trigger_job(job_id)
+            return json.dumps({"success": True, "job": _format_job(updated)}, indent=2)
+
+        if normalized == "update":
+            updates: Dict[str, Any] = {}
+            if prompt is not None:
+                scan_error = _scan_cron_prompt(prompt)
+                if scan_error:
+                    return json.dumps({"success": False, "error": scan_error}, indent=2)
+                updates["prompt"] = prompt
+            if name is not None:
+                updates["name"] = name
+            if deliver is not None:
+                updates["deliver"] = deliver
+            if skills is not None or skill is not None:
+                canonical_skills = _canonical_skills(skill, skills)
+                updates["skills"] = canonical_skills
+                updates["skill"] = canonical_skills[0] if canonical_skills else None
+            if model is not None:
+                updates["model"] = _normalize_optional_job_value(model)
+            if provider is not None:
+                updates["provider"] = _normalize_optional_job_value(provider)
+            if base_url is not None:
+                updates["base_url"] = _normalize_optional_job_value(base_url, strip_trailing_slash=True)
+            if repeat is not None:
+                repeat_state = dict(job.get("repeat") or {})
+                repeat_state["times"] = repeat
+                updates["repeat"] = repeat_state
+            if schedule is not None:
+                parsed_schedule = parse_schedule(schedule)
+                updates["schedule"] = parsed_schedule
+                updates["schedule_display"] = parsed_schedule.get("display", schedule)
+                if job.get("state") != "paused":
+                    updates["state"] = "scheduled"
+                    updates["enabled"] = True
+            if not updates:
+                return json.dumps({"success": False, "error": "No updates provided."}, indent=2)
+            updated = update_job(job_id, updates)
+            return json.dumps({"success": True, "job": _format_job(updated)}, indent=2)
+
+        return json.dumps({"success": False, "error": f"Unknown cron action '{action}'"}, indent=2)
+
+    except Exception as e:
+        return json.dumps({"success": False, "error": str(e)}, indent=2)
+
+
+# ---------------------------------------------------------------------------
+# Compatibility wrappers
+# ---------------------------------------------------------------------------
 
 def schedule_cronjob(
     prompt: str,
@@ -66,329 +298,129 @@ def schedule_cronjob(
     name: Optional[str] = None,
     repeat: Optional[int] = None,
     deliver: Optional[str] = None,
-    task_id: str = None
+    model: Optional[str] = None,
+    provider: Optional[str] = None,
+    base_url: Optional[str] = None,
+    task_id: str = None,
 ) -> str:
-    """
-    Schedule an automated task to run the agent on a schedule.
-    
-    IMPORTANT: When the cronjob runs, it starts a COMPLETELY FRESH session.
-    The agent will have NO memory of this conversation or any prior context.
-    Therefore, the prompt MUST contain ALL necessary information:
-    - Full context of what needs to be done
-    - Specific file paths, URLs, or identifiers
-    - Clear success criteria
-    - Any relevant background information
-    
-    BAD prompt:  "Check on that server issue"
-    GOOD prompt: "SSH into server 192.168.1.100 as user 'deploy', check if nginx 
-                  is running with 'systemctl status nginx', and verify the site 
-                  https://example.com returns HTTP 200. Report any issues found."
-    
-    Args:
-        prompt: Complete, self-contained instructions for the future agent.
-                Must include ALL context needed - the agent won't remember anything.
-        schedule: When to run. Either:
-                  - Duration for one-shot: "30m", "2h", "1d" (runs once)
-                  - Interval: "every 30m", "every 2h" (recurring)
-                  - Cron expression: "0 9 * * *" (daily at 9am)
-                  - ISO timestamp: "2026-02-03T14:00:00" (one-shot at specific time)
-        name: Optional human-friendly name for the job (for listing/management)
-        repeat: How many times to run. Omit for default behavior:
-                - One-shot schedules default to repeat=1 (run once)
-                - Intervals/cron default to forever
-                - Set repeat=5 to run 5 times then auto-delete
-        deliver: Where to send the output. Options:
-                 - "origin": Back to where this job was created (default)
-                 - "local": Save to local files only (~/.hermes/cron/output/)
-                 - "telegram": Send to Telegram home channel
-                 - "discord": Send to Discord home channel
-                 - "signal": Send to Signal home channel
-                 - "telegram:123456": Send to specific chat ID
-                 - "signal:+15551234567": Send to specific Signal number
-    
-    Returns:
-        JSON with job_id, next_run time, and confirmation
-    """
-    # Scan prompt for critical threats before scheduling
-    scan_error = _scan_cron_prompt(prompt)
-    if scan_error:
-        return json.dumps({"success": False, "error": scan_error}, indent=2)
-
-    # Get origin info from environment if available
-    origin = None
-    origin_platform = os.getenv("HERMES_SESSION_PLATFORM")
-    origin_chat_id = os.getenv("HERMES_SESSION_CHAT_ID")
-    if origin_platform and origin_chat_id:
-        origin = {
-            "platform": origin_platform,
-            "chat_id": origin_chat_id,
-            "chat_name": os.getenv("HERMES_SESSION_CHAT_NAME"),
-        }
-    
-    try:
-        job = create_job(
-            prompt=prompt,
-            schedule=schedule,
-            name=name,
-            repeat=repeat,
-            deliver=deliver,
-            origin=origin
-        )
-        
-        # Format repeat info for display
-        times = job["repeat"].get("times")
-        if times is None:
-            repeat_display = "forever"
-        elif times == 1:
-            repeat_display = "once"
-        else:
-            repeat_display = f"{times} times"
-        
-        return json.dumps({
-            "success": True,
-            "job_id": job["id"],
-            "name": job["name"],
-            "schedule": job["schedule_display"],
-            "repeat": repeat_display,
-            "deliver": job.get("deliver", "local"),
-            "next_run_at": job["next_run_at"],
-            "message": f"Cronjob '{job['name']}' created. It will run {repeat_display}, deliver to {job.get('deliver', 'local')}, next at {job['next_run_at']}."
-        }, indent=2)
-        
-    except Exception as e:
-        return json.dumps({
-            "success": False,
-            "error": str(e)
-        }, indent=2)
+    return cronjob(
+        action="create",
+        prompt=prompt,
+        schedule=schedule,
+        name=name,
+        repeat=repeat,
+        deliver=deliver,
+        model=model,
+        provider=provider,
+        base_url=base_url,
+        task_id=task_id,
+    )
 
 
-SCHEDULE_CRONJOB_SCHEMA = {
-    "name": "schedule_cronjob",
-    "description": """Schedule an automated task to run the agent on a schedule.
+def list_cronjobs(include_disabled: bool = False, task_id: str = None) -> str:
+    return cronjob(action="list", include_disabled=include_disabled, task_id=task_id)
 
-⚠️ CRITICAL: The cronjob runs in a FRESH SESSION with NO CONTEXT from this conversation.
-The prompt must be COMPLETELY SELF-CONTAINED with ALL necessary information including:
-- Full context and background
-- Specific file paths, URLs, server addresses
-- Clear instructions and success criteria
-- Any credentials or configuration details
 
-The future agent will NOT remember anything from the current conversation.
+def remove_cronjob(job_id: str, task_id: str = None) -> str:
+    return cronjob(action="remove", job_id=job_id, task_id=task_id)
 
-SCHEDULE FORMATS:
-- One-shot: "30m", "2h", "1d" (runs once after delay)
-- Interval: "every 30m", "every 2h" (recurring)  
-- Cron: "0 9 * * *" (cron expression for precise scheduling)
-- Timestamp: "2026-02-03T14:00:00" (specific date/time)
 
-REPEAT BEHAVIOR:
-- One-shot schedules: run once by default
-- Intervals/cron: run forever by default
-- Set repeat=N to run exactly N times then auto-delete
+CRONJOB_SCHEMA = {
+    "name": "cronjob",
+    "description": """Manage scheduled cron jobs with a single compressed tool.
 
-DELIVERY OPTIONS (where output goes):
-- "origin": Back to current chat (default if in messaging platform)
-- "local": Save to local files only (default if in CLI)
-- "telegram": Send to Telegram home channel
-- "discord": Send to Discord home channel
-- "telegram:123456": Send to specific chat (if user provides ID)
+Use action='create' to schedule a new job from a prompt or one or more skills.
+Use action='list' to inspect jobs.
+Use action='update', 'pause', 'resume', 'remove', or 'run' to manage an existing job.
+
+Jobs run in a fresh session with no current-chat context, so prompts must be self-contained.
+If skill or skills are provided on create, the future cron run loads those skills in order, then follows the prompt as the task instruction.
+On update, passing skills=[] clears attached skills.
 
 NOTE: The agent's final response is auto-delivered to the target — do NOT use
-send_message in the prompt. Just have the agent compose its response normally.
+send_message in the prompt for that same destination. Same-target send_message
+calls are skipped to avoid duplicate cron deliveries. Put the primary
+user-facing content in the final response, and use send_message only for
+additional or different targets.
 
-Use for: reminders, periodic checks, scheduled reports, automated maintenance.""",
+Important safety rule: cron-run sessions should not recursively schedule more cron jobs.""",
     "parameters": {
         "type": "object",
         "properties": {
+            "action": {
+                "type": "string",
+                "description": "One of: create, list, update, pause, resume, remove, run"
+            },
+            "job_id": {
+                "type": "string",
+                "description": "Required for update/pause/resume/remove/run"
+            },
             "prompt": {
                 "type": "string",
-                "description": "Complete, self-contained instructions. Must include ALL context - the future agent will have NO memory of this conversation."
+                "description": "For create: the full self-contained prompt. If skill or skills are also provided, this becomes the task instruction paired with those skills."
             },
             "schedule": {
                 "type": "string",
-                "description": "When to run: '30m' (once in 30min), 'every 30m' (recurring), '0 9 * * *' (cron), or ISO timestamp"
+                "description": "For create/update: '30m', 'every 2h', '0 9 * * *', or ISO timestamp"
             },
             "name": {
                 "type": "string",
-                "description": "Optional human-friendly name for the job"
+                "description": "Optional human-friendly name"
             },
             "repeat": {
                 "type": "integer",
-                "description": "How many times to run. Omit for default (once for one-shot, forever for recurring). Set to N for exactly N runs."
+                "description": "Optional repeat count. Omit for defaults (once for one-shot, forever for recurring)."
             },
             "deliver": {
                 "type": "string",
-                "description": "Where to send output: 'origin' (back to this chat), 'local' (files only), 'telegram', 'discord', 'signal', or 'platform:chat_id'"
-            }
-        },
-        "required": ["prompt", "schedule"]
-    }
-}
-
-
-# =============================================================================
-# Tool: list_cronjobs
-# =============================================================================
-
-def list_cronjobs(include_disabled: bool = False, task_id: str = None) -> str:
-    """
-    List all scheduled cronjobs.
-    
-    Returns information about each job including:
-    - Job ID (needed for removal)
-    - Name
-    - Schedule (human-readable)
-    - Repeat status (completed/total or 'forever')
-    - Next scheduled run time
-    - Last run time and status (if any)
-    
-    Args:
-        include_disabled: Whether to include disabled/completed jobs
-    
-    Returns:
-        JSON array of all scheduled jobs
-    """
-    try:
-        jobs = list_jobs(include_disabled=include_disabled)
-        
-        formatted_jobs = []
-        for job in jobs:
-            # Format repeat status
-            times = job["repeat"].get("times")
-            completed = job["repeat"].get("completed", 0)
-            if times is None:
-                repeat_status = "forever"
-            else:
-                repeat_status = f"{completed}/{times}"
-            
-            formatted_jobs.append({
-                "job_id": job["id"],
-                "name": job["name"],
-                "prompt_preview": job["prompt"][:100] + "..." if len(job["prompt"]) > 100 else job["prompt"],
-                "schedule": job["schedule_display"],
-                "repeat": repeat_status,
-                "deliver": job.get("deliver", "local"),
-                "next_run_at": job.get("next_run_at"),
-                "last_run_at": job.get("last_run_at"),
-                "last_status": job.get("last_status"),
-                "enabled": job.get("enabled", True)
-            })
-        
-        return json.dumps({
-            "success": True,
-            "count": len(formatted_jobs),
-            "jobs": formatted_jobs
-        }, indent=2)
-        
-    except Exception as e:
-        return json.dumps({
-            "success": False,
-            "error": str(e)
-        }, indent=2)
-
-
-LIST_CRONJOBS_SCHEMA = {
-    "name": "list_cronjobs",
-    "description": """List all scheduled cronjobs with their IDs, schedules, and status.
-
-Use this to:
-- See what jobs are currently scheduled
-- Find job IDs for removal with remove_cronjob
-- Check job status and next run times
-
-Returns job_id, name, schedule, repeat status, next/last run times.""",
-    "parameters": {
-        "type": "object",
-        "properties": {
+                "description": "Delivery target: origin, local, telegram, discord, signal, or platform:chat_id"
+            },
+            "model": {
+                "type": "string",
+                "description": "Optional per-job model override used when the cron job runs"
+            },
+            "provider": {
+                "type": "string",
+                "description": "Optional per-job provider override used when resolving runtime credentials"
+            },
+            "base_url": {
+                "type": "string",
+                "description": "Optional per-job base URL override paired with provider/model routing"
+            },
             "include_disabled": {
                 "type": "boolean",
-                "description": "Include disabled/completed jobs in the list (default: false)"
-            }
-        },
-        "required": []
-    }
-}
-
-
-# =============================================================================
-# Tool: remove_cronjob
-# =============================================================================
-
-def remove_cronjob(job_id: str, task_id: str = None) -> str:
-    """
-    Remove a scheduled cronjob by its ID.
-    
-    Use list_cronjobs first to find the job_id of the job you want to remove.
-    
-    Args:
-        job_id: The ID of the job to remove (from list_cronjobs output)
-    
-    Returns:
-        JSON confirmation of removal
-    """
-    try:
-        job = get_job(job_id)
-        if not job:
-            return json.dumps({
-                "success": False,
-                "error": f"Job with ID '{job_id}' not found. Use list_cronjobs to see available jobs."
-            }, indent=2)
-        
-        removed = remove_job(job_id)
-        if removed:
-            return json.dumps({
-                "success": True,
-                "message": f"Cronjob '{job['name']}' (ID: {job_id}) has been removed.",
-                "removed_job": {
-                    "id": job_id,
-                    "name": job["name"],
-                    "schedule": job["schedule_display"]
-                }
-            }, indent=2)
-        else:
-            return json.dumps({
-                "success": False,
-                "error": f"Failed to remove job '{job_id}'"
-            }, indent=2)
-            
-    except Exception as e:
-        return json.dumps({
-            "success": False,
-            "error": str(e)
-        }, indent=2)
-
-
-REMOVE_CRONJOB_SCHEMA = {
-    "name": "remove_cronjob",
-    "description": """Remove a scheduled cronjob by its ID.
-
-Use list_cronjobs first to find the job_id of the job you want to remove.
-Jobs that have completed their repeat count are auto-removed, but you can
-use this to cancel a job before it completes.""",
-    "parameters": {
-        "type": "object",
-        "properties": {
-            "job_id": {
+                "description": "For list: include paused/completed jobs"
+            },
+            "skill": {
                 "type": "string",
-                "description": "The ID of the cronjob to remove (from list_cronjobs output)"
+                "description": "Optional single skill name to load before executing the cron prompt"
+            },
+            "skills": {
+                "type": "array",
+                "items": {"type": "string"},
+                "description": "Optional ordered list of skills to load before executing the cron prompt. On update, pass an empty array to clear attached skills."
+            },
+            "reason": {
+                "type": "string",
+                "description": "Optional pause reason"
             }
         },
-        "required": ["job_id"]
+        "required": ["action"]
     }
 }
 
 
-# =============================================================================
-# Requirements check
-# =============================================================================
-
 def check_cronjob_requirements() -> bool:
     """
     Check if cronjob tools can be used.
-    
+
+    Requires 'crontab' executable to be present in the system PATH.
     Available in interactive CLI mode and gateway/messaging platforms.
-    Cronjobs are server-side scheduled tasks so they work from any interface.
     """
+    # Ensure the system can actually install and manage cron entries.
+    if not shutil.which("crontab"):
+        return False
+
     return bool(
         os.getenv("HERMES_INTERACTIVE")
         or os.getenv("HERMES_GATEWAY_SESSION")
@@ -396,66 +428,34 @@ def check_cronjob_requirements() -> bool:
     )
 
 
-# =============================================================================
-# Exports
-# =============================================================================
-
 def get_cronjob_tool_definitions():
     """Return tool definitions for cronjob management."""
-    return [
-        SCHEDULE_CRONJOB_SCHEMA,
-        LIST_CRONJOBS_SCHEMA,
-        REMOVE_CRONJOB_SCHEMA
-    ]
-
-
-# For direct testing
-if __name__ == "__main__":
-    # Test the tools
-    print("Testing schedule_cronjob:")
-    result = schedule_cronjob(
-        prompt="Test prompt for cron job",
-        schedule="5m",
-        name="Test Job"
-    )
-    print(result)
-    
-    print("\nTesting list_cronjobs:")
-    result = list_cronjobs()
-    print(result)
+    return [CRONJOB_SCHEMA]
 
 
 # --- Registry ---
 from tools.registry import registry
 
 registry.register(
-    name="schedule_cronjob",
+    name="cronjob",
     toolset="cronjob",
-    schema=SCHEDULE_CRONJOB_SCHEMA,
-    handler=lambda args, **kw: schedule_cronjob(
-        prompt=args.get("prompt", ""),
-        schedule=args.get("schedule", ""),
+    schema=CRONJOB_SCHEMA,
+    handler=lambda args, **kw: cronjob(
+        action=args.get("action", ""),
+        job_id=args.get("job_id"),
+        prompt=args.get("prompt"),
+        schedule=args.get("schedule"),
         name=args.get("name"),
         repeat=args.get("repeat"),
         deliver=args.get("deliver"),
-        task_id=kw.get("task_id")),
-    check_fn=check_cronjob_requirements,
-)
-registry.register(
-    name="list_cronjobs",
-    toolset="cronjob",
-    schema=LIST_CRONJOBS_SCHEMA,
-    handler=lambda args, **kw: list_cronjobs(
         include_disabled=args.get("include_disabled", False),
-        task_id=kw.get("task_id")),
-    check_fn=check_cronjob_requirements,
-)
-registry.register(
-    name="remove_cronjob",
-    toolset="cronjob",
-    schema=REMOVE_CRONJOB_SCHEMA,
-    handler=lambda args, **kw: remove_cronjob(
-        job_id=args.get("job_id", ""),
-        task_id=kw.get("task_id")),
+        skill=args.get("skill"),
+        skills=args.get("skills"),
+        model=args.get("model"),
+        provider=args.get("provider"),
+        base_url=args.get("base_url"),
+        reason=args.get("reason"),
+        task_id=kw.get("task_id"),
+    ),
     check_fn=check_cronjob_requirements,
 )

tools/delegate_tool.py

@@ -540,18 +540,51 @@ def delegate_task(
 def _resolve_delegation_credentials(cfg: dict, parent_agent) -> dict:
     """Resolve credentials for subagent delegation.
 
-    If ``delegation.provider`` is configured, resolves the full credential
-    bundle (base_url, api_key, api_mode, provider) via the runtime provider
-    system — the same path used by CLI/gateway startup.  This lets subagents
-    run on a completely different provider:model pair.
+    If ``delegation.base_url`` is configured, subagents use that direct
+    OpenAI-compatible endpoint. Otherwise, if ``delegation.provider`` is
+    configured, the full credential bundle (base_url, api_key, api_mode,
+    provider) is resolved via the runtime provider system — the same path used
+    by CLI/gateway startup. This lets subagents run on a completely different
+    provider:model pair.
 
-    If no provider is configured, returns None values so the child inherits
-    everything from the parent agent.
+    If neither base_url nor provider is configured, returns None values so the
+    child inherits everything from the parent agent.
 
     Raises ValueError with a user-friendly message on credential failure.
     """
-    configured_model = cfg.get("model") or None
-    configured_provider = cfg.get("provider") or None
+    configured_model = str(cfg.get("model") or "").strip() or None
+    configured_provider = str(cfg.get("provider") or "").strip() or None
+    configured_base_url = str(cfg.get("base_url") or "").strip() or None
+    configured_api_key = str(cfg.get("api_key") or "").strip() or None
+
+    if configured_base_url:
+        api_key = (
+            configured_api_key
+            or os.getenv("OPENAI_API_KEY", "").strip()
+        )
+        if not api_key:
+            raise ValueError(
+                "Delegation base_url is configured but no API key was found. "
+                "Set delegation.api_key or OPENAI_API_KEY."
+            )
+
+        base_lower = configured_base_url.lower()
+        provider = "custom"
+        api_mode = "chat_completions"
+        if "chatgpt.com/backend-api/codex" in base_lower:
+            provider = "openai-codex"
+            api_mode = "codex_responses"
+        elif "api.anthropic.com" in base_lower:
+            provider = "anthropic"
+            api_mode = "anthropic_messages"
+
+        return {
+            "model": configured_model,
+            "provider": provider,
+            "base_url": configured_base_url,
+            "api_key": api_key,
+            "api_mode": api_mode,
+        }
 
     if not configured_provider:
         # No provider override — child inherits everything from parent
@@ -570,7 +603,8 @@ def _resolve_delegation_credentials(cfg: dict, parent_agent) -> dict:
     except Exception as exc:
         raise ValueError(
             f"Cannot resolve delegation provider '{configured_provider}': {exc}. "
-            f"Check that the provider is configured (API key set, valid provider name). "
+            f"Check that the provider is configured (API key set, valid provider name), "
+            f"or set delegation.base_url/delegation.api_key for a direct endpoint. "
             f"Available providers: openrouter, nous, zai, kimi-coding, minimax."
         ) from exc
 

tools/environments/local.py

@@ -30,11 +30,12 @@ _HERMES_PROVIDER_ENV_FORCE_PREFIX = "_HERMES_FORCE_"
 
 
 def _build_provider_env_blocklist() -> frozenset:
-    """Derive the blocklist from the provider registry + known extras.
+    """Derive the blocklist from provider, tool, and gateway config.
 
     Automatically picks up api_key_env_vars and base_url_env_var from
-    every registered provider, so adding a new provider to auth.py is
-    enough — no manual list to keep in sync.
+    every registered provider, plus tool/messaging env vars from the
+    optional config registry, so new Hermes-managed secrets are blocked
+    in subprocesses without having to maintain multiple static lists.
     """
     blocked: set[str] = set()
 
@@ -47,7 +48,18 @@ def _build_provider_env_blocklist() -> frozenset:
     except ImportError:
         pass
 
-    # Vars not in the registry but still Hermes-internal / conflict-prone
+    try:
+        from hermes_cli.config import OPTIONAL_ENV_VARS
+        for name, metadata in OPTIONAL_ENV_VARS.items():
+            category = metadata.get("category")
+            if category in {"tool", "messaging"}:
+                blocked.add(name)
+            elif category == "setting" and metadata.get("password"):
+                blocked.add(name)
+    except ImportError:
+        pass
+
+    # Vars not covered above but still Hermes-internal / conflict-prone.
     blocked.update({
         "OPENAI_BASE_URL",
         "OPENAI_API_KEY",
@@ -59,6 +71,52 @@ def _build_provider_env_blocklist() -> frozenset:
         "ANTHROPIC_TOKEN",         # OAuth token (not in registry as env var)
         "CLAUDE_CODE_OAUTH_TOKEN",
         "LLM_MODEL",
+        # Expanded isolation for other major providers (Issue #1002)
+        "GOOGLE_API_KEY",          # Gemini / Google AI Studio
+        "DEEPSEEK_API_KEY",        # DeepSeek
+        "MISTRAL_API_KEY",         # Mistral AI
+        "GROQ_API_KEY",            # Groq
+        "TOGETHER_API_KEY",        # Together AI
+        "PERPLEXITY_API_KEY",      # Perplexity
+        "COHERE_API_KEY",          # Cohere
+        "FIREWORKS_API_KEY",       # Fireworks AI
+        "XAI_API_KEY",             # xAI (Grok)
+        "HELICONE_API_KEY",        # LLM Observability proxy
+        # Gateway/runtime config not represented in OPTIONAL_ENV_VARS.
+        "TELEGRAM_HOME_CHANNEL",
+        "TELEGRAM_HOME_CHANNEL_NAME",
+        "DISCORD_HOME_CHANNEL",
+        "DISCORD_HOME_CHANNEL_NAME",
+        "DISCORD_REQUIRE_MENTION",
+        "DISCORD_FREE_RESPONSE_CHANNELS",
+        "DISCORD_AUTO_THREAD",
+        "SLACK_HOME_CHANNEL",
+        "SLACK_HOME_CHANNEL_NAME",
+        "SLACK_ALLOWED_USERS",
+        "WHATSAPP_ENABLED",
+        "WHATSAPP_MODE",
+        "WHATSAPP_ALLOWED_USERS",
+        "SIGNAL_HTTP_URL",
+        "SIGNAL_ACCOUNT",
+        "SIGNAL_ALLOWED_USERS",
+        "SIGNAL_GROUP_ALLOWED_USERS",
+        "SIGNAL_HOME_CHANNEL",
+        "SIGNAL_HOME_CHANNEL_NAME",
+        "SIGNAL_IGNORE_STORIES",
+        "HASS_TOKEN",
+        "HASS_URL",
+        "EMAIL_ADDRESS",
+        "EMAIL_PASSWORD",
+        "EMAIL_IMAP_HOST",
+        "EMAIL_SMTP_HOST",
+        "EMAIL_HOME_ADDRESS",
+        "EMAIL_HOME_ADDRESS_NAME",
+        "GATEWAY_ALLOWED_USERS",
+        # Skills Hub / GitHub app auth paths and aliases.
+        "GH_TOKEN",
+        "GITHUB_APP_ID",
+        "GITHUB_APP_PRIVATE_KEY_PATH",
+        "GITHUB_APP_INSTALLATION_ID",
     })
     return frozenset(blocked)
 
@@ -66,6 +124,30 @@ def _build_provider_env_blocklist() -> frozenset:
 _HERMES_PROVIDER_ENV_BLOCKLIST = _build_provider_env_blocklist()
 
 
+def _sanitize_subprocess_env(base_env: dict | None, extra_env: dict | None = None) -> dict:
+    """Filter Hermes-managed secrets from a subprocess environment.
+
+    `_HERMES_FORCE_<VAR>` entries in ``extra_env`` opt a blocked variable back in
+    intentionally for callers that truly need it.
+    """
+    sanitized: dict[str, str] = {}
+
+    for key, value in (base_env or {}).items():
+        if key.startswith(_HERMES_PROVIDER_ENV_FORCE_PREFIX):
+            continue
+        if key not in _HERMES_PROVIDER_ENV_BLOCKLIST:
+            sanitized[key] = value
+
+    for key, value in (extra_env or {}).items():
+        if key.startswith(_HERMES_PROVIDER_ENV_FORCE_PREFIX):
+            real_key = key[len(_HERMES_PROVIDER_ENV_FORCE_PREFIX):]
+            sanitized[real_key] = value
+        elif key not in _HERMES_PROVIDER_ENV_BLOCKLIST:
+            sanitized[key] = value
+
+    return sanitized
+
+
 def _find_bash() -> str:
     """Find bash for command execution.
 

tools/process_registry.py

@@ -42,7 +42,7 @@ import time
 import uuid
 
 _IS_WINDOWS = platform.system() == "Windows"
-from tools.environments.local import _find_shell, _HERMES_PROVIDER_ENV_BLOCKLIST
+from tools.environments.local import _find_shell, _sanitize_subprocess_env
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Dict, List, Optional
@@ -155,9 +155,7 @@ class ProcessRegistry:
                 else:
                     from ptyprocess import PtyProcess as _PtyProcessCls
                 user_shell = _find_shell()
-                pty_env = {k: v for k, v in os.environ.items()
-                           if k not in _HERMES_PROVIDER_ENV_BLOCKLIST}
-                pty_env.update(env_vars or {})
+                pty_env = _sanitize_subprocess_env(os.environ, env_vars)
                 pty_env["PYTHONUNBUFFERED"] = "1"
                 pty_proc = _PtyProcessCls.spawn(
                     [user_shell, "-lic", command],
@@ -198,9 +196,7 @@ class ProcessRegistry:
         # Force unbuffered output for Python scripts so progress is visible
         # during background execution (libraries like tqdm/datasets buffer when
         # stdout is a pipe, hiding output from process(action="poll")).
-        bg_env = {k: v for k, v in os.environ.items()
-                  if k not in _HERMES_PROVIDER_ENV_BLOCKLIST}
-        bg_env.update(env_vars or {})
+        bg_env = _sanitize_subprocess_env(os.environ, env_vars)
         bg_env["PYTHONUNBUFFERED"] = "1"
         proc = subprocess.Popen(
             [user_shell, "-lic", command],

tools/send_message_tool.py

@@ -153,6 +153,10 @@ def _handle_send(args):
                 f"or set a home channel via: hermes config set {platform_name.upper()}_HOME_CHANNEL <channel_id>"
             })
 
+    duplicate_skip = _maybe_skip_cron_duplicate_send(platform_name, chat_id, thread_id)
+    if duplicate_skip:
+        return json.dumps(duplicate_skip)
+
     try:
         from model_tools import _run_async
         result = _run_async(
@@ -213,6 +217,51 @@ def _describe_media_for_mirror(media_files):
     return f"[Sent {len(media_files)} media attachments]"
 
 
+def _get_cron_auto_delivery_target():
+    """Return the cron scheduler's auto-delivery target for the current run, if any."""
+    platform = os.getenv("HERMES_CRON_AUTO_DELIVER_PLATFORM", "").strip().lower()
+    chat_id = os.getenv("HERMES_CRON_AUTO_DELIVER_CHAT_ID", "").strip()
+    if not platform or not chat_id:
+        return None
+    thread_id = os.getenv("HERMES_CRON_AUTO_DELIVER_THREAD_ID", "").strip() or None
+    return {
+        "platform": platform,
+        "chat_id": chat_id,
+        "thread_id": thread_id,
+    }
+
+
+def _maybe_skip_cron_duplicate_send(platform_name: str, chat_id: str, thread_id: str | None):
+    """Skip redundant cron send_message calls when the scheduler will auto-deliver there."""
+    auto_target = _get_cron_auto_delivery_target()
+    if not auto_target:
+        return None
+
+    same_target = (
+        auto_target["platform"] == platform_name
+        and str(auto_target["chat_id"]) == str(chat_id)
+        and auto_target.get("thread_id") == thread_id
+    )
+    if not same_target:
+        return None
+
+    target_label = f"{platform_name}:{chat_id}"
+    if thread_id is not None:
+        target_label += f":{thread_id}"
+
+    return {
+        "success": True,
+        "skipped": True,
+        "reason": "cron_auto_delivery_duplicate_target",
+        "target": target_label,
+        "note": (
+            f"Skipped send_message to {target_label}. This cron job will already auto-deliver "
+            "its final response to that same target. Put the intended user-facing content in "
+            "your final response instead, or use a different target if you want an additional message."
+        ),
+    }
+
+
 async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, media_files=None):
     """Route a message to the appropriate platform sender."""
     from gateway.config import Platform

tools/skills_hub.py

@@ -1156,11 +1156,176 @@ class ClawHubSource(SkillSource):
     def trust_level_for(self, identifier: str) -> str:
         return "community"
 
+    @staticmethod
+    def _normalize_tags(tags: Any) -> List[str]:
+        if isinstance(tags, list):
+            return [str(t) for t in tags]
+        if isinstance(tags, dict):
+            return [str(k) for k in tags.keys() if str(k) != "latest"]
+        return []
+
+    @staticmethod
+    def _coerce_skill_payload(data: Any) -> Optional[Dict[str, Any]]:
+        if not isinstance(data, dict):
+            return None
+        nested = data.get("skill")
+        if isinstance(nested, dict):
+            merged = dict(nested)
+            latest_version = data.get("latestVersion")
+            if latest_version is not None and "latestVersion" not in merged:
+                merged["latestVersion"] = latest_version
+            return merged
+        return data
+
+    @staticmethod
+    def _query_terms(query: str) -> List[str]:
+        return [term for term in re.split(r"[^a-z0-9]+", query.lower()) if term]
+
+    @classmethod
+    def _search_score(cls, query: str, meta: SkillMeta) -> int:
+        query_norm = query.strip().lower()
+        if not query_norm:
+            return 1
+
+        identifier = (meta.identifier or "").lower()
+        name = (meta.name or "").lower()
+        description = (meta.description or "").lower()
+        normalized_identifier = " ".join(cls._query_terms(identifier))
+        normalized_name = " ".join(cls._query_terms(name))
+        query_terms = cls._query_terms(query_norm)
+        identifier_terms = cls._query_terms(identifier)
+        name_terms = cls._query_terms(name)
+        score = 0
+
+        if query_norm == identifier:
+            score += 140
+        if query_norm == name:
+            score += 130
+        if normalized_identifier == query_norm:
+            score += 125
+        if normalized_name == query_norm:
+            score += 120
+        if normalized_identifier.startswith(query_norm):
+            score += 95
+        if normalized_name.startswith(query_norm):
+            score += 90
+        if query_terms and identifier_terms[: len(query_terms)] == query_terms:
+            score += 70
+        if query_terms and name_terms[: len(query_terms)] == query_terms:
+            score += 65
+        if query_norm in identifier:
+            score += 40
+        if query_norm in name:
+            score += 35
+        if query_norm in description:
+            score += 10
+
+        for term in query_terms:
+            if term in identifier_terms:
+                score += 15
+            if term in name_terms:
+                score += 12
+            if term in description:
+                score += 3
+
+        return score
+
+    @staticmethod
+    def _dedupe_results(results: List[SkillMeta]) -> List[SkillMeta]:
+        seen: set[str] = set()
+        deduped: List[SkillMeta] = []
+        for result in results:
+            key = (result.identifier or result.name).lower()
+            if key in seen:
+                continue
+            seen.add(key)
+            deduped.append(result)
+        return deduped
+
+    def _exact_slug_meta(self, query: str) -> Optional[SkillMeta]:
+        slug = query.strip().split("/")[-1]
+        query_terms = self._query_terms(query)
+        candidates: List[str] = []
+
+        if slug and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", slug):
+            candidates.append(slug)
+
+        if query_terms:
+            base_slug = "-".join(query_terms)
+            if len(query_terms) >= 2:
+                candidates.extend([
+                    f"{base_slug}-agent",
+                    f"{base_slug}-skill",
+                    f"{base_slug}-tool",
+                    f"{base_slug}-assistant",
+                    f"{base_slug}-playbook",
+                    base_slug,
+                ])
+            else:
+                candidates.append(base_slug)
+
+        seen: set[str] = set()
+        for candidate in candidates:
+            if candidate in seen:
+                continue
+            seen.add(candidate)
+            meta = self.inspect(candidate)
+            if meta:
+                return meta
+
+        return None
+
+    def _finalize_search_results(self, query: str, results: List[SkillMeta], limit: int) -> List[SkillMeta]:
+        query_norm = query.strip()
+        if not query_norm:
+            return self._dedupe_results(results)[:limit]
+
+        filtered = [meta for meta in results if self._search_score(query_norm, meta) > 0]
+        filtered.sort(
+            key=lambda meta: (
+                -self._search_score(query_norm, meta),
+                meta.name.lower(),
+                meta.identifier.lower(),
+            )
+        )
+        filtered = self._dedupe_results(filtered)
+
+        exact = self._exact_slug_meta(query_norm)
+        if exact:
+            filtered = [meta for meta in filtered if self._search_score(query_norm, meta) >= 20]
+            filtered = self._dedupe_results([exact] + filtered)
+
+        if filtered:
+            return filtered[:limit]
+
+        if re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/-]*", query_norm):
+            return []
+
+        return self._dedupe_results(results)[:limit]
+
     def search(self, query: str, limit: int = 10) -> List[SkillMeta]:
-        cache_key = f"clawhub_search_{hashlib.md5(query.encode()).hexdigest()}"
+        query = query.strip()
+
+        if query:
+            query_terms = self._query_terms(query)
+            if len(query_terms) >= 2:
+                direct = self._exact_slug_meta(query)
+                if direct:
+                    return [direct]
+
+            results = self._search_catalog(query, limit=limit)
+            if results:
+                return results
+
+        # Empty query or catalog fallback failure: use the lightweight listing API.
+        cache_key = f"clawhub_search_listing_v1_{hashlib.md5(query.encode()).hexdigest()}_{limit}"
         cached = _read_index_cache(cache_key)
         if cached is not None:
-            return [SkillMeta(**s) for s in cached][:limit]
+            return self._finalize_search_results(
+                query,
+                [SkillMeta(**s) for s in cached],
+                limit,
+            )
 
         try:
             resp = httpx.get(
@@ -1185,20 +1350,19 @@ class ClawHubSource(SkillSource):
                 continue
             display_name = item.get("displayName") or item.get("name") or slug
             summary = item.get("summary") or item.get("description") or ""
-            tags = item.get("tags", [])
-            if not isinstance(tags, list):
-                tags = []
+            tags = self._normalize_tags(item.get("tags", []))
             results.append(SkillMeta(
                 name=display_name,
                 description=summary,
                 source="clawhub",
                 identifier=slug,
                 trust_level="community",
-                tags=[str(t) for t in tags],
+                tags=tags,
             ))
 
-        _write_index_cache(cache_key, [_skill_meta_to_dict(s) for s in results])
-        return results
+        final_results = self._finalize_search_results(query, results, limit)
+        _write_index_cache(cache_key, [_skill_meta_to_dict(s) for s in final_results])
+        return final_results
 
     def fetch(self, identifier: str) -> Optional[SkillBundle]:
         slug = identifier.split("/")[-1]
@@ -1244,13 +1408,11 @@ class ClawHubSource(SkillSource):
 
     def inspect(self, identifier: str) -> Optional[SkillMeta]:
         slug = identifier.split("/")[-1]
-        data = self._get_json(f"{self.BASE_URL}/skills/{slug}")
+        data = self._coerce_skill_payload(self._get_json(f"{self.BASE_URL}/skills/{slug}"))
         if not isinstance(data, dict):
             return None
 
-        tags = data.get("tags", [])
-        if not isinstance(tags, list):
-            tags = []
+        tags = self._normalize_tags(data.get("tags", []))
 
         return SkillMeta(
             name=data.get("displayName") or data.get("name") or data.get("slug") or slug,
@@ -1258,9 +1420,75 @@ class ClawHubSource(SkillSource):
             source="clawhub",
             identifier=data.get("slug") or slug,
             trust_level="community",
-            tags=[str(t) for t in tags],
+            tags=tags,
         )
 
+    def _search_catalog(self, query: str, limit: int = 10) -> List[SkillMeta]:
+        cache_key = f"clawhub_search_catalog_v1_{hashlib.md5(f'{query}|{limit}'.encode()).hexdigest()}"
+        cached = _read_index_cache(cache_key)
+        if cached is not None:
+            return [SkillMeta(**s) for s in cached][:limit]
+
+        catalog = self._load_catalog_index()
+        if not catalog:
+            return []
+
+        results = self._finalize_search_results(query, catalog, limit)
+        _write_index_cache(cache_key, [_skill_meta_to_dict(s) for s in results])
+        return results
+
+    def _load_catalog_index(self) -> List[SkillMeta]:
+        cache_key = "clawhub_catalog_v1"
+        cached = _read_index_cache(cache_key)
+        if cached is not None:
+            return [SkillMeta(**s) for s in cached]
+
+        cursor: Optional[str] = None
+        results: List[SkillMeta] = []
+        seen: set[str] = set()
+        max_pages = 50
+
+        for _ in range(max_pages):
+            params: Dict[str, Any] = {"limit": 200}
+            if cursor:
+                params["cursor"] = cursor
+
+            try:
+                resp = httpx.get(f"{self.BASE_URL}/skills", params=params, timeout=30)
+                if resp.status_code != 200:
+                    break
+                data = resp.json()
+            except (httpx.HTTPError, json.JSONDecodeError):
+                break
+
+            items = data.get("items", []) if isinstance(data, dict) else []
+            if not isinstance(items, list) or not items:
+                break
+
+            for item in items:
+                slug = item.get("slug")
+                if not isinstance(slug, str) or not slug or slug in seen:
+                    continue
+                seen.add(slug)
+                display_name = item.get("displayName") or item.get("name") or slug
+                summary = item.get("summary") or item.get("description") or ""
+                tags = self._normalize_tags(item.get("tags", []))
+                results.append(SkillMeta(
+                    name=display_name,
+                    description=summary,
+                    source="clawhub",
+                    identifier=slug,
+                    trust_level="community",
+                    tags=tags,
+                ))
+
+            cursor = data.get("nextCursor") if isinstance(data, dict) else None
+            if not isinstance(cursor, str) or not cursor:
+                break
+
+        _write_index_cache(cache_key, [_skill_meta_to_dict(s) for s in results])
+        return results
+
     def _get_json(self, url: str, timeout: int = 20) -> Optional[Any]:
         try:
             resp = httpx.get(url, timeout=timeout)

tools/transcription_tools.py

@@ -93,6 +93,18 @@ def _load_stt_config() -> dict:
         return {}
 
 
+def is_stt_enabled(stt_config: Optional[dict] = None) -> bool:
+    """Return whether STT is enabled in config."""
+    if stt_config is None:
+        stt_config = _load_stt_config()
+    enabled = stt_config.get("enabled", True)
+    if isinstance(enabled, str):
+        return enabled.strip().lower() in ("true", "1", "yes", "on")
+    if enabled is None:
+        return True
+    return bool(enabled)
+
+
 def _get_provider(stt_config: dict) -> str:
     """Determine which STT provider to use.
 
@@ -101,6 +113,9 @@ def _get_provider(stt_config: dict) -> str:
       2. Auto-detect: local > groq (free) > openai (paid)
       3. Disabled (returns "none")
     """
+    if not is_stt_enabled(stt_config):
+        return "none"
+
     provider = stt_config.get("provider", DEFAULT_PROVIDER)
 
     if provider == "local":
@@ -334,6 +349,13 @@ def transcribe_audio(file_path: str, model: Optional[str] = None) -> Dict[str, A
 
     # Load config and determine provider
     stt_config = _load_stt_config()
+    if not is_stt_enabled(stt_config):
+        return {
+            "success": False,
+            "transcript": "",
+            "error": "STT is disabled in config.yaml (stt.enabled: false).",
+        }
+
     provider = _get_provider(stt_config)
 
     if provider == "local":

tools/vision_tools.py

@@ -3,7 +3,8 @@
 Vision Tools Module
 
 This module provides vision analysis tools that work with image URLs.
-Uses Gemini 3 Flash Preview via OpenRouter API for intelligent image understanding.
+Uses the centralized auxiliary vision router, which can select OpenRouter,
+Nous, Codex, native Anthropic, or a custom OpenAI-compatible endpoint.
 
 Available tools:
 - vision_analyze_tool: Analyze images from URLs with custom prompts
@@ -377,16 +378,11 @@ async def vision_analyze_tool(
 
 
 def check_vision_requirements() -> bool:
-    """Check if an auxiliary vision model is available."""
+    """Check if the configured runtime vision path can resolve a client."""
     try:
-        from agent.auxiliary_client import resolve_provider_client
-        client, _ = resolve_provider_client("openrouter")
-        if client is not None:
-            return True
-        client, _ = resolve_provider_client("nous")
-        if client is not None:
-            return True
-        client, _ = resolve_provider_client("custom")
+        from agent.auxiliary_client import resolve_vision_provider_client
+
+        _provider, client, _model = resolve_vision_provider_client()
         return client is not None
     except Exception:
         return False
@@ -414,7 +410,7 @@ if __name__ == "__main__":
     
     if not api_available:
         print("❌ No auxiliary vision model available")
-        print("Set OPENROUTER_API_KEY or configure Nous Portal to enable vision tools.")
+        print("Configure a supported multimodal backend (OpenRouter, Nous, Codex, Anthropic, or a custom OpenAI-compatible endpoint).")
         exit(1)
     else:
         print("✅ Vision model available")

tools/voice_mode.py

@@ -703,10 +703,11 @@ def check_voice_requirements() -> Dict[str, Any]:
         ``missing_packages``, and ``details``.
     """
     # Determine STT provider availability
-    from tools.transcription_tools import _get_provider, _load_stt_config, _HAS_FASTER_WHISPER
+    from tools.transcription_tools import _get_provider, _load_stt_config, is_stt_enabled, _HAS_FASTER_WHISPER
     stt_config = _load_stt_config()
+    stt_enabled = is_stt_enabled(stt_config)
     stt_provider = _get_provider(stt_config)
-    stt_available = stt_provider != "none"
+    stt_available = stt_enabled and stt_provider != "none"
 
     missing: List[str] = []
     has_audio = _audio_available()
@@ -725,7 +726,9 @@ def check_voice_requirements() -> Dict[str, Any]:
     else:
         details_parts.append("Audio capture: MISSING (pip install sounddevice numpy)")
 
-    if stt_provider == "local":
+    if not stt_enabled:
+        details_parts.append("STT provider: DISABLED in config (stt.enabled: false)")
+    elif stt_provider == "local":
         details_parts.append("STT provider: OK (local faster-whisper)")
     elif stt_provider == "groq":
         details_parts.append("STT provider: OK (Groq)")