coderai-system: dedicated worker for cache scan / HF downloads / cache mgmt

Phase 1 of moving non-GPU work off the engine. The front now spawns and
supervises a third process type — coderai-system — a lightweight (torch-free)
worker that owns the slow, I/O-bound, GPU-irrelevant endpoints: cache scan
(cached-models, cache-stats, cache delete), HF downloads (model-download,
download-stream SSE, downloads, cancel), HF lookups (hf-search/files/model-info/
model-files, ds4 defaults), and disk/bookkeeping (model-upload, model-free-disk,
model-add-known, model-mark/unmark-download). The front routes those /admin/api
paths to the worker instead of an engine, so they stay responsive during
generation and survive engine restarts — fixing the models/archive-style hangs
for the cache/download sections.

Mechanics:
- cli.py: --system-only flag. main.py: proc title coderai-system + run branch.
- codai/system_app.py: slim FastAPI = admin_router only + session/config init +
  internal-auth gate + /healthz + /internal/engine-state + reload-config. Stays
  torch-free (the cache/download/HF handlers import no torch).
- engine_supervisor: spawns one role="system" worker on the next internal port
  (_engine_cmd gains a mode arg), health-polls it, and excludes it from model
  assignment (it owns no models) while still queueing it model-json reloads.
- registry: Engine.role; can_serve()==False for role="system" so it's never an
  inference target. Front excludes it from the engine tiles.
- app.py: _SYSTEM_PATHS + _system_engine() + _proxy_passthrough() route the
  cache/download paths to the worker (long client handles JSON + SSE).
- routes.get_current_user now validates ANY session/session_<port> cookie by
  signature, so front/engine/system (different ports) all accept one browser cookie.

Engine still physically defines these handlers (now dead — front routes to the
worker); stripping them is a later cleanup. Verified: slim app serves cache-stats
with cross-port cookie + internal token (403 without), engine-state; front routing
picks the worker and excludes it from inference/tiles.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011DDv7BchtZQWsnPG6Jm49m

codai/admin/routes.py

@@ -225,19 +225,23 @@ def _sync_whisper_runner(model_path: str, model_entry: dict) -> bool:
 
 
 def get_current_user(request: Request) -> Optional[str]:
-    """Get the current logged-in user from session cookie."""
+    """Get the current logged-in user from the session cookie.
+
+    Validates whichever ``session`` / ``session_<port>`` cookie is present by HMAC
+    signature, so the exact (port-derived) cookie name doesn't matter — the front,
+    an engine and the coderai-system worker bind different ports yet must all accept
+    the same browser cookie."""
     if session_manager is None:
         return None
-    
-    cookie = request.cookies.get(SESSION_COOKIE_NAME)
-    if not cookie:
-        return None
-    
-    # Handle MUST_CHANGE flag
-    if cookie.endswith(".MUST_CHANGE"):
-        cookie = cookie[:-12]  # Remove .MUST_CHANGE suffix
-    
-    return session_manager.validate_session(cookie)
+    for k, v in request.cookies.items():
+        if k != "session" and not k.startswith("session_"):
+            continue
+        if v.endswith(".MUST_CHANGE"):
+            v = v[:-12]
+        user = session_manager.validate_session(v)
+        if user:
+            return user
+    return None
 
 
 def require_auth(request: Request) -> str:

codai/cli.py

@@ -310,11 +310,18 @@ configuration directory (--config DIR, default: OS-specific CoderAI directory).
              "front proxy). Normally launched automatically by the front; not "
              "intended to be run by hand.",
     )
