Add dedicated OCR subsystem + finalize multi-engine backends

OCR (v0.1.85): new codai/ocr/ subsystem — dedicated OCR engines (NOT a VLM):
- docTR (in-process, GPU) + PaddleOCR & Surya (isolated-venv subprocess workers
  via codai/ocr/workers/ocr_worker.py to avoid opencv-contrib/pillow<11 clashes;
  paddle-gpu bundles its own CUDA so a cu12x wheel runs on newer-CUDA hosts).
- /v1/ocr, /v1/ocr/batch, /v1/ocr/schemas CRUD; `ocr` pipeline step.
- Data-driven, user-extensible extraction schemas (named files + inline + auto),
  built-in seeds (italian_sentenza/generic_document/invoice); optional JSON Schema
  validation. Structured extraction via an existing coderai text model.
- Configurable stamp/signature detection (off|layout|detector|both); bundled
  signature YOLO default; layout markers EN+IT (timbro/sigillo/firma).
- Admin OCR card: engines/instances/detect/extraction + schema picker/raw-JSON/
  field-builder + isolated-venv build trigger (background build + status).
- Isolated venvs resolve baked(/opt) > /cache mount (persistent) > ~/.coderai.
- Deps: pypdfium2 in base requirements; requirements-ocr(.txt/-paddle.txt);
  requirements-surya.txt; build.sh --ocr.

Also lands the multi-engine backend work: colibri multi-family (GLM/DeepSeek/Kimi),
kimi-k3-in-c (k3), ktransformers (kt) — per-model selectable via resolve_engine_backend;
plus a deferred vLLM backend design note (docs/vllm.md).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Mw2KQiswmD69T45fTfjKwW

README.md

