test: resolve auxiliary client merge conflict

agent/auxiliary_client.py

@@ -30,6 +30,10 @@ Default "auto" follows the chains above.
 Per-task model overrides (e.g. AUXILIARY_VISION_MODEL,
 AUXILIARY_WEB_EXTRACT_MODEL) let callers use a different model slug
 than the provider's default.
+
+Per-task direct endpoint overrides (e.g. AUXILIARY_VISION_BASE_URL,
+AUXILIARY_VISION_API_KEY) let callers route a specific auxiliary task to a
+custom OpenAI-compatible endpoint without touching the main model settings.
 """
 
 import json
@@ -530,6 +534,17 @@ def _get_auxiliary_provider(task: str = "") -> str:
     return "auto"
 
 
+def _get_auxiliary_env_override(task: str, suffix: str) -> Optional[str]:
+    """Read an auxiliary env override from AUXILIARY_* or CONTEXT_* prefixes."""
+    if not task:
+        return None
+    for prefix in ("AUXILIARY_", "CONTEXT_"):
+        val = os.getenv(f"{prefix}{task.upper()}_{suffix}", "").strip()
+        if val:
+            return val
+    return None
+
+
 def _try_openrouter() -> Tuple[Optional[OpenAI], Optional[str]]:
     or_key = os.getenv("OPENROUTER_API_KEY")
     if not or_key:
@@ -577,9 +592,44 @@ def _read_main_model() -> str:
     return ""
 
 
+def _resolve_custom_runtime() -> Tuple[Optional[str], Optional[str]]:
+    """Resolve the active custom/main endpoint the same way the main CLI does.
+
+    This covers both env-driven OPENAI_BASE_URL setups and config-saved custom
+    endpoints where the base URL lives in config.yaml instead of the live
+    environment.
+    """
+    try:
+        from hermes_cli.runtime_provider import resolve_runtime_provider
+
+        runtime = resolve_runtime_provider(requested="custom")
+    except Exception as exc:
+        logger.debug("Auxiliary client: custom runtime resolution failed: %s", exc)
+        return None, None
+
+    custom_base = runtime.get("base_url")
+    custom_key = runtime.get("api_key")
+    if not isinstance(custom_base, str) or not custom_base.strip():
+        return None, None
+    if not isinstance(custom_key, str) or not custom_key.strip():
+        return None, None
+
+    custom_base = custom_base.strip().rstrip("/")
+    if "openrouter.ai" in custom_base.lower():
+        # requested='custom' falls back to OpenRouter when no custom endpoint is
+        # configured. Treat that as "no custom endpoint" for auxiliary routing.
+        return None, None
+
+    return custom_base, custom_key.strip()
+
+
+def _current_custom_base_url() -> str:
+    custom_base, _ = _resolve_custom_runtime()
+    return custom_base or ""
+
+
 def _try_custom_endpoint() -> Tuple[Optional[OpenAI], Optional[str]]:
-    custom_base = os.getenv("OPENAI_BASE_URL")
-    custom_key = os.getenv("OPENAI_API_KEY")
+    custom_base, custom_key = _resolve_custom_runtime()
     if not custom_base or not custom_key:
         return None, None
     model = _read_main_model() or "gpt-4o-mini"
@@ -694,6 +744,8 @@ def resolve_provider_client(
     model: str = None,
     async_mode: bool = False,
     raw_codex: bool = False,
+    explicit_base_url: str = None,
+    explicit_api_key: str = None,
 ) -> Tuple[Optional[Any], Optional[str]]:
     """Central router: given a provider name and optional model, return a
     configured client with the correct auth, base URL, and API format.
