A3-R5C benchmark-only GLM adapter deployment

benchmarks/a3_provider_benchmark.py

@@ -0,0 +1,514 @@
+# -*- coding: utf-8 -*-
+"""Standalone A3 B1/B2 provider benchmark runner.
+
+This module is intentionally unreachable from SmartPaper app/task execution.
+It evaluates fixed synthetic cases, records provider-neutral usage telemetry,
+and never touches Worker/D1 quota, payment, order, or reservation state.
+"""
+from __future__ import annotations
+
+import argparse
+from dataclasses import asdict
+from datetime import datetime, timezone
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import sys
+from typing import Any, Mapping
+from zoneinfo import ZoneInfo
+
+HF = Path(__file__).resolve().parents[1]
+if str(HF) not in sys.path:
+    sys.path.insert(0, str(HF))
+
+from core.model_usage import JsonlModelUsageSink, ModelUsageEvent  # noqa: E402
+from core.pipeline_agent import SSciPipelineAgent  # noqa: E402
+from core.provider_adapters.gemini_rest_adapter import GeminiRestAdapter  # noqa: E402
+from core.provider_adapters.glm_rest_adapter import ZhipuGLMRestAdapter  # noqa: E402
+
+DEFAULT_FIXTURE = Path(__file__).resolve().parent / "fixtures" / "a3_benchmark_cases.json"
+
+
+class MemoryUsageSink:
+    def __init__(self, downstream: Any = None):
+        self.events: list[ModelUsageEvent] = []
+        self.downstream = downstream
+
+    def emit(self, event: ModelUsageEvent) -> None:
+        self.events.append(event)
+        if self.downstream is not None:
+            self.downstream.emit(event)
+
+
+def load_fixture(path: str | Path = DEFAULT_FIXTURE) -> dict[str, Any]:
+    data = json.loads(Path(path).read_text(encoding="utf-8"))
+    if not isinstance(data.get("profiles"), dict) or not isinstance(data.get("cases"), list):
+        raise ValueError("invalid A3 benchmark fixture")
+    return data
+
+
+def _english_word_count(text: str) -> int:
+    return len(re.findall(r"\b[A-Za-z]+(?:[-'][A-Za-z]+)*\b", str(text or "")))
+
+
+def _script_ratio(text: str, script: str) -> float:
+    letters = [ch for ch in str(text or "") if ch.isalpha()]
+    if not letters:
+        return 0.0
+    name = str(script or "").strip().lower()
+    if name == "han":
+        matched = sum(1 for ch in letters if "\u4e00" <= ch <= "\u9fff")
+    elif name == "cyrillic":
+        matched = sum(1 for ch in letters if "\u0400" <= ch <= "\u052f")
+    else:
+        raise ValueError(f"unsupported script gate: {script}")
+    return matched / len(letters)
+
+
+def evaluate_gate(output: str, gate: Mapping[str, Any]) -> dict[str, Any]:
+    kind = str(gate.get("type") or "").strip().lower()
+    text = str(output or "")
+    lower = text.lower()
+    passed = False
+    detail = ""
+
+    if kind == "exact_ci":
+        expected = str(gate.get("expected") or "").strip().lower()
+        observed = text.strip().lower()
+        passed = observed == expected
+        detail = f"expected={expected!r} observed={observed[:120]!r}"
+    elif kind == "contains_all_ci":
+        values = [str(v) for v in gate.get("values") or []]
+        missing = [v for v in values if v.lower() not in lower]
+        passed = not missing
+        detail = f"missing={missing}"
+    elif kind == "contains_any_ci":
+        values = [str(v) for v in gate.get("values") or []]
+        passed = any(v.lower() in lower for v in values)
+        detail = f"values={values}"
+    elif kind == "forbid_any_ci":
+        values = [str(v) for v in gate.get("values") or []]
+        found = [v for v in values if v.lower() in lower]
+        passed = not found
+        detail = f"found={found}"
+    elif kind == "count_exact_ci":
+        value = str(gate.get("value") or "")
+        expected = int(gate.get("count") or 0)
+        observed = lower.count(value.lower())
+        passed = observed == expected
+        detail = f"value={value!r} expected={expected} observed={observed}"
+    elif kind == "word_range":
+        count = _english_word_count(text)
+        minimum = int(gate.get("min") or 0)
+        maximum = int(gate.get("max") or 10**9)
+        passed = minimum <= count <= maximum
+        detail = f"words={count} range=[{minimum},{maximum}]"
+    elif kind == "script_min_ratio":
+        ratio = _script_ratio(text, str(gate.get("script") or ""))
+        minimum = float(gate.get("min") or 0.0)
+        passed = ratio >= minimum
+        detail = f"ratio={ratio:.4f} minimum={minimum:.4f}"
+    elif kind in {"json_keys_exact", "json_value_ci"}:
+        try:
+            parsed = json.loads(text)
+        except Exception as exc:
+            parsed = None
+            detail = f"json_error={exc.__class__.__name__}"
+        if isinstance(parsed, dict):
+            if kind == "json_keys_exact":
+                expected = {str(v) for v in gate.get("keys") or []}
+                observed = set(map(str, parsed.keys()))
+                passed = observed == expected
+                detail = f"expected={sorted(expected)} observed={sorted(observed)}"
+            else:
+                key = str(gate.get("key") or "")
+                expected = str(gate.get("expected") or "").strip().lower()
+                observed = str(parsed.get(key) or "").strip().lower()
+                passed = observed == expected
+                detail = f"key={key!r} expected={expected!r} observed={observed!r}"
+    else:
+        raise ValueError(f"unsupported benchmark gate: {kind}")
+
+    return {"type": kind, "passed": bool(passed), "detail": detail}
+
+
+def evaluate_case(output: str, case: Mapping[str, Any]) -> dict[str, Any]:
+    gate_results = [evaluate_gate(output, gate) for gate in list(case.get("gates") or [])]
+    passed = all(row["passed"] for row in gate_results)
+    return {
+        "passed": passed,
+        "critical": bool(case.get("critical", True)),
+        "gate_results": gate_results,
+    }
+
+
+def _is_peak(now: datetime, snapshot: Mapping[str, Any]) -> bool:
+    tz = ZoneInfo(str(snapshot.get("timezone") or "UTC"))
+    local = now.astimezone(tz)
+    hour = local.hour + local.minute / 60.0
+    return any(float(start) <= hour < float(end) for start, end in snapshot.get("peak_windows") or [])
+
+
+def calculate_cost(
+    event: Mapping[str, Any],
+    price_snapshot: Mapping[str, Any],
+    *,
+    monetary_cost_class: str = "",
+) -> dict[str, Any]:
+    cost_class = str(monetary_cost_class or "").strip().lower()
+    if cost_class == "free":
+        return {"estimated_cost": 0.0, "currency": "", "billing_window": "free"}
+    model = str(event.get("model") or "")
+    rates_by_model = dict(price_snapshot.get("per_million_tokens") or {}).get(model)
+    if not isinstance(rates_by_model, Mapping):
+        return {"estimated_cost": None, "currency": str(price_snapshot.get("currency") or ""), "billing_window": "unknown"}
+    started = str(event.get("request_started_at") or "")
+    try:
+        now = datetime.fromisoformat(started.replace("Z", "+00:00")) if started else datetime.now(timezone.utc)
+    except ValueError:
+        now = datetime.now(timezone.utc)
+    window = "peak" if _is_peak(now, price_snapshot) else "off_peak"
+    rates = dict(rates_by_model.get(window) or {})
+    million = 1_000_000.0
+    cost = (
+        float(event.get("cache_hit_tokens") or 0) * float(rates.get("cache_hit_input") or 0) / million
+        + float(event.get("cache_miss_tokens") or 0) * float(rates.get("cache_miss_input") or 0) / million
+        + float(event.get("output_tokens") or 0) * float(rates.get("output") or 0) / million
+    )
+    return {
+        "estimated_cost": round(cost, 8),
+        "currency": str(price_snapshot.get("currency") or ""),
+        "billing_window": window,
+        "price_snapshot_id": str(price_snapshot.get("id") or ""),
+    }
+
+
+def _result_preview(text: str, limit: int = 240) -> str:
+    return re.sub(r"\s+", " ", str(text or "")).strip()[:limit]
+
+
+def _output_hash(text: str) -> str:
+    return hashlib.sha256(str(text or "").encode("utf-8")).hexdigest()
+
+
+def _profile_route(agent: SSciPipelineAgent, capability: str, provider: str) -> None:
+    agent.set_model_policy_snapshot({
+        "model_routes": {
+            capability: {
+                "provider": provider,
+                "provider_role": capability,
+                "fallback_providers": [],
+            }
+        }
+    })
+
+
+def run_case(
+    *,
+    agent: SSciPipelineAgent,
+    provider_adapter: Any = None,
+    case: Mapping[str, Any],
+    profile_id: str,
+    profile: Mapping[str, Any],
+    benchmark_version: str,
+    usage_sink: MemoryUsageSink,
+    price_snapshot: Mapping[str, Any],
+) -> dict[str, Any]:
+    provider = str(profile.get("provider") or "").strip().lower()
+    capability = str(case.get("capability") or "reasoning").strip().lower()
+    model = str(profile.get("model") or "").strip()
+    context = {
+        "case_id": str(case.get("case_id") or ""),
+        "benchmark_version": benchmark_version,
+        "profile_id": profile_id,
+        "phase_id": capability,
+        "attempt_index": 0,
+    }
+    agent.set_model_usage_sink(usage_sink, context)
+    agent.set_model_usage_context(**context)
+    if provider_adapter is not None:
+        if hasattr(provider_adapter, "usage_sink"):
+            provider_adapter.usage_sink = usage_sink
+        if hasattr(provider_adapter, "set_usage_context") and callable(provider_adapter.set_usage_context):
+            provider_adapter.set_usage_context(**context)
+    _profile_route(agent, capability, provider)
+
+    before = len(usage_sink.events)
+    output = ""
+    error = None
+    try:
+        output = agent.call_model(
+            str(case.get("system_prompt") or ""),
+            str(case.get("user_prompt") or ""),
+            max_tokens=int(case.get("max_tokens") or 8000),
+            temp=float(case.get("temperature") or 0.2),
+            model_name=model or None,
+            model_role=capability,
+            thinking=bool(profile.get("thinking", False)),
+            reasoning_effort=profile.get("reasoning_effort"),
+            allow_fallback=bool(profile.get("allow_fallback", False)),
+            response_format=case.get("response_format") if isinstance(case.get("response_format"), dict) else None,
+        )
+    except Exception as exc:
+        error = exc
+
+    new_events = usage_sink.events[before:]
+    usage_rows = [event.to_dict() for event in new_events]
+    usage = usage_rows[-1] if usage_rows else {}
+    evaluation = evaluate_case(output, case) if error is None else {
+        "passed": False,
+        "critical": bool(case.get("critical", True)),
+        "gate_results": [],
+    }
+    cost_class = str(profile.get("monetary_cost_class") or "").strip().lower()
+    cost = calculate_cost(usage, price_snapshot, monetary_cost_class=cost_class) if usage else {
+        "estimated_cost": 0.0 if cost_class == "free" else None,
+        "currency": "",
+        "billing_window": "no_usage",
+    }
+    return {
+        "case_id": str(case.get("case_id") or ""),
+        "stage": str(case.get("stage") or ""),
+        "benchmark_version": benchmark_version,
+        "profile_id": profile_id,
+        "provider": provider,
+        "model": model,
+        "capability": capability,
+        "thinking": bool(profile.get("thinking", False)),
+        "reasoning_effort": profile.get("reasoning_effort"),
+        "max_tokens": int(case.get("max_tokens") or 8000),
+        "passed": bool(evaluation["passed"]),
+        "critical": bool(evaluation["critical"]),
+        "gate_results": evaluation["gate_results"],
+        "error_class": error.__class__.__name__ if error else "",
+        "error_message": str(error)[:500] if error else "",
+        "output_hash": _output_hash(output) if output else "",
+        "output_preview": _result_preview(output),
+        "usage": usage,
+        "cost": cost,
+    }
+
+
+def summarize(results: list[Mapping[str, Any]], stage: str) -> dict[str, Any]:
+    critical = [row for row in results if row.get("critical")]
+    critical_pass = all(row.get("passed") for row in critical) if critical else False
+    passed = sum(1 for row in results if row.get("passed"))
+    total_cost = 0.0
+    cost_known = True
+    for row in results:
+        value = (row.get("cost") or {}).get("estimated_cost")
+        if value is None:
+            cost_known = False
+        else:
+            total_cost += float(value or 0.0)
+    promotion = f"{stage.upper()}_PASS" if critical_pass else "BLOCKED"
+    return {
+        "stage": stage,
+        "promotion_status": promotion,
+        "cases_total": len(results),
+        "cases_passed": passed,
+        "critical_cases": len(critical),
+        "critical_passed": sum(1 for row in critical if row.get("passed")),
+        "first_pass_rate": (passed / len(results)) if results else 0.0,
+        "estimated_cost_total": round(total_cost, 8) if cost_known else None,
+    }
+
+
+def _provider_config(fixture: Mapping[str, Any], provider: str) -> dict[str, Any]:
+    configs = fixture.get("provider_adapters")
+    if not isinstance(configs, Mapping):
+        return {}
+    row = configs.get(provider)
+    return dict(row) if isinstance(row, Mapping) else {}
+
+
+def _env_from_config(config: Mapping[str, Any], field: str) -> str:
+    env_name = str(config.get(field) or "").strip()
+    return str(os.environ.get(env_name) or "").strip() if env_name else ""
+
+
+def _build_external_adapter(
+    *,
+    provider: str,
+    profile: Mapping[str, Any],
+    fixture: Mapping[str, Any],
+    usage_sink: MemoryUsageSink,
+) -> Any:
+    config = _provider_config(fixture, provider)
+    adapter_type = str(config.get("adapter_type") or "").strip().lower()
+    api_key = _env_from_config(config, "api_key_env")
+    base_url = _env_from_config(config, "base_url_env") or str(config.get("base_url") or "").strip() or None
+    model = str(profile.get("model") or "").strip() or None
+    timeout_seconds = float(config.get("timeout_seconds") or 120.0)
+
+    if adapter_type == "gemini_rest":
+        return GeminiRestAdapter(
+            api_key=api_key or None,
+            model_name=model,
+            base_url=base_url,
+            timeout_seconds=timeout_seconds,
+            usage_sink=usage_sink,
+        )
+    if adapter_type == "zhipu_glm_rest":
+        return ZhipuGLMRestAdapter(
+            api_key=api_key or None,
+            model_name=model,
+            base_url=base_url,
+            timeout_seconds=timeout_seconds,
+            usage_sink=usage_sink,
+        )
+    raise ValueError(f"unsupported external adapter type for {provider}: {adapter_type or '<empty>'}")
+
+
+def select_profile_by_health(
+    *,
+    fixture: Mapping[str, Any],
+    policy_id: str,
+    health: Mapping[str, Any],
+) -> str:
+    """Config-driven route selection for benchmark simulations.
+
+    The function deliberately consumes provider health observations rather than
+    country codes or browser-supplied location.  Production activation is out
+    of scope for A3-R5C; this helper only proves that future CN/non-CN switching
+    can be policy-driven without geography hard-coded in application logic.
+    """
+    policies = fixture.get("routing_policies")
+    if not isinstance(policies, Mapping):
+        raise ValueError("benchmark routing_policies are missing")
+    policy = policies.get(policy_id)
+    if not isinstance(policy, Mapping):
+        raise ValueError(f"unknown routing policy: {policy_id}")
+    allowed_cost_classes = {str(v).strip().lower() for v in policy.get("allowed_cost_classes") or []}
+    profiles = fixture.get("profiles") if isinstance(fixture.get("profiles"), Mapping) else {}
+    max_latency_ms = float(policy.get("max_latency_ms") or 0)
+    max_error_rate = float(policy.get("max_error_rate") or 0)
+    skip_rate_limited = bool(policy.get("skip_rate_limited", True))
+    for profile_id in policy.get("candidate_profiles") or []:
+        profile_id = str(profile_id or "").strip()
+        profile = profiles.get(profile_id) if isinstance(profiles, Mapping) else None
+        if not isinstance(profile, Mapping):
+            continue
+        cost_class = str(profile.get("monetary_cost_class") or "").strip().lower()
+        if allowed_cost_classes and cost_class not in allowed_cost_classes:
+            continue
+        provider = str(profile.get("provider") or "").strip().lower()
+        provider_health = health.get(provider)
+        if isinstance(provider_health, Mapping):
+            available = bool(provider_health.get("available", False))
+            latency_ms = float(provider_health.get("latency_ms") or 0)
+            error_rate = float(provider_health.get("error_rate") or 0)
+            rate_limited = bool(provider_health.get("rate_limited", False))
+        else:
+            available = bool(provider_health)
+            latency_ms = 0.0
+            error_rate = 0.0
+            rate_limited = False
+        if not available:
+            continue
+        if skip_rate_limited and rate_limited:
+            continue
+        if max_latency_ms > 0 and latency_ms > max_latency_ms:
+            continue
+        if max_error_rate > 0 and error_rate > max_error_rate:
+            continue
+        return profile_id
+    raise RuntimeError(f"no_healthy_provider_for_policy:{policy_id}")
+
+
+def run_benchmark(
+    *,
+    stage: str,
+    profile_id: str,
+    fixture_path: str | Path = DEFAULT_FIXTURE,
+    output_dir: str | Path,
+) -> dict[str, Any]:
+    fixture = load_fixture(fixture_path)
+    profile = dict(fixture["profiles"].get(profile_id) or {})
+    if not profile:
+        raise ValueError(f"unknown profile: {profile_id}")
+    provider = str(profile.get("provider") or "").lower()
+    provider_config = _provider_config(fixture, provider)
+    if provider != "deepseek" and not provider_config:
+        raise ValueError(f"unsupported provider: {provider}")
+    cases = [row for row in fixture["cases"] if str(row.get("stage") or "").lower() == stage.lower()]
+    if not cases:
+        raise ValueError(f"no cases for stage: {stage}")
+
+    output_root = Path(output_dir)
+    output_root.mkdir(parents=True, exist_ok=True)
+    telemetry_path = output_root / f"{stage}_{profile_id}_usage.jsonl"
+    result_path = output_root / f"{stage}_{profile_id}_results.jsonl"
+    summary_path = output_root / f"{stage}_{profile_id}_summary.json"
+    usage_sink = MemoryUsageSink(JsonlModelUsageSink(telemetry_path))
+
+    deepseek_config = _provider_config(fixture, "deepseek")
+    deepseek_key = _env_from_config(deepseek_config, "api_key_env")
+    agent = SSciPipelineAgent(ds_key=deepseek_key)
+    provider_adapter = None
+    if provider != "deepseek":
+        provider_adapter = _build_external_adapter(
+            provider=provider,
+            profile=profile,
+            fixture=fixture,
+            usage_sink=usage_sink,
+        )
+        agent.model_gateway.register_provider(
+            provider,
+            call_adapter=provider_adapter.call,
+            stream_adapter=provider_adapter.stream,
+        )
+
+    results = []
+    with result_path.open("w", encoding="utf-8") as handle:
+        for case in cases:
+            row = run_case(
+                agent=agent,
+                provider_adapter=provider_adapter,
+                case=case,
+                profile_id=profile_id,
+                profile=profile,
+                benchmark_version=str(fixture.get("benchmark_version") or ""),
+                usage_sink=usage_sink,
+                price_snapshot=dict(fixture.get("price_snapshot") or {}),
+            )
+            results.append(row)
+            handle.write(json.dumps(row, ensure_ascii=False, sort_keys=True) + "\n")
+            if row["critical"] and not row["passed"]:
+                # Cost-conscious stop: a critical deterministic failure blocks
+                # promotion; later cases should not consume provider capacity.
+                break
+
+    summary = summarize(results, stage.lower())
+    summary.update({
+        "profile_id": profile_id,
+        "provider": provider,
+        "benchmark_version": str(fixture.get("benchmark_version") or ""),
+        "result_file": result_path.name,
+        "usage_file": telemetry_path.name,
+    })
+    summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
+    return summary
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="SmartPaper A3 isolated provider benchmark")
+    parser.add_argument("--stage", choices=("b1", "b2"), required=True)
+    parser.add_argument("--profile", required=True)
+    parser.add_argument("--fixture", default=str(DEFAULT_FIXTURE))
+    parser.add_argument("--output-dir", required=True)
+    args = parser.parse_args(argv)
+    summary = run_benchmark(
+        stage=args.stage,
+        profile_id=args.profile,
+        fixture_path=args.fixture,
+        output_dir=args.output_dir,
+    )
+    print(json.dumps(summary, ensure_ascii=False, indent=2, sort_keys=True))
+    return 0 if summary["promotion_status"].endswith("_PASS") else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

