A3-R5H Gemini free live probe

app.py

@@ -48,6 +48,13 @@ try:
 except Exception as _a3_probe_exc:
     logging.error("[A3_R5D_GLM_PROBE] bootstrap_failed error_class=%s", _a3_probe_exc.__class__.__name__)
 
+# A3-R5H: default-off Gemini Free-Tier startup probe; never blocks the UI.
+try:
+    from core.a3_gemini_live_probe import run_startup_probe_if_enabled as run_gemini_startup_probe_if_enabled
+    run_gemini_startup_probe_if_enabled()
+except Exception as _a3_gemini_probe_exc:
+    logging.error("[A3_R5H_GEMINI_PROBE] bootstrap_failed error_class=%s", _a3_gemini_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_gemini_live_probe.py

@@ -0,0 +1,227 @@
+# -*- coding: utf-8 -*-
+"""A3-R5H isolated Gemini Free-Tier 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 come from the versioned A3 benchmark fixture.
+A separate explicit Free-Tier confirmation flag is required before any request.
+"""
+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.gemini_rest_adapter import GeminiRestAdapter
+
+_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 {}
+    gemini_cfg = dict(adapters.get("gemini") or {})
+    if not gemini_cfg:
+        raise RuntimeError("gemini 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() != "gemini":
+            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 gemini mechanical and reasoning profiles are required")
+
+    pool = dict(data.get("free_provider_pool") or {})
+    if str(pool.get("price_contract") or "") != "input_and_output_zero_required":
+        raise RuntimeError("free-provider zero-price contract is missing")
+    if "gemini" not in {str(item).strip().lower() for item in list(pool.get("providers") or [])}:
+        raise RuntimeError("gemini is not authorized in the benchmark free-provider pool")
+    if bool(pool.get("paid_fallback_allowed")):
+        raise RuntimeError("paid fallback must remain disabled for the free-provider pool")
+
+    base_url_env = str(gemini_cfg.get("base_url_env") or "").strip()
+    base_url = str(os.getenv(base_url_env) or gemini_cfg.get("base_url") or "").strip()
+    api_key_env = str(gemini_cfg.get("api_key_env") or "").strip()
+    if not api_key_env:
+        raise RuntimeError("gemini api_key_env is not configured")
+    api_key = str(os.getenv(api_key_env) or "").strip()
+    if not api_key:
+        raise RuntimeError("configured Gemini API key environment variable is unavailable")
+    if not base_url:
+        raise RuntimeError("configured Gemini base URL is unavailable")
+
+    return {
+        "benchmark_version": str(data.get("benchmark_version") or ""),
+        "adapter": gemini_cfg,
+        "base_url": base_url,
+        "api_key": api_key,
+        "profiles": selected,
+    }
+
+
+def _zero_cost_guard() -> None:
+    # Human-confirmed project status is intentionally separate from the API key.
+    # Default false means an accidental deployment/restart cannot send a request.
+    if not _truthy(os.getenv("A3_R5H_GEMINI_FREE_TIER_CONFIRMED")):
+        raise RuntimeError("Gemini Free-Tier project status has not been explicitly confirmed")
+
+
+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(case_id: str, profile_id: str, event: Mapping[str, Any], *, content_ok: bool) -> dict[str, Any]:
+    return {
+        "case_id": case_id,
+        "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),
+        "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 _adapter_for(cfg: Mapping[str, Any], profile_id: str, profile: Mapping[str, Any], case_id: str, events: list[dict[str, Any]]) -> GeminiRestAdapter:
+    def sink(event: Any) -> None:
+        events.append(_event_dict(event))
+
+    return GeminiRestAdapter(
+        api_key=str(cfg["api_key"]),
+        model_name=str(profile.get("model") or "").strip(),
+        base_url=str(cfg["base_url"]),
+        timeout_seconds=float(dict(cfg["adapter"]).get("timeout_seconds") or 120),
+        usage_sink=sink,
+        usage_context={
+            "phase_id": "a3_r5h_gemini_free_live_probe",
+            "case_id": case_id,
+            "benchmark_version": str(cfg["benchmark_version"]),
+            "profile_id": profile_id,
+            "attempt_index": 1,
+        },
+    )
+
+
+def run_gemini_live_probe() -> dict[str, Any]:
+    """Run three tiny Free-Tier calls: ordinary, structured JSON, and stream."""
+    _zero_cost_guard()
+    cfg = _load_probe_config()
+    results: list[dict[str, Any]] = []
+
+    mechanical_id, mechanical = cfg["profiles"][False]
+    reasoning_id, reasoning = cfg["profiles"][True]
+
+    # 1) Ordinary generateContent.
+    events: list[dict[str, Any]] = []
+    adapter = _adapter_for(cfg, mechanical_id, mechanical, "ordinary", events)
+    text = adapter.call(
+        "Connectivity capability probe.",
+        "Reply with exactly OK.",
+        max_tokens=16,
+        temp=0.0,
+        model_name=adapter.model_name,
+        thinking=False,
+        allow_fallback=False,
+    )
+    results.append(_safe_summary("ordinary", mechanical_id, events[-1] if events else {}, content_ok=str(text or "").strip() == "OK"))
+
+    # 2) Structured JSON generateContent.
+    events = []
+    adapter = _adapter_for(cfg, mechanical_id, mechanical, "structured_json", events)
+    text = adapter.call(
+        "Structured-output capability probe.",
+        'Return a JSON object with exactly one field: {"ok": true}.',
+        max_tokens=32,
+        temp=0.0,
+        model_name=adapter.model_name,
+        thinking=False,
+        allow_fallback=False,
+        response_format={"type": "json_object"},
+    )
+    try:
+        parsed = json.loads(str(text or ""))
+        json_ok = isinstance(parsed, Mapping) and parsed.get("ok") is True and len(parsed) == 1
+    except Exception:
+        json_ok = False
+    results.append(_safe_summary("structured_json", mechanical_id, events[-1] if events else {}, content_ok=json_ok))
+
+    # 3) streamGenerateContent using the free reasoning profile.
+    events = []
+    adapter = _adapter_for(cfg, reasoning_id, reasoning, "stream", events)
+    chunks = list(adapter.stream(
+        "Streaming capability probe.",
+        "Reply with exactly OK.",
+        max_tokens=32,
+        temp=0.0,
+        model_name=adapter.model_name,
+        thinking=True,
+        reasoning_effort=str(reasoning.get("reasoning_effort") or "low"),
+        allow_fallback=False,
+    ))
+    results.append(_safe_summary("stream", reasoning_id, events[-1] if events else {}, content_ok="".join(chunks).strip() == "OK"))
+
+    passed = all(
+        row.get("status") == "success"
+        and row.get("http_status") == 200
+        and row.get("content_ok")
+        and int(row.get("total_tokens") or 0) > 0
+        for row in results
+    )
+    return {"probe": "A3-R5H", "status": "pass" if passed else "fail", "results": results}
+
+
+def run_startup_probe_if_enabled() -> None:
+    """Run at most once per process, only when both explicit safety flags allow it."""
+    global _RAN
+    if not _truthy(os.getenv("A3_R5H_GEMINI_PROBE_ENABLED")):
+        return
+    with _LOCK:
+        if _RAN:
+            return
+        _RAN = True
+
+    try:
+        summary = run_gemini_live_probe()
+        _LOG.warning("[A3_R5H_GEMINI_PROBE] %s", json.dumps(summary, ensure_ascii=False, sort_keys=True))
+    except Exception as exc:
+        # Never log provider response bodies, prompts, headers, API keys, or raw exception strings.
+        _LOG.error(
+            "[A3_R5H_GEMINI_PROBE] %s",
+            json.dumps({
+                "probe": "A3-R5H",
+                "status": "error",
+                "error_class": exc.__class__.__name__,
+            }, sort_keys=True),
+        )