feat(api): LLM 配置三端点+测试连接探针+draft/decompose 切三层解析——GET 打码视图带逐字段来源/PUT 掩码串跳过空串清键/POST test 单发探测(超时钳 10s 无重试);corpus_funnel 仍两层 env(data 域另提醒) (spec §13 P5.1) [vps] [no-doc]

sanguo_api/llm/probe.py

@@ -0,0 +1,40 @@
+# sanguo_api/llm/probe.py
+"""测试连接探针(P5.1,spec §13):单发最小 chat,区别于 client 三次重试.
+
+探针超时独立钳制(≤10s)——设置页点「测试连接」不该挂着等 60s。
+"""
+from __future__ import annotations
+
+import time
+
+import httpx
+
+from .provider import LLMConfig, chat_url
+
+_PROBE_TIMEOUT_CAP = 10.0
+
+
+async def probe_connection(config: LLMConfig,
+                           transport: httpx.AsyncBaseTransport | None = None
+                           ) -> dict:
+    """返回 {ok, latency_ms, error};任何失败都不 raise(前端直显)."""
+    start = time.perf_counter()
+    try:
+        async with httpx.AsyncClient(
+                timeout=min(config.timeout, _PROBE_TIMEOUT_CAP),
+                transport=transport) as client:
+            resp = await client.post(
+                chat_url(config.base_url),
+                headers={"Authorization": f"Bearer {config.api_key}"},
+                json={"model": config.model, "max_tokens": 8,
+                      "temperature": 0,
+                      "messages": [{"role": "user", "content": "ping"}]})
+        latency = round((time.perf_counter() - start) * 1000)
+        if resp.status_code == 200:
+            return {"ok": True, "latency_ms": latency, "error": ""}
+        return {"ok": False, "latency_ms": latency,
+                "error": f"HTTP {resp.status_code}: {resp.text[:200]}"}
+    except httpx.TransportError as e:
+        return {"ok": False,
+                "latency_ms": round((time.perf_counter() - start) * 1000),
+                "error": f"连接失败: {e}"}

sanguo_api/routes_pipeline.py

@@ -11,6 +11,7 @@ import hmac
 import json
 import logging
 import os
+import re
 import threading
 from dataclasses import asdict
 from datetime import date, datetime
@@ -24,7 +25,8 @@ from sanguo_api.hypothesis_card import (
     normalize_draft as _normalize_draft,
 )
 from sanguo_api.hypothesis_decompose import run_decompose as _run_decompose
-from sanguo_api.llm import LLMClient, LLMConfigError, LLMError, resolve_config
+from sanguo_api.llm import LLMClient, LLMConfigError, LLMError, config_store
+from sanguo_api.llm.probe import probe_connection
 from sanguo_api.routes import verify_token
 from sanguo_portfolio import pipeline_store
 
@@ -828,7 +830,7 @@ async def hypotheses_draft(body: dict) -> dict:
     if len(sentence) > _DRAFT_SENTENCE_MAX:
         raise HTTPException(400, f"sentence 超长(>{_DRAFT_SENTENCE_MAX} 字)")
     try:
-        config = resolve_config()
+        config = config_store.resolve_effective_config(_pipeline_db())
     except LLMConfigError as e:
         raise HTTPException(503, str(e))
     domains = _wizard_domains()
@@ -895,7 +897,7 @@ async def hypotheses_decompose(hyp_id: str) -> dict:
     if card["state"] not in _DECOMPOSE_ALLOWED_STATES:
         raise HTTPException(409, f"卡片状态 {card['state']} 不可分解")
     try:
-        config = resolve_config()
+        config = config_store.resolve_effective_config(_pipeline_db())
     except LLMConfigError as e:
         raise HTTPException(503, str(e))
 
@@ -950,3 +952,38 @@ async def hypotheses_decompose(hyp_id: str) -> dict:
     item = next(i for i in items if i["id"] == hyp_id)
     return {"registered": registered, "failed": result["failed"],
             "rounds": result["rounds"], "item": item}
