deploy: sync backend to Space root (learn-lesson HF cache fix)

.env.example

@@ -54,6 +54,15 @@ AI_TIMEOUT_SECONDS="120"
 # Server-side only — never put this in NEXT_PUBLIC_*.
 GROQ_API_KEY=""
 
+# Z.AI / GLM free Flash models (text + vision). Server-side only.
+# Text:  glm-4.7-flash   Vision: glm-4.6v-flash
+# https://docs.z.ai/  base: https://api.z.ai/api/paas/v4/
+ZAI_API_KEY=""
+ZAI_BASE_URL="https://api.z.ai/api/paas/v4"
+ZAI_MODEL_TEXT="glm-4.7-flash"
+ZAI_MODEL_VISION="glm-4.6v-flash"
+ZAI_TIMEOUT_SECONDS="90"
+
 # Scale knobs for ~1k concurrent students (raise in production Postgres deploys).
 DATABASE_POOL_SIZE="10"
 DATABASE_MAX_OVERFLOW="20"

.env.huggingface.example

@@ -20,6 +20,13 @@ AI_PROVIDER=sarvam
 SARVAM_API_KEY=
 AI_FALLBACK_TO_MOCK=false
 
+# Free Z.AI / GLM Flash (text + vision). Preferred cheap authoring path.
+ZAI_API_KEY=
+ZAI_BASE_URL=https://api.z.ai/api/paas/v4
+ZAI_MODEL_TEXT=glm-4.7-flash
+ZAI_MODEL_VISION=glm-4.6v-flash
+GROQ_API_KEY=
+
 SARVAM_BASE_URL=https://api.sarvam.ai/v1
 SARVAM_MODEL_MAIN=sarvam-30b
 SARVAM_MODEL_HEAVY=sarvam-105b

.env.production.example

@@ -57,6 +57,10 @@ AI_FALLBACK_TO_MOCK=false
 
 # Learn Anything lesson authoring (Groq). Server-only.
 GROQ_API_KEY=
+ZAI_API_KEY=
+ZAI_BASE_URL=https://api.z.ai/api/paas/v4
+ZAI_MODEL_TEXT=glm-4.7-flash
+ZAI_MODEL_VISION=glm-4.6v-flash
 
 # Postgres pool for ~1k concurrent students (tune against your plan limits).
 DATABASE_POOL_SIZE=15

app/core/config.py

@@ -107,6 +107,29 @@ class Settings(BaseSettings):
     groq_api_key: str | None = Field(default=None, validation_alias=AliasChoices("GROQ_API_KEY"))
     groq_base_url: str = "https://api.groq.com/openai/v1"
     groq_model_main: str = "llama-3.1-8b-instant"
+    # --- Z.AI / GLM (free text + vision Flash models) ---
+    # OpenAI-compatible: https://api.z.ai/api/paas/v4/
+    # Free: glm-4.7-flash (text), glm-4.6v-flash (vision / image understanding).
+    zai_api_key: str | None = Field(
+        default=None,
+        validation_alias=AliasChoices("ZAI_API_KEY", "Z_AI_API_KEY", "GLM_API_KEY"),
+    )
+    zai_base_url: str = Field(
+        default="https://api.z.ai/api/paas/v4",
+        validation_alias=AliasChoices("ZAI_BASE_URL", "Z_AI_BASE_URL"),
+    )
+    zai_model_text: str = Field(
+        default="glm-4.7-flash",
+        validation_alias=AliasChoices("ZAI_MODEL_TEXT", "GLM_MODEL_TEXT"),
+    )
+    zai_model_vision: str = Field(
+        default="glm-4.6v-flash",
+        validation_alias=AliasChoices("ZAI_MODEL_VISION", "GLM_MODEL_VISION"),
+    )
+    zai_timeout_seconds: int = Field(
+        default=90,
+        validation_alias=AliasChoices("ZAI_TIMEOUT_SECONDS"),
+    )
     nvidia_nim_api_key: str | None = Field(default=None, validation_alias=AliasChoices("NVIDIA_NIM_API_KEY"))
     nvidia_nim_base_url: str = "https://integrate.api.nvidia.com/v1"
     nvidia_nim_model_main: str = "meta/llama-3.1-8b-instruct"  # fast (~3-4s), good for quick answers