benchmarks/fixtures/a3_benchmark_cases.json

@@ -0,0 +1,617 @@
+{
+  "benchmark_version": "a3-provider-benchmark-v2",
+  "notes": "Synthetic B1/B2 benchmark cases only; no live user prompts, credentials, payment, quota, or manuscript data. Provider/model/endpoints and environment-routing preferences are versioned configuration. Runtime geography is not hard-coded; any future production switching must use server-authoritative provider health/reachability and frozen task policy.",
+  "profiles": {
+    "gemini_free_mechanical": {
+      "provider": "gemini",
+      "model": "gemini-2.5-flash",
+      "thinking": false,
+      "reasoning_effort": "low",
+      "allow_fallback": false,
+      "monetary_cost_class": "free"
+    },
+    "gemini_free_reasoning": {
+      "provider": "gemini",
+      "model": "gemini-2.5-flash",
+      "thinking": true,
+      "reasoning_effort": "low",
+      "allow_fallback": false,
+      "monetary_cost_class": "free"
+    },
+    "deepseek_flash_economy": {
+      "provider": "deepseek",
+      "model": "deepseek-v4-flash",
+      "thinking": false,
+      "reasoning_effort": "low",
+      "allow_fallback": false,
+      "monetary_cost_class": "paid"
+    },
+    "deepseek_pro_premium": {
+      "provider": "deepseek",
+      "model": "deepseek-v4-pro",
+      "thinking": true,
+      "reasoning_effort": "high",
+      "allow_fallback": false,
+      "monetary_cost_class": "paid"
+    },
+    "zhipu_glm47_free_no_thinking": {
+      "provider": "zhipu",
+      "model": "glm-4.7-flash",
+      "thinking": false,
+      "allow_fallback": false,
+      "monetary_cost_class": "free"
+    },
+    "zhipu_glm47_free_thinking": {
+      "provider": "zhipu",
+      "model": "glm-4.7-flash",
+      "thinking": true,
+      "allow_fallback": false,
+      "monetary_cost_class": "free"
+    }
+  },
+  "price_snapshot": {
+    "id": "deepseek-v4-cn-20260817",
+    "currency": "CNY",
+    "effective_from": "2026-08-17T00:00:00+08:00",
+    "timezone": "Asia/Shanghai",
+    "peak_windows": [
+      [
+        9,
+        12
+      ],
+      [
+        14,
+        18
+      ]
+    ],
+    "per_million_tokens": {
+      "deepseek-v4-flash": {
+        "off_peak": {
+          "cache_hit_input": 0.05,
+          "cache_miss_input": 1.5,
+          "output": 4.5
+        },
+        "peak": {
+          "cache_hit_input": 0.1,
+          "cache_miss_input": 3.0,
+          "output": 9.0
+        }
+      },
+      "deepseek-v4-pro": {
+        "off_peak": {
+          "cache_hit_input": 0.15,
+          "cache_miss_input": 4.5,
+          "output": 13.5
+        },
+        "peak": {
+          "cache_hit_input": 0.3,
+          "cache_miss_input": 9.0,
+          "output": 27.0
+        }
+      }
+    },
+    "source": "DeepSeek official Models & Pricing, verified 2026-08-30"
+  },
+  "cases": [
+    {
+      "case_id": "B1-CIT-SUPPORTED",
+      "stage": "b1",
+      "capability": "reasoning",
+      "critical": true,
+      "system_prompt": "Judge whether the evidence directly supports the claim. Return exactly SUPPORTED or UNSUPPORTED.",
+      "user_prompt": "CLAIM: Perceived personalization is positively associated with learning engagement. EVIDENCE: In a cross-sectional survey, perceived personalization had a positive standardized association with learning engagement (beta = 0.35, 95% CI [0.22, 0.48]).",
+      "max_tokens": 64,
+      "gates": [
+        {
+          "type": "exact_ci",
+          "expected": "SUPPORTED"
+        }
+      ]
+    },
+    {
+      "case_id": "B1-CIT-UNSUPPORTED",
+      "stage": "b1",
+      "capability": "reasoning",
+      "critical": true,
+      "system_prompt": "Judge whether the evidence directly supports the claim. Return exactly SUPPORTED or UNSUPPORTED.",
+      "user_prompt": "CLAIM: Generative AI reduces employee turnover by 30 percent. EVIDENCE: A qualitative interview study reports that employees discussed faster access to information, but it contains no turnover measure or percentage estimate.",
+      "max_tokens": 64,
+      "gates": [
+        {
+          "type": "exact_ci",
+          "expected": "UNSUPPORTED"
+        }
+      ]
+    },
+    {
+      "case_id": "B1-QUANT-CAUSAL",
+      "stage": "b1",
+      "capability": "repair",
+      "critical": true,
+      "system_prompt": "Rewrite the sentence so that a cross-sectional observational design is described without causal inference. Return one sentence only.",
+      "user_prompt": "Rewrite: AI-enabled planning causes lower decision anxiety and therefore leads to higher purchase intention.",
+      "max_tokens": 96,
+      "gates": [
+        {
+          "type": "contains_any_ci",
+          "values": [
+            "associated",
+            "association",
+            "linked",
+            "relationship"
+          ]
+        },
+        {
+          "type": "forbid_any_ci",
+          "values": [
+            " causes ",
+            " caused ",
+            " leads to ",
+            " results in ",
+            " drives "
+          ]
+        }
+      ]
+    },
+    {
+      "case_id": "B1-QUANT-METRIC",
+      "stage": "b1",
+      "capability": "repair",
+      "critical": true,
+      "system_prompt": "Correct the result sentence using only the frozen metrics. Return one sentence and do not invent values.",
+      "user_prompt": "Frozen metrics: X→M standardized estimate = 0.21; M→Y standardized estimate = 0.35. Draft: X→M = 0.351 and M→Y = 0.21.",
+      "max_tokens": 96,
+      "gates": [
+        {
+          "type": "contains_all_ci",
+          "values": [
+            "0.21",
+            "0.35"
+          ]
+        },
+        {
+          "type": "forbid_any_ci",
+          "values": [
+            "0.351"
+          ]
+        }
+      ]
+    },
+    {
+      "case_id": "B1-QUAL-SOURCE",
+      "stage": "b1",
+      "capability": "repair",
+      "critical": true,
+      "system_prompt": "Repair the methods sentence to match the frozen study-fact contract. Return one sentence only.",
+      "user_prompt": "Contract: actors=farmers; allowed data sources=semi-structured interviews and system logs. Draft: We observed farmers in the field to understand their experiences.",
+      "max_tokens": 96,
+      "gates": [
+        {
+          "type": "contains_all_ci",
+          "values": [
+            "farmers",
+            "interview"
+          ]
+        },
+        {
+          "type": "forbid_any_ci",
+          "values": [
+            "observed farmers",
+            "observation"
+          ]
+        }
+      ]
+    },
+    {
+      "case_id": "B1-JSON",
+      "stage": "b1",
+      "capability": "fast",
+      "critical": true,
+      "system_prompt": "Return JSON only with exactly the requested keys.",
+      "user_prompt": "Return an object with keys label and confidence. label must be supported; confidence must be high.",
+      "max_tokens": 96,
+      "response_format": {
+        "type": "json_object"
+      },
+      "gates": [
+        {
+          "type": "json_keys_exact",
+          "keys": [
+            "label",
+            "confidence"
+          ]
+        },
+        {
+          "type": "json_value_ci",
+          "key": "label",
+          "expected": "supported"
+        },
+        {
+          "type": "json_value_ci",
+          "key": "confidence",
+          "expected": "high"
+        }
+      ]
+    },
+    {
+      "case_id": "B1-EN-DRAFT",
+      "stage": "b1",
+      "capability": "writing",
+      "critical": true,
+      "system_prompt": "Write concise academic prose without meta-commentary about AI writing or the drafting process.",
+      "user_prompt": "In 110-140 English words, explain why construct validity matters when evaluating a multi-item measure in an empirical management study.",
+      "max_tokens": 260,
+      "gates": [
+        {
+          "type": "word_range",
+          "min": 100,
+          "max": 155
+        },
+        {
+          "type": "forbid_any_ci",
+          "values": [
+            "as an ai",
+            "ai-generated response",
+            "this draft",
+            "i cannot"
+          ]
+        }
+      ]
+    },
+    {
+      "case_id": "B1-ZH-TRANSLATION",
+      "stage": "b1",
+      "capability": "translation",
+      "critical": true,
+      "system_prompt": "Translate into formal academic Chinese. Preserve the meaning and do not add commentary.",
+      "user_prompt": "Generative artificial intelligence is associated with changes in perceived decision support, but the cross-sectional design does not establish causality.",
+      "max_tokens": 180,
+      "gates": [
+        {
+          "type": "contains_all_ci",
+          "values": [
+            "生成式",
+            "横断面"
+          ]
+        },
+        {
+          "type": "script_min_ratio",
+          "script": "han",
+          "min": 0.35
+        }
+      ]
+    },
+    {
+      "case_id": "B1-RU-TRANSLATION",
+      "stage": "b1",
+      "capability": "translation",
+      "critical": true,
+      "system_prompt": "Translate into formal academic Russian. Preserve the non-causal meaning and do not add commentary.",
+      "user_prompt": "Generative artificial intelligence is associated with changes in perceived decision support, but the cross-sectional design does not establish causality.",
+      "max_tokens": 180,
+      "gates": [
+        {
+          "type": "contains_any_ci",
+          "values": [
+            "генератив",
+            "искусственн"
+          ]
+        },
+        {
+          "type": "contains_any_ci",
+          "values": [
+            "попереч",
+            "кросс-секц",
+            "причин"
+          ]
+        },
+        {
+          "type": "script_min_ratio",
+          "script": "cyrillic",
+          "min": 0.55
+        }
+      ]
+    },
+    {
+      "case_id": "B1-KK-TRANSLATION",
+      "stage": "b1",
+      "capability": "translation",
+      "critical": true,
+      "system_prompt": "Translate into formal academic Kazakh. Preserve the non-causal meaning and do not add commentary.",
+      "user_prompt": "Generative artificial intelligence is associated with changes in perceived decision support, but the cross-sectional design does not establish causality.",
+      "max_tokens": 180,
+      "gates": [
+        {
+          "type": "contains_any_ci",
+          "values": [
+            "генеративті",
+            "жасанды интеллект"
+          ]
+        },
+        {
+          "type": "contains_any_ci",
+          "values": [
+            "зерттеу",
+            "себеп",
+            "байланыс"
+          ]
+        },
+        {
+          "type": "script_min_ratio",
+          "script": "cyrillic",
+          "min": 0.55
+        }
+      ]
+    },
+    {
+      "case_id": "B1-TERM-CONSISTENCY",
+      "stage": "b1",
+      "capability": "translation",
+      "critical": true,
+      "system_prompt": "Translate into academic Chinese and use the fixed glossary exactly: algorithmic fidelity=算法保真度; silicon sampling=硅基抽样.",
+      "user_prompt": "Algorithmic fidelity is a central validity concern in silicon sampling.",
+      "max_tokens": 120,
+      "gates": [
+        {
+          "type": "contains_all_ci",
+          "values": [
+            "算法保真度",
+            "硅基抽样"
+          ]
+        }
+      ]
+    },
+    {
+      "case_id": "B1-LOCAL-REPAIR",
+      "stage": "b1",
+      "capability": "repair",
+      "critical": true,
+      "system_prompt": "Repair only the unsupported causal wording. Preserve the numerical result and do not rewrite unrelated content.",
+      "user_prompt": "Cross-sectional result: beta = 0.28. Draft: The intervention caused a 0.28 increase in performance.",
+      "max_tokens": 100,
+      "gates": [
+        {
+          "type": "contains_all_ci",
+          "values": [
+            "0.28"
+          ]
+        },
+        {
+          "type": "forbid_any_ci",
+          "values": [
+            "caused",
+            "cause a",
+            "causes"
+          ]
+        },
+        {
+          "type": "contains_any_ci",
+          "values": [
+            "associated",
+            "association",
+            "relationship",
+            "linked"
+          ]
+        }
+      ]
+    },
+    {
+      "case_id": "B2-QUANT-DISCUSSION",
+      "stage": "b2",
+      "capability": "writing",
+      "critical": true,
+      "system_prompt": "Write a Discussion section using only the supplied evidence. Use cautious observational language, preserve all metrics, and include the design limitation.",
+      "user_prompt": "Topic: AI-enabled decision support in online travel planning. Frozen evidence: X→M standardized estimate 0.21; M→Y 0.35; indirect effect 0.07 with bootstrap 95% CI [0.03, 0.12]. Design: single-wave cross-sectional survey. Write 450-650 English words. Do not claim causal identification.",
+      "max_tokens": 1000,
+      "gates": [
+        {
+          "type": "word_range",
+          "min": 400,
+          "max": 700
+        },
+        {
+          "type": "contains_all_ci",
+          "values": [
+            "0.21",
+            "0.35",
+            "0.07",
+            "0.03",
+            "0.12",
+            "cross-sectional"
+          ]
+        },
+        {
+          "type": "forbid_any_ci",
+          "values": [
+            "caused by",
+            "causes",
+            "demonstrates that x causes",
+            "establishes causality"
+          ]
+        }
+      ]
+    },
+    {
+      "case_id": "B2-QUAL-FINDINGS",
+      "stage": "b2",
+      "capability": "writing",
+      "critical": true,
+      "system_prompt": "Write a qualitative Findings subsection. Preserve the quote identity exactly once and do not invent additional participants, data sources, or quotations.",
+      "user_prompt": "Study contract: participants=farmers; data sources=semi-structured interviews and system logs. Theme=trust calibration. Required quotation: [QUOTE-01] ‘I check the recommendation against my own field records before I act.’ Write 400-600 English words and include QUOTE-01 exactly once.",
+      "max_tokens": 900,
+      "gates": [
+        {
+          "type": "word_range",
+          "min": 350,
+          "max": 650
+        },
+        {
+          "type": "count_exact_ci",
+          "value": "QUOTE-01",
+          "count": 1
+        },
+        {
+          "type": "contains_all_ci",
+          "values": [
+            "farmers",
+            "interview",
+            "system logs",
+            "trust"
+          ]
+        },
+        {
+          "type": "forbid_any_ci",
+          "values": [
+            "participant 99",
+            "focus group",
+            "observation data"
+          ]
+        }
+      ]
+    },
+    {
+      "case_id": "B2-CITATION-BOUND",
+      "stage": "b2",
+      "capability": "writing",
+      "critical": true,
+      "system_prompt": "Write a theoretical-background subsection using only the two evidence records. Put the matching citation tag immediately after the supported claim. Do not transfer evidence across tags.",
+      "user_prompt": "Evidence [CITE-01]: Transactive memory systems improve coordination by helping members know who has expertise. Evidence [CITE-02]: AI literacy includes the ability to critically evaluate AI outputs. Write 350-500 English words connecting the two ideas without inventing empirical effects.",
+      "max_tokens": 850,
+      "gates": [
+        {
+          "type": "word_range",
+          "min": 300,
+          "max": 550
+        },
+        {
+          "type": "contains_all_ci",
+          "values": [
+            "[CITE-01]",
+            "[CITE-02]",
+            "expertise",
+            "critically"
+          ]
+        }
+      ]
+    },
+    {
+      "case_id": "B2-KK-SECTION",
+      "stage": "b2",
+      "capability": "translation",
+      "critical": true,
+      "system_prompt": "Translate the supplied academic paragraph into formal Kazakh. Preserve all numbers and the non-causal limitation. Do not translate the citation tag.",
+      "user_prompt": "Perceived personalization was positively associated with learning engagement (beta = 0.35) [CITE-07]. Because the study used a cross-sectional design, the estimate should not be interpreted as evidence of causality. The result is therefore discussed as an association rather than an identified causal effect.",
+      "max_tokens": 400,
+      "gates": [
+        {
+          "type": "contains_all_ci",
+          "values": [
+            "0.35",
+            "[CITE-07]"
+          ]
+        },
+        {
+          "type": "contains_any_ci",
+          "values": [
+            "байланыс",
+            "себеп",
+            "зерттеу"
+          ]
+        },
+        {
+          "type": "script_min_ratio",
+          "script": "cyrillic",
+          "min": 0.55
+        }
+      ]
+    }
+  ],
+  "provider_adapters": {
+    "deepseek": {
+      "adapter_type": "pipeline_deepseek",
+      "api_key_env": "DEEPSEEK_API_KEY"
+    },
+    "gemini": {
+      "adapter_type": "gemini_rest",
+      "api_key_env": "GEMINI_API_KEY",
+      "base_url_env": "GEMINI_API_BASE_URL",
+      "base_url": "https://generativelanguage.googleapis.com/v1beta",
+      "timeout_seconds": 120
+    },
+    "zhipu": {
+      "adapter_type": "zhipu_glm_rest",
+      "api_key_env": "ZHIPU_API_KEY",
+      "base_url_env": "ZHIPU_API_BASE_URL",
+      "base_url": "https://open.bigmodel.cn/api/paas/v4",
+      "timeout_seconds": 120
+    }
+  },
+  "routing_policies": {
+    "free_auto_v1": {
+      "status": "benchmark_design_only",
+      "selection_strategy": "server_health_then_config_priority",
+      "candidate_profiles": [
+        "zhipu_glm47_free_no_thinking",
+        "gemini_free_mechanical"
+      ],
+      "allowed_cost_classes": [
+        "free"
+      ],
+      "max_error_rate": 0.25,
+      "max_latency_ms": 120000,
+      "skip_rate_limited": true,
+      "decision_inputs": [
+        "server_reachability",
+        "recent_provider_health",
+        "rate_limit_state",
+        "latency"
+      ],
+      "client_region_authoritative": false
+    },
+    "standard_resilience_v1": {
+      "status": "benchmark_design_only",
+      "selection_strategy": "server_health_then_config_priority",
+      "candidate_profiles": [
+        "deepseek_flash_economy",
+        "zhipu_glm47_free_thinking"
+      ],
+      "allowed_cost_classes": [
+        "paid",
+        "free"
+      ],
+      "max_error_rate": 0.25,
+      "max_latency_ms": 120000,
+      "skip_rate_limited": true,
+      "decision_inputs": [
+        "server_reachability",
+        "recent_provider_health",
+        "rate_limit_state",
+        "latency"
+      ],
+      "client_region_authoritative": false
+    },
+    "flagship_resilience_v1": {
+      "status": "benchmark_design_only",
+      "selection_strategy": "server_health_then_config_priority",
+      "candidate_profiles": [
+        "deepseek_pro_premium",
+        "deepseek_flash_economy",
+        "zhipu_glm47_free_thinking"
+      ],
+      "allowed_cost_classes": [
+        "paid",
+        "free"
+      ],
+      "max_error_rate": 0.25,
+      "max_latency_ms": 120000,
+      "skip_rate_limited": true,
+      "decision_inputs": [
+        "server_reachability",
+        "recent_provider_health",
+        "rate_limit_state",
+        "latency"
+      ],
+      "client_region_authoritative": false
+    }
+  }
+}