@@ -715,6 +767,8 @@ def resolve_provider_client(
             instead of wrapping in CodexAuxiliaryClient.  Use this when
             the caller needs direct access to responses.stream() (e.g.,
             the main agent loop).
+        explicit_base_url: Optional direct OpenAI-compatible endpoint.
+        explicit_api_key: Optional API key paired with explicit_base_url.
 
     Returns:
         (client, resolved_model) or (None, None) if auth is unavailable.
@@ -791,6 +845,22 @@ def resolve_provider_client(
 
     # ── Custom endpoint (OPENAI_BASE_URL + OPENAI_API_KEY) ───────────
     if provider == "custom":
+        if explicit_base_url:
+            custom_base = explicit_base_url.strip()
+            custom_key = (
+                (explicit_api_key or "").strip()
+                or os.getenv("OPENAI_API_KEY", "").strip()
+            )
+            if not custom_base or not custom_key:
+                logger.warning(
+                    "resolve_provider_client: explicit custom endpoint requested "
+                    "but no API key was found (set explicit_api_key or OPENAI_API_KEY)"
+                )
+                return None, None
+            final_model = model or _read_main_model() or "gpt-4o-mini"
+            client = OpenAI(api_key=custom_key, base_url=custom_base)
+            return (_to_async_client(client, final_model) if async_mode
+                    else (client, final_model))
         # Try custom first, then codex, then API-key providers
         for try_fn in (_try_custom_endpoint, _try_codex,
                        _resolve_api_key_provider):
@@ -887,10 +957,13 @@ def get_text_auxiliary_client(task: str = "") -> Tuple[Optional[OpenAI], Optiona
     Callers may override the returned model with a per-task env var
     (e.g. CONTEXT_COMPRESSION_MODEL, AUXILIARY_WEB_EXTRACT_MODEL).
     """
-    forced = _get_auxiliary_provider(task)
-    if forced != "auto":
-        return resolve_provider_client(forced)
-    return resolve_provider_client("auto")
+    provider, model, base_url, api_key = _resolve_task_provider_model(task or None)
+    return resolve_provider_client(
+        provider,
+        model=model,
+        explicit_base_url=base_url,
+        explicit_api_key=api_key,
+    )
 
 
 def get_async_text_auxiliary_client(task: str = ""):
@@ -900,10 +973,14 @@ def get_async_text_auxiliary_client(task: str = ""):
     (AsyncCodexAuxiliaryClient, model) which wraps the Responses API.
     Returns (None, None) when no provider is available.
     """
-    forced = _get_auxiliary_provider(task)
-    if forced != "auto":
-        return resolve_provider_client(forced, async_mode=True)
-    return resolve_provider_client("auto", async_mode=True)
+    provider, model, base_url, api_key = _resolve_task_provider_model(task or None)
+    return resolve_provider_client(
+        provider,
+        model=model,
+        async_mode=True,
+        explicit_base_url=base_url,
+        explicit_api_key=api_key,
+    )
 
 
 _VISION_AUTO_PROVIDER_ORDER = (
@@ -979,26 +1056,43 @@ def resolve_vision_provider_client(
     provider: Optional[str] = None,
     model: Optional[str] = None,
     *,
+    base_url: Optional[str] = None,
+    api_key: Optional[str] = None,
     async_mode: bool = False,
 ) -> Tuple[Optional[str], Optional[Any], Optional[str]]:
     """Resolve the client actually used for vision tasks.
 
-    Explicit provider overrides still use the generic provider router for
-    non-standard backends, so users can intentionally force experimental
-    providers. Auto mode stays conservative and only tries vision backends
-    known to work today.
+    Direct endpoint overrides take precedence over provider selection. Explicit
+    provider overrides still use the generic provider router for non-standard
+    backends, so users can intentionally force experimental providers. Auto mode
+    stays conservative and only tries vision backends known to work today.
     """
-    requested = _normalize_vision_provider(provider or _get_auxiliary_provider("vision"))
+    requested, resolved_model, resolved_base_url, resolved_api_key = _resolve_task_provider_model(
+        "vision", provider, model, base_url, api_key
+    )
+    requested = _normalize_vision_provider(requested)
 
     def _finalize(resolved_provider: str, sync_client: Any, default_model: Optional[str]):
         if sync_client is None:
             return resolved_provider, None, None
-        final_model = model or default_model
+        final_model = resolved_model or default_model
         if async_mode:
             async_client, async_model = _to_async_client(sync_client, final_model)
             return resolved_provider, async_client, async_model
         return resolved_provider, sync_client, final_model
 
+    if resolved_base_url:
+        client, final_model = resolve_provider_client(
+            "custom",
+            model=resolved_model,
+            async_mode=async_mode,
+            explicit_base_url=resolved_base_url,
+            explicit_api_key=resolved_api_key,
+        )
+        if client is None:
+            return "custom", None, None
+        return "custom", client, final_model
+
     if requested == "auto":
         for candidate in get_available_vision_backends():
             sync_client, default_model = _resolve_strict_vision_backend(candidate)
@@ -1011,7 +1105,7 @@ def resolve_vision_provider_client(
         sync_client, default_model = _resolve_strict_vision_backend(requested)
         return _finalize(requested, sync_client, default_model)
 
-    client, final_model = _get_cached_client(requested, model, async_mode)
+    client, final_model = _get_cached_client(requested, resolved_model, async_mode)
     if client is None:
         return requested, None, None
     return requested, client, final_model
@@ -1046,7 +1140,7 @@ def auxiliary_max_tokens_param(value: int) -> dict:
     The Codex adapter translates max_tokens internally, so we use max_tokens
     for it as well.
     """
-    custom_base = os.getenv("OPENAI_BASE_URL", "")
+    custom_base = _current_custom_base_url()
     or_key = os.getenv("OPENROUTER_API_KEY")
     # Only use max_completion_tokens for direct OpenAI custom endpoints
     if (not or_key
@@ -1068,19 +1162,29 @@ def auxiliary_max_tokens_param(value: int) -> dict:
 # Every auxiliary LLM consumer should use these instead of manually
 # constructing clients and calling .chat.completions.create().
 
-# Client cache: (provider, async_mode) -> (client, default_model)
+# Client cache: (provider, async_mode, base_url, api_key) -> (client, default_model)
 _client_cache: Dict[tuple, tuple] = {}
 
 
 def _get_cached_client(
-    provider: str, model: str = None, async_mode: bool = False,
+    provider: str,
+    model: str = None,
+    async_mode: bool = False,
+    base_url: str = None,
+    api_key: str = None,
 ) -> Tuple[Optional[Any], Optional[str]]:
     """Get or create a cached client for the given provider."""
-    cache_key = (provider, async_mode)
+    cache_key = (provider, async_mode, base_url or "", api_key or "")
     if cache_key in _client_cache:
         cached_client, cached_default = _client_cache[cache_key]
         return cached_client, model or cached_default
-    client, default_model = resolve_provider_client(provider, model, async_mode)
+    client, default_model = resolve_provider_client(
+        provider,
+        model,
+        async_mode,
+        explicit_base_url=base_url,
+        explicit_api_key=api_key,
+    )
     if client is not None:
         _client_cache[cache_key] = (client, default_model)
     return client, model or default_model
@@ -1090,57 +1194,75 @@ def _resolve_task_provider_model(
     task: str = None,
     provider: str = None,
     model: str = None,
-) -> Tuple[str, Optional[str]]:
+    base_url: str = None,
+    api_key: str = None,
+) -> Tuple[str, Optional[str], Optional[str], Optional[str]]:
     """Determine provider + model for a call.
 
     Priority:
-      1. Explicit provider/model args (always win)
-      2. Env var overrides (AUXILIARY_{TASK}_PROVIDER, etc.)
-      3. Config file (auxiliary.{task}.provider/model or compression.*)
+      1. Explicit provider/model/base_url/api_key args (always win)
+      2. Env var overrides (AUXILIARY_{TASK}_*, CONTEXT_{TASK}_*)
+      3. Config file (auxiliary.{task}.* or compression.*)
       4. "auto" (full auto-detection chain)
 
-    Returns (provider, model) where model may be None (use provider default).
+    Returns (provider, model, base_url, api_key) where model may be None
+    (use provider default). When base_url is set, provider is forced to
+    "custom" and the task uses that direct endpoint.
     """
-    if provider:
-        return provider, model
+    config = {}
+    cfg_provider = None
+    cfg_model = None
+    cfg_base_url = None
+    cfg_api_key = None
 
     if task:
-        # Check env var overrides first
-        env_provider = _get_auxiliary_provider(task)
-        if env_provider != "auto":
-            # Check for env var model override too
-            env_model = None
-            for prefix in ("AUXILIARY_", "CONTEXT_"):
-                val = os.getenv(f"{prefix}{task.upper()}_MODEL", "").strip()
-                if val:
-                    env_model = val
-                    break
-            return env_provider, model or env_model
-
-        # Read from config file
         try:
             from hermes_cli.config import load_config
             config = load_config()
         except ImportError:
-            return "auto", model
+            config = {}
 
-        # Check auxiliary.{task} section
-        aux = config.get("auxiliary", {})
-        task_config = aux.get(task, {})
-        cfg_provider = task_config.get("provider", "").strip() or None
-        cfg_model = task_config.get("model", "").strip() or None
+        aux = config.get("auxiliary", {}) if isinstance(config, dict) else {}
+        task_config = aux.get(task, {}) if isinstance(aux, dict) else {}
+        if not isinstance(task_config, dict):
+            task_config = {}
+        cfg_provider = str(task_config.get("provider", "")).strip() or None
+        cfg_model = str(task_config.get("model", "")).strip() or None
+        cfg_base_url = str(task_config.get("base_url", "")).strip() or None
+        cfg_api_key = str(task_config.get("api_key", "")).strip() or None
 
         # Backwards compat: compression section has its own keys
         if task == "compression" and not cfg_provider:
-            comp = config.get("compression", {})
-            cfg_provider = comp.get("summary_provider", "").strip() or None
-            cfg_model = cfg_model or comp.get("summary_model", "").strip() or None
+            comp = config.get("compression", {}) if isinstance(config, dict) else {}
+            if isinstance(comp, dict):
+                cfg_provider = comp.get("summary_provider", "").strip() or None
+                cfg_model = cfg_model or comp.get("summary_model", "").strip() or None
 
+    env_model = _get_auxiliary_env_override(task, "MODEL") if task else None
+    resolved_model = model or env_model or cfg_model
+
+    if base_url:
+        return "custom", resolved_model, base_url, api_key
+    if provider:
+        return provider, resolved_model, base_url, api_key
+
+    if task:
+        env_base_url = _get_auxiliary_env_override(task, "BASE_URL")
+        env_api_key = _get_auxiliary_env_override(task, "API_KEY")
+        if env_base_url:
+            return "custom", resolved_model, env_base_url, env_api_key or cfg_api_key
+
+        env_provider = _get_auxiliary_provider(task)
+        if env_provider != "auto":
+            return env_provider, resolved_model, None, None
+
+        if cfg_base_url:
+            return "custom", resolved_model, cfg_base_url, cfg_api_key
         if cfg_provider and cfg_provider != "auto":
-            return cfg_provider, model or cfg_model
-        return "auto", model or cfg_model
+            return cfg_provider, resolved_model, None, None
+        return "auto", resolved_model, None, None
 
-    return "auto", model
+    return "auto", resolved_model, None, None
 
 
 def _build_call_kwargs(
@@ -1152,6 +1274,7 @@ def _build_call_kwargs(
     tools: Optional[list] = None,
     timeout: float = 30.0,
     extra_body: Optional[dict] = None,
+    base_url: Optional[str] = None,
 ) -> dict:
     """Build kwargs for .chat.completions.create() with model/provider adjustments."""
     kwargs: Dict[str, Any] = {
@@ -1167,7 +1290,7 @@ def _build_call_kwargs(
         # Codex adapter handles max_tokens internally; OpenRouter/Nous use max_tokens.
         # Direct OpenAI api.openai.com with newer models needs max_completion_tokens.
         if provider == "custom":
-            custom_base = os.getenv("OPENAI_BASE_URL", "")
+            custom_base = base_url or _current_custom_base_url()
             if "api.openai.com" in custom_base.lower():
                 kwargs["max_completion_tokens"] = max_tokens
             else:
@@ -1193,6 +1316,8 @@ def call_llm(
     *,
     provider: str = None,
     model: str = None,
+    base_url: str = None,
+    api_key: str = None,
     messages: list,
     temperature: float = None,
     max_tokens: int = None,
@@ -1224,16 +1349,18 @@ def call_llm(
     Raises:
         RuntimeError: If no provider is configured.
     """
-    resolved_provider, resolved_model = _resolve_task_provider_model(
-        task, provider, model)
+    resolved_provider, resolved_model, resolved_base_url, resolved_api_key = _resolve_task_provider_model(
+        task, provider, model, base_url, api_key)
 
     if task == "vision":
         effective_provider, client, final_model = resolve_vision_provider_client(
-            provider=resolved_provider,
-            model=resolved_model,
+            provider=provider,
+            model=model,
+            base_url=base_url,
+            api_key=api_key,
             async_mode=False,
         )
-        if client is None and resolved_provider != "auto":
+        if client is None and resolved_provider != "auto" and not resolved_base_url:
             logger.warning(
                 "Vision provider %s unavailable, falling back to auto vision backends",
                 resolved_provider,
@@ -1250,10 +1377,15 @@ def call_llm(
             )
         resolved_provider = effective_provider or resolved_provider
     else:
-        client, final_model = _get_cached_client(resolved_provider, resolved_model)
+        client, final_model = _get_cached_client(
+            resolved_provider,
+            resolved_model,
+            base_url=resolved_base_url,
+            api_key=resolved_api_key,
+        )
         if client is None:
             # Fallback: try openrouter
-            if resolved_provider != "openrouter":
+            if resolved_provider != "openrouter" and not resolved_base_url:
                 logger.warning("Provider %s unavailable, falling back to openrouter",
                                resolved_provider)
                 client, final_model = _get_cached_client(
@@ -1266,7 +1398,8 @@ def call_llm(
     kwargs = _build_call_kwargs(
         resolved_provider, final_model, messages,
         temperature=temperature, max_tokens=max_tokens,
-        tools=tools, timeout=timeout, extra_body=extra_body)
+        tools=tools, timeout=timeout, extra_body=extra_body,
+        base_url=resolved_base_url)
 
     # Handle max_tokens vs max_completion_tokens retry
     try:
@@ -1285,6 +1418,8 @@ async def async_call_llm(
     *,
     provider: str = None,
     model: str = None,
+    base_url: str = None,
+    api_key: str = None,
     messages: list,
     temperature: float = None,
     max_tokens: int = None,
@@ -1296,16 +1431,18 @@ async def async_call_llm(
 
     Same as call_llm() but async. See call_llm() for full documentation.
     """
-    resolved_provider, resolved_model = _resolve_task_provider_model(
-        task, provider, model)
+    resolved_provider, resolved_model, resolved_base_url, resolved_api_key = _resolve_task_provider_model(
+        task, provider, model, base_url, api_key)
 
     if task == "vision":
         effective_provider, client, final_model = resolve_vision_provider_client(
-            provider=resolved_provider,
-            model=resolved_model,
+            provider=provider,
+            model=model,
+            base_url=base_url,
+            api_key=api_key,
             async_mode=True,
         )
-        if client is None and resolved_provider != "auto":
+        if client is None and resolved_provider != "auto" and not resolved_base_url:
             logger.warning(
                 "Vision provider %s unavailable, falling back to auto vision backends",
                 resolved_provider,
@@ -1323,9 +1460,14 @@ async def async_call_llm(
         resolved_provider = effective_provider or resolved_provider
     else:
         client, final_model = _get_cached_client(
-            resolved_provider, resolved_model, async_mode=True)
+            resolved_provider,
+            resolved_model,
+            async_mode=True,
+            base_url=resolved_base_url,
+            api_key=resolved_api_key,
+        )
         if client is None:
-            if resolved_provider != "openrouter":
+            if resolved_provider != "openrouter" and not resolved_base_url:
                 logger.warning("Provider %s unavailable, falling back to openrouter",
                                resolved_provider)
                 client, final_model = _get_cached_client(
@@ -1339,7 +1481,8 @@ async def async_call_llm(
     kwargs = _build_call_kwargs(
         resolved_provider, final_model, messages,
         temperature=temperature, max_tokens=max_tokens,
-        tools=tools, timeout=timeout, extra_body=extra_body)
+        tools=tools, timeout=timeout, extra_body=extra_body,
+        base_url=resolved_base_url)
 
     try:
         return await client.chat.completions.create(**kwargs)

agent/display.py

@@ -80,7 +80,7 @@ def build_tool_preview(tool_name: str, args: dict, max_len: int = 40) -> str | N
         "image_generate": "prompt", "text_to_speech": "text",
         "vision_analyze": "question", "mixture_of_agents": "user_prompt",
         "skill_view": "name", "skills_list": "category",
-        "schedule_cronjob": "name",
+        "cronjob": "action",
         "execute_code": "code", "delegate_task": "goal",
         "clarify": "question", "skill_manage": "name",
     }
@@ -513,12 +513,15 @@ def get_cute_tool_message(
         return _wrap(f"┊ 🧠 reason    {_trunc(args.get('user_prompt', ''), 30)}  {dur}")
     if tool_name == "send_message":
         return _wrap(f"┊ 📨 send      {args.get('target', '?')}: \"{_trunc(args.get('message', ''), 25)}\"  {dur}")
-    if tool_name == "schedule_cronjob":
-        return _wrap(f"┊ ⏰ schedule  {_trunc(args.get('name', args.get('prompt', 'task')), 30)}  {dur}")
-    if tool_name == "list_cronjobs":
-        return _wrap(f"┊ ⏰ jobs      listing  {dur}")
-    if tool_name == "remove_cronjob":
-        return _wrap(f"┊ ⏰ remove    job {args.get('job_id', '?')}  {dur}")
+    if tool_name == "cronjob":
+        action = args.get("action", "?")
+        if action == "create":
+            skills = args.get("skills") or ([] if not args.get("skill") else [args.get("skill")])
+            label = args.get("name") or (skills[0] if skills else None) or args.get("prompt", "task")
+            return _wrap(f"┊ ⏰ cron      create {_trunc(label, 24)}  {dur}")
+        if action == "list":
+            return _wrap(f"┊ ⏰ cron      listing  {dur}")
+        return _wrap(f"┊ ⏰ cron      {action} {args.get('job_id', '')}  {dur}")
     if tool_name.startswith("rl_"):
         rl = {
             "rl_list_environments": "list envs", "rl_select_environment": f"select {args.get('name', '')}",

agent/skill_commands.py

@@ -1,17 +1,151 @@
-"""Skill slash commands — scan installed skills and build invocation messages.
+"""Shared slash command helpers for skills and built-in prompt-style modes.
 
 Shared between CLI (cli.py) and gateway (gateway/run.py) so both surfaces
-can invoke skills via /skill-name commands.
+can invoke skills via /skill-name commands and prompt-only built-ins like
+/plan.
 """
 
 import json
 import logging
+import re
+from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, Optional
 
 logger = logging.getLogger(__name__)
 
 _skill_commands: Dict[str, Dict[str, Any]] = {}
+_PLAN_SLUG_RE = re.compile(r"[^a-z0-9]+")
+
+
+def build_plan_path(
+    user_instruction: str = "",
+    *,
+    now: datetime | None = None,
+) -> Path:
+    """Return the default workspace-relative markdown path for a /plan invocation.
+
+    Relative paths are intentional: file tools are task/backend-aware and resolve
+    them against the active working directory for local, docker, ssh, modal,
+    daytona, and similar terminal backends. That keeps the plan with the active
+    workspace instead of the Hermes host's global home directory.
+    """
+    slug_source = (user_instruction or "").strip().splitlines()[0] if user_instruction else ""
+    slug = _PLAN_SLUG_RE.sub("-", slug_source.lower()).strip("-")
+    if slug:
+        slug = "-".join(part for part in slug.split("-")[:8] if part)[:48].strip("-")
+    slug = slug or "conversation-plan"
+    timestamp = (now or datetime.now()).strftime("%Y-%m-%d_%H%M%S")
+    return Path(".hermes") / "plans" / f"{timestamp}-{slug}.md"
+
+
+def _load_skill_payload(skill_identifier: str, task_id: str | None = None) -> tuple[dict[str, Any], Path | None, str] | None:
+    """Load a skill by name/path and return (loaded_payload, skill_dir, display_name)."""
+    raw_identifier = (skill_identifier or "").strip()
+    if not raw_identifier:
+        return None
+
+    try:
+        from tools.skills_tool import SKILLS_DIR, skill_view
+
+        identifier_path = Path(raw_identifier).expanduser()
+        if identifier_path.is_absolute():
+            try:
+                normalized = str(identifier_path.resolve().relative_to(SKILLS_DIR.resolve()))
+            except Exception:
+                normalized = raw_identifier
+        else:
+            normalized = raw_identifier.lstrip("/")
+
+        loaded_skill = json.loads(skill_view(normalized, task_id=task_id))
+    except Exception:
+        return None
+
+    if not loaded_skill.get("success"):
+        return None
+
+    skill_name = str(loaded_skill.get("name") or normalized)
+    skill_path = str(loaded_skill.get("path") or "")
+    skill_dir = None
+    if skill_path:
+        try:
+            skill_dir = SKILLS_DIR / Path(skill_path).parent
+        except Exception:
+            skill_dir = None
+
+    return loaded_skill, skill_dir, skill_name
+
+
+def _build_skill_message(
+    loaded_skill: dict[str, Any],
+    skill_dir: Path | None,
+    activation_note: str,
+    user_instruction: str = "",
+    runtime_note: str = "",
+) -> str:
+    """Format a loaded skill into a user/system message payload."""
+    from tools.skills_tool import SKILLS_DIR
+
+    content = str(loaded_skill.get("content") or "")
+
+    parts = [activation_note, "", content.strip()]
+
+    if loaded_skill.get("setup_skipped"):
+        parts.extend(
+            [
+                "",
+                "[Skill setup note: Required environment setup was skipped. Continue loading the skill and explain any reduced functionality if it matters.]",
+            ]
+        )
+    elif loaded_skill.get("gateway_setup_hint"):
+        parts.extend(
+            [
+                "",
+                f"[Skill setup note: {loaded_skill['gateway_setup_hint']}]",
+            ]
+        )
+    elif loaded_skill.get("setup_needed") and loaded_skill.get("setup_note"):
+        parts.extend(
+            [
+                "",
+                f"[Skill setup note: {loaded_skill['setup_note']}]",
+            ]
+        )
+
+    supporting = []
+    linked_files = loaded_skill.get("linked_files") or {}
+    for entries in linked_files.values():
+        if isinstance(entries, list):
+            supporting.extend(entries)
+
+    if not supporting and skill_dir:
+        for subdir in ("references", "templates", "scripts", "assets"):
+            subdir_path = skill_dir / subdir
+            if subdir_path.exists():
+                for f in sorted(subdir_path.rglob("*")):
+                    if f.is_file():
+                        rel = str(f.relative_to(skill_dir))
+                        supporting.append(rel)
+
+    if supporting and skill_dir:
+        skill_view_target = str(skill_dir.relative_to(SKILLS_DIR))
+        parts.append("")
+        parts.append("[This skill has supporting files you can load with the skill_view tool:]")
+        for sf in supporting:
+            parts.append(f"- {sf}")
+        parts.append(
+            f'\nTo view any of these, use: skill_view(name="{skill_view_target}", file_path="<path>")'
+        )
+
+    if user_instruction:
+        parts.append("")
+        parts.append(f"The user has provided the following instruction alongside the skill invocation: {user_instruction}")
+
+    if runtime_note:
+        parts.append("")
+        parts.append(f"[Runtime note: {runtime_note}]")
+
+    return "\n".join(parts)
 
 
 def scan_skill_commands() -> Dict[str, Dict[str, Any]]:
@@ -68,6 +202,7 @@ def build_skill_invocation_message(
     cmd_key: str,
     user_instruction: str = "",
     task_id: str | None = None,
+    runtime_note: str = "",
 ) -> Optional[str]:
     """Build the user message content for a skill slash command invocation.
 
@@ -83,77 +218,61 @@ def build_skill_invocation_message(
     if not skill_info:
         return None
 
-    skill_name = skill_info["name"]
-    skill_path = skill_info["skill_dir"]
+    loaded = _load_skill_payload(skill_info["skill_dir"], task_id=task_id)
+    if not loaded:
+        return f"[Failed to load skill: {skill_info['name']}]"
 
-    try:
-        from tools.skills_tool import SKILLS_DIR, skill_view
+    loaded_skill, skill_dir, skill_name = loaded
+    activation_note = (
+        f'[SYSTEM: The user has invoked the "{skill_name}" skill, indicating they want '
+        "you to follow its instructions. The full skill content is loaded below.]"
+    )
+    return _build_skill_message(
+        loaded_skill,
+        skill_dir,
+        activation_note,
+        user_instruction=user_instruction,
+        runtime_note=runtime_note,
+    )
 
-        loaded_skill = json.loads(skill_view(skill_path, task_id=task_id))
-    except Exception:
-        return f"[Failed to load skill: {skill_name}]"
 
-    if not loaded_skill.get("success"):
-        return f"[Failed to load skill: {skill_name}]"
+def build_preloaded_skills_prompt(
+    skill_identifiers: list[str],
+    task_id: str | None = None,
+) -> tuple[str, list[str], list[str]]:
+    """Load one or more skills for session-wide CLI preloading.
 
-    content = str(loaded_skill.get("content") or "")
-    skill_dir = Path(skill_info["skill_dir"])
+    Returns (prompt_text, loaded_skill_names, missing_identifiers).
+    """
+    prompt_parts: list[str] = []
+    loaded_names: list[str] = []
+    missing: list[str] = []
 
-    parts = [
-        f'[SYSTEM: The user has invoked the "{skill_name}" skill, indicating they want you to follow its instructions. The full skill content is loaded below.]',
-        "",
-        content.strip(),
-    ]
+    seen: set[str] = set()
+    for raw_identifier in skill_identifiers:
+        identifier = (raw_identifier or "").strip()
+        if not identifier or identifier in seen:
+            continue
+        seen.add(identifier)
 
-    if loaded_skill.get("setup_skipped"):
-        parts.extend(
-            [
-                "",
-                "[Skill setup note: Required environment setup was skipped. Continue loading the skill and explain any reduced functionality if it matters.]",
-            ]
+        loaded = _load_skill_payload(identifier, task_id=task_id)
+        if not loaded:
+            missing.append(identifier)
+            continue
+
+        loaded_skill, skill_dir, skill_name = loaded
+        activation_note = (
+            f'[SYSTEM: The user launched this CLI session with the "{skill_name}" skill '
+            "preloaded. Treat its instructions as active guidance for the duration of this "
+            "session unless the user overrides them.]"
         )
-    elif loaded_skill.get("gateway_setup_hint"):
-        parts.extend(
-            [
-                "",
-                f"[Skill setup note: {loaded_skill['gateway_setup_hint']}]",
-            ]
-        )
-    elif loaded_skill.get("setup_needed") and loaded_skill.get("setup_note"):
-        parts.extend(
-            [
-                "",
-                f"[Skill setup note: {loaded_skill['setup_note']}]",
-            ]
+        prompt_parts.append(
+            _build_skill_message(
+                loaded_skill,
+                skill_dir,
+                activation_note,
+            )
         )
+        loaded_names.append(skill_name)
 
-    supporting = []
-    linked_files = loaded_skill.get("linked_files") or {}
-    for entries in linked_files.values():
-        if isinstance(entries, list):
-            supporting.extend(entries)
-
-    if not supporting:
-        for subdir in ("references", "templates", "scripts", "assets"):
-            subdir_path = skill_dir / subdir
-            if subdir_path.exists():
-                for f in sorted(subdir_path.rglob("*")):
-                    if f.is_file():
-                        rel = str(f.relative_to(skill_dir))
-                        supporting.append(rel)
-
-    if supporting:
-        skill_view_target = str(Path(skill_path).relative_to(SKILLS_DIR))
-        parts.append("")
-        parts.append("[This skill has supporting files you can load with the skill_view tool:]")
-        for sf in supporting:
-            parts.append(f"- {sf}")
-        parts.append(
-            f'\nTo view any of these, use: skill_view(name="{skill_view_target}", file_path="<path>")'
-        )
-
-    if user_instruction:
-        parts.append("")
-        parts.append(f"The user has provided the following instruction alongside the skill invocation: {user_instruction}")
-
-    return "\n".join(parts)
+    return "\n\n".join(prompt_parts), loaded_names, missing