app/core/sanitizer.py

@@ -49,7 +49,8 @@ _INTERNAL_PATTERN: re.Pattern[str] = re.compile(
 # or "mock_" or word-boundary mock when followed by a dash/underscore.
 _PROVIDER_PATTERN: re.Pattern[str] = re.compile(
     r"\b(sarvam|gemini|openrouter|deepseek|nvidia|"
-    r"huggingface|cloudflare|groq|replicate|mock[-_])\b",
+    r"huggingface|cloudflare|groq|replicate|zai|z\.ai|glm|"
+    r"mock[-_])\b",
     re.IGNORECASE,
 )
 

app/routes/chat.py

@@ -515,27 +515,56 @@ def _chat_provider_candidates(settings: Any, intent: str) -> list[dict[str, Any]
             },
         })
 
+    def add_zai() -> None:
+        if not _has_configured_key(getattr(settings, "zai_api_key", None)):
+            return
+        base = (settings.zai_base_url or "https://api.z.ai/api/paas/v4").rstrip("/")
+        add({
+            "provider": "zai",
+            "base_url": base,
+            "api_key": settings.zai_api_key,
+            "model": settings.zai_model_text or "glm-4.7-flash",
+            "model_label": f"zai:{settings.zai_model_text or 'glm-4.7-flash'}",
+            "timeout": min(float(getattr(settings, "zai_timeout_seconds", 90) or 90), 60.0),
+            "token_param": "max_tokens",
+            "temperature": 0.3,
+            "system_role": "system",
+            # Free Flash is faster with thinking disabled.
+            "extra_body": {"thinking": {"type": "disabled"}},
+        })
+
     primary_adders = {
         "openai": add_openai,
         "sarvam": add_sarvam,
         "openrouter": add_openrouter,
+        "zai": add_zai,
+        "z_ai": add_zai,
+        "glm": add_zai,
     }
     # Sarvam's reasoning model is strong for deeper study work, but ordinary
     # explanations can spend close to a minute reasoning before showing text.
-    # Prefer the configured fast OpenRouter model for lightweight chat, then
-    # retain Sarvam as the first fallback. Heavy exam/PYQ work stays on Sarvam.
+    # Prefer free Z.AI Flash + OpenRouter for lightweight chat, then Sarvam.
+    # Heavy exam/PYQ work stays on Sarvam.
     if provider_name == "sarvam" and intent not in _CHAT_HEAVY_INTENTS:
+        add_zai()
         add_openrouter()
         add_sarvam()
     elif provider_name == "auto":
+        add_zai()
         add_openai()
         add_sarvam()
         add_openrouter()
+    elif provider_name in {"zai", "z_ai", "glm"}:
+        add_zai()
+        add_openrouter()
+        add_sarvam()
     else:
         primary_adders.get(provider_name, add_openai)()
 
     for name, adder in primary_adders.items():
-        if provider_name not in {name, "auto"}:
+        if provider_name not in {name, "auto", "z_ai", "glm"} and not (
+            provider_name == "zai" and name in {"zai", "z_ai", "glm"}
+        ):
             adder()
     return candidates
 