+    parser.add_argument(
+        "--system-only",
+        action="store_true",
+        help="Run this process as the coderai-system worker (cache scan, HF "
+             "downloads, cache management) on an internal localhost port. Launched "
+             "automatically by the front; not intended to be run by hand.",
+    )
     parser.add_argument(
         "--internal-port",
         type=int,
         default=None,
-        help="Internal port for --engine-only mode (the front assigns one per engine).",
+        help="Internal port for --engine-only / --system-only mode (assigned by the front).",
     )
     parser.add_argument(
         "--debug-engine",

codai/frontproxy/app.py

@@ -45,6 +45,20 @@ _DROP_REQ = _HOP_BY_HOP | {"host", "content-length", "x-coderai-internal",
 # request, flooding the terminal. Strip them here so each appears exactly once.
 _DROP_RESP = _HOP_BY_HOP | {"content-length", "date", "server"}
 
+# Admin paths handled by the coderai-system worker (cache scan, HF downloads, cache
+# management) rather than a GPU engine — so they stay responsive during generation
+# and survive engine restarts. Matched as substrings of the request path.
+_SYSTEM_PATHS = (
+    "/admin/api/cached-models", "/admin/api/cache-stats", "/admin/api/cache",
+    "/admin/api/model-download", "/admin/api/download-stream",
+    "/admin/api/downloads", "/admin/api/download-cancel",
+    "/admin/api/model-upload", "/admin/api/model-free-disk",
+    "/admin/api/hf-search", "/admin/api/hf-files", "/admin/api/hf-model-info",
+    "/admin/api/hf-model-files", "/admin/api/ds4/default-models",
+    "/admin/api/model-add-known", "/admin/api/model-mark-download",
+    "/admin/api/model-unmark-download",
+)
+
 
 class FrontProxy:
     def __init__(self, config, config_dir=None):
@@ -627,6 +641,8 @@ class FrontProxy:
     def engines_list(self) -> list:
         out = []
         for e in self.registry.all():
+            if getattr(e, "role", "engine") == "system":
+                continue   # the cache/downloads worker isn't a GPU engine tile
             try:
                 pid = e.proc.pid if e.proc else None
             except Exception:
@@ -926,11 +942,52 @@ class FrontProxy:
             })
         return merged
 
+    def _system_engine(self):
+        """The coderai-system worker, if it's up."""
+        for e in self.registry.all():
+            if getattr(e, "role", "engine") == "system" and e.is_alive():
+                return e
+        return None
+
+    async def _proxy_passthrough(self, request: Request, engine) -> Response:
+        """Stream a request through to a specific worker (used for the coderai-system
+        worker). The worker is always responsive, so the long (no-read-timeout)
+        client is safe and handles both buffered JSON and SSE (download-stream)."""
+        method = request.method
+        url = engine.url + request.url.path
+        headers = self._filter_headers(request.headers, _DROP_REQ)
+        content = (request.stream()
+                   if method in ("POST", "PUT", "PATCH") else None)
+        rp_req = self._long.build_request(method, url, headers=headers,
+                                          params=request.query_params, content=content)
+        try:
+            rp_resp = await self._long.send(rp_req, stream=True)
+        except Exception as exc:
+            return JSONResponse(
+                {"error": f"coderai-system worker unreachable: {exc}"}, status_code=502)
+
+        async def _release():
+            await rp_resp.aclose()
+
+        return StreamingResponse(
+            rp_resp.aiter_raw(), status_code=rp_resp.status_code,
+            headers=dict(self._filter_headers(rp_resp.headers, _DROP_RESP)),
+            media_type=rp_resp.headers.get("content-type"),
+            background=BackgroundTask(_release))
+
     # -------------------------------------------------------------------- proxy
     async def proxy(self, request: Request) -> Response:
         path = request.url.path
         method = request.method
 
+        # Cache scan / HF downloads / cache management → the coderai-system worker
+        # (off the GPU engines, always responsive). Falls through to normal routing
+        # if the worker isn't up yet.
+        if any(s in path for s in _SYSTEM_PATHS):
+            sysw = self._system_engine()
+            if sysw is not None:
+                return await self._proxy_passthrough(request, sysw)
+
         # Inference JSON bodies are small: buffer so we can route by `model`, then
         # forward the buffered bytes. Everything else streams through unbuffered.
         body_bytes: Optional[bytes] = None

