Stabilize chat and cookbook workflows

routes/cookbook_routes.py

@@ -9,6 +9,7 @@ import shlex
 import shutil
 import subprocess
 import sys
+import time
 import urllib.request
 import uuid
 from pathlib import Path
@@ -66,6 +67,9 @@ _HF_TOKEN_STATUS_SNIPPET = (
 def setup_cookbook_routes() -> APIRouter:
     router = APIRouter(tags=["cookbook"])
     _cookbook_state_path = Path(COOKBOOK_STATE_FILE)
+    _state_get_cache = {"ts": 0.0, "mtime": 0.0, "value": None}
+    _tasks_status_cache = {"ts": 0.0, "value": None}
+    _tasks_status_inflight = {"task": None}
 
     def _mask_secret(value: str) -> str:
         if not value:
@@ -2463,19 +2467,29 @@ def setup_cookbook_routes() -> APIRouter:
     async def get_cookbook_state(request: Request):
         """Load saved cookbook state (tasks, servers, presets, settings)."""
         require_admin(request)
+        now = time.monotonic()
+        try:
+            mtime = _cookbook_state_path.stat().st_mtime if _cookbook_state_path.exists() else 0.0
+        except Exception:
+            mtime = 0.0
+        cached = _state_get_cache.get("value")
+        if cached is not None and _state_get_cache.get("mtime") == mtime and now - float(_state_get_cache.get("ts") or 0) < 1.5:
+            return cached
         if _cookbook_state_path.exists():
             try:
                 state = json.loads(_cookbook_state_path.read_text(encoding="utf-8"))
                 saved_tasks = state.get("tasks", [])
                 tasks = saved_tasks if isinstance(saved_tasks, list) else list(saved_tasks.values()) if isinstance(saved_tasks, dict) else []
-                try:
-                    _maybe_sweep_orphans(tasks, state)
-                except Exception as _sweep_e:
-                    logger.warning(f"state orphan sweep failed (non-fatal): {_sweep_e!r}")
-                return _state_for_client(state)
+                client_state = _state_for_client(state)
+                _state_get_cache.update({"ts": now, "mtime": mtime, "value": client_state})
+                return client_state
             except Exception:
-                return _state_for_client({})
-        return _state_for_client({})
+                client_state = _state_for_client({})
+                _state_get_cache.update({"ts": now, "mtime": mtime, "value": client_state})
+                return client_state
+        client_state = _state_for_client({})
+        _state_get_cache.update({"ts": now, "mtime": mtime, "value": client_state})
+        return client_state
 
     @router.post("/api/cookbook/state")
     async def save_cookbook_state(request: Request):
@@ -2583,7 +2597,19 @@ def setup_cookbook_routes() -> APIRouter:
                             f"not in incoming body (race guard): "
                             f"{[t.get('sessionId') for t in preserved]}")
                 data["tasks"] = incoming_tasks + preserved
-            atomic_write_json(str(_cookbook_state_path), _state_for_storage(data, on_disk), indent=2)
+            storage_state = _state_for_storage(data, on_disk)
+            if storage_state == on_disk:
+                return {"ok": True, "preserved": len(preserved), "unchanged": True}
+            atomic_write_json(str(_cookbook_state_path), storage_state, indent=2)
+            try:
+                mtime = _cookbook_state_path.stat().st_mtime
+                _state_get_cache.update({
+                    "ts": time.monotonic(),
+                    "mtime": mtime,
+                    "value": _state_for_client(storage_state),
+                })
+            except Exception:
+                pass
             return {"ok": True, "preserved": len(preserved)}
         except Exception as e:
             return {"ok": False, "error": str(e)}
@@ -2705,10 +2731,10 @@ def setup_cookbook_routes() -> APIRouter:
 
         return {"models": out}
 
-    # Rate-limit for the orphan-tmux adoption sweep. 60s interval so SSH
+    # Rate-limit for the orphan-tmux adoption sweep. Five-minute interval so SSH
     # work is genuinely sparse even on an actively-polled cookbook page.
     _last_orphan_sweep_ts = [0.0]
-    _ORPHAN_SWEEP_MIN_INTERVAL_S = 60.0
+    _ORPHAN_SWEEP_MIN_INTERVAL_S = 300.0
     # Concurrency guard so two requests racing don't both spawn a sweep.
     _orphan_sweep_inflight = [False]
 