@@ -564,6 +593,8 @@ def _chat_completion_request(
         request["reasoning_effort"] = candidate["reasoning_effort"]
     if "extra_headers" in candidate:
         request["extra_headers"] = candidate["extra_headers"]
+    if "extra_body" in candidate:
+        request["extra_body"] = candidate["extra_body"]
     return request
 
 
@@ -571,13 +602,13 @@ def _create_chat_completion(client: Any, request: dict[str, Any]) -> Any:
     try:
         return client.chat.completions.create(**request)
     except TypeError:
-        if "reasoning_effort" in request:
-            request = dict(request)
-            request.pop("reasoning_effort", None)
-            if "max_completion_tokens" in request:
-                request["max_tokens"] = request.pop("max_completion_tokens")
-            return client.chat.completions.create(**request)
-        raise
+        request = dict(request)
+        # Older openai clients may not accept reasoning_effort / extra_body.
+        request.pop("reasoning_effort", None)
+        request.pop("extra_body", None)
+        if "max_completion_tokens" in request:
+            request["max_tokens"] = request.pop("max_completion_tokens")
+        return client.chat.completions.create(**request)
 
 
 def _call_ai_chat_sync(

app/services/ai_provider.py

@@ -2937,6 +2937,89 @@ class NvidiaNimAIProvider(SarvamAIProvider):
                 raise _FatalAPIError(f"NVIDIA NIM auth error {status}") from exc
             raise AIProviderError(f"NVIDIA NIM call failed: {_safe_error(exc)}") from exc
 
+
+class ZaiAIProvider(SarvamAIProvider):
+    """Z.AI free Flash text provider (OpenAI-compatible).
+
+    Uses glm-4.7-flash for all structured study tasks. Free tier is rate-limited;
+    keep this early in the chain for cost control, with Sarvam/OpenRouter as backup.
+    """
+
+    def __init__(self) -> None:
+        settings = get_settings()
+        if not settings.zai_api_key:
+            raise AIProviderError("ZAI_API_KEY is missing.")
+        try:
+            from openai import OpenAI
+        except ImportError as exc:
+            raise AIProviderError("openai package is not installed.") from exc
+        self._settings = settings
+        base = (settings.zai_base_url or "https://api.z.ai/api/paas/v4").rstrip("/")
+        self._client = OpenAI(base_url=base, api_key=settings.zai_api_key)
+        self._fallback = MockAIProvider()
+        self._text_model = settings.zai_model_text or "glm-4.7-flash"
+        self.model_name = f"zai:{self._text_model}"
+        self.is_fallback = False
+        self.fallback_reason = None
+        self.last_error_code: int | None = None
+
+    def _get_model_for_task(self, task_name: str) -> str:
+        return self._text_model
+
+    def _call_model(
+        self,
+        model_id: str,
+        system_msg: str,
+        prompt: str,
+        *,
+        max_tokens: int = 2048,
+        temperature: float = 0.2,
+        reasoning_effort: str = "low",  # ignored — we disable GLM thinking for speed
+    ) -> str:
+        started_at = time.perf_counter()
+        try:
+            # Z.AI free Flash: disable chain-of-thought for faster JSON tasks.
+            response = self._client.chat.completions.create(
+                model=model_id,
+                messages=[
+                    {"role": "system", "content": system_msg},
+                    {"role": "user", "content": prompt},
+                ],
+                temperature=temperature,
+                max_tokens=max_tokens,
+                timeout=float(self._settings.zai_timeout_seconds or 90),
+                extra_body={"thinking": {"type": "disabled"}},
+            )
+            msg = response.choices[0].message
+            content = (getattr(msg, "content", None) or "").strip()
+            reasoning = (getattr(msg, "reasoning_content", None) or "").strip()
+            text = content or reasoning
+            logger.info(
+                "AI call timing provider=zai model=%s elapsed_ms=%s max_tokens=%s",
+                model_id,
+                int((time.perf_counter() - started_at) * 1000),
+                max_tokens,
+            )
+            if not text:
+                raise AIProviderError("Z.AI returned empty content.")
+            return text
+        except _FatalAPIError:
+            raise
+        except Exception as exc:  # noqa: BLE001
+            logger.info(
+                "AI call timing provider=zai model=%s failed elapsed_ms=%s",
+                model_id,
+                int((time.perf_counter() - started_at) * 1000),
+            )
+            code = _http_status_code(exc)
+            if code in (401, 403):
+                self.last_error_code = code
+                raise _FatalAPIError(code, f"Z.AI API key error (HTTP {code})") from exc
+            if code:
+                self.last_error_code = code
+                logger.warning("Z.AI HTTP %s for model=%s", code, model_id)
+            raise AIProviderError(f"Z.AI call failed: {_safe_error(exc)}") from exc
+
     # ── Deep/quality tasks: defer to the next provider (Sarvam) in the chain ──
     def _defer(self, name: str):
         raise AIProviderError(f"NVIDIA NIM defers '{name}' to a quality provider.")
@@ -3659,12 +3742,25 @@ def get_ai_provider() -> BaseAIProvider:
             try: providers.append(NvidiaNimAIProvider())
             except Exception: pass
 
-    # 0. Fast path: NVIDIA NIM first for quick tasks (~3-4s). Deep tasks defer
+    def add_zai():
+        if has_configured_key(settings.zai_api_key):
+            try: providers.append(ZaiAIProvider())
+            except Exception: pass
+
+    # 0. Free Z.AI Flash first when available (cost control). Skip when primary
+    #    is explicitly mock. If AI_PROVIDER=zai, it is also the primary below.
+    if provider_name not in {"mock"} and provider_name != "zai":
+        # Prefer free Flash early for auto / cost-sensitive chains; for explicit
+        # paid primaries (openai/sarvam) keep Z.AI as a cheap backup later.
+        if provider_name in {"auto", "openrouter", "gemini"}:
+            add_zai()
+
+    # 0b. Fast path: NVIDIA NIM first for quick tasks (~3-4s). Deep tasks defer
     #    to Sarvam automatically. Only when FAST_PROVIDER_ENABLED=true.
     #    When the primary is explicitly Sarvam, do NOT let the fast path preempt
     #    it — Sarvam (sarvam-30b/105b) must answer the main study content, with
     #    NVIDIA demoted to a fallback at the end of the chain.
-    if settings.fast_provider_enabled and provider_name not in {"mock", "sarvam"}:
+    if settings.fast_provider_enabled and provider_name not in {"mock", "sarvam", "zai"}:
         add_nvidia_fast()
 
     # 1. Add the primary provider first
@@ -3675,6 +3771,7 @@ def get_ai_provider() -> BaseAIProvider:
     elif provider_name == "sarvam": add_sarvam()
     elif provider_name == "openrouter": add_openrouter()
     elif provider_name == "gemini": add_gemini()
+    elif provider_name in {"zai", "z_ai", "glm"}: add_zai()
     elif provider_name == "mock": return MockAIProvider()
 
     # 2. Add others as fallback backups
@@ -3682,6 +3779,8 @@ def get_ai_provider() -> BaseAIProvider:
     if provider_name not in {"sarvam", "auto"}: add_sarvam()
     if provider_name not in {"openrouter", "auto"}: add_openrouter()
     if provider_name not in {"gemini", "auto"}: add_gemini()
+    if provider_name not in {"zai", "z_ai", "glm", "auto", "openrouter", "gemini"}:
+        add_zai()
     # NVIDIA as resilience fallback when Sarvam is primary (it was skipped above)
     if provider_name == "sarvam" and settings.fast_provider_enabled:
         add_nvidia_fast()
@@ -3722,13 +3821,27 @@ def log_ai_mode() -> None:
             print(f">>> {msg}")
     elif provider_name == "auto":
         logger.info(
-            "AI MODE: auto (OpenAI=%s, Sarvam=%s, OpenRouter=%s, Gemini=%s) | fallback_to_mock=%s",
+            "AI MODE: auto (OpenAI=%s, Sarvam=%s, OpenRouter=%s, Gemini=%s, ZAI=%s) | fallback_to_mock=%s",
             bool(settings.openai_api_key),
             bool(settings.sarvam_api_key),
             bool(settings.openrouter_api_key),
             bool(settings.gemini_api_key),
+            bool(settings.zai_api_key),
             fallback,
         )
+    elif provider_name in {"zai", "z_ai", "glm"}:
+        has_key = bool(settings.zai_api_key)
+        if has_key:
+            msg = (
+                f"AI MODE: Z.AI (text={settings.zai_model_text}, "
+                f"vision={settings.zai_model_vision}) | fallback_to_mock={fallback}"
+            )
+            logger.info(msg)
+            print(f">>> {msg}")
+        else:
+            msg = "AI MODE: zai configured but ZAI_API_KEY is EMPTY — falling back to other providers"
+            logger.warning(msg)
+            print(f">>> {msg}")
     elif provider_name == "sarvam":
         has_key = bool(settings.sarvam_api_key)
         if has_key:

app/services/learn_lesson_builder.py

@@ -176,9 +176,47 @@ def _call_groq(prompt: str, api_key: str, max_tokens: int = 8000) -> dict[str, A
     )
 
 
+def _call_zai(prompt: str, api_key: str, max_tokens: int = 8000) -> dict[str, Any]:
+    """Free Z.AI GLM-4.7-Flash for lesson authoring (OpenAI-compatible)."""
+    base = (_env("ZAI_BASE_URL") or _env("Z_AI_BASE_URL") or "https://api.z.ai/api/paas/v4").rstrip("/")
+    model = _env("ZAI_MODEL_TEXT") or _env("GLM_MODEL_TEXT") or "glm-4.7-flash"
+    # Prefer the shared client when settings are loaded; fall back to raw HTTP.
+    try:
+        from app.services.zai_client import chat_completions, extract_message_text
+
+        result = chat_completions(
+            model=model,
+            messages=[{"role": "user", "content": prompt}],
+            max_tokens=max_tokens,
+            temperature=0.35,
+            response_format={"type": "json_object"},
+            thinking_disabled=True,
+        )
+        content = extract_message_text(result)
+        return json.loads(content)
+    except Exception:
+        return _call_openai_compatible_json(
+            url=f"{base}/chat/completions",
+            api_key=api_key,
+            model=model,
+            prompt=prompt,
+            max_tokens=max_tokens,
+            provider_label="Z.AI",
+        )
+
+
 def _script_llm_providers() -> list[dict[str, Any]]:
-    """Lesson-script providers: Groq OpenAI-compatible chat."""
+    """Lesson-script providers: free Z.AI Flash first, then Groq."""
     providers: list[dict[str, Any]] = []
+    zai = _env("ZAI_API_KEY") or _env("Z_AI_API_KEY") or _env("GLM_API_KEY")
+    if zai and len(zai) >= 16:
+        providers.append(
+            {
+                "name": "zai",
+                "key": zai,
+                "call": lambda prompt, key=zai: _call_zai(prompt, key),
+            }
+        )
     groq = _env("GROQ_API_KEY")
     if groq.startswith(("gsk-", "gsk_")) and len(groq) >= 20:
         providers.append(
@@ -511,7 +549,8 @@ def _generate_with_retries(
     chain = providers if providers is not None else _script_llm_providers()
     if not chain:
         raise LessonBuildError(
-            "No lesson LLM key configured. Set GROQ_API_KEY for Learn Anything authoring."
+            "No lesson LLM key configured. Set ZAI_API_KEY (free GLM-4.7-Flash) "
+            "or GROQ_API_KEY for Learn Anything authoring."
         )
     for provider in chain:
         name = str(provider.get("name") or "llm")
@@ -556,7 +595,7 @@ def generate_lesson_script(
     providers = _script_llm_providers()
     if not providers:
         logger.warning(
-            "GROQ_API_KEY missing/invalid; serving starter reading class."
+            "ZAI_API_KEY/GROQ_API_KEY missing/invalid; serving starter reading class."
         )
         return _starter_reading_script(
             topic=topic, lesson_title=lesson_title, level=level

app/services/provider_adapters.py

@@ -121,12 +121,48 @@ class CloudflareWorkersAITextAdapter(BaseTextAdapter):
         return TextProviderResponse(text=text.strip(), provider=self.provider_name, model=model, raw=data)
 
 
+class ZaiTextAdapter(BaseTextAdapter):
+    """Free Z.AI GLM-4.7-Flash text adapter (OpenAI-compatible)."""
+
+    provider_name = "zai"
+
+    def __init__(self) -> None:
+        settings = get_settings()
+        if not settings.zai_api_key:
+            raise ProviderAdapterError("Provider is not configured.", status_code=401)
+        try:
+            from openai import OpenAI
+        except ImportError as exc:
+            raise ProviderAdapterError("OpenAI-compatible client is not installed.") from exc
+        self._settings = settings
+        base = (settings.zai_base_url or "https://api.z.ai/api/paas/v4").rstrip("/")
+        self._client = OpenAI(base_url=base, api_key=settings.zai_api_key)
+        self._model = settings.zai_model_text or "glm-4.7-flash"
+
+    def generate_text(self, prompt: str, *, system: str | None = None, max_tokens: int = 2048) -> TextProviderResponse:
+        response = self._client.chat.completions.create(
+            model=self._model,
+            messages=[
+                {"role": "system", "content": system or "Return a concise, safe answer."},
+                {"role": "user", "content": prompt},
+            ],
+            max_tokens=max_tokens,
+            temperature=0.2,
+            timeout=float(self._settings.zai_timeout_seconds or 90),
+            extra_body={"thinking": {"type": "disabled"}},
+        )
+        text = response.choices[0].message.content or ""
+        return TextProviderResponse(text=text.strip(), provider=self.provider_name, model=self._model)
+
+
 def get_text_adapter(provider: str) -> BaseTextAdapter:
-    normalized = provider.strip().lower()
+    normalized = provider.strip().lower().replace("-", "_")
     if normalized == "openrouter":
         return OpenRouterTextAdapter()
     if normalized == "nvidia_nim":
         return NvidiaNimTextAdapter()
     if normalized == "cloudflare_workers_ai":
         return CloudflareWorkersAITextAdapter()
+    if normalized in {"zai", "z_ai", "glm"}:
+        return ZaiTextAdapter()
     raise ProviderAdapterError("Provider adapter is not implemented.")

app/services/provider_registry.py

@@ -73,6 +73,10 @@ def normalize_provider_name(provider: str) -> str:
         "nvidia": "nvidia_nim",
         "nim": "nvidia_nim",
         "hf": "huggingface",
+        "z_ai": "zai",
+        "z.ai": "zai",
+        "glm": "zai",
+        "zai_glm": "zai",
     }
     return aliases.get(normalized, normalized)
 
@@ -115,6 +119,17 @@ def get_provider_registry() -> dict[str, ProviderRegistryEntry]:
             required_settings=("groq_api_key",),
             estimated_cost_per_1k_units_usd=0.0001,
         ),
+        "zai": ProviderRegistryEntry(
+            name="zai",
+            kind="text",
+            # Free Flash: text (glm-4.7-flash) + vision OCR path (glm-4.6v-flash).
+            task_types=TEXT_TASKS | frozenset({"ocr"}),
+            quality_tiers=frozenset({"cheap", "premium"}),
+            required_settings=("zai_api_key",),
+            aliases=("z_ai", "glm"),
+            estimated_cost_per_1k_units_usd=0,
+            free_provider=True,
+        ),
         "nvidia_nim": ProviderRegistryEntry(
             name="nvidia_nim",
             kind="text",

app/services/provider_router.py

@@ -112,10 +112,13 @@ def _candidate_order(task_type: str, quality_tier: str, language: str) -> list[s
         return _dedupe(order)
 
     if task_type == "chat":
-        return _dedupe([current_default, "openrouter", "gemini", "sarvam", "groq", "mock"])
+        return _dedupe([current_default, "zai", "openrouter", "gemini", "sarvam", "groq", "mock"])
 
-    cheap_text = ["openrouter", "groq", "cloudflare_workers_ai", "huggingface", "gemini", "mock"]
-    premium_text = ["sarvam", "gemini", "nvidia_nim", "openrouter", "groq", "mock"]
+    if task_type == "ocr":
+        return _dedupe(["zai", "gemini", "openrouter", "mock"])
+
+    cheap_text = ["zai", "openrouter", "groq", "cloudflare_workers_ai", "huggingface", "gemini", "mock"]
+    premium_text = ["sarvam", "gemini", "nvidia_nim", "zai", "openrouter", "groq", "mock"]
     if task_type in {"scene_plan", "video_script"}:
         return _dedupe(cheap_text if quality_tier != "premium" else premium_text)
     if quality_tier == "premium":

app/services/zai_client.py

@@ -0,0 +1,172 @@
+"""Shared Z.AI / GLM OpenAI-compatible client (free Flash models).
+
+Text:  glm-4.7-flash
+Vision: glm-4.6v-flash (image understanding / OCR-style description)
+
+Docs: https://docs.z.ai/ — base URL https://api.z.ai/api/paas/v4/
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import urllib.error
+import urllib.request
+from typing import Any
+
+from app.core.config import get_settings
+
+logger = logging.getLogger(__name__)
+
+
+class ZaiClientError(RuntimeError):
+    def __init__(self, message: str, *, status_code: int | None = None) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+
+
+def zai_configured() -> bool:
+    settings = get_settings()
+    key = (settings.zai_api_key or "").strip()
+    return len(key) >= 16
+
+
+def _chat_url() -> str:
+    settings = get_settings()
+    base = (settings.zai_base_url or "https://api.z.ai/api/paas/v4").rstrip("/")
+    return f"{base}/chat/completions"
+
+
+def chat_completions(
+    *,
+    model: str,
+    messages: list[dict[str, Any]],
+    max_tokens: int = 2048,
+    temperature: float = 0.25,
+    response_format: dict[str, Any] | None = None,
+    thinking_disabled: bool = True,
+    timeout: float | None = None,
+) -> dict[str, Any]:
+    """Call Z.AI chat completions. Returns the parsed JSON body."""
+    settings = get_settings()
+    api_key = (settings.zai_api_key or "").strip()
+    if len(api_key) < 16:
+        raise ZaiClientError("ZAI_API_KEY is missing or invalid.", status_code=401)
+
+    payload: dict[str, Any] = {
+        "model": model,
+        "messages": messages,
+        "temperature": temperature,
+        "max_tokens": max_tokens,
+    }
+    # Flash free tier is faster with thinking off for structured tasks.
+    if thinking_disabled:
+        payload["thinking"] = {"type": "disabled"}
+    if response_format is not None:
+        payload["response_format"] = response_format
+
+    request = urllib.request.Request(
+        _chat_url(),
+        data=json.dumps(payload).encode("utf-8"),
+        headers={
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {api_key}",
+            "User-Agent": "DocDoe-ZAI/1.0",
+        },
+        method="POST",
+    )
+    wait = float(timeout if timeout is not None else settings.zai_timeout_seconds)
+    try:
+        with urllib.request.urlopen(request, timeout=wait) as response:
+            return json.loads(response.read().decode("utf-8"))
+    except urllib.error.HTTPError as exc:
+        body = ""
+        try:
+            body = exc.read().decode("utf-8", errors="replace")[:500]
+        except Exception:
+            body = ""
+        raise ZaiClientError(
+            f"Z.AI HTTP {exc.code}: {body or exc.reason}",
+            status_code=exc.code,
+        ) from exc
+    except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
+        raise ZaiClientError(f"Z.AI request failed: {exc}") from exc
+
+
+def extract_message_text(result: dict[str, Any]) -> str:
+    """Pull assistant text from a chat.completions response."""
+    try:
+        msg = result["choices"][0]["message"]
+    except (KeyError, IndexError, TypeError) as exc:
+        raise ZaiClientError("Z.AI response missing choices/message.") from exc
+    content = msg.get("content")
+    if isinstance(content, list):
+        content = "".join(
+            part.get("text", "") if isinstance(part, dict) else str(part)
+            for part in content
+        )
+    text = (content or "").strip()
+    if not text:
+        # Some GLM responses put useful text only in reasoning_content.
+        reasoning = (msg.get("reasoning_content") or "").strip()
+        if reasoning:
+            return reasoning
+    return text
+
+
+def generate_text(
+    prompt: str,
+    *,
+    system: str | None = None,
+    max_tokens: int = 2048,
+    temperature: float = 0.25,
+    json_mode: bool = False,
+) -> str:
+    settings = get_settings()
+    messages: list[dict[str, Any]] = []
+    if system:
+        messages.append({"role": "system", "content": system})
+    messages.append({"role": "user", "content": prompt})
+    result = chat_completions(
+        model=settings.zai_model_text,
+        messages=messages,
+        max_tokens=max_tokens,
+        temperature=temperature,
+        response_format={"type": "json_object"} if json_mode else None,
+        thinking_disabled=True,
+    )
+    text = extract_message_text(result)
+    if not text:
+        raise ZaiClientError("Z.AI returned empty content.")
+    return text
+
+
+def describe_image(
+    *,
+    image_url: str,
+    prompt: str = "Describe this image clearly for a student. Extract any text, diagrams, formulas, and labels.",
+    max_tokens: int = 1500,
+) -> str:
+    """Vision: GLM-4.6V-Flash free model. image_url may be https or data: URL."""
+    settings = get_settings()
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": prompt},
+                {"type": "image_url", "image_url": {"url": image_url}},
+            ],
+        }
+    ]
+    result = chat_completions(
+        model=settings.zai_model_vision,
+        messages=messages,
+        max_tokens=max_tokens,
+        temperature=0.2,
+        thinking_disabled=True,
+    )
+    text = extract_message_text(result)
+    if not text:
+        raise ZaiClientError("Z.AI vision returned empty content.")
+    logger.info("Z.AI vision ok model=%s chars=%s", settings.zai_model_vision, len(text))
+    return text