codai/frontproxy/engine_supervisor.py

@@ -272,10 +272,11 @@ class EngineSupervisor:
         return engines
 
     # ------------------------------------------------------------------ spawning
-    def _engine_cmd(self, port: int) -> list:
-        """Build the command to relaunch this codebase as an engine."""
+    def _engine_cmd(self, port: int, mode: str = "--engine-only") -> list:
+        """Build the command to relaunch this codebase as an engine (or, with
+        mode='--system-only', the coderai-system worker)."""
         # sys.argv[0] is the launcher script (``coderai``); preserve all original
-        # args (config dir, model selection, …) and append the engine flags. Strip
+        # args (config dir, model selection, …) and append the worker flags. Strip
         # any flag that would re-trigger front mode or fix a different port.
         passthrough = []
         skip_next = False
@@ -283,14 +284,14 @@ class EngineSupervisor:
             if skip_next:
                 skip_next = False
                 continue
-            if a == "--engine-only":
+            if a in ("--engine-only", "--system-only"):
                 continue
             if a == "--internal-port":
                 skip_next = True
                 continue
             passthrough.append(a)
         return [sys.executable, sys.argv[0], *passthrough,
-                "--engine-only", "--internal-port", str(port)]
+                mode, "--internal-port", str(port)]
 
     def _spawn(self, engine: Engine) -> None:
         env = dict(os.environ)
@@ -335,7 +336,10 @@ class EngineSupervisor:
             env["CODERAI_ENGINE_BACKEND"] = engine.backend
         # The engine names its own process (coderai-<name>) from this.
         env["CODERAI_ENGINE_NAME"] = str(engine.name)