+
+
+# —— P5.1 LLM 配置面(spec §13):三层解析的管理端点 ——
+
+
+@router.get("/pipeline/llm/config")
+def llm_config_get() -> dict:
+    """设置页 LLM 卡:打码视图+逐字段来源(db/env/default)+审计行."""
+    return {**config_store.get_masked_config(_pipeline_db()),
+            "audit": config_store.get_llm_audit(_pipeline_db())}
+
+
+@router.put("/pipeline/llm/config")
+def llm_config_put(body: dict) -> dict:
+    """写 DB 覆盖层;含 '*' 的 api_key(掩码回显)静默跳过;空串=清键回落."""
+    values = body.get("values")
+    changed_by = body.get("changed_by") or "console"
+    if not isinstance(values, dict):
+        raise HTTPException(422, "body 需要 {values: {...}, changed_by}")
+    try:
+        changed = config_store.set_llm_config(
+            _pipeline_db(), values, changed_by)
+    except ValueError as e:
+        raise HTTPException(422, str(e)) from e
+    return {"ok": True, "changed": changed}
+
+
+@router.post("/pipeline/llm/test")
+async def llm_test() -> dict:
+    """测试连接:按当前生效配置(三层解析后)单发最小探测,返回 ok/latency/error."""
+    try:
+        config = config_store.resolve_effective_config(_pipeline_db())
+    except LLMConfigError as e:
+        raise HTTPException(503, str(e))
+    return await probe_connection(config)

tests/api/test_llm_probe.py

@@ -0,0 +1,35 @@
+# tests/api/test_llm_probe.py
+"""probe(P5.1):单发最小 chat 探测,无重试无退避;MockTransport 零真调用."""
+import asyncio
+
+import httpx
+
+from sanguo_api.llm.probe import probe_connection
+from sanguo_api.llm.provider import LLMConfig
+
+CFG = LLMConfig("glm_cn", "http://x/v4/", "m", "sk-test", timeout=5.0)
+
+
+def _run(transport):
+    return asyncio.run(probe_connection(CFG, transport=transport))
+
+
+def test_ok_with_latency():
+    t = httpx.MockTransport(
+        lambda req: httpx.Response(200, json={"choices": []}))
+    r = _run(t)
+    assert r["ok"] is True and r["error"] == ""
+    assert 0 <= r["latency_ms"] < 5000
+
+
+def test_http_error_carries_status():
+    t = httpx.MockTransport(lambda req: httpx.Response(401, text="bad key"))
+    r = _run(t)
+    assert r["ok"] is False and "401" in r["error"]
+
+
+def test_transport_error():
+    def boom(req):
+        raise httpx.ConnectError("refused")
+    r = _run(httpx.MockTransport(boom))
+    assert r["ok"] is False and "refused" in r["error"]

tests/api/test_routes_pipeline_settings.py

