fix: route configured LiteLLM models by engine ownership (#714)

Previously the ws_bridge send loop blocked forever on queue.get(),
never learning the client left. When the server service stopped,
uvicorn waited for open WebSocket tasks until systemd SIGKILLed
after TimeoutStopSec=90s. Now each iteration races recv+send; a
completed receive means the client disconnected => break the loop.

.github/clone-stats/badge.json

@@ -1,7 +1,7 @@
 {
   "schemaVersion": 1,
   "label": "Git Clones",
-  "message": "176,576",
+  "message": "185,599",
   "color": "green",
   "namedLogo": "git"
 }

.github/clone-stats/clone-data.json

@@ -1,6 +1,6 @@
 {
-  "total_clones": 176576,
-  "last_updated": "2026-07-29T08:35:42Z",
+  "total_clones": 185599,
+  "last_updated": "2026-08-10T07:26:44Z",
   "daily": {
     "2026-03-27": 2189,
     "2026-03-28": 1874,
@@ -125,6 +125,18 @@
     "2026-07-25": 928,
     "2026-07-26": 740,
     "2026-07-27": 799,
-    "2026-07-28": 665
+    "2026-07-28": 665,
+    "2026-07-29": 745,
+    "2026-07-30": 591,
+    "2026-07-31": 783,
+    "2026-08-01": 567,
+    "2026-08-02": 1248,
+    "2026-08-03": 724,
+    "2026-08-04": 708,
+    "2026-08-05": 647,
+    "2026-08-06": 604,
+    "2026-08-07": 624,
+    "2026-08-08": 706,
+    "2026-08-09": 1076
   }
 }

frontend/src/components/Chat/InputArea.tsx

@@ -466,7 +466,10 @@ export function InputArea() {
       }
       const totalMs = Date.now() - startTime;
       const _CLOUD_PREFIXES = ['gpt-', 'o1-', 'o3-', 'o4-', 'claude-', 'gemini-', 'openrouter/', 'MiniMax-', 'chatgpt-'];
-      const engineLabel = _CLOUD_PREFIXES.some(p => selectedModel.startsWith(p)) ? 'cloud' : 'ollama';
+      const selectedOwner = useAppStore.getState().models.find((m) => m.id === selectedModel)?.owned_by;
+      const engineLabel = selectedOwner === 'litellm'
+        ? 'litellm'
+        : _CLOUD_PREFIXES.some(p => selectedModel.startsWith(p)) ? 'cloud' : 'ollama';
       const telemetry: MessageTelemetry = {
         engine: engineLabel,
         model_id: selectedModel,

frontend/src/components/CommandPalette.tsx

@@ -143,7 +143,7 @@ export function CommandPalette() {
     }
   }, [pullSuccess]);
 
-  const handleSelect = async (modelId: string) => {
+  const handleSelect = async (modelId: string, owner?: string) => {
     const previousModel = selectedModel;
     setSelectedModel(modelId);
     setCommandPaletteOpen(false);
@@ -153,7 +153,7 @@ export function CommandPalette() {
       setModelLoading(true);
       addLogEntry({ timestamp: Date.now(), level: 'info', category: 'model', message: `Switching to ${modelId}...` });
       try {
-        await preloadModel(modelId);
+        await preloadModel(modelId, owner);
         addLogEntry({ timestamp: Date.now(), level: 'info', category: 'model', message: `${modelId} loaded` });
       } catch (e: any) {
         addLogEntry({ timestamp: Date.now(), level: 'error', category: 'model', message: `Failed to load ${modelId}: ${e.message}` });
@@ -255,7 +255,8 @@ export function CommandPalette() {
       setSelectedIdx((i) => Math.max(i - 1, 0));
     } else if (e.key === 'Enter' && tab === 'installed' && filtered.length > 0) {
       e.preventDefault();
-      handleSelect((filtered[selectedIdx] as any).id);
+      const model = filtered[selectedIdx] as (typeof models)[number];
+      handleSelect(model.id, model.owned_by);
     }
   };
 
@@ -365,11 +366,15 @@ export function CommandPalette() {
                     onMouseEnter={() => setSelectedIdx(idx)}
                   >
                     <button
-                      onClick={() => handleSelect(model.id)}
+                      onClick={() => handleSelect(model.id, model.owned_by)}
                       className="flex items-center gap-3 flex-1 min-w-0 text-left cursor-pointer"
                       style={{ background: 'none', border: 'none', padding: 0 }}
                     >
-                      <Cpu size={16} style={{ color: isActive ? 'var(--color-accent)' : 'var(--color-text-tertiary)' }} />
+                      {model.owned_by === 'litellm' ? (
+                        <Cloud size={16} style={{ color: isActive ? 'var(--color-accent)' : 'var(--color-text-tertiary)' }} />
+                      ) : (
+                        <Cpu size={16} style={{ color: isActive ? 'var(--color-accent)' : 'var(--color-text-tertiary)' }} />
+                      )}
                       <div className="flex-1 min-w-0">
                         <div className="text-sm truncate" style={{ color: isActive ? 'var(--color-accent)' : 'var(--color-text)', fontWeight: isActive ? 500 : 400 }}>
                           {model.id}
@@ -381,17 +386,19 @@ export function CommandPalette() {
                         </span>
                       )}
                     </button>
-                    <button
-                      onClick={() => handleDelete(model.id)}
-                      disabled={isDeleting}
-                      className="p-1 rounded transition-colors cursor-pointer"
-                      style={{ color: 'var(--color-text-tertiary)', opacity: 0 }}
-                      title="Delete model"
-                      onMouseEnter={(e) => { e.currentTarget.style.opacity = '1'; e.currentTarget.style.color = 'var(--color-error)'; }}
-                      onMouseLeave={(e) => { e.currentTarget.style.opacity = '0'; e.currentTarget.style.color = 'var(--color-text-tertiary)'; }}
-                    >
-                      {isDeleting ? <Loader2 size={14} className="animate-spin" /> : <Trash2 size={14} />}
-                    </button>
+                    {model.owned_by !== 'litellm' && (
+                      <button
+                        onClick={() => handleDelete(model.id)}
+                        disabled={isDeleting}
+                        className="p-1 rounded transition-colors cursor-pointer"
+                        style={{ color: 'var(--color-text-tertiary)', opacity: 0 }}
+                        title="Delete model"
+                        onMouseEnter={(e) => { e.currentTarget.style.opacity = '1'; e.currentTarget.style.color = 'var(--color-error)'; }}
+                        onMouseLeave={(e) => { e.currentTarget.style.opacity = '0'; e.currentTarget.style.color = 'var(--color-text-tertiary)'; }}
+                      >
+                        {isDeleting ? <Loader2 size={14} className="animate-spin" /> : <Trash2 size={14} />}
+                      </button>
+                    )}
                   </div>
                 );
               })