-        cmd = self._engine_cmd(engine.port)
+        cmd = self._engine_cmd(
+            engine.port,
+            mode="--system-only" if getattr(engine, "role", "engine") == "system"
+                 else "--engine-only")
         tag = engine.name + (f"(gpu{engine.gpu})" if engine.gpu is not None else "")
         print(f"[front] launching {tag} on port {engine.port}: {' '.join(cmd)}", flush=True)
         proc = subprocess.Popen(
@@ -464,6 +468,13 @@ class EngineSupervisor:
         for engine in engines:
             self.registry.add(engine)
             self._spawn(engine)
+        # The coderai-system worker: cache scan + HF downloads + cache management,
+        # off the GPU engines so they never block generation. One instance.
+        sysw = Engine(id=900, gpu=None, port=self._alloc_port(), role="system",
+                      name="system", backend="none")
+        self._system_worker = sysw
+        self.registry.add(sysw)
+        self._spawn(sysw)
         self._poll_thread = threading.Thread(target=self._poll_loop, daemon=True)
         self._poll_thread.start()
         atexit.register(self.stop_all)
@@ -472,8 +483,13 @@ class EngineSupervisor:
         """When models.json changes (admin add/remove model), re-compute the
         per-engine assignment and push it live to every engine + the front's
         router, so /v1/models and routing reflect the change without a restart."""
-        if not self.models_path or len(self.registry.all()) < 2:
+        if not self.models_path:
             return
+        # The coderai-system worker isn't an inference engine — exclude it from
+        # assignment (it owns no models), but still queue it a reload so it re-reads
+        # models.json (its cache view reflects admin add/remove).
+        real = [e for e in self.registry.all()
+                if getattr(e, "role", "engine") != "system"]
         try:
             mtime = os.path.getmtime(self.models_path)
         except OSError:
@@ -481,26 +497,26 @@ class EngineSupervisor:
         if mtime == self._assign_mtime:
             return
         self._assign_mtime = mtime
-        try:
-            from codai.frontproxy.assignment import compute_assignment
-            default_engine = getattr(self.config.server, "default_engine", None)
-            ds4 = getattr(self.config, "ds4", None)
-            assignment = compute_assignment(self.registry.all(), self.models_path,
-                                            default_engine, ds4)
-        except Exception as exc:
-            print(f"[front] live reassignment skipped: {exc}", flush=True)
-            return
+        assignment = {}
+        if len(real) >= 2:
+            try:
+                from codai.frontproxy.assignment import compute_assignment
+                default_engine = getattr(self.config.server, "default_engine", None)
+                ds4 = getattr(self.config, "ds4", None)
+                assignment = compute_assignment(real, self.models_path,
+                                                default_engine, ds4)
+            except Exception as exc:
+                print(f"[front] live reassignment skipped: {exc}", flush=True)
+                assignment = {}
+            for e in real:
+                e.assigned_models = set(assignment.get(e.name, []))
+        # Queue a reload to EVERY worker (engines + system) so they re-read
+        # models.json. Pushing to a mid-generation engine would block on its GIL, so
+        # _flush_pending_reloads() delivers each once that worker is idle.
         for e in self.registry.all():
-            owned = assignment.get(e.name, [])
-            e.assigned_models = set(owned)   # update the front's router (local, instant)
-            # Pushing /internal/reload-config to an engine that's mid-generation would
-            # block on its GIL and time out, freezing this poll thread. Queue the push
-            # instead and let _flush_pending_reloads() deliver it once the engine idles.
-            self._pending_reload[e.name] = owned
-        print(f"[front] models.json changed — assignment recomputed "
-              f"({', '.join(f'{e.name}:{len(assignment.get(e.name, []))}' for e in self.registry.all())}); "
-              f"reload queued until each engine is idle",
-              flush=True)
+            self._pending_reload[e.name] = assignment.get(e.name, [])
+        print(f"[front] models.json changed — reload queued for "
+              f"{', '.join(e.name for e in self.registry.all())}", flush=True)
         self._flush_pending_reloads(client)
 
     def _flush_pending_reloads(self, client) -> None:

codai/frontproxy/registry.py

@@ -55,6 +55,7 @@ class Engine:
     gpu: Optional[int]             # device hint for logs (CUDA/Vulkan index; None = n/a)
     port: int
     primary: bool = False          # the engine that owns admin/auth/config traffic
+    role: str = "engine"           # "engine" (GPU/inference) or "system" (cache/downloads worker)
     name: str = ""                 # human label for logs
     backend: str = "auto"          # nvidia | vulkan | … (forced for this engine)
     env: dict = field(default_factory=dict)        # extra env applied at spawn
@@ -89,6 +90,10 @@ class Engine:
             self.capabilities = set(_DEFAULT_CAPS.get(self.backend, {"transformers", "gguf"}))
 
     def can_serve(self, required_cap: Optional[str]) -> bool:
+        # The system worker (cache/downloads) never serves inference, so it must
+        # never be picked as an inference target — even for cap-less requests.
+        if self.role == "system":
+            return False
         return (not required_cap) or (required_cap in self.capabilities)
 
     def is_alive(self) -> bool:

codai/main.py

@@ -345,6 +345,8 @@ def _set_proc_title():
         _ename = (os.environ.get("CODERAI_ENGINE_NAME")
                   or os.environ.get("CODERAI_ENGINE_BACKEND") or "engine")
         name = f"coderai-{_ename}"
+    elif "--system-only" in _argv:
+        name = "coderai-system"
     else:
         name = "coderai-front"
     try:
@@ -603,10 +605,23 @@ def main():
     #   --engine-only     → this process IS an engine: bind an internal localhost
     #                       port and run the full app below (the front spawns these).
     _engine_only = getattr(args, "engine_only", False)
-    if not _engine_only:
+    _system_only = getattr(args, "system_only", False)
+    if not _engine_only and not _system_only:
         from codai.frontproxy import run_front
         run_front(config, args)
         return
+    if _system_only:
+        # The coderai-system worker: cache scan + HF downloads + cache mgmt only.
+        import uvicorn
+        from codai.system_app import build_system_app
+        _sport = int(getattr(args, "internal_port", None)
+                     or config.server.internal_port_base)
+        _set_proc_title()
+        _sapp = build_system_app(config, config_dir, internal_port=_sport)
+        print(f"[system] coderai-system serving on http://127.0.0.1:{_sport} "
+              f"(internal — reach it via the front)", flush=True)
+        uvicorn.run(_sapp, host="127.0.0.1", port=_sport, log_config=None)
+        return
     if _engine_only:
         # Engines bind plain localhost HTTP; the front owns the public host + TLS.
         # NOTE: don't mutate config.server here — the settings API reads it, and it

codai/system_app.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+"""The ``coderai-system`` worker app.
+
+A lightweight process (spawned + supervised by the front like an engine) that
+owns the slow, I/O-bound, non-GPU work: model cache scanning, HuggingFace
+downloads and lookups, and GGUF/HF cache management. Keeping it out of the engine
+means those operations never block generation and survive engine restarts; the
+front routes the relevant /admin/api/* paths here instead of to an engine.
+
+It mounts only the admin router (whose cache/download/HF handlers import no torch),
+so the worker stays light — but it does NOT need to be torch-free; it simply never
+does GPU work. Sessions and config are initialized from the shared config_dir, the
+same way an engine does.
+"""
+import os
+from pathlib import Path
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+
+
+def build_system_app(config, config_dir, internal_port: int = 0) -> FastAPI:
+    from codai.admin.routes import router as admin_router
+    from codai.admin.routes import init_session_manager, set_config_manager
+    from codai.api.app import _InternalAuthMiddleware, _ForwardedPrefixMiddleware
+    from codai.config import ConfigManager
+
+    app = FastAPI(title="CoderAI System", docs_url=None, redoc_url=None,
+                  openapi_url=None)
+
+    # Sessions + config from the shared config_dir (same auth.json/config.json the
+    # engine and front use), so the cookie the front forwards validates here too.
+    try:
+        init_session_manager(Path(config_dir), port=internal_port or 0)
+    except Exception as exc:
+        print(f"[system] session manager init failed: {exc}", flush=True)
+    try:
+        cm = ConfigManager(str(config_dir))
+        cm.load()
+        set_config_manager(cm)
+    except Exception as exc:
+        print(f"[system] config manager init failed: {exc}", flush=True)
+
+    app.add_middleware(_ForwardedPrefixMiddleware)
+    app.add_middleware(_InternalAuthMiddleware)   # only the front (internal token) may call us
+
+    static_dir = Path(__file__).resolve().parent / "admin" / "static"
+    if static_dir.exists():
+        from fastapi.staticfiles import StaticFiles
+        app.mount("/static/admin", StaticFiles(directory=str(static_dir)),
+                  name="admin_static")
+
+    @app.get("/healthz", include_in_schema=False)
+    async def _healthz():
+        return {"ok": True, "role": "system"}
+
+    @app.get("/internal/engine-state", include_in_schema=False)
+    async def _engine_state():
+        # The front's supervisor health-polls this. The system worker holds no
+        # models and no GPU, so report an empty, always-cheap snapshot.
+        return {"ok": True, "pid": os.getpid(), "role": "system",
+                "loaded_models": [], "vram": None, "tasks": [], "cooling": None}
+
+    @app.post("/internal/reload-config", include_in_schema=False)
+    async def _reload_config(request: Request):
+        # Re-read config/models.json so cache scans + downloads see admin edits.
+        try:
+            cm2 = ConfigManager(str(config_dir))
+            cm2.load()
+            set_config_manager(cm2)
+        except Exception as exc:
+            return JSONResponse({"ok": False, "error": str(exc)}, status_code=500)
+        return JSONResponse({"ok": True})
+
+    app.include_router(admin_router, tags=["Admin"])
+    return app