@@ -0,0 +1,119 @@
+# tests/api/test_routes_pipeline_settings.py
+"""设置页端点(P5.1/P5.2/P5.6):llm config GET/PUT/test + env-status
++ monthly-status。全 mock LLM/探针,零真调用;fixture 逐字对齐
+test_routes_pipeline_hypotheses.py 既有范式。"""
+import os
+
+import pytest
+from fastapi.testclient import TestClient
+
+from sanguo_api.app import create_app
+from sanguo_api.auth import create_token, set_jwt_config
+
+REAL_KEY = "sk-test-1234567890abcd"
+
+
+@pytest.fixture()
+def client(tmp_path, monkeypatch):
+    monkeypatch.setenv("SANGUO_PIPELINE_DB", str(tmp_path / "pipeline.db"))
+    monkeypatch.setenv("SANGUO_FACTOR_MONTHLY_DIR",
+                       str(tmp_path / "factor_monthly"))
+    for k in ("SANGUO_LLM_API_KEY", "SANGUO_LLM_BASE_URL",
+              "SANGUO_LLM_MODEL", "SANGUO_LLM_PROVIDER",
+              "SANGUO_PROMOTION_PASSPHRASE", "SANGUO_GITEA_TOKEN"):
+        monkeypatch.delenv(k, raising=False)
+    set_jwt_config(secret="test", expire_minutes=60)
+    app = create_app(db_path=str(tmp_path / "t.db"), file_dir=None)
+    c = TestClient(app)
+    c.headers.update({"Authorization": f"Bearer {create_token('admin')}"})
+    return c
+
+
+class TestLlmConfigEndpoints:
+    def test_get_initially_env_layer_unresolved(self, client):
+        r = client.get("/api/v1/pipeline/llm/config")
+        assert r.status_code == 200
+        body = r.json()
+        assert body["resolved"] is False and body["missing"]
+        assert body["config"]["api_key"] == ""   # 未配置任一层=空串非打码
+        assert REAL_KEY not in str(body)
+
+    def test_put_then_get_masked_and_resolved(self, client):
+        r = client.put("/api/v1/pipeline/llm/config", json={
+            "values": {"api_key": REAL_KEY, "model": "db-model"},
+            "changed_by": "admin"})
+        assert r.status_code == 200 and r.json()["ok"] is True
+        body = client.get("/api/v1/pipeline/llm/config").json()
+        assert body["config"]["api_key"] == "sk-t****abcd"
+        assert body["sources"]["api_key"] == "db"
+        assert body["resolved"] is True
+        assert body["audit"][0]["key"] == "api_key"   # 倒序:后改的在前
+
+    def test_put_masked_apikey_skipped(self, client):
+        client.put("/api/v1/pipeline/llm/config", json={
+            "values": {"api_key": REAL_KEY}, "changed_by": "a"})
+        r = client.put("/api/v1/pipeline/llm/config", json={
+            "values": {"api_key": "sk-t****abcd"}, "changed_by": "a"})
+        body = r.json()
+        assert body["changed"] == []
+        view = client.get("/api/v1/pipeline/llm/config").json()
+        assert view["config"]["api_key"] == "sk-t****abcd"
+
+    def test_put_bad_values_422(self, client):
+        for bad in ({"provider": "nope"}, {"timeout": "abc"},
+                    {"timeout": "900"}, {"evil": "x"}):
+            r = client.put("/api/v1/pipeline/llm/config", json={
+                "values": bad, "changed_by": "a"})
+            assert r.status_code == 422, bad
+        r = client.put("/api/v1/pipeline/llm/config", json={"values": "x"})
+        assert r.status_code == 422
+
+    def test_env_fallback_visible_in_sources(self, client, monkeypatch):
+        monkeypatch.setenv("SANGUO_LLM_API_KEY", "sk-env-12345678")
+        body = client.get("/api/v1/pipeline/llm/config").json()
+        assert body["sources"]["api_key"] == "env"
+        assert body["config"]["api_key"] == "sk-e****5678"
+        assert body["resolved"] is True
+
+
+class TestLlmTest:
+    def test_unconfigured_503(self, client):
+        assert client.post("/api/v1/pipeline/llm/test").status_code == 503
+
+    def test_probe_ok_and_latency(self, client, monkeypatch):
+        import sanguo_api.routes_pipeline as rp
+        client.put("/api/v1/pipeline/llm/config", json={
+            "values": {"api_key": REAL_KEY}, "changed_by": "a"})
+
+        async def fake_probe(cfg, transport=None):
+            return {"ok": True, "latency_ms": 42, "error": ""}
+        monkeypatch.setattr(rp, "probe_connection", fake_probe)
+        r = client.post("/api/v1/pipeline/llm/test")
+        assert r.status_code == 200
+        assert r.json() == {"ok": True, "latency_ms": 42, "error": ""}
+
+
+class TestDraftUsesDbLayer:
+    """draft 换三层入口后,DB 配置在 env 全空时也能出草稿(P5.1 主收益)."""
+
+    def test_draft_resolves_from_db(self, client, monkeypatch):
+        client.put("/api/v1/pipeline/llm/config", json={
+            "values": {"api_key": REAL_KEY}, "changed_by": "a"})
+        from sanguo_api import hypothesis_card as hc
+        monkeypatch.setattr(hc, "load_domains",
+                            lambda: ["corpus_sentiment", "bars_daily"])
+        import sanguo_api.routes_pipeline as rp
+        monkeypatch.setattr(rp, "_wizard_domains", hc.load_domains)
+
+        class Fake:
+            def __init__(self, *a, **k):
+                pass
+
+            async def chat_json(self, messages, **k):
+                return {"title": "t", "logic": "l", "expected_sign":
+                        "positive", "falsifiable": "f",
+                        "data_needs": ["corpus_sentiment"]}
+        monkeypatch.setattr(rp, "LLMClient", Fake)
+        r = client.post("/api/v1/pipeline/hypotheses/draft",
+                        json={"sentence": "x"})
+        assert r.status_code == 200 and r.json()["draft"]["title"] == "t"