@@ -699,13 +699,23 @@ thanks to the developers of the native inference engines that let it punch far a
 its hardware — running models that otherwise simply couldn't run on a single GPU:
 
 - **[colibri](https://github.com/JustVugg/colibri)** by **JustVugg** — a brilliant
-  pure-C MoE engine that streams **GLM-5.2 (744B params)** across VRAM/RAM/disk to run
-  it on a single consumer GPU.
+  pure-C multi-family MoE engine that streams **GLM-5.2, DeepSeek-V4 and Kimi-K3**
+  across VRAM/RAM/disk to run them on a single consumer GPU (or CPU).
 - **[ds4 / DwarfStar](https://github.com/antirez/ds4)** by **Salvatore Sanfilippo
   (antirez)** — a superb from-scratch **DeepSeek-V4** inference engine.
+- **[kimi-k3-in-c](https://github.com/FareedKhan-dev/kimi-k3-in-c)** by
+  **Fareed Khan** — a portable-C engine that runs **Kimi-K3 (2.78T params)** on CPU by
+  streaming the dense trunk + routed experts from disk in as little as ~8 GB RAM.
+- **[ktransformers](https://github.com/kvcache-ai/ktransformers)** by **KVCache.AI** —
+  a CPU+GPU heterogeneous engine (served via SGLang) for large MoE models
+  (DeepSeek / Kimi / Qwen / GLM / MiniMax).
 - **[llama.cpp](https://github.com/ggml-org/llama.cpp)** &
   **[whisper.cpp](https://github.com/ggml-org/whisper.cpp)** by **Georgi Gerganov**
   and contributors — the foundational GGUF LLM inference and Whisper STT engines.
+- **[PaddleOCR](https://github.com/PaddlePaddle/PaddleOCR)** (Baidu),
+  **[docTR](https://github.com/mindee/doctr)** (Mindee) and
+  **[Surya](https://github.com/VikParuchuri/surya)** — the dedicated OCR engines behind
+  the `/v1/ocr` document-transcription subsystem (text + layout + structured extraction).
 
 And the libraries, models and research CoderAI builds on:
 
@@ -717,6 +727,7 @@ And the libraries, models and research CoderAI builds on:
 - [F5-TTS](https://github.com/SWivid/F5-TTS) — voice cloning
 - [Seed-VC](https://github.com/Plachta/Seed-VC) — singing voice conversion
 - [Real-ESRGAN](https://github.com/xinntao/Real-ESRGAN) — image/video upscaling
+- [pypdfium2](https://github.com/pypdfium2-team/pypdfium2) — PDF rasterisation for OCR; [Ultralytics YOLO](https://github.com/ultralytics/ultralytics) — stamp/signature detection
 - [Wav2Lip](https://github.com/Rudrabha/Wav2Lip) — audio-driven lip sync
 - [SadTalker](https://github.com/OpenTalker/SadTalker) — talking head / lip sync generation
 - Visual place recognition & geolocation research — [EigenPlaces](https://github.com/gmberton/EigenPlaces) (Gabriele Berton et al.), [DINOv2-SALAD](https://github.com/serizba/salad) (Sergio Izquierdo, Javier Civera), [GeoCLIP](https://github.com/VicenteVivan/geo-clip) (Vicente Vivanco et al.), [DINOv2](https://github.com/facebookresearch/dinov2) (Meta AI)

build.sh

@@ -37,6 +37,8 @@ CUSTOM_VENV=""
 PACKAGE=false
 DS4=false
 COLIBRI=false
+K3=false
+OCR=false
 
 # Parse arguments
 i=1
@@ -58,6 +60,12 @@ for arg in "$@"; do
         --colibri)
             COLIBRI=true
             ;;
+        --k3)
+            K3=true
+            ;;
+        --ocr)
+            OCR=true
+            ;;
     esac
     i=$((i + 1))
 done
@@ -77,7 +85,9 @@ if [[ "$BACKEND" != "nvidia" && "$BACKEND" != "vulkan" && "$BACKEND" != "vulkan-
     echo "Options:"
     echo "  --flash     - Install Flash Attention 2 for faster inference (NVIDIA only)"
     echo "  --ds4       - Clone + build the ds4 (DeepSeek V4) native engine"
-    echo "  --colibri   - Clone + build the colibri (GLM-5.2) native engine"
+    echo "  --colibri   - Clone + build the colibri (GLM-5.2 / DeepSeek-V4 / Kimi-K3) engines"
+    echo "  --k3        - Clone + patch + build the kimi-k3-in-c (Kimi-K3) native engine"
+    echo "  --ocr       - Install the dedicated OCR engine stack (requirements-ocr.txt: PaddleOCR/docTR/YOLO)"
     exit 1
 fi
 
@@ -827,15 +837,81 @@ build_colibri() {
     ( cd "$COLIBRI_DIR/c" && make -s $MAKE_ARGS ) || {
         echo -e "${YELLOW}Warning: colibri build failed; it can still be built at runtime.${NC}"; return 0; }
     if [ -x "$COLIBRI_DIR/c/colibri" ]; then
-        echo -e "${GREEN}✓ colibri built at $COLIBRI_DIR/c/colibri${NC}"
+        echo -e "${GREEN}✓ colibri (GLM-5.2) built at $COLIBRI_DIR/c/colibri${NC}"
         echo -e "${YELLOW}Note: the GLM-5.2 int4 container (~372 GB dir) is NOT downloaded; point colibri.model_path at it.${NC}"
     fi
+
+    # colibri v1.5.0 ships sibling engines for other families, each its own binary +
+    # make target. Build them too (non-fatal) so a single install serves DeepSeek-V4 and
+    # Kimi-K3 as well as GLM-5.2. Both are CPU engines that stream experts from disk:
+    # DeepSeek-V4 is AVX2 (no CUDA path, distinct from the separate ds4 CUDA engine) and
+    # Kimi-K3 is CPU + an opt-in Vulkan tier (COLI_K3_VK=1 → make VK=1, needs
+    # glslc/shaderc). The multi-GB model containers are NOT downloaded here.
+    ( cd "$COLIBRI_DIR/c" && make -s deepseek-v4 ) \
+        && [ -x "$COLIBRI_DIR/c/deepseek_v4" ] \
+        && echo -e "${GREEN}✓ colibri deepseek_v4 (CPU/AVX2) built at $COLIBRI_DIR/c/deepseek_v4${NC}" \
+        || echo -e "${YELLOW}Warning: colibri deepseek_v4 build skipped/failed (built on demand at runtime).${NC}"
+
+    local K3_ARGS="kimi_k3"
+    if [ "${COLI_K3_VK:-0}" = "1" ]; then K3_ARGS="kimi_k3 VK=1"; fi
+    ( cd "$COLIBRI_DIR/c" && make -s $K3_ARGS ) \
+        && [ -x "$COLIBRI_DIR/c/kimi_k3" ] \
+        && echo -e "${GREEN}✓ colibri kimi_k3 built at $COLIBRI_DIR/c/kimi_k3${NC}" \
+        || echo -e "${YELLOW}Warning: colibri kimi_k3 build skipped/failed (built on demand at runtime).${NC}"
 }
 
 if [ "$COLIBRI" = true ]; then
     build_colibri
 fi
 
+# Optionally clone + patch + build kimi-k3-in-c (Kimi-K3 native engine). Opt-in via
+# --k3. coderai patches in a resident serve loop (packaging/patch-k3.py) that speaks the
+# colibri mux protocol, then drives it via MuxEngine. CPU-only (AVX2+FMA); the ~1.56 TB
+# checkpoint + ~109 GB packed trunk are NOT downloaded here.
+build_k3() {
+    local K3_DIR="${CODERAI_K3_DIR:-$HOME/.coderai/kimi-k3-in-c}"
+    echo -e "${YELLOW}Building kimi-k3-in-c (Kimi-K3 engine) → $K3_DIR ...${NC}"
+    if [ ! -e "$K3_DIR/Makefile" ]; then
+        mkdir -p "$(dirname "$K3_DIR")"
+        git clone --depth 1 https://github.com/FareedKhan-dev/kimi-k3-in-c "$K3_DIR" || {
+            echo -e "${YELLOW}Warning: could not clone kimi-k3-in-c; skipping.${NC}"; return 0; }
+    fi
+    # Apply coderai's resident serve-loop patch (idempotent) so the engine stays warm.
+    if [ -f "$(dirname "$0")/packaging/patch-k3.py" ] && [ -f "$K3_DIR/src/cli/k3_run.c" ]; then
+        python3 "$(dirname "$0")/packaging/patch-k3.py" "$K3_DIR/src/cli/k3_run.c" || \
+            echo -e "${YELLOW}Warning: k3 serve-loop patch failed (serve mode will not work).${NC}"
+    fi
+    ( cd "$K3_DIR" && make -j ) || {
+        echo -e "${YELLOW}Warning: kimi-k3-in-c build failed; it can still be built at runtime.${NC}"; return 0; }
+    if [ -x "$K3_DIR/bin/k3" ]; then
+        echo -e "${GREEN}✓ kimi-k3-in-c built at $K3_DIR/bin/k3${NC}"
+        echo -e "${YELLOW}Note: the Kimi-K3 checkpoint (~1.56 TB) + packed trunk are NOT downloaded; point k3.model_path/k3.trunk_dir at them.${NC}"
+    fi
+}
+
+if [ "$K3" = true ]; then
+    build_k3
+fi
+
+# Optionally install the dedicated OCR engine stack (PaddleOCR/docTR/YOLO). Opt-in via
+# --ocr. Kept out of the base requirements so the default install/image stay lean; the
+# /v1/ocr subsystem returns HTTP 503 for any engine whose dependency is absent.
+install_ocr() {
+    echo -e "${YELLOW}Installing OCR engine stack (requirements-ocr.txt)...${NC}"
+    if [ -f "$(dirname "$0")/requirements-ocr.txt" ]; then
+        pip install -r "$(dirname "$0")/requirements-ocr.txt" || {
+            echo -e "${YELLOW}Warning: some OCR deps failed (PaddlePaddle is CUDA-version-specific).${NC}"
+            echo -e "${YELLOW}         See requirements-ocr.txt for the correct paddlepaddle-gpu wheel.${NC}"
+        }
+    else
+        echo -e "${YELLOW}Warning: requirements-ocr.txt not found; skipping OCR stack.${NC}"
+    fi
+}
+
+if [ "$OCR" = true ]; then
+    install_ocr
+fi
+
 # Create .backend file to track which backend was used
 echo "$BACKEND" > .backend
 

codai/__init__.py

@@ -16,7 +16,7 @@
 
 # Canonical product version for CoderAI — single source of truth. Both the API
 # metadata and the admin web UI read from here.
-__version__ = "0.1.83"
+__version__ = "0.1.85"
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.
 # expandable_segments lets the allocator return freed pages to the driver even

codai/admin/templates/models.html

@@ -551,7 +551,10 @@ window.__DEFAULT_WHISPER_SERVER_PATH__ = {{ default_whisper_server_path|tojson }
             <option value="vulkan">Vulkan</option>
             <option value="opencl">OpenCL</option>
             <option value="cpu">CPU only</option>
-            <option value="colibri">colibri (GLM-5.2 C engine)</option>
+            <option value="colibri">colibri (GLM-5.2 / DeepSeek-V4 / Kimi-K3 C engine)</option>
+            <option value="ds4">ds4 (DeepSeek-V4 CUDA engine)</option>
+            <option value="k3">k3 (Kimi-K3 / kimi-k3-in-c)</option>
+            <option value="kt">ktransformers (SGLang, CPU+GPU)</option>
           </select>
         </div>
         <div class="form-row" style="margin:0">

codai/api/app.py

@@ -142,6 +142,7 @@ from codai.api.characters import router as characters_router
 from codai.api.loras import router as loras_router
 from codai.api.spatial import router as spatial_router
 from codai.api.environments import router as environments_router
+from codai.api.ocr import router as ocr_router
 from codai.admin.routes import router as admin_router
 
 # Import and add middleware
@@ -574,6 +575,7 @@ app.include_router(characters_router, tags=["Characters"])
 app.include_router(loras_router, tags=["LoRAs"])
 app.include_router(environments_router, tags=["Environments"])
 app.include_router(spatial_router, tags=["Spatial / 3D"])
+app.include_router(ocr_router, tags=["OCR"])
 app.include_router(admin_router, tags=["Admin"])
 
 

codai/api/custom_pipelines.py

@@ -89,6 +89,7 @@ STEP_TYPES = {
     "audio_gen":      ("codai.api.audio_gen",     "audio_generate",         None),
     "voice_clone":    ("codai.api.voice_clone",   "clone_voice",            None),
     "voice_convert":  ("codai.api.voice_convert", "convert_voice",          None),
+    "ocr":            ("codai.api.ocr",           "run_ocr_step",           None),
 }
 
 # Human-readable labels for the UI
@@ -112,6 +113,7 @@ STEP_TYPE_LABELS = {
     "audio_gen":      "Audio/Music Generation",
     "voice_clone":    "Voice Clone (TTS)",
     "voice_convert":  "Voice Convert (SVC)",
+    "ocr":            "OCR (document → text)",
 }
 
 # Which params each step type accepts (for the UI form builder)
@@ -136,6 +138,7 @@ STEP_PARAMS = {
     "audio_cleanup":  [("audio","ref","Source audio"),("noise_reduction","checkbox","Reduce background noise"),("normalize","checkbox","Normalize levels"),("remove_hum","checkbox","Remove hum"),("repair_clicks","checkbox","Repair clicks")],
     "voice_clone":    [("text","textarea","Text to synthesize"),("voice_name","text","Voice profile name"),("ref_text","text","Reference transcript"),("speed","number","Speed","1.0")],
     "voice_convert":  [("source_audio","ref","Source audio"),("voice_name","text","Voice profile name"),("f0_condition","checkbox","Singing mode"),("pitch_shift","number","Pitch shift","0"),("diffusion_steps","number","Steps","10")],
+    "ocr":            [("image","ref","Document image/PDF ({{stepN.url}})"),("engine","select:paddle|doctr|surya","OCR engine","paddle"),("detect","select:off|layout|detector|both","Stamp/signature detect","off"),("structured","checkbox","Structured JSON extraction")],
 }
 
 

codai/api/k3_worker.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.
+
+"""Fully-managed Kimi-K3 worker — the kimi-k3-in-c C engine, driven directly.
+
+kimi-k3-in-c (https://github.com/FareedKhan-dev/kimi-k3-in-c) is a portable-C CPU
+engine for Kimi-K3 that streams the dense trunk + routed experts from disk. Upstream it
+is a one-shot batch CLI; coderai applies ``packaging/patch-k3.py`` to add a resident
+serve loop that speaks the SAME mux stdin/stdout protocol as colibri, so we reuse
+:class:`~codai.api.colibri_worker.MuxEngine` as the protocol client. This module owns the
+build + process lifecycle; the Kimi-K3 chat template lives in :mod:`codai.backends.k3`.
+
+Lifecycle (mirrors the other managed engines):
+
+* :func:`ensure_built` clones the repo, applies the serve-loop patch, and runs ``make``
+  so the ``bin/k3`` binary exists (idempotent).
+* :func:`ensure_engine` launches the engine on the configured checkpoint in serve mode
+  (``SERVE=1`` + the model dir / ``--trunk`` / ``--tok`` / budgets as argv), completes
+  the ``READY`` handshake, and returns a live :class:`MuxEngine`.
+"""
+
+import collections
+import os
+import shutil
+import subprocess
+import threading
+import time
+from pathlib import Path
+from typing import Optional
+
+from codai.api.colibri_worker import MuxEngine, _run_logged
+
+_lock = threading.RLock()
+_services: dict[str, MuxEngine] = {}
+
+
+def default_install_dir() -> Path:
+    return Path(os.environ.get("CODERAI_K3_DIR")
+               or os.path.expanduser("~/.coderai/kimi-k3-in-c"))
+
+
+def _install_dir(cfg) -> Path:
+    return Path(cfg.install_dir).expanduser() if getattr(cfg, "install_dir", None) \
+        else default_install_dir()
+
+
+def _engine_bin(install_dir: Path) -> Path:
+    """The k3 engine binary (``bin/k3`` in the repo root)."""
+    for name in ("k3", "k3.exe"):
+        cand = install_dir / "bin" / name
+        if cand.exists():
+            return cand
+    return install_dir / "bin" / "k3"
+
+
+def _patch_script() -> Optional[str]:
+    """Path to coderai's serve-loop patch (packaging/patch-k3.py), or None."""
+    here = Path(__file__).resolve()
+    for base in (here.parents[2], here.parents[1]):   # repo root, then codai/
+        cand = base / "packaging" / "patch-k3.py"
+        if cand.exists():
+            return str(cand)
+    return None
+
+
+def ensure_built(cfg) -> Path:
+    """Clone + patch + build kimi-k3-in-c if the ``bin/k3`` binary is missing."""
+    install_dir = _install_dir(cfg)
+    binary = _engine_bin(install_dir)
+    if binary.exists():
+        return binary
+    if not getattr(cfg, "auto_build", True):
+        raise RuntimeError(
+            f"k3 engine not found at {binary} and auto_build is disabled. Build it "
+            f"manually (git clone {cfg.repo_url}; python3 packaging/patch-k3.py "
+            f"src/cli/k3_run.c; make) or enable auto_build.")
+    tail = collections.deque(maxlen=40)
+    install_dir.parent.mkdir(parents=True, exist_ok=True)
+    if not (install_dir / ".git").exists() and not (install_dir / "Makefile").exists():
+        print(f"[k3] cloning {cfg.repo_url} → {install_dir} …", flush=True)
+        _run_logged(["git", "clone", "--depth", "1", cfg.repo_url, str(install_dir)],
+                    cwd=install_dir.parent, label="git clone", tail=tail)
+    # Apply the resident serve-loop patch (idempotent) so the engine can stay warm.
+    patch = _patch_script()
+    k3_run = install_dir / "src" / "cli" / "k3_run.c"
+    if patch and k3_run.exists():
+        try:
+            _run_logged(["python3", patch, str(k3_run)],
+                        cwd=install_dir, label="patch-k3", tail=tail)
+        except Exception as exc:
+            print(f"[k3] warning: serve-loop patch failed ({exc}); building unpatched "
+                  "(serve mode will not work)", flush=True)
+    print("[k3] building engine (make) — CPU/AVX2, this can take a moment …", flush=True)
+    _run_logged(["make", "-j"], cwd=install_dir, label="make", tail=tail)
+    binary = _engine_bin(install_dir)
+    if not binary.exists():
+        raise RuntimeError("k3 build completed but bin/k3 is missing. Last output: "
+                           + " | ".join(list(tail)[-5:]))
+    print(f"[k3] built {binary}", flush=True)
+    return binary
+
+
+def _resolve_checkpoint(cfg, model_dir: Optional[str]) -> str:
+    """The Kimi-K3 checkpoint directory (config.json + tokenizer + shards)."""
+    for cand in (model_dir, getattr(cfg, "model_path", "") or ""):
+        cand = os.path.expanduser((cand or "").strip())
+        if cand and os.path.isdir(cand):
+            return os.path.abspath(cand)
+    return ""
+
+
+def resolve_service_key(cfg, model_dir: Optional[str] = None):
+    ckpt = _resolve_checkpoint(cfg, model_dir)
+    return ckpt, (ckpt or (getattr(cfg, "model_id", "kimi-k3") or "kimi-k3"))
+
+
+def _build_argv(cfg, binary: Path, ckpt: str) -> list:
+    """The k3 serve command line: model dir + trunk/tok/budget flags."""
+    argv = [str(binary), ckpt]
+    trunk = os.path.expanduser((getattr(cfg, "trunk_dir", "") or "").strip())
+    if trunk:
+        argv += ["--trunk", trunk]
+    tok = os.path.expanduser((getattr(cfg, "tok_dir", "") or "").strip()) or ckpt
+    argv += ["--tok", tok]
+    preset = (getattr(cfg, "preset", "") or "").strip()
+    if preset:
+        argv += ["--preset", preset]
+    else:
+        argv += ["--trunk-gb", str(getattr(cfg, "trunk_gb", 16.0) or 16.0),
+                 "--cache-gb", str(getattr(cfg, "cache_gb", 64.0) or 64.0)]
+    extra = (getattr(cfg, "extra_args", "") or "").strip()
+    if extra:
+        import shlex
+        argv += shlex.split(extra)
+    return argv
+
+
+def _build_env(cfg, ctx: int) -> dict:
+    env = os.environ.copy()
+    env["SERVE"] = "1"
+    env["K3_MAXT"] = str(int(ctx) if ctx else int(getattr(cfg, "ctx", 4096) or 4096))
+    extra_env = (getattr(cfg, "extra_env", "") or "").strip()
+    if extra_env:
+        import shlex
+        for tok in shlex.split(extra_env):
+            if "=" in tok:
+                k, v = tok.split("=", 1)
+                if k.strip():
+                    env[k.strip()] = v
+    return env
+
+
+def ensure_engine(cfg, model_dir: Optional[str] = None, ctx: Optional[int] = None,
+                  ready_timeout: float = 3600.0) -> MuxEngine:
+    """Build (as needed), then start (or reuse) the k3 engine for a checkpoint."""
+    ckpt, svc_key = resolve_service_key(cfg, model_dir)
+    with _lock:
+        eng = _services.get(svc_key)
+        if eng and eng.is_alive():
+            return eng
+        if eng and not eng.is_alive():
+            eng.close()
+            _services.pop(svc_key, None)
+
+        binary = ensure_built(cfg)
+        if not ckpt:
+            raise RuntimeError(
+                "k3: no Kimi-K3 checkpoint resolved for this request. Point the model at "
+                "the checkpoint directory (or set k3.model_path). There is no auto-"
+                "download of the ~1.56 TB checkpoint.")
+        try:
+            maxt = int(ctx) if ctx else 0
+        except (TypeError, ValueError):
+            maxt = 0
+        if maxt <= 0:
+            maxt = int(getattr(cfg, "ctx", 4096) or 4096)
+
+        argv = _build_argv(cfg, binary, ckpt)
+        env = _build_env(cfg, maxt)
+        print(f"[k3] launching engine {' '.join(argv)} (K3_MAXT={maxt})", flush=True)
+        eng = MuxEngine(binary, ckpt, max_tokens=maxt, kv_slots=1, env=env,
+                        force_mux=True, family="kimi_k3", argv=argv,
+                        inject_colibri_env=False)
+        _services[svc_key] = eng
+
+    deadline = time.time() + ready_timeout
+    while time.time() < deadline:
+        if not eng.is_alive():
+            tail = eng.log_tail()
+            stop_service(svc_key)
+            raise RuntimeError("k3 engine exited before serving"
+                               + (f". Last output: {tail}" if tail else ""))
+        print(f"[k3] engine ready for {svc_key}", flush=True)
+        return eng
+    stop_service(svc_key)
+    raise RuntimeError(f"k3 engine for {svc_key} did not become ready in time")
+
+
+def stop_service(model_id: str) -> None:
+    with _lock:
+        eng = _services.pop(model_id, None)
+    if not eng:
+        return
+    try:
+        eng.close()
+    except Exception:
+        pass
+    print(f"[k3] engine for {model_id} stopped", flush=True)
+
+
+def stop_all() -> None:
+    for mid in list(_services.keys()):
+        stop_service(mid)
+
+
+import atexit as _atexit
+_atexit.register(stop_all)

codai/api/kt_worker.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.
+
+"""Fully-managed ktransformers worker — the SGLang OpenAI server, driven as a subprocess.
+
+ktransformers (https://github.com/kvcache-ai/ktransformers) serves through SGLang's
+OpenAI-compatible HTTP server, so — like :mod:`codai.api.ds4_worker` — coderai launches
+``python -m sglang.launch_server`` as a managed subprocess, health-checks its
+``/v1/models`` endpoint, and the backend (:mod:`codai.backends.ktransformers`) proxies
+chat/completions to it. This module owns the process lifecycle only.
+
+Heavy dependencies (SGLang + kt-kernel) are expected to be installed out of band; with
+``auto_build`` a best-effort ``pip install`` is attempted, else a clear error is raised.
+"""
+
+import collections
+import os
+import shlex
+import socket
+import subprocess
+import sys
+import threading
+import time
+from typing import Optional
+
+_lock = threading.RLock()
+_services: dict[str, dict] = {}   # svc_key -> {"proc","port","url"}
+
+
+def _free_port() -> int:
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.bind(("127.0.0.1", 0))
+    port = s.getsockname()[1]
+    s.close()
+    return port
+
+
+def _pump_logs(proc, tail):
+    for line in proc.stdout:
+        line = line.rstrip()
+        if line:
+            tail.append(line)
+            print(f"[kt] {line}", flush=True)
+
+
+def _health_ok(url: str) -> bool:
+    import requests
+    try:
+        r = requests.get(url + "/v1/models", timeout=3)
+        return r.status_code == 200
+    except Exception:
+        return False
+
+
+def _sglang_available() -> bool:
+    import importlib.util
+    return importlib.util.find_spec("sglang") is not None
+
+
+def ensure_built(cfg) -> None:
+    """Ensure SGLang (the ktransformers serving frontend) is importable.
+
+    ktransformers' kt-kernel is a native build; coderai does not compile it per-request.
+    With ``auto_build`` we attempt a best-effort ``pip install sglang``, but the kt-kernel
+    wheel/build must be provided out of band — otherwise raise a clear, actionable error.
+    """
+    if _sglang_available():
+        return
+    if not getattr(cfg, "auto_build", False):
+        raise RuntimeError(
+            "ktransformers needs SGLang (python -m sglang.launch_server) which is not "
+            "installed, and auto_build is disabled. Install SGLang + kt-kernel out of "
+            "band (see the ktransformers docs), or enable ktransformers.auto_build.")
+    print("[kt] SGLang not found; attempting pip install (heavy) …", flush=True)
+    tail = collections.deque(maxlen=40)
+    try:
+        proc = subprocess.Popen([sys.executable, "-m", "pip", "install", "sglang"],
+                                stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
+                                bufsize=1)
+        for line in proc.stdout:
+            line = line.rstrip()
+            if line:
+                tail.append(line)
+                print(f"[kt] {line}", flush=True)
+        proc.wait()
+    except Exception as exc:
+        raise RuntimeError(f"ktransformers: pip install sglang failed: {exc}")
+    if not _sglang_available():
+        raise RuntimeError(
+            "ktransformers: SGLang still not importable after pip install. Install "
+            "SGLang + kt-kernel manually per the ktransformers docs. Last output: "
+            + " | ".join(list(tail)[-5:]))
+
+
+def resolve_service_key(cfg, model_path: Optional[str] = None):
+    mp = os.path.expanduser((model_path or getattr(cfg, "model_path", "") or "").strip())
+    key = mp or (getattr(cfg, "model_id", "ktransformers") or "ktransformers")
+    return mp, key
+
+
+def _launch_cmd(cfg, host: str, port: int, model_path: str) -> list:
+    mid = (getattr(cfg, "model_id", "ktransformers") or "ktransformers")
+    cmd = [sys.executable, "-m", "sglang.launch_server",
+           "--host", host, "--port", str(port),
+           "--model", model_path,
+           "--served-model-name", mid]
+    ktw = os.path.expanduser((getattr(cfg, "kt_weight_path", "") or "").strip())
+    if ktw:
+        cmd += ["--kt-weight-path", ktw]
+    ctx = int(getattr(cfg, "ctx", 0) or 0)
+    if ctx > 0:
+        cmd += ["--context-length", str(ctx)]
+    extra = (getattr(cfg, "extra_args", "") or "").strip()
+    if extra:
+        cmd += shlex.split(extra)
+    return cmd
+
+
+def ensure_service(cfg, model_path: Optional[str] = None,
+                   ready_timeout: float = 3600.0) -> str:
+    """Launch (or reuse) the SGLang server for a ktransformers model; return its base URL."""
+    resolved, svc_key = resolve_service_key(cfg, model_path)
+    with _lock:
+        svc = _services.get(svc_key)
+        if svc and svc["proc"].poll() is None and _health_ok(svc["url"]):
+            return svc["url"]
+        if svc:
+            _services.pop(svc_key, None)   # died — restart below
+
+        ensure_built(cfg)
+        if not resolved or not os.path.isdir(resolved):
+            raise RuntimeError(
+                "ktransformers: no model directory resolved for this request. Point the "
+                "model at the HF model dir (or set ktransformers.model_path); also set "
+                "ktransformers.kt_weight_path. There is no auto-download.")
+
+        host = (getattr(cfg, "host", "127.0.0.1") or "127.0.0.1").strip()
+        port = int(getattr(cfg, "port", 0) or 0) or _free_port()
+        # Health/proxy over loopback even when SGLang is bound to 0.0.0.0.
+        url_host = "127.0.0.1" if host in ("0.0.0.0", "") else host
+        url = f"http://{url_host}:{port}"
+        cmd = _launch_cmd(cfg, host, port, resolved)
+
+        env = os.environ.copy()
+        applied_env = {}
+        extra_env = (getattr(cfg, "extra_env", "") or "").strip()
+        if extra_env:
+            for tok in shlex.split(extra_env):
+                if "=" in tok:
+                    k, v = tok.split("=", 1)
+                    if k.strip():
+                        env[k.strip()] = v
+                        applied_env[k.strip()] = v
+        env_note = ("  (" + " ".join(f"{k}={v}" for k, v in applied_env.items()) + ")"
+                    if applied_env else "")
+        print(f"[kt] launching SGLang: {' '.join(cmd)}{env_note}", flush=True)
+        tail = collections.deque(maxlen=60)
+        proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                                text=True, bufsize=1, env=env)
+        threading.Thread(target=_pump_logs, args=(proc, tail), daemon=True).start()
+        _services[svc_key] = {"proc": proc, "port": port, "url": url}
+
+    def _tail_msg():
+        joined = " | ".join(list(tail)[-5:]).strip()
+        return f". Last output: {joined}" if joined else ""
+
+    deadline = time.time() + ready_timeout
+    while time.time() < deadline:
+        if proc.poll() is not None:
+            stop_service(svc_key)
+            raise RuntimeError(
+                f"SGLang exited (code {proc.returncode}) before becoming ready" + _tail_msg())
+        if _health_ok(url):
+            print(f"[kt] service ready for {svc_key} at {url}", flush=True)
+            return url
+        time.sleep(2)
+    stop_service(svc_key)
+    raise RuntimeError(f"SGLang for {svc_key} did not become ready in time" + _tail_msg())
+
+
+def stop_service(model_id: str) -> None:
+    with _lock:
+        svc = _services.pop(model_id, None)
+    if not svc:
+        return
+    proc = svc["proc"]
+    if proc.poll() is None:
+        try:
+            proc.terminate()
+            proc.wait(timeout=10)
+        except Exception:
+            pass
+    if proc.poll() is None:
+        try:
+            proc.kill()
+        except Exception:
+            pass
+    print(f"[kt] service for {model_id} stopped", flush=True)
+
+
+def stop_all() -> None:
+    for mid in list(_services.keys()):
+        stop_service(mid)
+
+
+import atexit as _atexit
+_atexit.register(stop_all)

codai/api/ocr.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.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""Dedicated OCR endpoint.
+
+POST /v1/ocr — OCR an uploaded image or PDF with a purpose-built OCR engine
+(PaddleOCR / docTR / Surya), returning faithful text plus per-line boxes and layout.
+NOT a vision LLM. Structured-JSON extraction and stamp/signature detection are layered
+on in later milestones (fields already accepted, no-op until then).
+"""
+
+import asyncio
+from typing import List, Optional
+
+from fastapi import APIRouter, HTTPException, UploadFile, File, Form
+from pydantic import BaseModel, ConfigDict, Field
+
+from codai.ocr.base import OcrError
+from codai.ocr.manager import ocr_manager
+from codai.ocr.schemas import schema_store
+
+router = APIRouter()
+
+# Maximum upload size: 100 MB (multi-page scanned PDFs can be large).
+_MAX_OCR_BYTES = 100 * 1024 * 1024
+
+global_args = None
+
+
+def set_global_args(args):
+    """Set global args from coderai."""
+    global global_args
+    global_args = args
+
+
+def _ocr_config():
+    """Return the live OcrConfig from the admin config manager (or None)."""
+    try:
+        from codai.admin.routes import config_manager
+        if config_manager is not None and getattr(config_manager, "config", None) is not None:
+            return config_manager.config.ocr
+    except Exception:
+        pass
+    return None
+
+
+@router.post("/v1/ocr", summary="OCR an image or PDF with a dedicated OCR engine")
+async def create_ocr(
+    file: UploadFile = File(...),
+    engine: Optional[str] = Form(None),        # paddle|doctr|surya (default from config)
+    dpi: Optional[int] = Form(None),           # PDF rasterisation DPI
+    lang: Optional[str] = Form(None),          # reserved (per-engine language override)
+    structured: Optional[bool] = Form(False),  # structured JSON extraction via text model
+    schema_: Optional[str] = Form(None, alias="schema"),  # override extraction schema (JSON string)
+    detect: Optional[str] = Form(None),        # off|layout|detector|both
+):
+    cfg = _ocr_config()
+    if cfg is None:
+        raise HTTPException(status_code=503, detail="OCR subsystem is not available")
+    ocr_manager.configure(cfg)
+
+    data = await file.read()
+    if not data:
+        raise HTTPException(status_code=400, detail="empty upload")
+    if len(data) > _MAX_OCR_BYTES:
+        raise HTTPException(status_code=413, detail="file too large (max 100 MB)")
+
+    try:
+        return await ocr_manager.ocr_document(
+            data,
+            filename=file.filename or "",
+            content_type=file.content_type or "",
+            engine=engine,
+            dpi=dpi,
+            detect=detect,
+            structured=bool(structured),
+            schema=schema_,
+        )
+    except OcrError as e:
+        raise HTTPException(status_code=e.status, detail=str(e))
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"OCR failed: {e}")
+
+
+@router.post("/v1/ocr/batch", summary="OCR many images/PDFs concurrently")
+async def create_ocr_batch(
+    files: List[UploadFile] = File(...),
+    engine: Optional[str] = Form(None),
+    dpi: Optional[int] = Form(None),
+    structured: Optional[bool] = Form(False),
+    schema_: Optional[str] = Form(None, alias="schema"),
+    detect: Optional[str] = Form(None),
+):
+    """Fan a batch of documents across the engine's instance pool. Per-file failures are
+    reported inline (not fatal) so one bad scan doesn't sink the batch."""
+    cfg = _ocr_config()
+    if cfg is None:
+        raise HTTPException(status_code=503, detail="OCR subsystem is not available")
+    ocr_manager.configure(cfg)
+
+    payloads = []
+    for f in files:
+        data = await f.read()
+        payloads.append((f.filename or "", f.content_type or "", data))
+
+    async def _one(name, ctype, data):
+        if not data:
+            return {"filename": name, "error": "empty upload"}
+        if len(data) > _MAX_OCR_BYTES:
+            return {"filename": name, "error": "file too large (max 100 MB)"}
+        try:
+            res = await ocr_manager.ocr_document(
+                data, filename=name, content_type=ctype, engine=engine, dpi=dpi,
+                detect=detect, structured=bool(structured), schema=schema_,
+            )
+            res["filename"] = name
+            return res
+        except OcrError as e:
+            return {"filename": name, "error": str(e), "status": e.status}
+        except Exception as e:
+            return {"filename": name, "error": f"OCR failed: {e}"}
+
+    results = await asyncio.gather(*[_one(n, c, d) for (n, c, d) in payloads])
+    ok = sum(1 for r in results if "error" not in r)
+    return {"num_files": len(results), "num_ok": ok, "results": list(results)}
+
+
+# ---------------------------------------------------------------------------
+# Extraction schema management (schemas are data, user-extensible)
+# ---------------------------------------------------------------------------
+
+class SchemaPayload(BaseModel):
+    name: Optional[str] = None
+    description: Optional[str] = ""
+    language: Optional[str] = ""
+    type: Optional[str] = "template"          # template | jsonschema
+    prompt_hint: Optional[str] = ""
+    schema_: dict = Field(default_factory=dict, alias="schema")
+    model_config = ConfigDict(extra="allow", populate_by_name=True)
+
+
+@router.get("/v1/ocr/schemas", summary="List available extraction schemas")
+async def list_schemas():
+    return {"schemas": schema_store.list()}
+
+
+@router.get("/v1/ocr/schemas/{name}", summary="Get one extraction schema")
+async def get_schema(name: str):
+    try:
+        return schema_store.get(name)
+    except OcrError as e:
+        raise HTTPException(status_code=e.status, detail=str(e))
+
+
+@router.post("/v1/ocr/schemas", summary="Create or update a user extraction schema")
+async def save_schema(payload: SchemaPayload):
+    name = payload.name
+    if not name:
+        raise HTTPException(status_code=400, detail="schema 'name' is required")
+    data = {
+        "name": name,
+        "description": payload.description or "",
+        "language": payload.language or "",
+        "type": payload.type or "template",
+        "prompt_hint": payload.prompt_hint or "",
+        "schema": getattr(payload, "schema_", None) or getattr(payload, "schema", None) or {},
+    }
+    try:
+        return schema_store.save(name, data)
+    except OcrError as e:
+        raise HTTPException(status_code=e.status, detail=str(e))
+
+
+@router.delete("/v1/ocr/schemas/{name}", summary="Delete a user extraction schema")
+async def delete_schema(name: str):
+    try:
+        schema_store.delete(name)
+        return {"deleted": name}
+    except OcrError as e:
+        raise HTTPException(status_code=e.status, detail=str(e))
+
+
+# ---------------------------------------------------------------------------
+# Pipeline step handler (registered as step type "ocr" in custom_pipelines)
+# ---------------------------------------------------------------------------
+
+class OcrStepRequest(BaseModel):
+    image: str = ""                       # data: URI, base64, or {{stepN.url}} reference
+    engine: Optional[str] = None
+    dpi: Optional[int] = None
+    detect: Optional[str] = None
+    structured: Optional[bool] = False
+    schema_: Optional[str] = Field(default=None, alias="schema")
+    model_config = ConfigDict(extra="allow", populate_by_name=True)
+
+
+async def run_ocr_step(request: OcrStepRequest, http_request=None):
+    """Pipeline step: OCR an image reference and return {text, structured, ...}."""
+    import base64
+
+    cfg = _ocr_config()
+    if cfg is None:
+        raise HTTPException(status_code=503, detail="OCR subsystem is not available")
+    ocr_manager.configure(cfg)
+
+    ref = (getattr(request, "image", "") or "").strip()
+    filename = "input.png"
+    if ref.startswith("data:"):
+        header, encoded = ref.split(",", 1)
+        data = base64.b64decode(encoded)
+        if "application/pdf" in header:
+            filename = "input.pdf"
+    elif ref.startswith("http://") or ref.startswith("https://"):
+        import urllib.request
+        with urllib.request.urlopen(ref) as resp:      # nosec - user-provided pipeline input
+            data = resp.read()
+        filename = ref.rsplit("/", 1)[-1] or filename
+    else:
+        data = base64.b64decode(ref) if ref else b""
+
+    if not data:
+        raise HTTPException(status_code=400, detail="ocr step: no image provided")
+
+    schema = getattr(request, "schema_", None) or getattr(request, "schema", None)
+    try:
+        return await ocr_manager.ocr_document(
+            data, filename=filename, engine=request.engine, dpi=request.dpi,
+            detect=request.detect, structured=bool(request.structured), schema=schema,
+        )
+    except OcrError as e:
+        raise HTTPException(status_code=e.status, detail=str(e))

codai/backends/colibri.py

@@ -26,6 +26,7 @@ import threading
 from typing import AsyncGenerator, Dict, List, Optional
 
 from codai.backends.base import ModelBackend
+from codai.backends.colibri_families import render_deepseek_chat, build_kimi_wire
 
 # GLM-5.2 chat-template markers (from colibri openai_server.py — the model expresses
 # tool calls as ordinary text, so we render them into the prompt and let the parser
@@ -62,6 +63,70 @@ def clean_glm_output(text: str) -> str:
     return text
 
 
+# --- DeepSeek-V4 / Kimi-K3 output trimming (multi-family colibri, v1.5.0) -------- #
+# DeepSeek uses the official markers (｜=U+FF5C, ▁=U+2581); Kimi emits XTML specials
+# that its serve loop already suppresses, but we strip any residual defensively.
+_DS_STOP_MARKERS = ("<｜end▁of▁sentence｜>", "<｜User｜>", "<｜Assistant｜>",
+                    "<｜begin▁of▁sentence｜>")
+_KIMI_STRIP_TOKENS = ("<|open|>", "<|close|>", "<|sep|>", "<|end_of_msg|>")
+
+
+def _ds_cut_index(text: str) -> int:
+    cut = len(text)
+    for m in _DS_STOP_MARKERS:
+        i = text.find(m)
+        if i != -1 and i < cut:
+            cut = i
+    return cut
+
+
+def clean_deepseek_output(text: str) -> str:
+    """Trim a DeepSeek-V4 completion at its turn boundary."""
+    if not text:
+        return text
+    text = text[:_ds_cut_index(text)]
+    return text
+
+
+def clean_kimi_output(text: str) -> str:
+    """Strip any residual Kimi-K3 XTML control tokens from the visible content (the
+    engine's serve loop already suppresses structural runs, so this is defensive)."""
+    if not text:
+        return text
+    for t in _KIMI_STRIP_TOKENS:
+        text = text.replace(t, "")
+    return text
+
+
+# Family → (payload builder, output cleaner, turn-boundary cut). The payload builder
+# takes (messages, enable_thinking, tools) and returns str (GLM/DeepSeek rendered
+# prompt) or bytes (Kimi K3CHAT1 wire). See project_colibri_v15_families.
+def _payload_for(family: str, messages, enable_thinking: bool, tools):
+    if family == "deepseek_v4":
+        return render_deepseek_chat(messages, enable_thinking=enable_thinking)
+    if family == "kimi_k3":
+        return build_kimi_wire(messages, enable_thinking=enable_thinking)
+    return render_chat(messages, enable_thinking=enable_thinking, tools=tools)
+
+
+def _clean_for(family: str, text: str) -> str:
+    if family == "deepseek_v4":
+        return clean_deepseek_output(text)
+    if family == "kimi_k3":
+        return clean_kimi_output(text)
+    return clean_glm_output(text)
+
+
+def _cut_index_for(family: str, text: str) -> int:
+    if family == "deepseek_v4":
+        return _ds_cut_index(text)
+    if family == "kimi_k3":
+        # The Kimi engine emits already-trimmed visible text and stops on its own EOS;
+        # no extra server-side turn cut is needed.
+        return len(text)
+    return _glm_cut_index(text)
+
+
 def _content_text(content) -> str:
     """Flatten OpenAI message content (string or list of text parts) to a string."""
     if content is None:
@@ -401,7 +466,8 @@ class ColibriBackend(ModelBackend):
         self._enter_request()
         try:
             engine = self._need_engine()
-            prompt = render_chat(messages, enable_thinking=self._enable_thinking, tools=tools)
+            family = getattr(engine, "family", "glm")
+            prompt = _payload_for(family, messages, self._enable_thinking, tools)
             chunks: List[str] = []
 
             def _hit_turn_boundary():
@@ -409,13 +475,13 @@ class ColibriBackend(ModelBackend):
                 # ramble into a new turn. Stop it the moment a turn marker appears —
                 # a big saving at streaming-bound decode speeds.
                 joined = "".join(chunks)
-                return _glm_cut_index(joined) < len(joined)
+                return _cut_index_for(family, joined) < len(joined)
 
             stats = engine.run(prompt, int(max_tokens or 1024), float(temperature),
                                float(top_p), on_text=chunks.append,
                                cancelled=_hit_turn_boundary)
             self._store_usage(stats)
-            return clean_glm_output("".join(chunks))
+            return _clean_for(family, "".join(chunks))
         finally:
             self._exit_request()
 
@@ -425,9 +491,10 @@ class ColibriBackend(ModelBackend):
         self._enter_request()
         try:
             engine = self._need_engine()
-            prompt = render_chat(messages, enable_thinking=self._enable_thinking, tools=tools)
+            family = getattr(engine, "family", "glm")
+            prompt = _payload_for(family, messages, self._enable_thinking, tools)
             async for chunk in self._stream(engine, prompt, int(max_tokens or 1024),
-                                            float(temperature), float(top_p)):
+                                            float(temperature), float(top_p), family=family):
                 yield chunk
         finally:
             self._exit_request()
@@ -453,8 +520,8 @@ class ColibriBackend(ModelBackend):
     # SSE streaming: the engine's blocking run() streams tokens to a callback on a
     # worker thread; bridge them to the event loop through an asyncio.Queue.
     # ------------------------------------------------------------------ #
-    async def _stream(self, engine, prompt: str, max_tokens: int, temperature: float,
-                      top_p: float) -> AsyncGenerator[str, None]:
+    async def _stream(self, engine, prompt, max_tokens: int, temperature: float,
+                      top_p: float, family: str = "glm") -> AsyncGenerator[str, None]:
         loop = asyncio.get_event_loop()
         out_queue: asyncio.Queue = asyncio.Queue()
         _SENTINEL = object()
@@ -462,7 +529,7 @@ class ColibriBackend(ModelBackend):
 
         def _hit_turn_boundary():
             joined = "".join(raw)
-            return _glm_cut_index(joined) < len(joined)
+            return _cut_index_for(family, joined) < len(joined)
 
         def _on_text(text: str):
             if text:
@@ -480,9 +547,9 @@ class ColibriBackend(ModelBackend):
                 loop.call_soon_threadsafe(out_queue.put_nowait, _SENTINEL)
 
         threading.Thread(target=_worker, daemon=True).start()
-        # Emit the CLEANED text incrementally. Hold back the last few chars so a GLM
-        # control token split across chunks (e.g. "<th"+"ink>") is never emitted raw;
-        # clean_glm_output also cuts at the turn boundary, so the stream ends there.
+        # Emit the CLEANED text incrementally. Hold back the last few chars so a control
+        # token split across chunks (e.g. "<th"+"ink>") is never emitted raw; the
+        # per-family cleaner also cuts at the turn boundary, so the stream ends there.
         _HOLD = 16
         yielded = 0
         while True:
@@ -491,11 +558,11 @@ class ColibriBackend(ModelBackend):
                 break
             if isinstance(item, Exception):
                 raise item
-            cleaned = clean_glm_output("".join(raw))
+            cleaned = _clean_for(family, "".join(raw))
             safe = max(yielded, len(cleaned) - _HOLD)
             if safe > yielded:
                 yield cleaned[yielded:safe]
                 yielded = safe
-        cleaned = clean_glm_output("".join(raw))
+        cleaned = _clean_for(family, "".join(raw))
         if len(cleaned) > yielded:
             yield cleaned[yielded:]

codai/frontproxy/app.py

@@ -908,6 +908,7 @@ class FrontProxy:
             new = cm.config
             for f in ("server", "backend", "models", "offload", "vulkan", "image",
                       "whisper", "archive", "thermal", "jobs", "enhance", "ds4", "colibri",
+                      "k3", "ktransformers",
                       "compaction", "broker", "system_prompt", "tools_closer_prompt",
                       "grammar_guided", "parser", "tmp_dir"):
                 if hasattr(new, f):
@@ -919,6 +920,8 @@ class FrontProxy:
     def _required_cap(self, path: str, model: Optional[str]) -> Optional[str]:
         ds4 = getattr(self.config, "ds4", None)
         colibri = getattr(self.config, "colibri", None)
+        k3 = getattr(self.config, "k3", None)
+        kt = getattr(self.config, "ktransformers", None)
         info = self._model_info(model)
         cap = _router.required_capability(
             model, path=path,
@@ -926,7 +929,11 @@ class FrontProxy:
             ds4_model_id=getattr(ds4, "model_id", None) if ds4 else None,
             ds4_enabled=bool(getattr(ds4, "enabled", False)) if ds4 else False,
             colibri_model_id=getattr(colibri, "model_id", None) if colibri else None,
-            colibri_enabled=bool(getattr(colibri, "enabled", False)) if colibri else False)
+            colibri_enabled=bool(getattr(colibri, "enabled", False)) if colibri else False,
+            k3_model_id=getattr(k3, "model_id", None) if k3 else None,
+            k3_enabled=bool(getattr(k3, "enabled", False)) if k3 else False,
+            kt_model_id=getattr(kt, "model_id", None) if kt else None,
+            kt_enabled=bool(getattr(kt, "enabled", False)) if kt else False)
         # The name heuristic can't see that a bare alias (e.g. '…-q4_k_m', no
         # literal 'gguf') backs a .gguf file, so it falls through to
         # 'transformers' (CUDA-only) and the request never reaches a Vulkan/AMD

codai/frontproxy/assignment.py

@@ -54,7 +54,7 @@ def _route_key(entry):
     return None
 
 
-def _required_cap(entry, ds4_cfg, colibri_cfg=None):
+def _required_cap(entry, ds4_cfg, colibri_cfg=None, k3_cfg=None, kt_cfg=None):
     from codai.frontproxy.router import required_capability
     path = _entry_path(entry) or ""
     backend = entry.get("backend") if isinstance(entry, dict) else None
@@ -63,11 +63,15 @@ def _required_cap(entry, ds4_cfg, colibri_cfg=None):
         ds4_model_id=getattr(ds4_cfg, "model_id", None) if ds4_cfg else None,
         ds4_enabled=bool(getattr(ds4_cfg, "enabled", False)) if ds4_cfg else False,
         colibri_model_id=getattr(colibri_cfg, "model_id", None) if colibri_cfg else None,
-        colibri_enabled=bool(getattr(colibri_cfg, "enabled", False)) if colibri_cfg else False)
+        colibri_enabled=bool(getattr(colibri_cfg, "enabled", False)) if colibri_cfg else False,
+        k3_model_id=getattr(k3_cfg, "model_id", None) if k3_cfg else None,
+        k3_enabled=bool(getattr(k3_cfg, "enabled", False)) if k3_cfg else False,
+        kt_model_id=getattr(kt_cfg, "model_id", None) if kt_cfg else None,
+        kt_enabled=bool(getattr(kt_cfg, "enabled", False)) if kt_cfg else False)
 
 
 def compute_assignment(engines, models_path, default_engine=None, ds4_cfg=None,
-                       colibri_cfg=None):
+                       colibri_cfg=None, k3_cfg=None, kt_cfg=None):
     """Return {engine_name: [model_identifiers]} — each model owned by one engine."""
     assignment = {e.name: [] for e in engines}
     if not engines or not models_path:
@@ -87,7 +91,7 @@ def compute_assignment(engines, models_path, default_engine=None, ds4_cfg=None,
             ident = _route_key(entry)
             if not ident or ident in seen:
                 continue
-            cap = _required_cap(entry, ds4_cfg, colibri_cfg)
+            cap = _required_cap(entry, ds4_cfg, colibri_cfg, k3_cfg, kt_cfg)
             candidates = [e for e in engines if e.can_serve(cap)]
             if not candidates:
                 continue   # nothing can run it — leave unassigned

codai/frontproxy/engine_supervisor.py

@@ -156,8 +156,10 @@ class EngineSupervisor:
             default_engine = getattr(self.config.server, "default_engine", None)
             ds4 = getattr(self.config, "ds4", None)
             colibri = getattr(self.config, "colibri", None)
+            k3 = getattr(self.config, "k3", None)
+            kt = getattr(self.config, "ktransformers", None)
             assignment = compute_assignment(engines, self.models_path,
-                                            default_engine, ds4, colibri)
+                                            default_engine, ds4, colibri, k3, kt)
             for e in engines:
                 owned = assignment.get(e.name, [])
                 e.assigned_models = set(owned)   # the front's router enforces this
@@ -604,8 +606,10 @@ class EngineSupervisor:
                 default_engine = getattr(self.config.server, "default_engine", None)
                 ds4 = getattr(self.config, "ds4", None)
                 colibri = getattr(self.config, "colibri", None)
+                k3 = getattr(self.config, "k3", None)
+                kt = getattr(self.config, "ktransformers", None)
                 assignment = compute_assignment(real, self.models_path,
-                                                default_engine, ds4, colibri)
+                                                default_engine, ds4, colibri, k3, kt)
             except Exception as exc:
                 print(f"[front] live reassignment skipped: {exc}", flush=True)
                 assignment = {}

codai/frontproxy/registry.py

@@ -41,12 +41,14 @@ def _short_stem(key: str) -> str:
 #   colibri      — GLM-5.2 via the native colibri C engine (CUDA build here)
 # An NVIDIA engine can do all of them; a Vulkan (e.g. Radeon) engine does GGUF and
 # whisper, but not transformers, ds4 or colibri.
+# k3 (kimi-k3-in-c) is a CPU engine → any node can host it; kt (ktransformers/SGLang)
+# is CPU+GPU → the GPU-capable nodes.
 _DEFAULT_CAPS = {
-    "nvidia": {"transformers", "gguf", "whisper", "ds4", "colibri"},
-    "cuda": {"transformers", "gguf", "whisper", "ds4", "colibri"},
-    "vulkan": {"gguf", "whisper"},
-    "opencl": {"gguf", "whisper"},
-    "auto": {"transformers", "gguf", "whisper", "ds4", "colibri"},
+    "nvidia": {"transformers", "gguf", "whisper", "ds4", "colibri", "k3", "kt"},
+    "cuda": {"transformers", "gguf", "whisper", "ds4", "colibri", "k3", "kt"},
+    "vulkan": {"gguf", "whisper", "k3"},
+    "opencl": {"gguf", "whisper", "k3"},
+    "auto": {"transformers", "gguf", "whisper", "ds4", "colibri", "k3", "kt"},
 }
 
 

codai/frontproxy/router.py

@@ -76,32 +76,59 @@ def required_capability(model: Optional[str], path: Optional[str] = None,
                         ds4_model_id: Optional[str] = None,
                         ds4_enabled: bool = False,
                         colibri_model_id: Optional[str] = None,
-                        colibri_enabled: bool = False) -> Optional[str]:
+                        colibri_enabled: bool = False,
+                        k3_model_id: Optional[str] = None,
+                        k3_enabled: bool = False,
+                        kt_model_id: Optional[str] = None,
+                        kt_enabled: bool = False) -> Optional[str]:
     """The capability an engine must have to serve this request.
 
     * ``whisper``      — whisper.cpp STT (transcription endpoint or a
                          ``whisper-server`` model). Runs on CUDA or Vulkan.
     * ``ds4``          — DeepSeek V4 via the native ds4 engine. CUDA-only.
-    * ``colibri``      — GLM-5.2 via the native colibri C engine. CUDA-only here.
+    * ``colibri``      — GLM-5.2 / DeepSeek-V4 / Kimi-K3 via the native colibri C engine.
+    * ``k3``           — Kimi-K3 via kimi-k3-in-c. CPU.
+    * ``kt``           — many families via ktransformers/SGLang. CPU+GPU.
     * ``gguf``         — llama.cpp model. Runs on CUDA or Vulkan.
     * ``transformers`` — safetensors/HF model. CUDA-only.
 
-    Signals are combined: the request path and the model's configured ``backend``
-    (from models.json) take precedence over the name heuristic, so whisper works
-    even when the model id isn't in the request body (multipart upload)."""
+    This mirrors the in-process resolver (:func:`codai.models.manager.resolve_engine_backend`):
+    an explicit ``backend`` pin is authoritative, then an enabled engine's ``model_id``
+    alias, then an unambiguous name marker (a Kimi name claimed by BOTH colibri and k3
+    falls through — the model must pin a backend)."""
     p = (path or "").split("?", 1)[0].rstrip("/")
     if p == "/v1/audio/transcriptions" or (backend or "") == "whisper-server":
         return "whisper"
+    # 1. Explicit engine-backend pin wins (authoritative), like the manager resolver.
+    b = (backend or "").lower()
+    if b in ("colibri", "ds4", "k3", "kt"):
+        return b
     m = (model or "").lower()
-    if (backend or "") == "colibri" or (colibri_enabled and m and (
-            ((colibri_model_id or "").lower()
-             and (m == (colibri_model_id or "").lower()
-                  or m.split("/")[-1] == (colibri_model_id or "").lower()))
-            or "glm-5.2" in m or "glm5.2" in m or "colibri" in m)):
+
+    def _alias(mid: Optional[str]) -> bool:
+        mid = (mid or "").lower()
+        return bool(mid) and (m == mid or m.split("/")[-1] == mid)
+
+    # 2. An enabled engine's model_id alias.
+    if kt_enabled and _alias(kt_model_id):
+        return "kt"
+    if k3_enabled and _alias(k3_model_id):
+        return "k3"
+    if colibri_enabled and _alias(colibri_model_id):
         return "colibri"
-    if ds4_enabled and m:
-        mid = (ds4_model_id or "").lower()
-        if (mid and (m == mid or m.split("/")[-1] == mid)) or "deepseek-v4" in m:
+    if ds4_enabled and _alias(ds4_model_id):
+        return "ds4"
+
+    # 3. Name markers (unambiguous only). GLM → colibri; DeepSeek → ds4 (default owner);
+    #    Kimi → colibri XOR k3 (ambiguous when both enabled → fall through to a pin).
+    if m:
+        if colibri_enabled and ("glm-5.2" in m or "glm5.2" in m or "colibri" in m):
+            return "colibri"
+        if "kimi-k3" in m or "kimi_k3" in m or "kimik3" in m:
+            claimers = [c for c, en in (("colibri", colibri_enabled), ("k3", k3_enabled)) if en]
+            if len(claimers) == 1:
+                return claimers[0]
+        if ds4_enabled and "deepseek-v4" in m:
             return "ds4"
     if m.endswith(".gguf") or "gguf" in m:
         return "gguf"

codai/main.py

@@ -1271,6 +1271,10 @@ def main():
     from codai.api.characters import set_global_args as set_chars_global_args
     set_chars_global_args(global_args)
 
+    # Set OCR module global args
+    from codai.api.ocr import set_global_args as set_ocr_global_args
+    set_ocr_global_args(global_args)
+
     # Set LoRA training module global args. Resolve job-recovery first (the
     # --no-resume-jobs flag overrides the persisted config setting), then call
     # set_global_args, which runs _load_jobs_on_start and honours the flag.

codai/ocr/__init__.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.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""Dedicated OCR subsystem for coderai.
+
+Purpose-built OCR engines (detection + recognition), NOT a vision LLM. Engines are
+pluggable behind :class:`~codai.ocr.base.OcrEngine` and selected per request; the
+:class:`~codai.ocr.manager.OcrManager` holds per-engine instance pools for concurrent
+GPU OCR. See docs/ocr.md.
+"""
+
+from codai.ocr.base import OcrEngine, OcrPage, OcrLine, OcrRegion, OcrError, load_pages
+
+__all__ = [
+    "OcrEngine",
+    "OcrPage",
+    "OcrLine",
+    "OcrRegion",
+    "OcrError",
+    "load_pages",
+]

codai/ocr/base.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.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""OCR engine interface + shared result types + input loading.
+
+Every OCR engine (PaddleOCR, docTR, Surya) subclasses :class:`OcrEngine` and returns
+:class:`OcrPage` objects. The rest of the subsystem (manager, API, detection,
+extraction) speaks only these types, so engines stay interchangeable.
+"""
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Any, List, Optional
+
+
+class OcrError(RuntimeError):
+    """Raised for OCR failures the API should surface (missing deps, bad input, …).
+
+    ``status`` maps to the HTTP status the endpoint should return (503 when an engine's
+    optional dependency is not installed, 400 for bad input, 500 otherwise).
+    """
+
+    def __init__(self, message: str, status: int = 500):
+        super().__init__(message)
+        self.status = status
+
+
+# ---------------------------------------------------------------------------
+# Result types
+# ---------------------------------------------------------------------------
+
+@dataclass
+class OcrLine:
+    """A recognised text line (or word), with its polygon bounding box."""
+    text: str
+    bbox: List[float]              # [x0, y0, x1, y1] in pixels
+    conf: float = 0.0
+
+    def to_dict(self) -> dict:
+        return {"text": self.text, "bbox": self.bbox, "conf": round(self.conf, 4)}
+
+
+@dataclass
+class OcrRegion:
+    """A layout region (title/text/table/figure/seal/…) from layout analysis.
+
+    Used both for structure output and — in the ``layout`` stamp/signature detection
+    mode — to flag likely stamps/seals/figures without a separate detector.
+    """
+    label: str
+    bbox: List[float]             # [x0, y0, x1, y1] in pixels
+    conf: float = 0.0
+
+    def to_dict(self) -> dict:
+        return {"label": self.label, "bbox": self.bbox, "conf": round(self.conf, 4)}
+
+
+@dataclass
+class OcrPage:
+    """OCR result for a single page/image."""
+    index: int
+    text: str = ""
+    width: int = 0
+    height: int = 0
+    lines: List[OcrLine] = field(default_factory=list)
+    regions: List[OcrRegion] = field(default_factory=list)
+    tables: List[Any] = field(default_factory=list)     # list of {bbox, html?, cells?}
+
+    def to_dict(self) -> dict:
+        return {
+            "index": self.index,
+            "text": self.text,
+            "width": self.width,
+            "height": self.height,
+            "lines": [l.to_dict() for l in self.lines],
+            "regions": [r.to_dict() for r in self.regions],
+            "tables": self.tables,
+        }
+
+
+# ---------------------------------------------------------------------------
+# Engine interface
+# ---------------------------------------------------------------------------
+
+class OcrEngine(ABC):
+    """Base class for a dedicated OCR engine.
+
+    Subclasses import their heavy dependency LAZILY inside :meth:`load` (never at module
+    import time) and raise :class:`OcrError` with ``status=503`` if it is missing, so the
+    server runs fine without any OCR library installed.
+    """
+
+    name: str = "base"
+
+    def __init__(self, cfg):
+        self.cfg = cfg
+        self._loaded = False
+
+    @abstractmethod
+    def load(self) -> None:
+        """Load models into memory (called once per instance, lazily). Idempotent."""
+
+    @abstractmethod
+    def recognize_image(self, image) -> OcrPage:
+        """OCR a single page image (a PIL.Image.Image) → :class:`OcrPage`."""
+
+    def ensure_loaded(self) -> None:
+        if not self._loaded:
+            self.load()
+            self._loaded = True
+
+
+# ---------------------------------------------------------------------------
+# Input loading (images + PDF) → list of PIL pages
+# ---------------------------------------------------------------------------
+
+# Extensions we accept directly as raster images.
+_IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".tif", ".tiff", ".bmp", ".webp", ".gif", ".ppm"}
+
+
+def load_pages(data: bytes, filename: str = "", content_type: str = "", dpi: int = 200) -> List["Any"]:
+    """Decode ``data`` (an image or a PDF) into a list of RGB ``PIL.Image`` pages.
+
+    PDFs are rasterised at ``dpi`` via pypdfium2 (preferred) or pdf2image/poppler.
+    Raises :class:`OcrError` (status 400) on undecodable input, 503 if PDF support is
+    needed but unavailable.
+    """
+    try:
+        from PIL import Image
+    except Exception as e:  # pragma: no cover - Pillow is a base dep
+        raise OcrError(f"Pillow (PIL) is required for OCR: {e}", status=503)
+
+    import io
+
+    name = (filename or "").lower()
+    is_pdf = (content_type or "").lower() == "application/pdf" or name.endswith(".pdf") \
+        or data[:5] == b"%PDF-"
+
+    if is_pdf:
+        return _rasterize_pdf(data, dpi)
+
+    try:
+        img = Image.open(io.BytesIO(data))
+        img.load()
+    except Exception as e:
+        raise OcrError(f"could not decode image: {e}", status=400)
+
+    # Multi-frame images (multi-page TIFF) → one page per frame.
+    pages = []
+    n = getattr(img, "n_frames", 1)
+    for i in range(n):
+        try:
+            img.seek(i)
+        except Exception:
+            pass
+        pages.append(img.convert("RGB"))
+    return pages
+
+
+def _rasterize_pdf(data: bytes, dpi: int) -> List["Any"]:
+    """Rasterise a PDF to a list of RGB PIL pages (pypdfium2 first, then pdf2image)."""
+    from PIL import Image  # noqa: F401  (imported for type parity / already validated)
+
+    # Preferred: pypdfium2 (self-contained, no system poppler).
+    try:
+        import pypdfium2 as pdfium
+    except Exception:
+        pdfium = None
+
+    if pdfium is not None:
+        try:
+            pdf = pdfium.PdfDocument(data)
+            scale = max(0.5, dpi / 72.0)
+            pages = []
+            for i in range(len(pdf)):
+                page = pdf[i]
+                bitmap = page.render(scale=scale)
+                pages.append(bitmap.to_pil().convert("RGB"))
+            return pages
+        except Exception as e:
+            raise OcrError(f"failed to rasterise PDF (pypdfium2): {e}", status=400)
+
+    # Fallback: pdf2image (needs system poppler).
+    try:
+        from pdf2image import convert_from_bytes
+    except Exception:
+        raise OcrError(
+            "PDF OCR needs 'pypdfium2' (pip install pypdfium2) or 'pdf2image' + poppler",
+            status=503,
+        )
+    try:
+        return [im.convert("RGB") for im in convert_from_bytes(data, dpi=dpi)]
+    except Exception as e:
+        raise OcrError(f"failed to rasterise PDF (pdf2image): {e}", status=400)

codai/ocr/detect.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.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""Stamp / signature detection for the OCR subsystem.
+
+Configurable via ``ocr.detect_mode``:
+
+- ``off``       — no detection.
+- ``layout``    — reuse the OCR engine's layout regions (PP-Structure etc.); non-text
+  regions (figure/seal/stamp) are flagged as candidate stamps. No extra model; coarse,
+  and signatures are rarely separable this way.
+- ``detector``  — run a small YOLO model (``ocr.detect_model_path``) trained to spot
+  signatures/stamps; returns tight boxes + confidence + class.
+- ``both``      — run both and merge.
+
+The detector model is loaded once per weights path and reused across requests.
+"""
+
+import threading
+from typing import Dict, List, Optional
+
+from codai.ocr.base import OcrPage, OcrError
+
+
+VALID_MODES = ("off", "layout", "detector", "both")
+
+# Class-name → bucket heuristics (English + Italian markers).
+_SIGNATURE_MARKERS = ("sign", "signature", "firma", "autograph")
+_STAMP_MARKERS = ("stamp", "seal", "timbro", "sigillo", "bollo")
+# Layout labels (PP-Structure / docTR / Surya) that plausibly cover a stamp/seal.
+_LAYOUT_STAMP_LABELS = ("figure", "seal", "stamp", "image", "graphic")
+
+
+def resolve_detect_mode(cfg, override: Optional[str]) -> str:
+    """Per-request override wins over the configured default; validated."""
+    mode = (override or getattr(cfg, "detect_mode", "off") or "off").strip().lower()
+    if mode not in VALID_MODES:
+        raise OcrError(f"invalid detect mode '{mode}' (use {'/'.join(VALID_MODES)})", status=400)
+    return mode
+
+
+def _classify(label: str) -> str:
+    lo = (label or "").lower()
+    if any(m in lo for m in _SIGNATURE_MARKERS):
+        return "signature"
+    if any(m in lo for m in _STAMP_MARKERS):
+        return "stamp"
+    return "stamp"   # unknown detector classes default to the stamp bucket (label kept)
+
+
+# ---------------------------------------------------------------------------
+# YOLO detector (lazy, cached per weights path)
+# ---------------------------------------------------------------------------
+
+_detector_cache: Dict[str, "YoloDetector"] = {}
+_cache_lock = threading.Lock()
+
+
+class YoloDetector:
+    def __init__(self, model_path: str, conf: float):
+        self.model_path = model_path
+        self.conf = conf
+        self._model = None
+        self._load_lock = threading.Lock()
+
+    def ensure_loaded(self):
+        if self._model is not None:
+            return
+        with self._load_lock:
+            if self._model is not None:
+                return
+            try:
+                from ultralytics import YOLO
+            except Exception as e:
+                raise OcrError(
+                    "ultralytics is not installed (needed for detector mode). Install: "
+                    f"`pip install ultralytics`. Import error: {e}",
+                    status=503,
+                )
+            import os
+            if not self.model_path or not os.path.isfile(self.model_path):
+                raise OcrError(
+                    "stamp/signature detector weights not found. Set ocr.detect_model_path "
+                    "to a YOLO .pt file trained on signatures/stamps.",
+                    status=400,
+                )
+            try:
+                self._model = YOLO(self.model_path)
+            except Exception as e:
+                raise OcrError(f"failed to load YOLO detector: {e}", status=500)
+
+    def detect(self, image) -> List[dict]:
+        self.ensure_loaded()
+        import numpy as np
+
+        arr = np.array(image.convert("RGB"))
+        out: List[dict] = []
+        try:
+            results = self._model.predict(arr, conf=self.conf, verbose=False)
+        except Exception as e:
+            raise OcrError(f"YOLO detection failed: {e}", status=500)
+        for res in results:
+            names = getattr(res, "names", {}) or {}
+            boxes = getattr(res, "boxes", None)
+            if boxes is None:
+                continue
+            for b in boxes:
+                try:
+                    xyxy = [float(v) for v in b.xyxy[0].tolist()]
+                    conf = float(b.conf[0].item()) if b.conf is not None else 0.0
+                    cls_id = int(b.cls[0].item()) if b.cls is not None else -1
+                    label = str(names.get(cls_id, cls_id))
+                    out.append({
+                        "bbox": xyxy, "conf": round(conf, 4),
+                        "label": label, "type": _classify(label), "source": "detector",
+                    })
+                except Exception:
+                    continue
+        return out
+
+
+# Default detector weight locations (baked in image, then persistent cache, then host).
+# A bundled signature detector ships at the first path; stamps are best covered by the
+# 'layout' mode or a custom multi-class model via ocr.detect_model_path.
+_DEFAULT_DETECTOR_NAME = "signature-yolo.pt"
+
+
+def _resolve_detector_path(cfg) -> str:
+    import os
+    if cfg.detect_model_path:
+        return cfg.detect_model_path
+    cache = os.environ.get("CODERAI_CACHE_DIR") or ("/cache" if os.path.isdir("/cache") else "")
+    cands = [f"/opt/coderai/models/ocr/{_DEFAULT_DETECTOR_NAME}"]
+    if cache:
+        cands.append(os.path.join(cache, "ocr", _DEFAULT_DETECTOR_NAME))
+    cands.append(os.path.expanduser(f"~/.coderai/models/ocr/{_DEFAULT_DETECTOR_NAME}"))
+    for c in cands:
+        if os.path.isfile(c):
+            return c
+    return ""   # none found → YoloDetector raises a clear 400
+
+
+def _get_detector(cfg) -> YoloDetector:
+    path = _resolve_detector_path(cfg)
+    key = f"{path}|{cfg.detect_conf}"
+    with _cache_lock:
+        det = _detector_cache.get(key)
+        if det is None:
+            det = YoloDetector(path, float(cfg.detect_conf))
+            _detector_cache[key] = det
+        return det
+
+
+# ---------------------------------------------------------------------------
+# Layout-based flagging (no extra model)
+# ---------------------------------------------------------------------------
+
+def _detect_from_layout(page: OcrPage) -> List[dict]:
+    out: List[dict] = []
+    for reg in page.regions:
+        lo = (reg.label or "").lower()
+        hit = (any(m in lo for m in _LAYOUT_STAMP_LABELS)
+               or any(m in lo for m in _STAMP_MARKERS)
+               or any(m in lo for m in _SIGNATURE_MARKERS))
+        if hit:
+            out.append({
+                "bbox": list(reg.bbox), "conf": round(float(reg.conf), 4),
+                "label": reg.label, "type": _classify(reg.label), "source": "layout",
+            })
+    return out
+
+
+# ---------------------------------------------------------------------------
+# Entry point (runs in a worker thread; sync)
+# ---------------------------------------------------------------------------
+
+def detect_page(image, page: OcrPage, cfg, mode: str) -> Dict[str, List[dict]]:
+    """Return {'stamps': [...], 'signatures': [...]} for one page under ``mode``."""
+    dets: List[dict] = []
+    if mode in ("layout", "both"):
+        dets.extend(_detect_from_layout(page))
+    if mode in ("detector", "both"):
+        dets.extend(_get_detector(cfg).detect(image))
+
+    stamps = [d for d in dets if d["type"] == "stamp"]
+    signatures = [d for d in dets if d["type"] == "signature"]
+    return {"stamps": stamps, "signatures": signatures}

codai/ocr/doctr.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.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""docTR (Mindee) OCR engine.
+
+Apache-2.0, pure-PyTorch (uses the torch already in the stack). Detection +
+recognition predictor; docTR groups words into blocks/lines with reading order, so
+transcription comes out ordered. Weaker on complex table layout than PaddleOCR/Surya.
+"""
+
+from typing import List
+
+from codai.ocr.base import OcrEngine, OcrPage, OcrLine, OcrError
+
+
+class DoctrEngine(OcrEngine):
+    name = "doctr"
+
+    def __init__(self, cfg):
+        super().__init__(cfg)
+        self._model = None
+
+    def load(self) -> None:
+        try:
+            from doctr.models import ocr_predictor
+        except Exception as e:
+            raise OcrError(
+                "docTR is not installed. Install it out of band: "
+                f"`pip install python-doctr[torch]`. Import error: {e}",
+                status=503,
+            )
+        try:
+            self._model = ocr_predictor(
+                det_arch=self.cfg.doctr_det_arch or "db_resnet50",
+                reco_arch=self.cfg.doctr_reco_arch or "crnn_vgg16_bn",
+                pretrained=True,
+            )
+        except Exception as e:
+            raise OcrError(f"failed to initialise docTR predictor: {e}", status=500)
+
+        # Device placement: default is GPU when torch sees CUDA. Force CPU if requested.
+        if not self.cfg.doctr_use_gpu:
+            try:
+                self._model = self._model.cpu()
+            except Exception:
+                pass
+        else:
+            try:
+                import torch
+                if torch.cuda.is_available():
+                    self._model = self._model.cuda()
+            except Exception:
+                pass
+
+    def recognize_image(self, image) -> OcrPage:
+        import numpy as np
+
+        arr = np.array(image.convert("RGB"))
+        h, w = arr.shape[0], arr.shape[1]
+        page = OcrPage(index=0, width=int(w), height=int(h))
+
+        try:
+            result = self._model([arr])
+            export = result.export()
+        except Exception as e:
+            raise OcrError(f"docTR recognition failed: {e}", status=500)
+
+        lines: List[OcrLine] = []
+        pages = export.get("pages") or []
+        if pages:
+            pg = pages[0]
+            dims = pg.get("dimensions") or (h, w)   # (height, width)
+            ph, pw = float(dims[0]), float(dims[1])
+            for block in pg.get("blocks", []):
+                for line in block.get("lines", []):
+                    words = line.get("words", [])
+                    text = " ".join(str(wd.get("value", "")) for wd in words).strip()
+                    if not text:
+                        continue
+                    confs = [float(wd.get("confidence", 0.0)) for wd in words] or [0.0]
+                    geom = line.get("geometry") or ((0, 0), (0, 0))
+                    (x0, y0), (x1, y1) = geom
+                    bbox = [x0 * pw, y0 * ph, x1 * pw, y1 * ph]
+                    lines.append(OcrLine(text=text, bbox=bbox, conf=sum(confs) / len(confs)))
+
+        page.lines = lines
+        page.text = "\n".join(l.text for l in lines)
+        return page

codai/ocr/extract.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.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""Structured field extraction from OCR text via an existing coderai text model.
+
+This is NOT a separate model — it reuses whatever text LLM is configured
+(``ocr.extract_model_id``) through the in-process ``/v1/chat/completions`` handler. The
+target schema is not hardcoded: it is resolved from the schema store (named schema,
+inline schema, or generic "auto" mode). See :mod:`codai.ocr.schemas`.
+"""
+
+import json
+from typing import Optional
+
+from codai.ocr.base import OcrError
+from codai.ocr.schemas import schema_store
+
+
+_SYSTEM_PROMPT = (
+    "You extract structured data from OCR'd documents. The OCR text may contain "
+    "recognition errors; correct them implicitly where obvious. Respond with a SINGLE "
+    "valid JSON object ONLY — no prose, no explanations, no code fences. Use null for "
+    "fields not present in the document."
+)
+
+_AUTO_INSTRUCTION = (
+    "No fixed schema was given. Extract the salient fields of this document as a flat "
+    "JSON object of key/value pairs (include a \"document_type\" key), plus a \"summary\"."
+)
+
+
+def _build_user_prompt(text: str, schema_obj: Optional[dict]) -> str:
+    parts = []
+    if schema_obj is None:
+        parts.append(_AUTO_INSTRUCTION)
+    else:
+        stype = schema_obj.get("type", "template")
+        hint = schema_obj.get("prompt_hint", "")
+        schema_json = json.dumps(schema_obj.get("schema", {}), ensure_ascii=False, indent=2)
+        if stype == "jsonschema":
+            parts.append(
+                "Return a JSON object conforming to this JSON Schema (same property "
+                "names and types):"
+            )
+        else:
+            parts.append(
+                "Extract the fields following EXACTLY this JSON template (same keys):"
+            )
+        parts.append(schema_json)
+        if hint:
+            parts.append(hint)
+    parts.append("Document OCR text:\n-----8<-----\n" + text + "\n-----8<-----")
+    parts.append("Return only the filled JSON.")
+    return "\n\n".join(parts)
+
+
+def _content_from_result(result) -> str:
+    """Pull the assistant message content from a chat_completions return value,
+    tolerating dict / pydantic model / JSONResponse shapes."""
+    r = None
+    if isinstance(result, dict):
+        r = result
+    elif hasattr(result, "model_dump"):
+        try:
+            r = result.model_dump()
+        except Exception:
+            r = None
+    if r is None and hasattr(result, "body"):
+        try:
+            r = json.loads(result.body)
+        except Exception:
+            r = None
+    if r is None and hasattr(result, "__dict__"):
+        r = result.__dict__
+    if not isinstance(r, dict):
+        return ""
+    choices = r.get("choices") or []
+    if not choices:
+        return ""
+    msg = choices[0].get("message") if isinstance(choices[0], dict) else None
+    if isinstance(msg, dict):
+        return msg.get("content") or ""
+    return ""
+
+
+def _parse_json_lenient(content: str):
+    """Best-effort JSON parse: strip code fences, then fall back to the outermost {...}."""
+    s = (content or "").strip()
+    if s.startswith("```"):
+        s = s.split("```", 2)
+        s = s[1] if len(s) > 1 else content
+        if s.lstrip().lower().startswith("json"):
+            s = s.lstrip()[4:]
+    s = s.strip()
+    try:
+        return json.loads(s)
+    except Exception:
+        pass
+    a, b = s.find("{"), s.rfind("}")
+    if a != -1 and b != -1 and b > a:
+        try:
+            return json.loads(s[a:b + 1])
+        except Exception:
+            pass
+    return None
+
+
+def _validate(parsed, schema_obj: Optional[dict]) -> Optional[str]:
+    """Validate ``parsed`` against a JSON Schema if applicable. Returns an error string
+    or None. Best-effort: silently skips when jsonschema is absent or type != jsonschema."""
+    if not schema_obj or schema_obj.get("type") != "jsonschema":
+        return None
+    try:
+        import jsonschema
+    except Exception:
+        return None
+    try:
+        jsonschema.validate(parsed, schema_obj.get("schema", {}))
+        return None
+    except Exception as e:
+        return str(getattr(e, "message", e))
+
+
+async def extract_fields(text: str, cfg, schema: Optional[str] = None) -> dict:
+    """Run field extraction. ``schema`` is a per-request spec (name, inline JSON, or
+    auto); falls back to ``cfg.extract_schema``. Returns {"schema": name, "fields": {...}}
+    or, on parse failure, {"_parse_error", "_raw"}. Raises OcrError(400) if not configured.
+    """
+    if not getattr(cfg, "extract_enabled", False):
+        raise OcrError("structured extraction is disabled (set ocr.extract_enabled)", status=400)
+    model_id = getattr(cfg, "extract_model_id", "") or ""
+    if not model_id:
+        raise OcrError("no extraction model set (ocr.extract_model_id)", status=400)
+
+    spec = schema if (schema is not None and schema != "") else getattr(cfg, "extract_schema", "")
+    schema_obj, _ = schema_store.resolve(spec)
+    schema_name = (schema_obj or {}).get("name") or ("auto" if schema_obj is None else "inline")
+
+    if not (text or "").strip():
+        return {"schema": schema_name, "fields": None, "_note": "empty OCR text"}
+
+    from codai.api.text import chat_completions
+    from codai.pydantic.textrequest import ChatCompletionRequest
+
+    req = ChatCompletionRequest(
+        model=model_id,
+        messages=[
+            {"role": "system", "content": _SYSTEM_PROMPT},
+            {"role": "user", "content": _build_user_prompt(text, schema_obj)},
+        ],
+        stream=False,
+        temperature=0.0,
+        max_tokens=int(getattr(cfg, "extract_max_tokens", 2048) or 2048),
+    )
+    try:
+        result = await chat_completions(req, None)
+    except Exception as e:
+        raise OcrError(f"extraction model call failed: {e}", status=500)
+
+    content = _content_from_result(result)
+    parsed = _parse_json_lenient(content)
+    if parsed is None:
+        return {"schema": schema_name, "fields": None,
+                "_parse_error": "model did not return valid JSON", "_raw": content}
+
+    out = {"schema": schema_name, "fields": parsed}
+    if bool(getattr(cfg, "extract_validate", True)):
+        err = _validate(parsed, schema_obj)
+        if err:
+            out["_validation_error"] = err
+    return out

codai/ocr/manager.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.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""OCR engine manager — per-engine instance pools for concurrent GPU OCR.
+
+Each engine is loaded as N resident instances (``*_instances`` in
+:class:`~codai.config.OcrConfig`); pages are fanned across the pool so many documents
+OCR concurrently on one GPU. An engine instance handles one page at a time (OCR runtimes
+are not reentrant), so pool size bounds that engine's concurrency; ``max_concurrency``
+bounds the whole subsystem.
+"""
+
+import asyncio
+from typing import Dict, List, Optional
+
+from codai.ocr.base import OcrEngine, OcrPage, OcrError, load_pages
+
+
+# Engine registry: name → factory(cfg) → OcrEngine. Imports are deferred inside each
+# factory so a missing OCR dependency never breaks manager import.
+def _paddle_factory(cfg):
+    from codai.ocr.paddle import PaddleEngine
+    return PaddleEngine(cfg)
+
+
+def _doctr_factory(cfg):
+    from codai.ocr.doctr import DoctrEngine
+    return DoctrEngine(cfg)
+
+
+def _surya_factory(cfg):
+    from codai.ocr.surya import SuryaEngine
+    return SuryaEngine(cfg)
+
+
+_ENGINE_FACTORIES = {
+    "paddle": _paddle_factory,
+    "doctr": _doctr_factory,
+    "surya": _surya_factory,
+}
+
+
+def _engine_enabled(cfg, name: str) -> bool:
+    if name == "paddle":
+        return bool(cfg.paddle_enabled)
+    if name == "doctr":
+        return bool(cfg.doctr_enabled)
+    if name == "surya":
+        return bool(cfg.surya_enabled and cfg.surya_accept_license)
+    return False
+
+
+def _engine_instances(cfg, name: str) -> int:
+    n = getattr(cfg, f"{name}_instances", 1)
+    try:
+        return max(1, int(n))
+    except Exception:
+        return 1
+
+
+class _Pool:
+    """A fixed-size pool of loaded engine instances, served via an asyncio.Queue."""
+
+    def __init__(self, name: str, cfg, size: int):
+        self.name = name
+        self.cfg = cfg
+        self.size = size
+        self._q: asyncio.Queue = asyncio.Queue()
+        self._built = False
+        self._build_lock = asyncio.Lock()
+
+    async def _ensure_built(self):
+        if self._built:
+            return
+        async with self._build_lock:
+            if self._built:
+                return
+            factory = _ENGINE_FACTORIES[self.name]
+            for _ in range(self.size):
+                eng = factory(self.cfg)
+                # Load the first instance synchronously-in-thread so a missing
+                # dependency surfaces as OcrError(503) before we spawn the rest.
+                await asyncio.to_thread(eng.ensure_loaded)
+                await self._q.put(eng)
+            self._built = True
+            print(f"[ocr] engine '{self.name}': {self.size} instance(s) ready")
+
+    async def recognize(self, image) -> OcrPage:
+        await self._ensure_built()
+        eng: OcrEngine = await self._q.get()
+        try:
+            return await asyncio.to_thread(eng.recognize_image, image)
+        finally:
+            await self._q.put(eng)
+
+
+class OcrManager:
+    """Holds per-engine pools; (re)configured from OcrConfig on demand."""
+
+    def __init__(self):
+        self._cfg = None
+        self._pools: Dict[str, _Pool] = {}
+        self._sem: Optional[asyncio.Semaphore] = None
+        self._lock = asyncio.Lock()
+
+    def configure(self, cfg) -> None:
+        """Point the manager at the current OcrConfig. Rebuilds pools if params changed."""
+        prev = self._cfg
+        self._cfg = cfg
+        if prev is None or self._pool_params(prev) != self._pool_params(cfg):
+            # Drop stale pools (instances are GC'd; their models freed on unload/GC).
+            self._pools = {}
+            self._sem = asyncio.Semaphore(max(1, int(getattr(cfg, "max_concurrency", 4))))
+
+    @staticmethod
+    def _pool_params(cfg) -> tuple:
+        return (
+            cfg.max_concurrency,
+            cfg.paddle_enabled, cfg.paddle_instances, cfg.paddle_use_gpu,
+            cfg.paddle_structure, cfg.paddle_lang,
+            cfg.paddle_det_model_dir, cfg.paddle_rec_model_dir,
+            cfg.paddle_venv, cfg.paddle_auto_build,
+            cfg.doctr_enabled, cfg.doctr_instances, cfg.doctr_use_gpu,
+            cfg.doctr_det_arch, cfg.doctr_reco_arch,
+            cfg.surya_enabled, cfg.surya_accept_license, cfg.surya_instances,
+            cfg.surya_langs, cfg.surya_venv, cfg.surya_auto_build,
+        )
+
+    def resolve_engine(self, requested: Optional[str]) -> str:
+        cfg = self._require_cfg()
+        name = (requested or cfg.default_engine or "paddle").strip().lower()
+        if name not in _ENGINE_FACTORIES:
+            raise OcrError(f"unknown OCR engine '{name}'", status=400)
+        if not _engine_enabled(cfg, name):
+            raise OcrError(
+                f"OCR engine '{name}' is not enabled (or license not accepted). "
+                f"Enable it in Settings → OCR.",
+                status=400,
+            )
+        return name
+
+    async def _get_pool(self, name: str) -> _Pool:
+        async with self._lock:
+            pool = self._pools.get(name)
+            if pool is None:
+                cfg = self._require_cfg()
+                pool = _Pool(name, cfg, _engine_instances(cfg, name))
+                self._pools[name] = pool
+            return pool
+
+    async def ocr_pages(self, images: List, engine: Optional[str] = None) -> (str, List[OcrPage]):
+        """OCR a list of PIL page images. Returns (engine_name, [OcrPage])."""
+        name = self.resolve_engine(engine)
+        pool = await self._get_pool(name)
+        sem = self._sem or asyncio.Semaphore(4)
+
+        async def _one(idx, img):
+            async with sem:
+                page = await pool.recognize(img)
+                page.index = idx
+                return page
+
+        pages = await asyncio.gather(*[_one(i, im) for i, im in enumerate(images)])
+        return name, list(pages)
+
+    async def ocr_document(self, data: bytes, filename: str = "", content_type: str = "",
+                           engine: Optional[str] = None, dpi: Optional[int] = None,
+                           detect: Optional[str] = None, structured: bool = False,
+                           schema: Optional[str] = None) -> dict:
+        """OCR raw bytes (image/PDF) end-to-end → response dict.
+
+        ``detect`` overrides the configured stamp/signature detection mode
+        (off|layout|detector|both). ``structured`` triggers field extraction via the
+        configured text model; ``schema`` overrides the extraction schema for this call.
+        """
+        cfg = self._require_cfg()
+        use_dpi = int(dpi) if dpi else int(cfg.dpi)
+        images = load_pages(data, filename=filename, content_type=content_type, dpi=use_dpi)
+        name, pages = await self.ocr_pages(images, engine=engine)
+        full_text = "\n\n".join(p.text for p in pages)
+
+        stamps, signatures = await self._detect_all(images, pages, detect)
+
+        structured_out = None
+        if structured:
+            from codai.ocr.extract import extract_fields
+            structured_out = await extract_fields(full_text, cfg, schema=schema)
+
+        return {
+            "engine": name,
+            "num_pages": len(pages),
+            "text": full_text,
+            "pages": [p.to_dict() for p in pages],
+            "stamps": stamps,
+            "signatures": signatures,
+            "structured": structured_out,
+        }
+
+    async def _detect_all(self, images, pages, detect: Optional[str]):
+        """Run stamp/signature detection across pages; returns (stamps, signatures)
+        with each detection tagged by page index. Empty when mode is 'off'."""
+        from codai.ocr.detect import resolve_detect_mode, detect_page
+
+        cfg = self._require_cfg()
+        mode = resolve_detect_mode(cfg, detect)
+        if mode == "off":
+            return [], []
+
+        sem = self._sem or asyncio.Semaphore(4)
+
+        async def _one(idx, img, page):
+            async with sem:
+                res = await asyncio.to_thread(detect_page, img, page, cfg, mode)
+                for d in res["stamps"]:
+                    d["page"] = idx
+                for d in res["signatures"]:
+                    d["page"] = idx
+                return res
+
+        results = await asyncio.gather(*[_one(i, im, pg) for i, (im, pg) in enumerate(zip(images, pages))])
+        stamps, signatures = [], []
+        for r in results:
+            stamps.extend(r["stamps"])
+            signatures.extend(r["signatures"])
+        return stamps, signatures
+
+    def _require_cfg(self):
+        if self._cfg is None:
+            raise OcrError("OCR subsystem is not configured", status=503)
+        if not getattr(self._cfg, "enabled", False):
+            raise OcrError("OCR subsystem is disabled (enable it in Settings → OCR)", status=400)
+        return self._cfg
+
+
+# Module-level singleton (mirrors multi_model_manager).
+ocr_manager = OcrManager()

codai/ocr/paddle.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+# GPLv3 - see the project LICENSE.
+
+"""PaddleOCR engine — runs in an ISOLATED venv subprocess.
+
+PaddleOCR pulls opencv-contrib-python (clashes with the main venv's opencv-python/cv2) and
+its paddlepaddle-gpu wheel bundles its own CUDA runtime, so it lives in a dedicated venv
+(``ocr.paddle_venv``) driven by codai/ocr/workers/ocr_worker.py. See
+:mod:`codai.ocr.subprocess_engine`.
+"""
+
+import os
+
+from codai.ocr.subprocess_engine import SubprocessOcrEngine, _resolve_venv_dir
+
+# Repo-root requirements file for the paddle isolated venv (auto-build).
+_REQ = os.path.join(os.path.dirname(__file__), "..", "..", "requirements-ocr-paddle.txt")
+
+
+class PaddleEngine(SubprocessOcrEngine):
+    name = "paddle"
+    _worker_engine = "paddle"
+
+    def _venv_dir(self) -> str:
+        return _resolve_venv_dir(self.cfg.paddle_venv, "paddle_venv")
+
+    def _requirements(self) -> str:
+        return os.path.abspath(_REQ)
+
+    def _auto_build(self) -> bool:
+        return bool(getattr(self.cfg, "paddle_auto_build", False))
+
+    def _worker_opts(self) -> dict:
+        return {
+            "lang": self.cfg.paddle_lang or self.cfg.lang or "it",
+            "use_gpu": bool(self.cfg.paddle_use_gpu),
+            "structure": bool(self.cfg.paddle_structure),
+            "det_model_dir": self.cfg.paddle_det_model_dir or "",
+            "rec_model_dir": self.cfg.paddle_rec_model_dir or "",
+        }

codai/ocr/schemas.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.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""OCR extraction schema store.
+
+Schemas are DATA, not code — so any document type can be supported without touching the
+engine. A schema is resolved three ways (in precedence order), which is how we avoid
+trying to enumerate "every schema out there":
+
+1. **Inline** — the spec string starts with ``{`` / ``[``: used verbatim (a fill-in
+   template or a JSON Schema).
+2. **Named** — a bare name (``italian_sentenza``): looked up in the user schema dir
+   (``<config_dir>/ocr_schemas/*.json``) first, then the built-in seeds shipped with
+   coderai. User files shadow built-ins of the same name.
+3. **Auto** — empty / ``auto`` / ``generic``: no schema; the model is asked to extract
+   the salient key/value fields itself.
+
+A schema file is JSON:
+    {
+      "name": "italian_sentenza",
+      "description": "Italian judicial decision (sentenza)",
+      "language": "it",
+      "type": "template" | "jsonschema",   # template = fill-in example; jsonschema = JSON Schema
+      "schema": { ... },
+      "prompt_hint": "optional extra instruction for the model"
+    }
+"""
+
+import json
+import os
+import re
+from typing import List, Optional, Tuple
+
+from codai.ocr.base import OcrError
+
+_BUILTIN_DIR = os.path.join(os.path.dirname(__file__), "schemas_builtin")
+_NAME_RE = re.compile(r"^[A-Za-z0-9._-]+$")
+_AUTO_NAMES = ("", "auto", "generic")
+
+
+def _user_dir() -> str:
+    """User schema directory: ``<config_dir>/ocr_schemas`` (created on demand)."""
+    base = ""
+    try:
+        from codai.admin.routes import config_manager
+        if config_manager is not None:
+            base = str(getattr(config_manager, "config_dir", "") or "")
+    except Exception:
+        base = ""
+    if not base:
+        base = os.path.expanduser("~/.coderai")
+    return os.path.join(base, "ocr_schemas")
+
+
+def _valid_name(name: str) -> str:
+    if not name or not _NAME_RE.match(name):
+        raise OcrError(
+            "invalid schema name (use letters, digits, '.', '_', '-')", status=400
+        )
+    return name
+
+
+def _read_json(path: str) -> Optional[dict]:
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            return json.load(f)
+    except Exception:
+        return None
+
+
+class SchemaStore:
+    def list(self) -> List[dict]:
+        """Return schema summaries (built-ins + user), user shadowing built-ins."""
+        out = {}
+        for src, d in (("builtin", _BUILTIN_DIR), ("user", _user_dir())):
+            if not os.path.isdir(d):
+                continue
+            for fn in sorted(os.listdir(d)):
+                if not fn.endswith(".json"):
+                    continue
+                data = _read_json(os.path.join(d, fn)) or {}
+                name = data.get("name") or fn[:-5]
+                out[name] = {
+                    "name": name,
+                    "description": data.get("description", ""),
+                    "language": data.get("language", ""),
+                    "type": data.get("type", "template"),
+                    "source": src,
+                }
+        return sorted(out.values(), key=lambda x: x["name"])
+
+    def get(self, name: str) -> dict:
+        """Load a named schema (user dir wins over built-in). Raises 400 if missing."""
+        _valid_name(name)
+        for d in (_user_dir(), _BUILTIN_DIR):
+            path = os.path.join(d, f"{name}.json")
+            if os.path.isfile(path):
+                data = _read_json(path)
+                if data is None:
+                    raise OcrError(f"schema '{name}' is not valid JSON", status=500)
+                return data
+        avail = ", ".join(s["name"] for s in self.list()) or "(none)"
+        raise OcrError(f"unknown schema '{name}'. Available: {avail}", status=400)
+
+    def save(self, name: str, data: dict) -> dict:
+        """Create/overwrite a USER schema (built-ins are read-only)."""
+        _valid_name(name)
+        if not isinstance(data, dict) or "schema" not in data:
+            raise OcrError("schema payload must be an object with a 'schema' field", status=400)
+        data.setdefault("name", name)
+        data.setdefault("type", "template")
+        d = _user_dir()
+        os.makedirs(d, exist_ok=True)
+        with open(os.path.join(d, f"{name}.json"), "w", encoding="utf-8") as f:
+            json.dump(data, f, ensure_ascii=False, indent=2)
+        return data
+
+    def delete(self, name: str) -> None:
+        """Delete a USER schema. Built-ins cannot be deleted."""
+        _valid_name(name)
+        path = os.path.join(_user_dir(), f"{name}.json")
+        if not os.path.isfile(path):
+            raise OcrError(f"user schema '{name}' not found (built-ins are read-only)", status=404)
+        os.remove(path)
+
+    # -- resolution --------------------------------------------------------
+
+    def resolve(self, spec: Optional[str]) -> Tuple[Optional[dict], Optional[str]]:
+        """Resolve a spec → (schema_dict_or_None, error_note).
+
+        Returns ``(None, None)`` for auto/generic mode (no schema). For inline or named
+        schemas returns the loaded/parsed schema dict.
+        """
+        s = (spec or "").strip()
+        if s.lower() in _AUTO_NAMES:
+            return None, None
+        if s[:1] in ("{", "["):
+            try:
+                parsed = json.loads(s)
+            except Exception as e:
+                raise OcrError(f"inline schema is not valid JSON: {e}", status=400)
+            # Wrap a bare schema object into the standard envelope.
+            if isinstance(parsed, dict) and "schema" in parsed and (
+                    "type" in parsed or "name" in parsed):
+                return parsed, None
+            inferred = "jsonschema" if isinstance(parsed, dict) and (
+                "$schema" in parsed or "properties" in parsed) else "template"
+            return {"name": "inline", "type": inferred, "schema": parsed}, None
+        return self.get(s), None
+
+
+schema_store = SchemaStore()

codai/ocr/schemas_builtin/generic_document.json

+{
+  "name": "generic_document",
+  "description": "Generic document metadata + free key/value fields (any language)",
+  "language": "",
+  "type": "template",
+  "prompt_hint": "Add any additional salient fields you find under 'fields' as key/value pairs.",
+  "schema": {
+    "document_type": "string|null",
+    "title": "string|null",
+    "date": "YYYY-MM-DD|null",
+    "authors_or_parties": ["string"],
+    "reference_numbers": ["string"],
+    "summary": "string|null",
+    "fields": {"<detected_field_name>": "<value>"}
+  }
+}

codai/ocr/schemas_builtin/invoice.json

+{
+  "name": "invoice",
+  "description": "Invoice / fattura — demonstrates a strict JSON Schema (validated when jsonschema is installed)",
+  "language": "",
+  "type": "jsonschema",
+  "prompt_hint": "Amounts as numbers without currency symbols; put the currency in 'currency'.",
+  "schema": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "type": "object",
+    "properties": {
+      "invoice_number": {"type": ["string", "null"]},
+      "issue_date": {"type": ["string", "null"], "description": "YYYY-MM-DD"},
+      "seller": {"type": ["string", "null"]},
+      "buyer": {"type": ["string", "null"]},
+      "currency": {"type": ["string", "null"]},
+      "subtotal": {"type": ["number", "null"]},
+      "tax": {"type": ["number", "null"]},
+      "total": {"type": ["number", "null"]},
+      "line_items": {
+        "type": "array",
+        "items": {
+          "type": "object",
+          "properties": {
+            "description": {"type": "string"},
+            "quantity": {"type": ["number", "null"]},
+            "unit_price": {"type": ["number", "null"]},
+            "amount": {"type": ["number", "null"]}
+          }
+        }
+      }
+    }
+  }
+}

codai/ocr/workers/__init__.py

+# OCR isolated-venv subprocess workers (self-contained; no coderai imports).

packaging/linux/smoke_test_services.sh

@@ -61,6 +61,16 @@ if "${DK[@]}" exec "$NAME" sh -lc "test -x /cache/ds4/ds4-server"; then ok "/cac
 echo "== colibri seeded on the cache volume =="
 if "${DK[@]}" exec "$NAME" sh -lc "test -x /cache/colibri/c/colibri"; then ok "/cache/colibri/c/colibri"; else bad "/cache/colibri/c/colibri" "missing (optional — bundle with build.sh --colibri)"; fi
 
+echo "== kimi-k3-in-c seeded on the cache volume =="
+if "${DK[@]}" exec "$NAME" sh -lc "test -x /cache/kimi-k3-in-c/bin/k3"; then ok "/cache/kimi-k3-in-c/bin/k3"; else bad "/cache/kimi-k3-in-c/bin/k3" "missing (optional — bundle with build.sh --k3)"; fi
+
+echo "== OCR subsystem present (core imports; engine deps optional) =="
+if "${DK[@]}" exec "$NAME" sh -lc "cd /opt/coderai/app && python -c 'import codai.ocr; from codai.ocr.schemas import schema_store; assert any(s[\"name\"]==\"italian_sentenza\" for s in schema_store.list())'"; then
+  ok "codai.ocr imports + built-in schemas present"
+else
+  bad "codai.ocr" "import or built-in schema check failed"
+fi
+
 echo "== shared lip-sync venv (py3.10 + torch) =="
 if "${DK[@]}" exec "$NAME" /opt/coderai/lipsync_venv/bin/python -c "import torch,sys; print(sys.version.split()[0], torch.__version__)" >/dev/null 2>&1; then
   ok "lipsync venv imports torch"