A3-R5D GLM live capability probe harness

app.py

@@ -41,6 +41,13 @@ st.set_page_config(
     initial_sidebar_state="collapsed",
 )
 
+# A3-R5D: default-off, benchmark-only startup probe; never blocks the UI.
+try:
+    from core.a3_glm_live_probe import run_startup_probe_if_enabled
+    run_startup_probe_if_enabled()
+except Exception as _a3_probe_exc:
+    logging.error("[A3_R5D_GLM_PROBE] bootstrap_failed error_class=%s", _a3_probe_exc.__class__.__name__)
+
 WORKSPACE_DIR = APP_CONFIG.get("workspace_dir", ".workspace")
 os.makedirs(WORKSPACE_DIR, exist_ok=True)
 # The display-only preview cache must exist from the first Streamlit script run.

core/a3_glm_live_probe.py

@@ -0,0 +1,189 @@
+# -*- coding: utf-8 -*-
+"""A3-R5D isolated GLM live-capability startup probe.
+
+Default-off, benchmark-only diagnostic.  It never changes product routing,
+quota/payment state, manuscript generation, or UI.  Provider/model/base URL and
+API-key environment names are resolved from the versioned A3 benchmark fixture
+rather than hard-coded in this module.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import os
+from pathlib import Path
+import threading
+from typing import Any, Mapping
+
+from core.provider_adapters.glm_rest_adapter import ZhipuGLMRestAdapter
+
+_LOG = logging.getLogger(__name__)
+_LOCK = threading.Lock()
+_RAN = False
+
+
+def _truthy(value: Any) -> bool:
+    return str(value or "").strip().lower() in {"1", "true", "yes", "on", "enabled"}
+
+
+def _fixture_path() -> Path:
+    return Path(__file__).resolve().parents[1] / "benchmarks" / "fixtures" / "a3_benchmark_cases.json"
+
+
+def _load_probe_config() -> dict[str, Any]:
+    data = json.loads(_fixture_path().read_text(encoding="utf-8"))
+    profiles = data.get("profiles") or {}
+    adapters = data.get("provider_adapters") or {}
+    zhipu_cfg = dict(adapters.get("zhipu") or {})
+    if not zhipu_cfg:
+        raise RuntimeError("zhipu benchmark adapter configuration is missing")
+
+    selected: dict[bool, tuple[str, Mapping[str, Any]]] = {}
+    for profile_id, raw in profiles.items():
+        profile = dict(raw or {})
+        if str(profile.get("provider") or "").strip().lower() != "zhipu":
+            continue
+        if str(profile.get("monetary_cost_class") or "").strip().lower() != "free":
+            continue
+        thinking = bool(profile.get("thinking"))
+        selected.setdefault(thinking, (str(profile_id), profile))
+
+    if False not in selected or True not in selected:
+        raise RuntimeError("both free zhipu thinking profiles are required for the live probe")
+
+    base_url_env = str(zhipu_cfg.get("base_url_env") or "").strip()
+    base_url = str(os.getenv(base_url_env) or zhipu_cfg.get("base_url") or "").strip()
+    api_key_env = str(zhipu_cfg.get("api_key_env") or "").strip()
+    if not api_key_env:
+        raise RuntimeError("zhipu api_key_env is not configured")
+    api_key = str(os.getenv(api_key_env) or "").strip()
+    if not api_key:
+        raise RuntimeError("configured zhipu API key environment variable is unavailable")
+    if not base_url:
+        raise RuntimeError("configured zhipu base URL is unavailable")
+
+    return {
+        "benchmark_version": str(data.get("benchmark_version") or ""),
+        "adapter": zhipu_cfg,
+        "base_url": base_url,
+        "api_key": api_key,
+        "profiles": selected,
+    }
+
+
+def _event_dict(event: Any) -> dict[str, Any]:
+    if hasattr(event, "to_dict") and callable(event.to_dict):
+        return dict(event.to_dict())
+    if isinstance(event, Mapping):
+        return dict(event)
+    return {}
+
+
+def _safe_summary(profile_id: str, event: Mapping[str, Any], *, content_ok: bool) -> dict[str, Any]:
+    return {
+        "profile_id": profile_id,
+        "status": str(event.get("status") or ""),
+        "provider": str(event.get("provider") or ""),
+        "model": str(event.get("provider_model_version") or event.get("model") or ""),
+        "streaming": bool(event.get("streaming")),
+        "thinking_mode": str(event.get("thinking_mode") or ""),
+        "finish_reason": str(event.get("finish_reason") or ""),
+        "http_status": event.get("http_status"),
+        "latency_ms": int(event.get("latency_ms") or 0),
+        "prompt_input_tokens": int(event.get("prompt_input_tokens") or 0),
+        "cache_hit_tokens": int(event.get("cache_hit_tokens") or 0),
+        "cache_miss_tokens": int(event.get("cache_miss_tokens") or 0),
+        "output_tokens": int(event.get("output_tokens") or 0),
+        "reasoning_or_thought_tokens": int(event.get("reasoning_or_thought_tokens") or 0),
+        "total_tokens": int(event.get("total_tokens") or 0),
+        "content_ok": bool(content_ok),
+    }
+
+
+def run_glm_live_probe() -> dict[str, Any]:
+    """Run two tiny real calls: static/no-thinking and stream/thinking."""
+    cfg = _load_probe_config()
+    results: list[dict[str, Any]] = []
+
+    for thinking, streaming in ((False, False), (True, True)):
+        profile_id, profile = cfg["profiles"][thinking]
+        events: list[dict[str, Any]] = []
+
+        def sink(event: Any) -> None:
+            events.append(_event_dict(event))
+
+        adapter = ZhipuGLMRestAdapter(
+            api_key=cfg["api_key"],
+            model_name=str(profile.get("model") or "").strip(),
+            base_url=cfg["base_url"],
+            timeout_seconds=float(cfg["adapter"].get("timeout_seconds") or 120),
+            usage_sink=sink,
+            usage_context={
+                "phase_id": "a3_r5d_live_probe",
+                "case_id": "stream_thinking" if streaming else "static_no_thinking",
+                "benchmark_version": cfg["benchmark_version"],
+                "profile_id": profile_id,
+                "attempt_index": 1,
+            },
+        )
+        if not adapter.model_name:
+            raise RuntimeError("zhipu benchmark profile model is missing")
+
+        system_prompt = "Connectivity capability probe."
+        user_prompt = "Reply with exactly OK."
+        if streaming:
+            chunks = list(adapter.stream(
+                system_prompt,
+                user_prompt,
+                max_tokens=128,
+                model_name=adapter.model_name,
+                thinking=True,
+                temp=0.0,
+                allow_fallback=False,
+            ))
+            content_ok = bool("".join(chunks).strip())
+        else:
+            text = adapter.call(
+                system_prompt,
+                user_prompt,
+                max_tokens=16,
+                temp=0.0,
+                model_name=adapter.model_name,
+                thinking=False,
+                allow_fallback=False,
+            )
+            content_ok = bool(str(text or "").strip())
+
+        event = events[-1] if events else {}
+        results.append(_safe_summary(profile_id, event, content_ok=content_ok))
+
+    return {
+        "probe": "A3-R5D",
+        "status": "pass" if all(row.get("status") == "success" and row.get("content_ok") for row in results) else "fail",
+        "results": results,
+    }
+
+
+def run_startup_probe_if_enabled() -> None:
+    """Run at most once per process, only when the explicit probe flag is enabled."""
+    global _RAN
+    if not _truthy(os.getenv("A3_R5D_GLM_PROBE_ENABLED")):
+        return
+    with _LOCK:
+        if _RAN:
+            return
+        _RAN = True
+
+    try:
+        summary = run_glm_live_probe()
+        _LOG.warning("[A3_R5D_GLM_PROBE] %s", json.dumps(summary, ensure_ascii=False, sort_keys=True))
+    except Exception as exc:
+        # Do not log provider response text, prompts, headers, API keys, or raw exception strings.
+        _LOG.error(
+            "[A3_R5D_GLM_PROBE] %s",
+            json.dumps({
+                "probe": "A3-R5D",
+                "status": "error",
+                "error_class": exc.__class__.__name__,
+            }, sort_keys=True),
+        )