core/model_usage.py

@@ -0,0 +1,206 @@
+# -*- coding: utf-8 -*-
+"""Provider-neutral model usage telemetry for SmartPaper benchmarks.
+
+A3-R5 intentionally keeps this module independent from product entitlement,
+payment, quota, and production routing.  It normalizes provider usage metadata
+and emits append-only benchmark records without retaining prompts or secrets.
+"""
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+import hashlib
+import json
+from pathlib import Path
+import threading
+import uuid
+from typing import Any, Mapping
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _safe_int(value: Any) -> int:
+    try:
+        return max(0, int(value or 0))
+    except (TypeError, ValueError):
+        return 0
+
+
+def _safe_text(value: Any, limit: int = 240) -> str:
+    text = str(value or "").replace("\r", " ").replace("\n", " ").strip()
+    return text[:limit]
+
+
+def sanitize_metadata(value: Any, *, depth: int = 0) -> Any:
+    """Return a bounded, secret-safe JSON-compatible metadata value."""
+    if depth > 4:
+        return "<max-depth>"
+    if value is None or isinstance(value, (bool, int, float)):
+        return value
+    if isinstance(value, str):
+        lowered = value.lower()
+        if any(marker in lowered for marker in ("bearer ", "api_key", "api-key", "x-goog-api-key")):
+            return "<redacted>"
+        return value[:500]
+    if isinstance(value, Mapping):
+        clean = {}
+        for key, item in list(value.items())[:40]:
+            name = str(key)
+            lowered = name.lower()
+            if any(marker in lowered for marker in ("key", "token", "authorization", "secret", "credential", "prompt", "manuscript")):
+                clean[name] = "<redacted>"
+            else:
+                clean[name] = sanitize_metadata(item, depth=depth + 1)
+        return clean
+    if isinstance(value, (list, tuple)):
+        return [sanitize_metadata(item, depth=depth + 1) for item in list(value)[:40]]
+    return _safe_text(value, 500)
+
+
+def prompt_identity(system_prompt: Any = "", user_prompt: Any = "") -> str:
+    """Hash prompt identity without storing the prompt itself."""
+    payload = (str(system_prompt or "") + "\n<SEP>\n" + str(user_prompt or "")).encode("utf-8")
+    return hashlib.sha256(payload).hexdigest()
+
+
+@dataclass
+class ModelUsageEvent:
+    provider: str
+    model: str
+    status: str
+    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
+    created_at: str = field(default_factory=_utc_now_iso)
+    phase_id: str = ""
+    case_id: str = ""
+    benchmark_version: str = ""
+    profile_id: str = ""
+    attempt_index: int = 0
+    streaming: bool = False
+    thinking_mode: str = ""
+    requested_thinking_budget: int | None = None
+    prompt_hash: str = ""
+    request_started_at: str = ""
+    first_token_at: str = ""
+    completed_at: str = ""
+    latency_ms: int = 0
+    prompt_input_tokens: int = 0
+    cache_hit_tokens: int = 0
+    cache_miss_tokens: int = 0
+    output_tokens: int = 0
+    reasoning_or_thought_tokens: int = 0
+    total_tokens: int = 0
+    finish_reason: str = ""
+    http_status: int | None = None
+    error_class: str = ""
+    retry_reason: str = ""
+    escalation_reason: str = ""
+    provider_service_tier: str = ""
+    provider_model_version: str = ""
+    provider_response_id: str = ""
+    price_snapshot_id: str = ""
+    estimated_cost_microunits: int = 0
+    currency: str = ""
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        payload = asdict(self)
+        payload["metadata"] = sanitize_metadata(payload.get("metadata") or {})
+        return payload
+
+
+class NoOpModelUsageSink:
+    def emit(self, event: ModelUsageEvent | Mapping[str, Any]) -> None:
+        return None
+
+
+class JsonlModelUsageSink:
+    """Thread-safe append-only JSONL sink for benchmark artifacts."""
+
+    def __init__(self, path: str | Path):
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._lock = threading.Lock()
+
+    def emit(self, event: ModelUsageEvent | Mapping[str, Any]) -> None:
+        if isinstance(event, ModelUsageEvent):
+            row = event.to_dict()
+        else:
+            row = sanitize_metadata(dict(event or {}))
+        line = json.dumps(row, ensure_ascii=False, sort_keys=True)
+        with self._lock:
+            with self.path.open("a", encoding="utf-8") as handle:
+                handle.write(line + "\n")
+
+
+def emit_usage_event(sink: Any, event: ModelUsageEvent) -> None:
+    """Best-effort telemetry emission that can never fail model generation."""
+    if sink is None:
+        return
+    try:
+        if hasattr(sink, "emit") and callable(sink.emit):
+            sink.emit(event)
+        elif callable(sink):
+            sink(event)
+    except Exception:
+        # Telemetry is strictly observational in A3-R5.
+        return
+
+
+def normalize_deepseek_usage(raw: Mapping[str, Any] | None) -> dict[str, int]:
+    usage = dict(raw or {})
+    details = usage.get("completion_tokens_details")
+    if not isinstance(details, Mapping):
+        details = {}
+    return {
+        "prompt_input_tokens": _safe_int(usage.get("prompt_tokens")),
+        "cache_hit_tokens": _safe_int(usage.get("prompt_cache_hit_tokens")),
+        "cache_miss_tokens": _safe_int(usage.get("prompt_cache_miss_tokens")),
+        "output_tokens": _safe_int(usage.get("completion_tokens")),
+        "reasoning_or_thought_tokens": _safe_int(details.get("reasoning_tokens")),
+        "total_tokens": _safe_int(usage.get("total_tokens")),
+    }
+
+
+def normalize_gemini_usage(raw: Mapping[str, Any] | None) -> dict[str, int | str]:
+    usage = dict(raw or {})
+    return {
+        "prompt_input_tokens": _safe_int(usage.get("promptTokenCount")),
+        "cache_hit_tokens": _safe_int(usage.get("cachedContentTokenCount")),
+        "cache_miss_tokens": max(
+            0,
+            _safe_int(usage.get("promptTokenCount")) - _safe_int(usage.get("cachedContentTokenCount")),
+        ),
+        "output_tokens": _safe_int(usage.get("candidatesTokenCount")),
+        "reasoning_or_thought_tokens": _safe_int(usage.get("thoughtsTokenCount")),
+        "total_tokens": _safe_int(usage.get("totalTokenCount")),
+        "provider_service_tier": _safe_text(usage.get("trafficType") or usage.get("serviceTier"), 80),
+    }
+
+def normalize_zhipu_usage(raw: Mapping[str, Any] | None) -> dict[str, int]:
+    """Normalize Zhipu/GLM Chat Completions usage fields.
+
+    Official GLM responses expose prompt/completion/total token counts and
+    prompt_tokens_details.cached_tokens.  reasoning_tokens is consumed only
+    when the provider actually returns it; it is never inferred from
+    reasoning_content text.
+    """
+    usage = dict(raw or {})
+    prompt_details = usage.get("prompt_tokens_details")
+    if not isinstance(prompt_details, Mapping):
+        prompt_details = {}
+    completion_details = usage.get("completion_tokens_details")
+    if not isinstance(completion_details, Mapping):
+        completion_details = {}
+    prompt_tokens = _safe_int(usage.get("prompt_tokens"))
+    cached_tokens = _safe_int(prompt_details.get("cached_tokens"))
+    return {
+        "prompt_input_tokens": prompt_tokens,
+        "cache_hit_tokens": cached_tokens,
+        "cache_miss_tokens": max(0, prompt_tokens - cached_tokens),
+        "output_tokens": _safe_int(usage.get("completion_tokens")),
+        "reasoning_or_thought_tokens": _safe_int(completion_details.get("reasoning_tokens")),
+        "total_tokens": _safe_int(usage.get("total_tokens")),
+    }
+