@@ -3263,7 +3289,26 @@ def setup_cookbook_routes() -> APIRouter:
         event loop. Now the whole body runs in a worker thread via
         asyncio.to_thread so other requests stay responsive."""
         require_admin(request)
-        return await asyncio.to_thread(_cookbook_tasks_status_sync)
+        now = time.monotonic()
+        cached = _tasks_status_cache.get("value")
+        if cached is not None and now - float(_tasks_status_cache.get("ts") or 0) < 2.0:
+            return cached
+        inflight = _tasks_status_inflight.get("task")
+        if inflight and not inflight.done():
+            return await inflight
+
+        async def _compute():
+            data = await asyncio.to_thread(_cookbook_tasks_status_sync)
+            _tasks_status_cache.update({"ts": time.monotonic(), "value": data})
+            return data
+
+        task = asyncio.create_task(_compute())
+        _tasks_status_inflight["task"] = task
+        try:
+            return await task
+        finally:
+            if _tasks_status_inflight.get("task") is task:
+                _tasks_status_inflight["task"] = None
 
     def _cookbook_tasks_status_sync():
         import subprocess

routes/model_routes.py

@@ -1479,6 +1479,7 @@ def setup_model_routes(model_discovery):
     # within ~8s of the user noticing.
     _LOCAL_PROBE_TTL = 8.0
     _local_probe_cache: Dict[str, Any] = {"data": None, "time": 0.0}
+    _local_probe_inflight: Dict[str, Any] = {"task": None}
 
     @router.get("/model-endpoints/probe-local")
     async def probe_local_endpoints(request: Request):
@@ -1493,60 +1494,72 @@ def setup_model_routes(model_discovery):
                 (now - _local_probe_cache["time"]) < _LOCAL_PROBE_TTL):
             return _local_probe_cache["data"]
 
-        db = SessionLocal()
-        try:
-            if _disable_stale_cookbook_local_endpoints(db):
-                _invalidate_models_cache()
-            endpoints = db.query(ModelEndpoint).filter(ModelEndpoint.is_enabled == True).all()
-            local_eps = []
-            for ep in endpoints:
-                base = _normalize_base(ep.base_url)
-                kind = _effective_endpoint_kind(ep, base)
-                if _classify_endpoint(base, kind) == "local":
-                    local_eps.append((ep.id, base, ep.api_key))
-        finally:
-            db.close()
-
-        grouped: Dict[str, Dict[str, Any]] = {}
-        for ep_id, base, api_key in local_eps:
-            key = _refresh_key(base, api_key)
-            grouped.setdefault(key, {"base": base, "api_key": api_key, "endpoint_ids": []})["endpoint_ids"].append(ep_id)
-
-        async def _probe_one(data: Dict[str, Any]) -> Dict[str, Any]:
-            t0 = _time.time()
-            try:
-                import asyncio as _asyncio
-                # Bumped 1.5s → 3.5s. The previous 1.5s budget was clipping
-                # local vLLM endpoints on Tailscale links where the model
-                # server is still loading (Qwen3.5-122B takes 2–3 min to
-                # warm); /v1/models can take 500–2500 ms on a busy box,
-                # which pushed _ping_endpoint's full path-discovery sweep
-                # past the cap and marked the row offline despite the
-                # user actively chatting with it.
-                ping = await _asyncio.to_thread(_ping_endpoint, data["base"], data.get("api_key"), 3.5)
-                lat = round((_time.time() - t0) * 1000)
-                return {
-                    "alive": bool(ping.get("reachable")),
-                    "latency_ms": lat,
-                    "status_code": ping.get("status_code"),
-                    "error": ping.get("error"),
-                }
-            except Exception as e:
-                return {"alive": False, "latency_ms": None, "status_code": None, "error": str(e)[:120]}
-
         import asyncio as _asyncio
-        results_list = await _asyncio.gather(
-            *[_probe_one(data) for data in grouped.values()],
-            return_exceptions=False,
-        )
-        results: Dict[str, Any] = {}
-        for data, r in zip(grouped.values(), results_list):
-            for eid in data["endpoint_ids"]:
-                results[eid] = r
+        task = _local_probe_inflight.get("task")
+        if task is not None and not task.done():
+            return await task
 
-        _local_probe_cache["data"] = results
-        _local_probe_cache["time"] = now
-        return results
+        async def _compute_local_probe() -> Dict[str, Any]:
+            db = SessionLocal()
+            try:
+                if _disable_stale_cookbook_local_endpoints(db):
+                    _invalidate_models_cache()
+                endpoints = db.query(ModelEndpoint).filter(ModelEndpoint.is_enabled == True).all()
+                local_eps = []
+                for ep in endpoints:
+                    base = _normalize_base(ep.base_url)
+                    kind = _effective_endpoint_kind(ep, base)
+                    if _classify_endpoint(base, kind) == "local":
+                        local_eps.append((ep.id, base, ep.api_key))
+            finally:
+                db.close()
+
+            grouped: Dict[str, Dict[str, Any]] = {}
+            for ep_id, base, api_key in local_eps:
+                key = _refresh_key(base, api_key)
+                grouped.setdefault(key, {"base": base, "api_key": api_key, "endpoint_ids": []})["endpoint_ids"].append(ep_id)
+
+            async def _probe_one(data: Dict[str, Any]) -> Dict[str, Any]:
+                t0 = _time.time()
+                try:
+                    # Bumped 1.5s → 3.5s. The previous 1.5s budget was clipping
+                    # local vLLM endpoints on Tailscale links where the model
+                    # server is still loading (Qwen3.5-122B takes 2–3 min to
+                    # warm); /v1/models can take 500–2500 ms on a busy box,
+                    # which pushed _ping_endpoint's full path-discovery sweep
+                    # past the cap and marked the row offline despite the
+                    # user actively chatting with it.
+                    ping = await _asyncio.to_thread(_ping_endpoint, data["base"], data.get("api_key"), 3.5)
+                    lat = round((_time.time() - t0) * 1000)
+                    return {
+                        "alive": bool(ping.get("reachable")),
+                        "latency_ms": lat,
+                        "status_code": ping.get("status_code"),
+                        "error": ping.get("error"),
+                    }
+                except Exception as e:
+                    return {"alive": False, "latency_ms": None, "status_code": None, "error": str(e)[:120]}
+
+            results_list = await _asyncio.gather(
+                *[_probe_one(data) for data in grouped.values()],
+                return_exceptions=False,
+            )
+            results: Dict[str, Any] = {}
+            for data, r in zip(grouped.values(), results_list):
+                for eid in data["endpoint_ids"]:
+                    results[eid] = r
+
+            _local_probe_cache["data"] = results
+            _local_probe_cache["time"] = _time.time()
+            return results
+
+        task = _asyncio.create_task(_compute_local_probe())
+        _local_probe_inflight["task"] = task
+        try:
+            return await task
+        finally:
+            if _local_probe_inflight.get("task") is task:
+                _local_probe_inflight["task"] = None
 
     @router.get("/ping")
     def ping_endpoints(request: Request):