frontend/src/lib/api.ts

@@ -218,9 +218,9 @@ export async function deleteModel(modelName: string): Promise<void> {
 
 const _CLOUD_PREFIXES = ['gpt-', 'o1-', 'o3-', 'o4-', 'claude-', 'gemini-', 'openrouter/'];
 
-export async function preloadModel(modelName: string): Promise<void> {
+export async function preloadModel(modelName: string, owner?: string): Promise<void> {
   // Cloud models don't need Ollama preloading
-  if (_CLOUD_PREFIXES.some(p => modelName.startsWith(p))) {
+  if (owner === 'litellm' || _CLOUD_PREFIXES.some(p => modelName.startsWith(p))) {
     return;
   }
   // Trigger Ollama to load the model into memory (empty prompt, no generation).

frontend/src/pages/AgentsPage.tsx

@@ -574,7 +574,7 @@ function ToolsPicker({
           </div>
           {/* Live description strip */}
           <div
-            className="flex items-center gap-2 px-2.5 py-1.5"
+            className="flex items-start gap-2 px-2.5 py-1.5"
             style={{
               borderTop: '1px solid var(--color-border)',
               background: 'var(--color-bg)',
@@ -608,10 +608,11 @@ function ToolsPicker({
               </span>
             )}
             <span
-              className="truncate"
+              className="min-w-0 whitespace-normal break-words"
               style={{
                 flex: 1,
                 color: 'var(--color-text-tertiary)',
+                lineHeight: 1.4,
               }}
             >
               {hovered ? `— ${hint}` : hint}

src/openjarvis/agents/proactive_agent.py

@@ -37,6 +37,7 @@ called from your app startup:
 from __future__ import annotations
 
 import json
+import logging
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Set
@@ -56,6 +57,15 @@ from openjarvis.tools.approval_store import (
 )
 from openjarvis.tools.proactive_tools import get_store
 
+logger = logging.getLogger(__name__)
+
+_PROACTIVE_CRON_PROMPT = (
+    "Run the proactive agent: collect overnight data, execute approved actions, "
+    "notify pending approvals."
+)
+_PROACTIVE_TASK_KEY = "proactive-daily"
+_PROACTIVE_TASK_KEY_FIELD = "openjarvis_task_key"
+
 _SYSTEM_PROMPT = """You are a proactive personal assistant agent. You have already collected
 data from the user's connected sources (email, messages, calendar). Your job is to:
 
@@ -252,14 +262,31 @@ def _build_notification_channel(channel_spec: str) -> Optional[Any]:
 
         if ChannelRegistry.contains(channel_type):
             channel_cls = ChannelRegistry.get(channel_type)
-            instance = channel_cls()
+            # Load credentials from config so the channel uses bot_token from
+            # config.toml rather than falling back to a bare env var.
             try:
-                instance.connect()
+                from openjarvis.core.config import load_config
+                from openjarvis.system._channel_kwargs import build_channel_kwargs
+
+                _cfg = load_config()
+                _kwargs = build_channel_kwargs(_cfg.channel, channel_type)
             except Exception:
-                pass
+                _kwargs = {}
+            instance = channel_cls(**_kwargs)
+            # Telegram.send() is self-contained, while connect() starts a
+            # getUpdates loop.  A second loop for the same bot token conflicts
+            # with the server's main listener.  Other channel implementations
+            # may initialize resources required by send() in connect(), so keep
+            # their established lifecycle intact.
+            if channel_type != "telegram":
+                instance.connect()
             return instance
     except Exception:
-        pass
+        logger.warning(
+            "Failed to build proactive notification channel %s",
+            channel_type,
+            exc_info=True,
+        )
 
     return None
 
@@ -299,6 +326,7 @@ class ProactiveAgent(ToolUsingAgent):
             self._notification_channel_id
         )
         self._notification_channel = notification_channel
+        self._notification_destination = self._notification_channel_id.partition(":")[2]
 
         from openjarvis.tools.channel_tools import ChannelSendTool
         from openjarvis.tools.digest_collect import DigestCollectTool
@@ -484,13 +512,13 @@ class ProactiveAgent(ToolUsingAgent):
         # --- Step 5: Build and send notification ---
         notification = self._build_notification(executed_results, pending_actions)
 
-        if notification and self._notification_channel_id:
+        if notification and self._notification_destination:
             send_call = ToolCall(
                 id="proactive-notify-1",
                 name="channel_send",
                 arguments=json.dumps(
                     {
-                        "channel": self._notification_channel_id,
+                        "channel": self._notification_destination,
                         "content": notification,
                     }
                 ),
@@ -592,15 +620,74 @@ def register_cron(
         hours_back = hours_back or 24
         timezone = timezone or "America/Los_Angeles"
 
+    metadata = {
+        "notification_channel_id": notification_channel_id,
+        "hours_back": hours_back,
+        "timezone": timezone,
+        _PROACTIVE_TASK_KEY_FIELD: _PROACTIVE_TASK_KEY,
+    }
+
+    # Match the stable key for tasks created by this version and the historical
+    # agent+prompt signature so existing installations are migrated on startup.
+    existing = [
+        task
+        for task in scheduler.list_tasks()
+        if task.status in {"active", "paused"}
+        and task.agent == "proactive"
+        and (
+            task.metadata.get(_PROACTIVE_TASK_KEY_FIELD) == _PROACTIVE_TASK_KEY
+            or (task.prompt == _PROACTIVE_CRON_PROMPT and task.schedule_type == "cron")
+        )
+    ]
+
+    # A scheduler pause is an explicit user choice and must survive restart.
+    # Keep one deterministically and remove any active or paused duplicates.
+    paused = [task for task in existing if task.status == "paused"]
+    if paused:
+        keep = min(paused, key=lambda task: task.id)
+        _cancel_proactive_duplicates(scheduler, existing, keep=keep)
+        return keep
+
+    matching = [
+        task
+        for task in existing
+        if task.prompt == _PROACTIVE_CRON_PROMPT
+        and task.schedule_type == "cron"
+        and task.schedule_value == cron_expr
+        and task.context_mode == "isolated"
+        and task.metadata == metadata
+    ]
+    if matching:
+        keep = min(matching, key=lambda task: task.id)
+        _cancel_proactive_duplicates(scheduler, existing, keep=keep)
+        return keep
+
+    # Configuration changed. Replace stale active tasks so the schedule and
+    # notification settings from config.toml take effect on this startup.
+    _cancel_proactive_duplicates(scheduler, existing)
+
     return scheduler.create_task(
-        prompt="Run the proactive agent: collect overnight data, execute approved actions, notify pending approvals.",
+        prompt=_PROACTIVE_CRON_PROMPT,
         schedule_type="cron",
         schedule_value=cron_expr,
         agent="proactive",
         context_mode="isolated",
-        metadata={
-            "notification_channel_id": notification_channel_id,
-            "hours_back": hours_back,
-            "timezone": timezone,
-        },
+        metadata=metadata,
     )
+
+
+def _cancel_proactive_duplicates(
+    scheduler: Any, tasks: List[Any], *, keep: Optional[Any] = None
+) -> None:
+    """Cancel managed proactive tasks other than *keep*."""
+    for task in tasks:
+        if keep is not None and task.id == keep.id:
+            continue
+        try:
+            scheduler.cancel_task(task.id)
+        except Exception:
+            logger.warning(
+                "Failed to cancel duplicate proactive task %s",
+                task.id,
+                exc_info=True,
+            )

src/openjarvis/core/config.py

@@ -12,9 +12,18 @@ import os
 import platform
 import shutil
 import subprocess
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, is_dataclass
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Dict, List, Optional
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Dict,
+    List,
+    Optional,
+    get_args,
+    get_origin,
+    get_type_hints,
+)
 
 from openjarvis.core.paths import (
     ConfigurationError,
@@ -1710,10 +1719,16 @@ def _apply_toml_section(target: Any, section: Dict[str, Any]) -> None:
     """Overlay TOML key/value pairs onto a dataclass instance.
 
     Recursively handles nested dicts when the target attribute is itself
-    a dataclass.  Normalises TOML arrays to comma-separated strings — both
-    for dataclass fields annotated as ``str`` and for backward-compat
-    property setters that expect string input.
+    a dataclass, including dict entries in lists of dataclasses.  Normalises
+    TOML arrays to comma-separated strings — both for dataclass fields annotated
+    as ``str`` and for backward-compat property setters that expect string input.
     """
+    try:
+        type_hints = get_type_hints(type(target))
+    except (NameError, TypeError):
+        # Some config types contain optional runtime-only forward references.
+        type_hints = {}
+
     for key, value in section.items():
         if hasattr(target, key):
             if isinstance(value, dict):
@@ -1728,14 +1743,35 @@ def _apply_toml_section(target: Any, section: Dict[str, Any]) -> None:
                 # property setters (e.g. reward_weights, default_tools).
                 if isinstance(value, list):
                     is_str_field = False
+                    item_dataclass = None
                     if hasattr(target, "__dataclass_fields__"):
                         field_obj = target.__dataclass_fields__.get(key)
-                        if field_obj is not None and field_obj.type in ("str", str):
-                            is_str_field = True
-                        elif field_obj is None:
+                        if field_obj is not None:
+                            field_type = type_hints.get(key, field_obj.type)
+                            type_args = get_args(field_type)
+                            if (
+                                get_origin(field_type) is list
+                                and len(type_args) == 1
+                                and is_dataclass(type_args[0])
+                            ):
+                                item_dataclass = type_args[0]
+                            elif field_obj.type in ("str", str):
+                                is_str_field = True
+                        else:
                             # Property, not a real field — normalise to string
                             is_str_field = True
-                    if is_str_field:
+
+                    if item_dataclass is not None:
+                        converted = []
+                        for item in value:
+                            if isinstance(item, dict):
+                                nested = item_dataclass()
+                                _apply_toml_section(nested, item)
+                                converted.append(nested)
+                            else:
+                                converted.append(item)
+                        value = converted
+                    elif is_str_field:
                         value = ",".join(str(v) for v in value)
                 setattr(target, key, value)
 

src/openjarvis/engine/_discovery.py

@@ -35,6 +35,12 @@ def _make_engine(key: str, config: JarvisConfig) -> InferenceEngine:
     """Instantiate a registered engine with the appropriate config host."""
     cls = EngineRegistry.get(key)
 
+    # LiteLLM cannot enumerate every model supported by every provider.  Its
+    # list_models() contract therefore advertises the configured default
+    # model, which must be supplied when discovery constructs the engine.
+    if key == "litellm":
+        return cls(default_model=config.intelligence.default_model or None)
+
     # gemma_cpp: pass config fields instead of host
     if key == "gemma_cpp":
         cfg = config.engine.gemma_cpp

src/openjarvis/engine/multi.py

@@ -26,16 +26,19 @@ class MultiEngine(InferenceEngine):
     def __init__(self, engines: list[tuple[str, InferenceEngine]]) -> None:
         self._engines = engines
         self._model_map: Dict[str, InferenceEngine] = {}
+        self._model_key_map: Dict[str, str] = {}
         self._refresh_map()
 
     def _refresh_map(self) -> None:
         self._model_map.clear()
-        for _key, engine in self._engines:
+        self._model_key_map.clear()
+        for key, engine in self._engines:
             try:
                 for model_id in engine.list_models():
                     self._model_map[model_id] = engine
+                    self._model_key_map[model_id] = key
             except Exception as exc:
-                logger.debug("Failed to list models for %s: %s", _key, exc)
+                logger.debug("Failed to list models for %s: %s", key, exc)
 
     _CLOUD_PREFIXES = ("gpt-", "o1-", "o3-", "o4-", "claude-", "gemini-", "openrouter/")
 
@@ -117,6 +120,14 @@ class MultiEngine(InferenceEngine):
         self._refresh_map()
         return list(self._model_map.keys())
 
+    def engine_key_for(self, model: str) -> str | None:
+        """Return the registry key of the engine advertising *model*."""
+        key = self._model_key_map.get(model)
+        if key is not None:
+            return key
+        self._refresh_map()
+        return self._model_key_map.get(model)
+
     def health(self) -> bool:
         return any(engine.health() for _key, engine in self._engines)
 

src/openjarvis/server/routes.py

@@ -336,6 +336,34 @@ def _remember_exchange(
     )
 
 
+def _engine_key_for_model(engine: Any, model: str) -> str | None:
+    """Resolve the engine that advertised *model* through wrapper layers."""
+    from openjarvis.engine.multi import MultiEngine
+    from openjarvis.security.guardrails import GuardrailsEngine
+    from openjarvis.telemetry.instrumented_engine import InstrumentedEngine
+
+    current = engine
+    while current is not None:
+        if isinstance(current, MultiEngine):
+            return current.engine_key_for(model)
+        if isinstance(current, InstrumentedEngine):
+            current = current._inner
+            continue
+        if isinstance(current, GuardrailsEngine):
+            current = current._engine
+            continue
+        engine_id = getattr(current, "engine_id", None)
+        return engine_id if isinstance(engine_id, str) else None
+    return None
+
+
+def _uses_direct_cloud_router(engine: Any, model: str) -> bool:
+    """Whether *model* should bypass the configured engine for direct cloud."""
+    from openjarvis.server.cloud_router import is_cloud_model
+
+    return is_cloud_model(model) and _engine_key_for_model(engine, model) != "litellm"
+
+
 def _handle_direct(
     engine,
     model: str,
@@ -541,12 +569,13 @@ async def _handle_stream_tools(
     tool_calls) — identical to the prior plain-stream behaviour, so this never
     regresses non-tool-capable engines.
     """
-    from openjarvis.server.cloud_router import is_cloud_model
-
     messages = _to_messages(req.messages)
     messages = _ensure_identity_prompt(messages, app_config)
     chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"
-    use_cloud = is_cloud_model(model)
+    use_cloud = _uses_direct_cloud_router(engine, model)
+    telemetry_engine = (
+        "cloud" if use_cloud else (_engine_key_for_model(engine, model) or "ollama")
+    )
     query_text = ""
     for _m in reversed(req.messages):
         if _m.role == "user" and _m.content:
@@ -626,7 +655,7 @@ async def _handle_stream_tools(
         # Tag the finish chunk with the engine label, matching _handle_stream
         # so UI/telemetry consumers see the same field on the tools path.
         finish_dict.setdefault("telemetry", {})
-        finish_dict["telemetry"]["engine"] = "cloud" if use_cloud else "ollama"
+        finish_dict["telemetry"]["engine"] = telemetry_engine
         if complexity_info is not None:
             finish_dict["complexity"] = complexity_info.model_dump()
         yield f"data: {_json.dumps(finish_dict)}\n\n"
@@ -668,11 +697,7 @@ async def _handle_stream(
     """
     import time
 
-    from openjarvis.server.cloud_router import (
-        is_cloud_model,
-        stream_cloud,
-        stream_local,
-    )
+    from openjarvis.server.cloud_router import stream_cloud, stream_local
 
     messages = _to_messages(req.messages)
     messages = _ensure_identity_prompt(messages, app_config)
@@ -687,7 +712,10 @@ async def _handle_stream(
 
     # Route directly to the right backend — bypasses engine routing entirely
     # so broken MultiEngine state can never misdirect requests.
-    use_cloud = is_cloud_model(model)
+    use_cloud = _uses_direct_cloud_router(engine, model)
+    telemetry_engine = (
+        "cloud" if use_cloud else (_engine_key_for_model(engine, model) or "ollama")
+    )
 
     async def generate():
         started_at = time.time()
@@ -792,7 +820,7 @@ async def _handle_stream(
                 query=query_text,
                 result=full_content,
                 model=model,
-                engine="cloud" if use_cloud else "ollama",
+                engine=telemetry_engine,
                 started_at=started_at,
                 ended_at=time.time(),
             )
@@ -825,7 +853,7 @@ async def _handle_stream(
         # We use the routing decision (use_cloud) directly rather than
         # unwrapping the engine chain, which can be in a broken state.
         finish_dict.setdefault("telemetry", {})
-        finish_dict["telemetry"]["engine"] = "cloud" if use_cloud else "ollama"
+        finish_dict["telemetry"]["engine"] = telemetry_engine
 
         if complexity_info is not None:
             finish_dict["complexity"] = complexity_info.model_dump()
@@ -842,24 +870,40 @@ async def _handle_stream(
 
 @router.get("/v1/models")
 async def list_models(request: Request) -> ModelListResponse:
-    """List locally installed models (Ollama).
+    """List selectable engine models for the installed-model picker.
 
-    Cloud models are not included here — they live in the Cloud Models tab
-    of the UI and are selected there, not from this endpoint.
+    Direct cloud models live in the Cloud Models tab. Models advertised by a
+    configured LiteLLM engine remain here because LiteLLM owns their routing
+    and may use provider-qualified IDs that resemble OpenRouter IDs.
     """
     from openjarvis.server.cloud_router import is_cloud_model, list_local_models
 
     # Prefer engine.list_models() so mock engines work in tests.
-    # Filter out any cloud model IDs that may appear via MultiEngine.
+    # Filter out direct-cloud model IDs that may appear via MultiEngine, but
+    # retain provider-qualified IDs owned by the configured LiteLLM engine.
     # Fall back to direct Ollama query only when the engine returns nothing.
     engine = request.app.state.engine
     all_ids = await asyncio.to_thread(engine.list_models)
-    model_ids = [m for m in all_ids if not is_cloud_model(m)]
+    model_ids = [
+        m
+        for m in all_ids
+        if not is_cloud_model(m) or _engine_key_for_model(engine, m) == "litellm"
+    ]
     if not model_ids:
         model_ids = await list_local_models()
 
     return ModelListResponse(
-        data=[ModelObject(id=mid) for mid in model_ids],
+        data=[
+            ModelObject(
+                id=mid,
+                owned_by=(
+                    "litellm"
+                    if _engine_key_for_model(engine, mid) == "litellm"
+                    else "openjarvis"
+                ),
+            )
+            for mid in model_ids
+        ],
     )
 
 

src/openjarvis/server/ws_bridge.py

@@ -79,14 +79,41 @@ def create_ws_router(event_bus: EventBus) -> Any:
         queue: asyncio.Queue = asyncio.Queue(maxsize=100)
         loop = asyncio.get_running_loop()
         clients[websocket] = (queue, loop)
+        recv: asyncio.Task | None = None
+        payload: asyncio.Task | None = None
+        disconnected = False
         try:
+            recv = asyncio.create_task(websocket.receive())
+            payload = asyncio.create_task(queue.get())
             while True:
-                payload = await queue.get()
-                await websocket.send_json(payload)
+                done, _ = await asyncio.wait(
+                    {recv, payload}, return_when=asyncio.FIRST_COMPLETED
+                )
+                if recv in done:
+                    # Starlette surfaces a disconnect message only when the app
+                    # reads from the socket. Without this receive, the handler
+                    # can stay parked on queue.get() after the client leaves.
+                    message = await recv
+                    if message.get("type") == "websocket.disconnect":
+                        disconnected = True
+                        break
+                    recv = asyncio.create_task(websocket.receive())
+                if payload in done:
+                    await websocket.send_json(payload.result())
+                    payload = asyncio.create_task(queue.get())
         except WebSocketDisconnect:
-            pass
+            disconnected = True
         finally:
             clients.pop(websocket, None)
+            pending = [task for task in (recv, payload) if task is not None]
+            for task in pending:
+                task.cancel()
+            cleanup = asyncio.gather(*pending, return_exceptions=True)
+            try:
+                await asyncio.shield(cleanup)
+            except asyncio.CancelledError:
+                if not disconnected:
+                    raise
 
     return router
 

src/openjarvis/telemetry/instrumented_engine.py

@@ -212,6 +212,7 @@ class InstrumentedEngine(InferenceEngine):
             completion_tokens=completion_tokens,
             total_tokens=prompt_tok + completion_tokens,
             latency_seconds=latency,
+            cost_usd=result.get("cost_usd", 0.0),
             ttft=ttft,
             throughput_tok_per_sec=throughput,
             energy_per_output_token_joules=energy_per_output_token,

src/openjarvis/tools/__init__.py

@@ -142,4 +142,14 @@ try:
 except ImportError:
     pass
 
+try:
+    import openjarvis.tools.scan_chunks  # noqa: F401
+except ImportError:
+    pass
+
+try:
+    import openjarvis.tools.knowledge_sql  # noqa: F401
+except ImportError:
+    pass
+
 __all__ = ["BaseTool", "ToolExecutor", "ToolSpec"]

tests/agents/test_proactive_agent.py

@@ -0,0 +1,131 @@
+"""Regression tests for proactive scheduling and notification setup."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from openjarvis.agents.proactive_agent import (
+    _PROACTIVE_CRON_PROMPT,
+    _build_notification_channel,
+    register_cron,
+)
+from openjarvis.core.registry import ChannelRegistry
+from openjarvis.scheduler.scheduler import TaskScheduler
+from openjarvis.scheduler.store import SchedulerStore
+
+
+@pytest.fixture()
+def scheduler(tmp_path):
+    store = SchedulerStore(tmp_path / "scheduler.db")
+    scheduler = TaskScheduler(store)
+    yield scheduler
+    scheduler.stop()
+    store.close()
+
+
+def _register(scheduler, *, schedule="0 5 * * *", channel="telegram:123"):
+    return register_cron(
+        scheduler,
+        notification_channel_id=channel,
+        cron_expr=schedule,
+        hours_back=24,
+        timezone="UTC",
+    )
+
+
+class TestRegisterCron:
+    def test_reuses_exact_task_and_cancels_duplicates(self, scheduler):
+        first = _register(scheduler)
+        duplicate = scheduler.create_task(
+            _PROACTIVE_CRON_PROMPT,
+            "cron",
+            "0 5 * * *",
+            agent="proactive",
+            metadata=first.metadata,
+        )
+
+        returned = _register(scheduler)
+
+        assert returned.id in {first.id, duplicate.id}
+        assert [task.id for task in scheduler.list_tasks(status="active")] == [
+            returned.id
+        ]
+        cancelled_id = scheduler.list_tasks(status="cancelled")[0].id
+        assert cancelled_id == ({first.id, duplicate.id} - {returned.id}).pop()
+
+    def test_replaces_task_when_configuration_changes(self, scheduler):
+        old = _register(scheduler, schedule="0 5 * * *", channel="telegram:old")
+
+        new = _register(scheduler, schedule="0 7 * * *", channel="telegram:new")
+
+        assert new.id != old.id
+        assert new.schedule_value == "0 7 * * *"
+        assert new.metadata["notification_channel_id"] == "telegram:new"
+        assert scheduler.list_tasks(status="cancelled")[0].id == old.id
+
+    def test_preserves_pause_across_restart(self, scheduler):
+        paused = _register(scheduler)
+        scheduler.pause_task(paused.id)
+
+        returned = _register(scheduler, schedule="0 7 * * *")
+
+        assert returned.id == paused.id
+        assert returned.status == "paused"
+        assert scheduler.list_tasks(status="active") == []
+
+    def test_migrates_legacy_tasks_without_stable_key(self, scheduler):
+        legacy = scheduler.create_task(
+            _PROACTIVE_CRON_PROMPT,
+            "cron",
+            "0 5 * * *",
+            agent="proactive",
+            metadata={
+                "notification_channel_id": "telegram:123",
+                "hours_back": 24,
+                "timezone": "UTC",
+            },
+        )
+
+        current = _register(scheduler)
+
+        assert current.id != legacy.id
+        assert current.metadata["openjarvis_task_key"] == "proactive-daily"
+        assert scheduler.list_tasks(status="cancelled")[0].id == legacy.id
+
+
+class TestNotificationChannel:
+    def test_telegram_is_configured_without_starting_polling(self):
+        class FakeTelegram:
+            def __init__(self, **kwargs):
+                self.kwargs = kwargs
+                self.connect = MagicMock()
+
+        config = MagicMock()
+        with (
+            patch.object(ChannelRegistry, "contains", return_value=True),
+            patch.object(ChannelRegistry, "get", return_value=FakeTelegram),
+            patch("openjarvis.core.config.load_config", return_value=config),
+            patch(
+                "openjarvis.system._channel_kwargs.build_channel_kwargs",
+                return_value={"bot_token": "configured-token"},
+            ),
+        ):
+            channel = _build_notification_channel("telegram:123")
+
+        assert channel.kwargs == {"bot_token": "configured-token"}
+        channel.connect.assert_not_called()
+
+    def test_non_telegram_channel_keeps_connect_lifecycle(self):
+        class FakeChannel:
+            def __init__(self, **kwargs):
+                self.connect = MagicMock()
+
+        with (
+            patch.object(ChannelRegistry, "contains", return_value=True),
+            patch.object(ChannelRegistry, "get", return_value=FakeChannel),
+        ):
+            channel = _build_notification_channel("twilio:15551234567")
+
+        channel.connect.assert_called_once_with()

tests/core/test_config_skills_sources.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
-from openjarvis.core.config import SkillsConfig, SkillSourceConfig
+from pathlib import Path
+
+from openjarvis.core.config import SkillsConfig, SkillSourceConfig, load_config
 
 
 class TestSkillSourceConfig:
@@ -41,3 +43,36 @@ class TestSkillsConfigWithSources:
         )
         assert len(cfg.sources) == 2
         assert cfg.sources[0].source == "hermes"
+
+    def test_loads_source_tables_as_config_objects(
+        self, tmp_path: Path, monkeypatch
+    ) -> None:
+        monkeypatch.setenv("OPENJARVIS_HOME", str(tmp_path / "home"))
+        toml_file = tmp_path / "config.toml"
+        toml_file.write_text(
+            "[[skills.sources]]\n"
+            'source = "hermes"\n'
+            'filter = { category = ["productivity"] }\n\n'
+            "[[skills.sources]]\n"
+            'source = "github"\n'
+            'url = "https://github.com/example/skill-library"\n'
+            "auto_update = true\n"
+        )
+
+        load_config.cache_clear()
+        try:
+            cfg = load_config(toml_file)
+        finally:
+            load_config.cache_clear()
+
+        assert cfg.skills.sources == [
+            SkillSourceConfig(
+                source="hermes",
+                filter={"category": ["productivity"]},
+            ),
+            SkillSourceConfig(
+                source="github",
+                url="https://github.com/example/skill-library",
+                auto_update=True,
+            ),
+        ]

tests/engine/test_discovery.py

@@ -8,10 +8,12 @@ from openjarvis.core.config import JarvisConfig
 from openjarvis.core.registry import EngineRegistry
 from openjarvis.engine._base import InferenceEngine
 from openjarvis.engine._discovery import (
+    _make_engine,
     discover_engines,
     discover_models,
     get_engine,
 )
+from openjarvis.engine.litellm import LiteLLMEngine
 
 
 class _FakeEngine(InferenceEngine):
@@ -131,6 +133,24 @@ class TestDiscoverModels:
         assert result == {"ollama": ["m1", "m2"], "vllm": ["m3"]}
 
 
+class TestLiteLLMDiscovery:
+    def test_configured_default_model_is_advertised(self) -> None:
+        """Regression for #713: discovery must configure LiteLLM's model.
+
+        LiteLLM cannot enumerate every model supported by every provider, so
+        ``LiteLLMEngine.list_models()`` advertises the configured default
+        model.  Dropping that value while constructing the engine leaves the
+        API and Web UI with an empty model list.
+        """
+        cfg = JarvisConfig()
+        cfg.intelligence.default_model = "groq/llama-3.3-70b-versatile"
+        EngineRegistry.register_value("litellm", LiteLLMEngine)
+
+        engine = _make_engine("litellm", cfg)
+
+        assert engine.list_models() == ["groq/llama-3.3-70b-versatile"]
+
+
 class TestGetEngine:
     def test_fallback_when_default_unhealthy(self) -> None:
         _reg("bad", "bad")

tests/engine/test_engine_wrappers.py

@@ -120,6 +120,9 @@ async def test_multi_routes_stream_full_by_model():
     engine_b.list_models = lambda: ["model-b"]
 
     multi = MultiEngine([("a", engine_a), ("b", engine_b)])
+    assert multi.engine_key_for("model-a") == "a"
+    assert multi.engine_key_for("model-b") == "b"
+    assert multi.engine_key_for("missing") is None
 
     # Route to engine A
     result_a = []

tests/server/test_routes.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import json
-from unittest.mock import MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -883,6 +883,64 @@ class TestModelsEndpoint:
         data = resp.json()
         assert len(data["data"]) == 3
 
+    def test_configured_litellm_model_is_listed(self):
+        """Regression for #713: LiteLLM models must reach the Web UI."""
+        model = "groq/llama-3.3-70b-versatile"
+        engine = _make_engine(models=[model])
+        engine.engine_id = "litellm"
+        app = create_app(
+            engine,
+            model,
+            engine_name="litellm",
+            config=_test_config(),
+        )
+
+        with patch(
+            "openjarvis.server.cloud_router.list_local_models",
+            new_callable=AsyncMock,
+        ) as list_local_models:
+            list_local_models.return_value = []
+            client = TestClient(app)
+            resp = client.get("/v1/models")
+
+        assert resp.status_code == 200
+        assert [item["id"] for item in resp.json()["data"]] == [model]
+        assert resp.json()["data"][0]["owned_by"] == "litellm"
+
+    def test_litellm_provider_model_streams_through_active_engine(self):
+        """A LiteLLM ``provider/model`` ID must not bypass its engine."""
+        model = "groq/llama-3.3-70b-versatile"
+        engine = _make_engine(models=[model])
+        engine.engine_id = "litellm"
+        app = create_app(
+            engine,
+            model,
+            engine_name="litellm",
+            config=_test_config(),
+        )
+
+        async def direct_cloud_tokens():
+            yield "wrong backend"
+
+        with patch(
+            "openjarvis.server.cloud_router.stream_cloud",
+            return_value=direct_cloud_tokens(),
+        ) as stream_cloud:
+            client = TestClient(app)
+            resp = client.post(
+                "/v1/chat/completions",
+                json={
+                    "model": model,
+                    "messages": [{"role": "user", "content": "hello"}],
+                    "stream": True,
+                },
+            )
+
+        assert resp.status_code == 200
+        stream_cloud.assert_not_called()
+        assert "Hello" in resp.text
+        assert '"engine": "litellm"' in resp.text
+
 
 # ---------------------------------------------------------------------------
 # Health endpoint tests

tests/server/test_ws_bridge.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import asyncio
 import time
+from types import SimpleNamespace
 
 import pytest
 
@@ -60,3 +62,102 @@ class TestWSBridge:
             time.sleep(0.05)  # Let call_soon_threadsafe deliver to queue
             data = ws.receive_json()
             assert data["data"]["agent_id"] == "agent-A"
+
+    def test_client_disconnect_stops_handler(self, event_bus):
+        async def exercise():
+            from openjarvis.server.ws_bridge import create_ws_router
+
+            class FakeWebSocket:
+                app = SimpleNamespace(state=SimpleNamespace(api_key=""))
+                query_params = {}
+                headers = {}
+
+                async def accept(self):
+                    pass
+
+                async def receive(self):
+                    return {"type": "websocket.disconnect"}
+
+            endpoint = create_ws_router(event_bus).routes[0].endpoint
+            await asyncio.wait_for(endpoint(FakeWebSocket()), timeout=1)
+
+        asyncio.run(exercise())
+
+    def test_simultaneous_client_message_does_not_drop_event(self, event_bus):
+        async def exercise():
+            from openjarvis.server.ws_bridge import create_ws_router
+
+            class FakeWebSocket:
+                def __init__(self):
+                    self.app = SimpleNamespace(state=SimpleNamespace(api_key=""))
+                    self.query_params = {}
+                    self.headers = {}
+                    self.sent = []
+                    self.receive_count = 0
+                    self.disconnect = asyncio.Event()
+
+                async def accept(self):
+                    pass
+
+                async def receive(self):
+                    self.receive_count += 1
+                    if self.receive_count == 1:
+                        event_bus.publish(
+                            EventType.AGENT_TICK_START, {"agent_id": "not-dropped"}
+                        )
+                        return {"type": "websocket.receive", "text": "client message"}
+                    await self.disconnect.wait()
+                    return {"type": "websocket.disconnect"}
+
+                async def send_json(self, payload):
+                    self.sent.append(payload)
+                    self.disconnect.set()
+
+            websocket = FakeWebSocket()
+            endpoint = create_ws_router(event_bus).routes[0].endpoint
+
+            await asyncio.wait_for(endpoint(websocket), timeout=1)
+
+            assert websocket.sent[0]["data"]["agent_id"] == "not-dropped"
+
+        asyncio.run(exercise())
+
+    def test_cancelling_handler_cleans_up_child_tasks(self, event_bus):
+        async def exercise():
+            from openjarvis.server.ws_bridge import create_ws_router
+
+            class FakeWebSocket:
+                def __init__(self):
+                    self.app = SimpleNamespace(state=SimpleNamespace(api_key=""))
+                    self.query_params = {}
+                    self.headers = {}
+                    self.receiving = asyncio.Event()
+                    self.receive_cancelled = asyncio.Event()
+
+                async def accept(self):
+                    pass
+
+                async def receive(self):
+                    self.receiving.set()
+                    try:
+                        await asyncio.Event().wait()
+                    finally:
+                        self.receive_cancelled.set()
+
+            websocket = FakeWebSocket()
+            endpoint = create_ws_router(event_bus).routes[0].endpoint
+            handler = asyncio.create_task(endpoint(websocket))
+            await websocket.receiving.wait()
+
+            handler.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await handler
+
+            assert websocket.receive_cancelled.is_set()
+            assert not [
+                task
+                for task in asyncio.all_tasks()
+                if task is not asyncio.current_task() and not task.done()
+            ]
+
+        asyncio.run(exercise())

tests/telemetry/test_instrumented_engine.py

@@ -71,6 +71,17 @@ class TestInstrumentedEngine:
         assert record.prompt_tokens == 10
         assert record.completion_tokens == 5
 
+    def test_generate_records_cost(self, mock_engine, bus):
+        mock_engine.generate.return_value["cost_usd"] = 0.0015
+        ie = InstrumentedEngine(mock_engine, bus)
+        messages = [Message(role=Role.USER, content="Hi")]
+        ie.generate(messages, model="test")
+
+        event = next(
+            e for e in bus.history if e.event_type == EventType.TELEMETRY_RECORD
+        )
+        assert event.data["record"].cost_usd == pytest.approx(0.0015)
+
     def test_list_models_delegates(self, mock_engine, bus):
         ie = InstrumentedEngine(mock_engine, bus)
         assert ie.list_models() == ["test-model"]

tests/tools/test_tool_registration.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import importlib
+import subprocess
 import sys
 
 from openjarvis.core.registry import ToolRegistry
@@ -68,6 +69,10 @@ EXPECTED_TOOLS = {
     "kg_add_relation",
     "kg_query",
     "kg_neighbors",
+    # knowledge_sql.py
+    "knowledge_sql",
+    # scan_chunks.py
+    "scan_chunks",
 }
 
 
@@ -100,3 +105,25 @@ def test_all_builtin_tools_registered():
     assert not missing, (
         f"Tools not registered (missing import in __init__.py?): {sorted(missing)}"
     )
+
+
+def test_package_import_registers_deep_research_tools():
+    """Registration must not depend on another module being imported first."""
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            (
+                "import openjarvis.tools; "
+                "from openjarvis.core.registry import ToolRegistry; "
+                "expected = {'knowledge_sql', 'scan_chunks'}; "
+                "missing = expected - set(ToolRegistry.keys()); "
+                "assert not missing, f'Missing tools: {sorted(missing)}'"
+            ),
+        ],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    assert result.returncode == 0, result.stderr