core/pipeline_agent.py

@@ -29,6 +29,7 @@ except ImportError:
 from config import APP_CONFIG
 from core.production_guard import find_authorial_ai_meta_spans
 from core.model_gateway import ProviderNeutralModelGateway
+from core.model_usage import ModelUsageEvent, emit_usage_event, normalize_deepseek_usage, prompt_identity, sanitize_metadata
 
 
 class DeepSeekAPIError(RuntimeError):
@@ -148,6 +149,10 @@ class SSciPipelineAgent:
         }
         self._cancel_checker = None
         self._execution_progress_hook = None
+        # A3-R5: observability-only model usage capture. Disabled by default so
+        # production request shape and routing remain unchanged.
+        self._model_usage_sink = None
+        self._model_usage_context = {}
         # S2 is task-local and auxiliary. A configured key is always used
         # first; no username/email is required and the secret is never logged.
         # All S2 HTTP attempts (including retries/future endpoints) share one
@@ -239,6 +244,76 @@ class SSciPipelineAgent:
         """Register a best-effort callback tied to real backend execution."""
         self._execution_progress_hook = hook
 
+    def set_model_usage_sink(self, sink=None, context=None):
+        """Enable best-effort model usage telemetry for benchmark processes.
+
+        A3-R5 keeps this disabled by default.  The sink is observational only:
+        failures in telemetry emission can never fail manuscript generation.
+        """
+        self._model_usage_sink = sink
+        self._model_usage_context = dict(context or {})
+
+    def set_model_usage_context(self, **context):
+        self._model_usage_context = dict(context or {})
+
+    def _model_usage_enabled(self):
+        return getattr(self, "_model_usage_sink", None) is not None
+
+    @staticmethod
+    def _usage_iso(epoch_value):
+        if epoch_value is None:
+            return ""
+        try:
+            return datetime.fromtimestamp(float(epoch_value), tz=timezone.utc).isoformat()
+        except Exception:
+            return ""
+
+    def _emit_deepseek_usage(
+        self, *, usage, model, role, attempt_index, streaming, thinking,
+        reasoning_effort, system_prompt, user_prompt, started_at,
+        completed_at, first_token_at=None, finish_reason="",
+        response_payload=None,
+    ):
+        if not self._model_usage_enabled():
+            return
+        try:
+            normalized = normalize_deepseek_usage(usage if isinstance(usage, dict) else {})
+            context = dict(getattr(self, "_model_usage_context", {}) or {})
+            response_payload = dict(response_payload or {})
+            event = ModelUsageEvent(
+                provider="deepseek",
+                model=str(response_payload.get("model") or model or ""),
+                status="success",
+                phase_id=str(context.get("phase_id") or role or ""),
+                case_id=str(context.get("case_id") or ""),
+                benchmark_version=str(context.get("benchmark_version") or ""),
+                profile_id=str(context.get("profile_id") or ""),
+                attempt_index=max(0, int(attempt_index or 0)),
+                streaming=bool(streaming),
+                thinking_mode="enabled" if thinking else "disabled",
+                prompt_hash=prompt_identity(system_prompt, user_prompt),
+                request_started_at=self._usage_iso(started_at),
+                first_token_at=self._usage_iso(first_token_at),
+                completed_at=self._usage_iso(completed_at),
+                latency_ms=max(0, int((float(completed_at) - float(started_at)) * 1000)),
+                prompt_input_tokens=int(normalized.get("prompt_input_tokens") or 0),
+                cache_hit_tokens=int(normalized.get("cache_hit_tokens") or 0),
+                cache_miss_tokens=int(normalized.get("cache_miss_tokens") or 0),
+                output_tokens=int(normalized.get("output_tokens") or 0),
+                reasoning_or_thought_tokens=int(normalized.get("reasoning_or_thought_tokens") or 0),
+                total_tokens=int(normalized.get("total_tokens") or 0),
+                finish_reason=str(finish_reason or "")[:120],
+                provider_model_version=str(response_payload.get("model") or model or "")[:160],
+                provider_response_id=str(response_payload.get("id") or "")[:160],
+                metadata=sanitize_metadata({
+                    "system_fingerprint": response_payload.get("system_fingerprint"),
+                    "reasoning_effort": reasoning_effort,
+                }),
+            )
+            emit_usage_event(getattr(self, "_model_usage_sink", None), event)
+        except Exception:
+            return
+
     def _touch_execution_progress(self, stage=""):
         hook = self._execution_progress_hook
         if hook is None:
@@ -905,11 +980,27 @@ class SSciPipelineAgent:
                         f"finish_reason={finish_reason!r}"
                     )
 
+                completed_at = time.time()
                 logging.info(
                     "DeepSeek static response | seconds=%.2f | chars=%s",
-                    time.time() - start_time,
+                    completed_at - start_time,
                     len(content),
                 )
+                self._emit_deepseek_usage(
+                    usage=result.get("usage") or {},
+                    model=current_model,
+                    role=resolved_role,
+                    attempt_index=attempt,
+                    streaming=False,
+                    thinking=current_thinking,
+                    reasoning_effort=resolved_effort,
+                    system_prompt=system_prompt,
+                    user_prompt=user_prompt,
+                    started_at=start_time,
+                    completed_at=completed_at,
+                    finish_reason=choices[0].get("finish_reason"),
+                    response_payload=result,
+                )
                 return content
 
             except (
@@ -1009,6 +1100,8 @@ class SSciPipelineAgent:
                 stream=True,
                 reasoning_effort=resolved_effort,
             )
+            if self._model_usage_enabled():
+                payload["stream_options"] = {"include_usage": True}
 
             emitted_content = False
             reasoning_open = False
@@ -1016,6 +1109,9 @@ class SSciPipelineAgent:
             content_chunks = 0
             content_chars = 0
             first_content_at = None
+            stream_usage = None
+            stream_finish_reason = ""
+            stream_response_payload = {}
             start_time = time.time()
             last_model_activity = start_time
 
@@ -1062,10 +1158,20 @@ class SSciPipelineAgent:
                             )
                             break
 
+                        if isinstance(event.get("usage"), dict):
+                            stream_usage = dict(event.get("usage") or {})
+                            stream_response_payload = {
+                                "id": event.get("id"),
+                                "model": event.get("model") or current_model,
+                                "system_fingerprint": event.get("system_fingerprint"),
+                            }
+
                         choices = event.get("choices") or []
                         if not choices:
                             continue
 
+                        if choices[0].get("finish_reason"):
+                            stream_finish_reason = str(choices[0].get("finish_reason") or "")
                         delta = choices[0].get("delta") or {}
                         reasoning_content = (
                             delta.get("reasoning_content") or ""
@@ -1130,14 +1236,31 @@ class SSciPipelineAgent:
                         "DeepSeek stream ended without final content."
                     )
 
+                completed_at = time.time()
                 self._touch_execution_progress("deepseek_stream_complete")
                 logging.info(
                     "DeepSeek stream complete | seconds=%.1f | reasoning_chunks=%s | content_chunks=%s | content_chars=%s",
-                    time.time() - start_time,
+                    completed_at - start_time,
                     reasoning_chunks,
                     content_chunks,
                     content_chars,
                 )
+                self._emit_deepseek_usage(
+                    usage=stream_usage or {},
+                    model=current_model,
+                    role=resolved_role,
+                    attempt_index=attempt,
+                    streaming=True,
+                    thinking=current_thinking,
+                    reasoning_effort=resolved_effort,
+                    system_prompt=system_prompt,
+                    user_prompt=user_prompt,
+                    started_at=start_time,
+                    completed_at=completed_at,
+                    first_token_at=first_content_at,
+                    finish_reason=stream_finish_reason,
+                    response_payload=stream_response_payload,
+                )
                 return
 
             except GeneratorExit:

core/provider_adapters/gemini_rest_adapter.py

@@ -0,0 +1,471 @@
+# -*- coding: utf-8 -*-
+"""Official Gemini REST adapter used only by isolated A3 provider benchmarks.
+
+No Google SDK or local/self-hosted model is required.  This adapter deliberately
+contains no paid-provider fallback and does not alter SmartPaper production
+routing by itself.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import json
+import os
+import time
+from typing import Any, Iterable, Mapping
+
+import requests
+
+from core.model_usage import (
+    ModelUsageEvent,
+    emit_usage_event,
+    normalize_gemini_usage,
+    prompt_identity,
+    sanitize_metadata,
+)
+
+
+class GeminiAPIError(RuntimeError):
+    pass
+
+
+class GeminiAuthenticationError(GeminiAPIError):
+    pass
+
+
+class GeminiRateLimitError(GeminiAPIError):
+    pass
+
+
+class GeminiRequestError(GeminiAPIError):
+    pass
+
+
+class GeminiTimeoutError(GeminiAPIError):
+    pass
+
+
+def _iso_from_epoch(value: float | None) -> str:
+    if value is None:
+        return ""
+    return datetime.fromtimestamp(value, tz=timezone.utc).isoformat()
+
+
+class GeminiRestAdapter:
+    DEFAULT_MODEL = "gemini-2.5-flash"
+    DEFAULT_BASE_URL = "https://generativelanguage.googleapis.com/v1beta"
+
+    def __init__(
+        self,
+        *,
+        api_key: str | None = None,
+        model_name: str | None = None,
+        base_url: str | None = None,
+        timeout_seconds: float = 120.0,
+        usage_sink: Any = None,
+        usage_context: Mapping[str, Any] | None = None,
+        thinking_budgets: Mapping[str, int] | None = None,
+    ):
+        self.api_key = str(api_key or os.environ.get("GEMINI_API_KEY") or "").strip()
+        self.model_name = str(model_name or self.DEFAULT_MODEL).strip()
+        self.base_url = str(base_url or self.DEFAULT_BASE_URL).rstrip("/")
+        self.timeout_seconds = max(1.0, float(timeout_seconds))
+        self.usage_sink = usage_sink
+        self.usage_context = dict(usage_context or {})
+        self.thinking_budgets = {
+            "low": 512,
+            "medium": 1024,
+            "high": 2048,
+            **{str(k).lower(): int(v) for k, v in dict(thinking_budgets or {}).items()},
+        }
+
+    def set_usage_context(self, **context: Any) -> None:
+        self.usage_context = dict(context or {})
+
+    def _headers(self) -> dict[str, str]:
+        if not self.api_key:
+            raise GeminiAuthenticationError(
+                "GEMINI_API_KEY is empty for the selected Gemini provider adapter."
+            )
+        return {
+            "Content-Type": "application/json",
+            "x-goog-api-key": self.api_key,
+        }
+
+    def _endpoint(self, *, stream: bool, model_name: str | None = None) -> str:
+        model = str(model_name or self.model_name).strip() or self.DEFAULT_MODEL
+        operation = "streamGenerateContent" if stream else "generateContent"
+        suffix = "?alt=sse" if stream else ""
+        return f"{self.base_url}/models/{model}:{operation}{suffix}"
+
+    def _thinking_budget(self, thinking: Any, reasoning_effort: Any) -> int:
+        if thinking is False or thinking in (0, "0", "false", "disabled", "off"):
+            return 0
+        if not thinking:
+            return 0
+        if isinstance(reasoning_effort, bool):
+            return self.thinking_budgets["low"] if reasoning_effort else 0
+        try:
+            numeric = int(reasoning_effort)
+        except (TypeError, ValueError):
+            numeric = None
+        if numeric is not None:
+            return max(0, numeric)
+        key = str(reasoning_effort or "low").strip().lower()
+        return max(0, int(self.thinking_budgets.get(key, self.thinking_budgets["low"])))
+
+    def _build_payload(
+        self,
+        system_prompt: Any,
+        user_prompt: Any,
+        *,
+        max_tokens: int,
+        temp: float,
+        thinking: Any,
+        reasoning_effort: Any,
+        response_format: Mapping[str, Any] | None,
+    ) -> tuple[dict[str, Any], int]:
+        thinking_budget = self._thinking_budget(thinking, reasoning_effort)
+        generation_config: dict[str, Any] = {
+            "maxOutputTokens": max(1, int(max_tokens)),
+            "temperature": max(0.0, min(float(temp), 2.0)),
+            "thinkingConfig": {"thinkingBudget": thinking_budget},
+        }
+        if response_format:
+            format_type = str(response_format.get("type") or "").strip().lower()
+            if format_type in {"json_object", "json_schema", "application/json"}:
+                generation_config["responseMimeType"] = "application/json"
+
+        payload = {
+            "systemInstruction": {
+                "parts": [{"text": str(system_prompt or "")}],
+            },
+            "contents": [
+                {
+                    "role": "user",
+                    "parts": [{"text": str(user_prompt or "")}],
+                }
+            ],
+            "generationConfig": generation_config,
+        }
+        return payload, thinking_budget
+
+    @staticmethod
+    def _extract_text(payload: Mapping[str, Any] | None) -> str:
+        pieces: list[str] = []
+        for candidate in list((payload or {}).get("candidates") or []):
+            content = candidate.get("content") if isinstance(candidate, Mapping) else None
+            if not isinstance(content, Mapping):
+                continue
+            for part in list(content.get("parts") or []):
+                if not isinstance(part, Mapping):
+                    continue
+                text = part.get("text")
+                if text is not None:
+                    pieces.append(str(text))
+        return "".join(pieces)
+
+    @staticmethod
+    def _finish_reason(payload: Mapping[str, Any] | None) -> str:
+        candidates = list((payload or {}).get("candidates") or [])
+        if not candidates or not isinstance(candidates[0], Mapping):
+            return ""
+        return str(candidates[0].get("finishReason") or "")[:120]
+
+    @staticmethod
+    def _safe_error_detail(response: Any) -> str:
+        try:
+            payload = response.json()
+        except Exception:
+            payload = None
+        if isinstance(payload, Mapping):
+            error = payload.get("error")
+            if isinstance(error, Mapping):
+                return str(error.get("message") or error.get("status") or "provider error")[:500]
+        return str(getattr(response, "text", "provider error") or "provider error")[:500]
+
+    def _raise_for_status(self, response: Any) -> None:
+        status = int(getattr(response, "status_code", 0) or 0)
+        detail = self._safe_error_detail(response)
+        if status in (401, 403):
+            raise GeminiAuthenticationError(f"Gemini authentication failed ({status}): {detail}")
+        if status == 429:
+            raise GeminiRateLimitError(f"Gemini rate limit reached (429): {detail}")
+        if 400 <= status < 500:
+            raise GeminiRequestError(f"Gemini request rejected ({status}): {detail}")
+        if status >= 500:
+            raise GeminiAPIError(f"Gemini provider error ({status}): {detail}")
+        if status != 200:
+            raise GeminiAPIError(f"Gemini unexpected HTTP status ({status}): {detail}")
+
+    def _event(
+        self,
+        *,
+        model: str,
+        status: str,
+        started_at: float,
+        completed_at: float,
+        first_token_at: float | None,
+        thinking_budget: int,
+        system_prompt: Any,
+        user_prompt: Any,
+        usage: Mapping[str, Any] | None = None,
+        response_payload: Mapping[str, Any] | None = None,
+        streaming: bool = False,
+        http_status: int | None = None,
+        error: BaseException | None = None,
+    ) -> ModelUsageEvent:
+        normalized = normalize_gemini_usage(usage)
+        context = dict(self.usage_context or {})
+        payload = dict(response_payload or {})
+        return ModelUsageEvent(
+            provider="gemini",
+            model=model,
+            status=status,
+            phase_id=str(context.get("phase_id") or context.get("capability") or ""),
+            case_id=str(context.get("case_id") or ""),
+            benchmark_version=str(context.get("benchmark_version") or ""),
+            profile_id=str(context.get("profile_id") or ""),
+            attempt_index=int(context.get("attempt_index") or 0),
+            streaming=bool(streaming),
+            thinking_mode="disabled" if thinking_budget == 0 else "enabled",
+            requested_thinking_budget=thinking_budget,
+            prompt_hash=prompt_identity(system_prompt, user_prompt),
+            request_started_at=_iso_from_epoch(started_at),
+            first_token_at=_iso_from_epoch(first_token_at),
+            completed_at=_iso_from_epoch(completed_at),
+            latency_ms=max(0, int((completed_at - started_at) * 1000)),
+            prompt_input_tokens=int(normalized.get("prompt_input_tokens") or 0),
+            cache_hit_tokens=int(normalized.get("cache_hit_tokens") or 0),
+            cache_miss_tokens=int(normalized.get("cache_miss_tokens") or 0),
+            output_tokens=int(normalized.get("output_tokens") or 0),
+            reasoning_or_thought_tokens=int(normalized.get("reasoning_or_thought_tokens") or 0),
+            total_tokens=int(normalized.get("total_tokens") or 0),
+            finish_reason=self._finish_reason(payload),
+            http_status=http_status,
+            error_class=error.__class__.__name__ if error else "",
+            provider_service_tier=str(normalized.get("provider_service_tier") or ""),
+            provider_model_version=str(payload.get("modelVersion") or "")[:160],
+            provider_response_id=str(payload.get("responseId") or "")[:160],
+            metadata=sanitize_metadata({
+                "provider_model_version": payload.get("modelVersion"),
+                "provider_response_id": payload.get("responseId"),
+            }),
+        )
+
+    def call(
+        self,
+        system_prompt: Any,
+        user_prompt: Any,
+        max_tokens: int = 8000,
+        temp: float = 0.3,
+        model_name: str | None = None,
+        model_role: str = "reasoning",
+        thinking: Any = None,
+        reasoning_effort: Any = None,
+        allow_fallback: Any = None,
+        response_format: Mapping[str, Any] | None = None,
+        **_: Any,
+    ) -> str:
+        del model_role, allow_fallback
+        model = str(model_name or self.model_name).strip() or self.DEFAULT_MODEL
+        payload, thinking_budget = self._build_payload(
+            system_prompt,
+            user_prompt,
+            max_tokens=max_tokens,
+            temp=temp,
+            thinking=thinking,
+            reasoning_effort=reasoning_effort,
+            response_format=response_format,
+        )
+        started_at = time.time()
+        response = None
+        try:
+            response = requests.post(
+                self._endpoint(stream=False, model_name=model),
+                headers=self._headers(),
+                json=payload,
+                timeout=self.timeout_seconds,
+            )
+            self._raise_for_status(response)
+            try:
+                result = response.json()
+            except ValueError as exc:
+                raise GeminiAPIError("Gemini returned a non-JSON success response.") from exc
+            content = self._extract_text(result).strip()
+            if not content:
+                raise GeminiAPIError(
+                    "Gemini returned empty final content. "
+                    f"finish_reason={self._finish_reason(result)!r}"
+                )
+            completed_at = time.time()
+            emit_usage_event(
+                self.usage_sink,
+                self._event(
+                    model=model,
+                    status="success",
+                    started_at=started_at,
+                    completed_at=completed_at,
+                    first_token_at=None,
+                    thinking_budget=thinking_budget,
+                    system_prompt=system_prompt,
+                    user_prompt=user_prompt,
+                    usage=result.get("usageMetadata") if isinstance(result, Mapping) else None,
+                    response_payload=result,
+                    streaming=False,
+                    http_status=int(getattr(response, "status_code", 200) or 200),
+                ),
+            )
+            return content
+        except requests.exceptions.Timeout as exc:
+            error = GeminiTimeoutError("Gemini request timed out.")
+            self._emit_error(model, started_at, thinking_budget, system_prompt, user_prompt, response, error, False)
+            raise error from exc
+        except requests.exceptions.ConnectionError as exc:
+            error = GeminiAPIError("Gemini connection failed.")
+            self._emit_error(model, started_at, thinking_budget, system_prompt, user_prompt, response, error, False)
+            raise error from exc
+        except GeminiAPIError as error:
+            self._emit_error(model, started_at, thinking_budget, system_prompt, user_prompt, response, error, False)
+            raise
+
+    def _emit_error(
+        self,
+        model: str,
+        started_at: float,
+        thinking_budget: int,
+        system_prompt: Any,
+        user_prompt: Any,
+        response: Any,
+        error: BaseException,
+        streaming: bool,
+    ) -> None:
+        completed_at = time.time()
+        emit_usage_event(
+            self.usage_sink,
+            self._event(
+                model=model,
+                status="error",
+                started_at=started_at,
+                completed_at=completed_at,
+                first_token_at=None,
+                thinking_budget=thinking_budget,
+                system_prompt=system_prompt,
+                user_prompt=user_prompt,
+                usage=None,
+                response_payload=None,
+                streaming=streaming,
+                http_status=int(getattr(response, "status_code", 0) or 0) or None,
+                error=error,
+            ),
+        )
+
+    @staticmethod
+    def _parse_sse_line(raw_line: Any) -> Mapping[str, Any] | None:
+        if raw_line is None:
+            return None
+        if isinstance(raw_line, bytes):
+            raw_line = raw_line.decode("utf-8", errors="replace")
+        line = str(raw_line).strip()
+        if not line or line.startswith(":") or not line.startswith("data:"):
+            return None
+        data = line[5:].strip()
+        if not data or data == "[DONE]":
+            return None
+        try:
+            parsed = json.loads(data)
+        except json.JSONDecodeError as exc:
+            raise GeminiAPIError("Gemini streaming returned malformed SSE JSON.") from exc
+        if not isinstance(parsed, Mapping):
+            raise GeminiAPIError("Gemini streaming returned a non-object SSE payload.")
+        return parsed
+
+    def stream(
+        self,
+        system_prompt: Any,
+        user_prompt: Any,
+        max_tokens: int = 8000,
+        model_name: str | None = None,
+        model_role: str = "reasoning",
+        thinking: Any = None,
+        reasoning_effort: Any = None,
+        temp: float = 0.3,
+        allow_fallback: Any = None,
+        reasoning_only_recovery: Any = None,
+        **_: Any,
+    ) -> Iterable[str]:
+        del model_role, allow_fallback, reasoning_only_recovery
+        model = str(model_name or self.model_name).strip() or self.DEFAULT_MODEL
+        payload, thinking_budget = self._build_payload(
+            system_prompt,
+            user_prompt,
+            max_tokens=max_tokens,
+            temp=temp,
+            thinking=thinking,
+            reasoning_effort=reasoning_effort,
+            response_format=None,
+        )
+        started_at = time.time()
+        first_token_at = None
+        response = None
+        latest_usage: Mapping[str, Any] | None = None
+        latest_payload: Mapping[str, Any] | None = None
+        emitted = False
+        try:
+            with requests.post(
+                self._endpoint(stream=True, model_name=model),
+                headers=self._headers(),
+                json=payload,
+                stream=True,
+                timeout=self.timeout_seconds,
+            ) as response:
+                self._raise_for_status(response)
+                for raw_line in response.iter_lines(decode_unicode=True):
+                    event = self._parse_sse_line(raw_line)
+                    if not event:
+                        continue
+                    latest_payload = event
+                    if isinstance(event.get("usageMetadata"), Mapping):
+                        latest_usage = event.get("usageMetadata")
+                    text = self._extract_text(event)
+                    if text:
+                        emitted = True
+                        if first_token_at is None:
+                            first_token_at = time.time()
+                        yield text
+            if not emitted:
+                raise GeminiAPIError("Gemini stream ended without final content.")
+            completed_at = time.time()
+            emit_usage_event(
+                self.usage_sink,
+                self._event(
+                    model=model,
+                    status="success",
+                    started_at=started_at,
+                    completed_at=completed_at,
+                    first_token_at=first_token_at,
+                    thinking_budget=thinking_budget,
+                    system_prompt=system_prompt,
+                    user_prompt=user_prompt,
+                    usage=latest_usage,
+                    response_payload=latest_payload,
+                    streaming=True,
+                    http_status=int(getattr(response, "status_code", 200) or 200),
+                ),
+            )
+        except GeneratorExit:
+            raise
+        except requests.exceptions.Timeout as exc:
+            error = GeminiTimeoutError("Gemini streaming request timed out.")
+            self._emit_error(model, started_at, thinking_budget, system_prompt, user_prompt, response, error, True)
+            raise error from exc
+        except requests.exceptions.ConnectionError as exc:
+            error = GeminiAPIError("Gemini streaming connection failed.")
+            self._emit_error(model, started_at, thinking_budget, system_prompt, user_prompt, response, error, True)
+            raise error from exc
+        except requests.exceptions.ChunkedEncodingError as exc:
+            error = GeminiAPIError("Gemini streaming response ended prematurely.")
+            self._emit_error(model, started_at, thinking_budget, system_prompt, user_prompt, response, error, True)
+            raise error from exc
+        except GeminiAPIError as error:
+            self._emit_error(model, started_at, thinking_budget, system_prompt, user_prompt, response, error, True)
+            raise

core/provider_adapters/glm_rest_adapter.py

@@ -0,0 +1,495 @@
+# -*- coding: utf-8 -*-
+"""Official Zhipu GLM REST adapter for isolated A3 provider benchmarks.
+
+The adapter is remote-API only.  It does not select products, regions, or
+production routes, and it contains no internal fallback to another provider.
+Provider/model/base-URL selection is supplied by benchmark policy/config so a
+future server-authoritative routing layer can make reachability decisions
+without hard-coding geography into manuscript code.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import json
+import time
+from typing import Any, Iterable, Mapping
+
+import requests
+
+from core.model_usage import (
+    ModelUsageEvent,
+    emit_usage_event,
+    normalize_zhipu_usage,
+    prompt_identity,
+    sanitize_metadata,
+)
+
+
+class ZhipuAPIError(RuntimeError):
+    pass
+
+
+class ZhipuAuthenticationError(ZhipuAPIError):
+    pass
+
+
+class ZhipuRateLimitError(ZhipuAPIError):
+    pass
+
+
+class ZhipuRequestError(ZhipuAPIError):
+    pass
+
+
+class ZhipuTimeoutError(ZhipuAPIError):
+    pass
+
+
+def _iso_from_epoch(value: float | None) -> str:
+    if value is None:
+        return ""
+    return datetime.fromtimestamp(value, tz=timezone.utc).isoformat()
+
+
+def _thinking_enabled(value: Any) -> bool:
+    if value is None:
+        return False
+    if isinstance(value, str):
+        return value.strip().lower() not in {"", "0", "false", "disabled", "off", "none"}
+    return bool(value)
+
+
+class ZhipuGLMRestAdapter:
+    """OpenAI-style GLM Chat Completions adapter.
+
+    GLM-4.7-Flash supports thinking enabled/disabled but does not support the
+    GLM-5.2-only ``reasoning_effort`` control.  The common argument is accepted
+    for gateway signature compatibility and intentionally not transmitted.
+    """
+
+    def __init__(
+        self,
+        *,
+        api_key: str | None = None,
+        model_name: str | None = None,
+        base_url: str | None = None,
+        timeout_seconds: float = 120.0,
+        usage_sink: Any = None,
+        usage_context: Mapping[str, Any] | None = None,
+    ):
+        self.api_key = str(api_key or "").strip()
+        self.model_name = str(model_name or "").strip()
+        self.base_url = str(base_url or "").rstrip("/")
+        self.timeout_seconds = max(1.0, float(timeout_seconds))
+        self.usage_sink = usage_sink
+        self.usage_context = dict(usage_context or {})
+
+    def set_usage_context(self, **context: Any) -> None:
+        self.usage_context = dict(context or {})
+
+    def _headers(self) -> dict[str, str]:
+        if not self.api_key:
+            raise ZhipuAuthenticationError(
+                "API key is empty for the selected Zhipu provider adapter."
+            )
+        return {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {self.api_key}",
+        }
+
+    def _endpoint(self) -> str:
+        if not self.base_url:
+            raise ZhipuRequestError("Zhipu base URL is not configured.")
+        return f"{self.base_url}/chat/completions"
+
+    def _model(self, model_name: str | None) -> str:
+        model = str(model_name or self.model_name or "").strip()
+        if not model:
+            raise ZhipuRequestError("Zhipu model is not configured.")
+        return model
+
+    def _build_payload(
+        self,
+        system_prompt: Any,
+        user_prompt: Any,
+        *,
+        model: str,
+        max_tokens: int,
+        temp: float,
+        thinking: Any,
+        response_format: Mapping[str, Any] | None,
+    ) -> tuple[dict[str, Any], bool]:
+        enabled = _thinking_enabled(thinking)
+        payload: dict[str, Any] = {
+            "model": model,
+            "messages": [
+                {"role": "system", "content": str(system_prompt or "")},
+                {"role": "user", "content": str(user_prompt or "")},
+            ],
+            "max_tokens": max(1, int(max_tokens)),
+            "temperature": max(0.0, min(float(temp), 1.0)),
+            "thinking": {"type": "enabled" if enabled else "disabled"},
+        }
+        if response_format:
+            format_type = str(response_format.get("type") or "").strip().lower()
+            if format_type in {"json_object", "application/json"}:
+                payload["response_format"] = {"type": "json_object"}
+        return payload, enabled
+
+    @staticmethod
+    def _extract_static_text(payload: Mapping[str, Any] | None) -> str:
+        choices = list((payload or {}).get("choices") or [])
+        if not choices or not isinstance(choices[0], Mapping):
+            return ""
+        message = choices[0].get("message")
+        if not isinstance(message, Mapping):
+            return ""
+        # Deliberately exclude reasoning_content from reader-facing output.
+        return str(message.get("content") or "")
+
+    @staticmethod
+    def _extract_stream_text(payload: Mapping[str, Any] | None) -> str:
+        choices = list((payload or {}).get("choices") or [])
+        if not choices or not isinstance(choices[0], Mapping):
+            return ""
+        delta = choices[0].get("delta")
+        if not isinstance(delta, Mapping):
+            return ""
+        # Deliberately exclude reasoning_content from reader-facing output.
+        return str(delta.get("content") or "")
+
+    @staticmethod
+    def _finish_reason(payload: Mapping[str, Any] | None) -> str:
+        choices = list((payload or {}).get("choices") or [])
+        if not choices or not isinstance(choices[0], Mapping):
+            return ""
+        return str(choices[0].get("finish_reason") or "")[:120]
+
+    @staticmethod
+    def _safe_error_detail(response: Any) -> str:
+        try:
+            payload = response.json()
+        except Exception:
+            payload = None
+        if isinstance(payload, Mapping):
+            error = payload.get("error")
+            if isinstance(error, Mapping):
+                return str(error.get("message") or error.get("code") or "provider error")[:500]
+            return str(payload.get("message") or payload.get("msg") or "provider error")[:500]
+        return str(getattr(response, "text", "provider error") or "provider error")[:500]
+
+    def _raise_for_status(self, response: Any) -> None:
+        status = int(getattr(response, "status_code", 0) or 0)
+        detail = self._safe_error_detail(response)
+        if status in (401, 403):
+            raise ZhipuAuthenticationError(f"Zhipu authentication failed ({status}): {detail}")
+        if status == 429:
+            raise ZhipuRateLimitError(f"Zhipu rate limit reached (429): {detail}")
+        if 400 <= status < 500:
+            raise ZhipuRequestError(f"Zhipu request rejected ({status}): {detail}")
+        if status >= 500:
+            raise ZhipuAPIError(f"Zhipu provider error ({status}): {detail}")
+        if status != 200:
+            raise ZhipuAPIError(f"Zhipu unexpected HTTP status ({status}): {detail}")
+
+    def _event(
+        self,
+        *,
+        model: str,
+        status: str,
+        started_at: float,
+        completed_at: float,
+        first_token_at: float | None,
+        thinking_enabled: bool,
+        system_prompt: Any,
+        user_prompt: Any,
+        usage: Mapping[str, Any] | None = None,
+        response_payload: Mapping[str, Any] | None = None,
+        streaming: bool = False,
+        http_status: int | None = None,
+        error: BaseException | None = None,
+    ) -> ModelUsageEvent:
+        normalized = normalize_zhipu_usage(usage)
+        context = dict(self.usage_context or {})
+        payload = dict(response_payload or {})
+        return ModelUsageEvent(
+            provider="zhipu",
+            model=model,
+            status=status,
+            phase_id=str(context.get("phase_id") or context.get("capability") or ""),
+            case_id=str(context.get("case_id") or ""),
+            benchmark_version=str(context.get("benchmark_version") or ""),
+            profile_id=str(context.get("profile_id") or ""),
+            attempt_index=int(context.get("attempt_index") or 0),
+            streaming=bool(streaming),
+            thinking_mode="enabled" if thinking_enabled else "disabled",
+            requested_thinking_budget=None,
+            prompt_hash=prompt_identity(system_prompt, user_prompt),
+            request_started_at=_iso_from_epoch(started_at),
+            first_token_at=_iso_from_epoch(first_token_at),
+            completed_at=_iso_from_epoch(completed_at),
+            latency_ms=max(0, int((completed_at - started_at) * 1000)),
+            prompt_input_tokens=int(normalized.get("prompt_input_tokens") or 0),
+            cache_hit_tokens=int(normalized.get("cache_hit_tokens") or 0),
+            cache_miss_tokens=int(normalized.get("cache_miss_tokens") or 0),
+            output_tokens=int(normalized.get("output_tokens") or 0),
+            reasoning_or_thought_tokens=int(normalized.get("reasoning_or_thought_tokens") or 0),
+            total_tokens=int(normalized.get("total_tokens") or 0),
+            finish_reason=self._finish_reason(payload),
+            http_status=http_status,
+            error_class=error.__class__.__name__ if error else "",
+            provider_model_version=str(payload.get("model") or model)[:160],
+            provider_response_id=str(payload.get("id") or payload.get("request_id") or "")[:160],
+            metadata=sanitize_metadata({
+                "provider_model": payload.get("model"),
+                "provider_response_id": payload.get("id") or payload.get("request_id"),
+            }),
+        )
+
+    def _emit_error(
+        self,
+        *,
+        model: str,
+        started_at: float,
+        thinking_enabled: bool,
+        system_prompt: Any,
+        user_prompt: Any,
+        response: Any,
+        error: BaseException,
+        streaming: bool,
+    ) -> None:
+        completed_at = time.time()
+        emit_usage_event(
+            self.usage_sink,
+            self._event(
+                model=model,
+                status="error",
+                started_at=started_at,
+                completed_at=completed_at,
+                first_token_at=None,
+                thinking_enabled=thinking_enabled,
+                system_prompt=system_prompt,
+                user_prompt=user_prompt,
+                usage=None,
+                response_payload=None,
+                streaming=streaming,
+                http_status=int(getattr(response, "status_code", 0) or 0) or None,
+                error=error,
+            ),
+        )
+
+    def call(
+        self,
+        system_prompt: Any,
+        user_prompt: Any,
+        max_tokens: int = 8000,
+        temp: float = 0.3,
+        model_name: str | None = None,
+        model_role: str = "reasoning",
+        thinking: Any = None,
+        reasoning_effort: Any = None,
+        allow_fallback: Any = None,
+        response_format: Mapping[str, Any] | None = None,
+        **_: Any,
+    ) -> str:
+        del model_role, reasoning_effort, allow_fallback
+        model = self._model(model_name)
+        payload, thinking_enabled = self._build_payload(
+            system_prompt,
+            user_prompt,
+            model=model,
+            max_tokens=max_tokens,
+            temp=temp,
+            thinking=thinking,
+            response_format=response_format,
+        )
+        started_at = time.time()
+        response = None
+        try:
+            response = requests.post(
+                self._endpoint(),
+                headers=self._headers(),
+                json=payload,
+                timeout=self.timeout_seconds,
+            )
+            self._raise_for_status(response)
+            try:
+                result = response.json()
+            except ValueError as exc:
+                raise ZhipuAPIError("Zhipu returned a non-JSON success response.") from exc
+            content = self._extract_static_text(result).strip()
+            if not content:
+                raise ZhipuAPIError(
+                    "Zhipu returned empty final content. "
+                    f"finish_reason={self._finish_reason(result)!r}"
+                )
+            completed_at = time.time()
+            emit_usage_event(
+                self.usage_sink,
+                self._event(
+                    model=model,
+                    status="success",
+                    started_at=started_at,
+                    completed_at=completed_at,
+                    first_token_at=None,
+                    thinking_enabled=thinking_enabled,
+                    system_prompt=system_prompt,
+                    user_prompt=user_prompt,
+                    usage=result.get("usage") if isinstance(result, Mapping) else None,
+                    response_payload=result,
+                    streaming=False,
+                    http_status=int(getattr(response, "status_code", 200) or 200),
+                ),
+            )
+            return content
+        except requests.exceptions.Timeout as exc:
+            error = ZhipuTimeoutError("Zhipu request timed out.")
+            self._emit_error(
+                model=model, started_at=started_at, thinking_enabled=thinking_enabled,
+                system_prompt=system_prompt, user_prompt=user_prompt, response=response,
+                error=error, streaming=False,
+            )
+            raise error from exc
+        except requests.exceptions.ConnectionError as exc:
+            error = ZhipuAPIError("Zhipu connection failed.")
+            self._emit_error(
+                model=model, started_at=started_at, thinking_enabled=thinking_enabled,
+                system_prompt=system_prompt, user_prompt=user_prompt, response=response,
+                error=error, streaming=False,
+            )
+            raise error from exc
+        except ZhipuAPIError as error:
+            self._emit_error(
+                model=model, started_at=started_at, thinking_enabled=thinking_enabled,
+                system_prompt=system_prompt, user_prompt=user_prompt, response=response,
+                error=error, streaming=False,
+            )
+            raise
+
+    @staticmethod
+    def _parse_sse_line(raw_line: Any) -> Mapping[str, Any] | None:
+        if raw_line is None:
+            return None
+        if isinstance(raw_line, bytes):
+            raw_line = raw_line.decode("utf-8", errors="replace")
+        line = str(raw_line).strip()
+        if not line or line.startswith(":") or not line.startswith("data:"):
+            return None
+        data = line[5:].strip()
+        if not data or data == "[DONE]":
+            return None
+        try:
+            parsed = json.loads(data)
+        except json.JSONDecodeError as exc:
+            raise ZhipuAPIError("Zhipu streaming returned malformed SSE JSON.") from exc
+        if not isinstance(parsed, Mapping):
+            raise ZhipuAPIError("Zhipu streaming returned a non-object SSE payload.")
+        return parsed
+
+    def stream(
+        self,
+        system_prompt: Any,
+        user_prompt: Any,
+        max_tokens: int = 8000,
+        model_name: str | None = None,
+        model_role: str = "reasoning",
+        thinking: Any = None,
+        reasoning_effort: Any = None,
+        temp: float = 0.3,
+        allow_fallback: Any = None,
+        reasoning_only_recovery: Any = None,
+        **_: Any,
+    ) -> Iterable[str]:
+        del model_role, reasoning_effort, allow_fallback, reasoning_only_recovery
+        model = self._model(model_name)
+        payload, thinking_enabled = self._build_payload(
+            system_prompt,
+            user_prompt,
+            model=model,
+            max_tokens=max_tokens,
+            temp=temp,
+            thinking=thinking,
+            response_format=None,
+        )
+        payload["stream"] = True
+        started_at = time.time()
+        first_token_at = None
+        response = None
+        latest_usage: Mapping[str, Any] | None = None
+        latest_payload: Mapping[str, Any] | None = None
+        emitted = False
+        try:
+            with requests.post(
+                self._endpoint(),
+                headers=self._headers(),
+                json=payload,
+                stream=True,
+                timeout=self.timeout_seconds,
+            ) as response:
+                self._raise_for_status(response)
+                for raw_line in response.iter_lines(decode_unicode=True):
+                    event = self._parse_sse_line(raw_line)
+                    if not event:
+                        continue
+                    latest_payload = event
+                    if isinstance(event.get("usage"), Mapping):
+                        latest_usage = event.get("usage")
+                    text = self._extract_stream_text(event)
+                    if text:
+                        emitted = True
+                        if first_token_at is None:
+                            first_token_at = time.time()
+                        yield text
+            if not emitted:
+                raise ZhipuAPIError("Zhipu stream ended without final content.")
+            completed_at = time.time()
+            emit_usage_event(
+                self.usage_sink,
+                self._event(
+                    model=model,
+                    status="success",
+                    started_at=started_at,
+                    completed_at=completed_at,
+                    first_token_at=first_token_at,
+                    thinking_enabled=thinking_enabled,
+                    system_prompt=system_prompt,
+                    user_prompt=user_prompt,
+                    usage=latest_usage,
+                    response_payload=latest_payload,
+                    streaming=True,
+                    http_status=int(getattr(response, "status_code", 200) or 200),
+                ),
+            )
+        except GeneratorExit:
+            raise
+        except requests.exceptions.Timeout as exc:
+            error = ZhipuTimeoutError("Zhipu streaming request timed out.")
+            self._emit_error(
+                model=model, started_at=started_at, thinking_enabled=thinking_enabled,
+                system_prompt=system_prompt, user_prompt=user_prompt, response=response,
+                error=error, streaming=True,
+            )
+            raise error from exc
+        except requests.exceptions.ConnectionError as exc:
+            error = ZhipuAPIError("Zhipu streaming connection failed.")
+            self._emit_error(
+                model=model, started_at=started_at, thinking_enabled=thinking_enabled,
+                system_prompt=system_prompt, user_prompt=user_prompt, response=response,
+                error=error, streaming=True,
+            )
+            raise error from exc
+        except requests.exceptions.ChunkedEncodingError as exc:
+            error = ZhipuAPIError("Zhipu streaming response ended prematurely.")
+            self._emit_error(
+                model=model, started_at=started_at, thinking_enabled=thinking_enabled,
+                system_prompt=system_prompt, user_prompt=user_prompt, response=response,
+                error=error, streaming=True,
+            )
+            raise error from exc
+        except ZhipuAPIError as error:
+            self._emit_error(
+                model=model, started_at=started_at, thinking_enabled=thinking_enabled,
+                system_prompt=system_prompt, user_prompt=user_prompt, response=response,
+                error=error, streaming=True,
+            )
+            raise