test(rag): bounded real-provider Hybrid runtime smoke

README.md

@@ -251,6 +251,20 @@ API 默认保持 **Dense Chroma 检索**；`/rag/search` 另提供默认关闭
 
 本阶段不调用答案生成、不改变拒答逻辑、不创建工单；完整问答与审批演示仍是后续独立验收范围。
 
+小范围接线验收脚本（默认只展示计划，零模型调用且不读取 `.env`）：
+
+```bash
+python -m scripts.smoke_hybrid_runtime
+# 仅在明确批准本次数据外发和费用后执行；此参数本身不构成授权：
+python -m scripts.smoke_hybrid_runtime --execute --env-file .env
+```
+
+真实模式只发送脚本内固定的四篇虚构说明及两个查询，使用独立临时 SQLite、Chroma 与上传目录，经过 ASGI 路由及 Demo JWT 认证，不连接已运行的服务、不改现有数据库和开关。检查默认 Dense、Hybrid 简单排序、租户/分类隔离、删除生效、空语料零调用及检索审计；不是 HTTP 网络部署验收或准确率评测。
+固定使用北京 `text-embedding-v4`（最多 8 次）及新加坡 `qwen3-rerank`（最多 3 次），禁用 SDK/传输层重试与重定向。发送前校验端点、模型及文本白名单；单次请求体最多 4 KiB、累计 16 KiB。控制台只记录尝试次数、字节、状态、耗时和服务端 usage，不打印密钥或提供方错误正文。任何失败即停止，不自动重新运行。
+预算按 [官方价格](https://www.alibabacloud.com/help/en/model-studio/model-pricing) 保守预留 ¥1，短文本预计不足 ¥0.1；次数/字节限制不是提供方账单硬限额。中断后的请求可能已计费，重新执行须重新确认，不能把脚本重跑当成免费续跑。
+
+2026-10-16 已获授权完成一次真实提供方冒烟：8 次 embedding + 3 次重排全部 HTTP 200，服务端 usage 分别为 126 / 162 tokens，9 项接线检查通过；[逐次结果与边界](docs/reports/hybrid_runtime_smoke_2026-10-16.json)。这是临时 ASGI 环境的小范围验证，不证明 Hybrid 优于 Dense，也不代表已开启现有服务或验收问答、拒答、工单整链。
+
 ---
 
 # 4. RAG 评测与迭代：从 Dense 基线到当前状态

docs/reports/hybrid_runtime_smoke_2026-10-16.json

@@ -0,0 +1,46 @@
+{
+  "date": "2026-10-16",
+  "status": "PASS",
+  "runtime_base_commit": "1557d670c8ce4a5aea8f6b436c0ed0502009e3e9",
+  "script": "scripts/smoke_hybrid_runtime.py",
+  "script_sha256": "392fee1dd5386e044a34aea238cce63b13f4a05cbbb22a32a57fe1b90638351f",
+  "execution_count": 1,
+  "scope": "ASGI TestClient, real provider HTTP, temporary SQLite/Chroma/storage, four fixed synthetic documents",
+  "models": {
+    "embedding": {"name": "text-embedding-v4", "region": "Beijing", "attempts": 8, "total_tokens": 126},
+    "rerank": {"name": "qwen3-rerank", "region": "Singapore", "attempts": 3, "total_tokens": 162}
+  },
+  "total_request_body_bytes": 1925,
+  "retries": 0,
+  "checks": [
+    "upload/index",
+    "JWT tenant scope",
+    "body tenant ignored",
+    "default Dense",
+    "Hybrid VPN ranking",
+    "category",
+    "delete",
+    "empty corpus",
+    "typed audit"
+  ],
+  "provider_attempts": [
+    {"kind": "embedding", "bytes": 155, "status": 200, "total_tokens": 21, "latency_ms": 335},
+    {"kind": "embedding", "bytes": 163, "status": 200, "total_tokens": 21, "latency_ms": 146},
+    {"kind": "embedding", "bytes": 159, "status": 200, "total_tokens": 19, "latency_ms": 136},
+    {"kind": "embedding", "bytes": 154, "status": 200, "total_tokens": 21, "latency_ms": 146},
+    {"kind": "embedding", "bytes": 99, "status": 200, "total_tokens": 12, "latency_ms": 160},
+    {"kind": "embedding", "bytes": 99, "status": 200, "total_tokens": 12, "latency_ms": 151},
+    {"kind": "rerank", "bytes": 356, "status": 200, "total_tokens": 84, "latency_ms": 394},
+    {"kind": "embedding", "bytes": 98, "status": 200, "total_tokens": 8, "latency_ms": 125},
+    {"kind": "rerank", "bytes": 270, "status": 200, "total_tokens": 36, "latency_ms": 357},
+    {"kind": "embedding", "bytes": 99, "status": 200, "total_tokens": 12, "latency_ms": 130},
+    {"kind": "rerank", "bytes": 273, "status": 200, "total_tokens": 42, "latency_ms": 334}
+  ],
+  "limitations": [
+    "Synthetic wiring smoke, not independent retrieval quality or refusal/answer acceptance",
+    "No running-server HTTP, Docker, public deployment or production-load acceptance",
+    "Latency is per provider HTTP attempt, not end-to-end request latency",
+    "Provider usage is not a billing receipt; no actual billed amount is asserted",
+    "Default Hybrid switch, existing data, answer/refusal and ticket paths were not changed"
+  ]
+}

scripts/smoke_hybrid_runtime.py

@@ -0,0 +1,216 @@
+"""Isolated ASGI/SQL/Chroma smoke; default is a zero-call plan, not a live server."""
+
+import argparse
+from contextlib import ExitStack
+import json
+import logging
+import os
+from pathlib import Path
+import secrets
+import sys
+import tempfile
+import time
+from unittest.mock import patch
+from urllib.parse import urlsplit
+from uuid import uuid4
+
+import httpx
+
+
+DOCS = (
+    ("vpn.md", "it", "support", "VPN error 809: reconnect the VPN client and check the network firewall settings."),
+    ("printer.md", "it", "support", "Printer paper jam: switch off the printer and remove the jammed paper from the tray."),
+    ("leave.md", "hr", "support", "Annual leave: submit a leave request in the HR portal and obtain manager approval."),
+    ("foreign.md", "it", "support_other", "FOREIGN_ONLY VPN error 809: this synthetic guide belongs to another tenant."),
+)
+VPN_QUERY = "How do I resolve VPN error 809?"
+HR_QUERY = "How do I request annual leave?"
+QUERIES = {VPN_QUERY, HR_QUERY}
+LIMITS = {"embedding": 8, "rerank": 3}
+MAX_BODY_BYTES = 4096
+MAX_TOTAL_BYTES = 16384
+
+
+class BoundedTransport(httpx.BaseTransport):
+    """Bound actual wire attempts, including failures; never log credentials or bodies."""
+
+    def __init__(self, kind, base_url, records, delegate):
+        self.kind, self.base_url, self.records, self.delegate = kind, base_url, records, delegate
+
+    def handle_request(self, request):
+        expected = self.base_url.rstrip("/") + ("/embeddings" if self.kind == "embedding" else "/reranks")
+        body = json.loads(request.content)
+        if request.method != "POST" or str(request.url) != expected:
+            raise RuntimeError("Unapproved model endpoint")
+        if body.get("model") != ("text-embedding-v4" if self.kind == "embedding" else "qwen3-rerank"):
+            raise RuntimeError("Unapproved model")
+        if self.kind == "embedding":
+            allowed = QUERIES | {f"{Path(name).stem}\n\n{text}" for name, _, _, text in DOCS}
+            if not body.get("input") or not set(body["input"]) <= allowed:
+                raise RuntimeError("Unapproved embedding text")
+        else:
+            allowed = {text for _, _, user, text in DOCS if user == "support"}
+            if body.get("query") not in QUERIES or not body.get("documents") or not set(body["documents"]) <= allowed:
+                raise RuntimeError("Unapproved rerank text")
+        size = len(request.content)
+        if (sum(r["kind"] == self.kind for r in self.records) >= LIMITS[self.kind]
+                or size > MAX_BODY_BYTES or sum(r["bytes"] for r in self.records) + size > MAX_TOTAL_BYTES):
+            raise RuntimeError("Smoke call/size budget exhausted")
+        row = {"kind": self.kind, "bytes": size, "status": None, "total_tokens": None}
+        self.records.append(row)  # Count an attempted send even if its response is lost.
+        started = time.perf_counter()
+        try:
+            response = self.delegate.handle_request(request)
+            response.read()
+            row["status"] = response.status_code
+            try:
+                row["total_tokens"] = response.json().get("usage", {}).get("total_tokens")
+            except (ValueError, AttributeError):
+                pass
+            return response
+        finally:
+            row["latency_ms"] = round((time.perf_counter() - started) * 1000)
+
+    def close(self):
+        self.delegate.close()
+
+
+def run_flow(client):
+    def request(method, path, expected=200, **kwargs):
+        response = client.request(method, path, **kwargs)
+        if response.status_code != expected:
+            raise RuntimeError(f"{method} {path}: HTTP {response.status_code}")
+        return response.json()
+
+    headers = {}
+    for username in ("support", "support_other"):
+        token = request("POST", "/auth/token", json={"username": username, "password": "support"})["access_token"]
+        headers[username] = {"Authorization": f"Bearer {token}"}
+    ids = {}
+    for name, category, user, text in DOCS:
+        doc = request("POST", "/documents/upload", 201, headers=headers[user],
+                      data={"category": category}, files={"file": (name, text.encode(), "text/markdown")})
+        ids[name] = doc["id"]
+        indexed = request("POST", f"/documents/{doc['id']}/index", headers=headers[user])
+        if indexed["status"] != "indexed" or indexed["chunk_count"] != 1:
+            raise RuntimeError("Synthetic document did not index as one chunk")
+    own = headers["support"]
+    if client.get(f"/documents/{ids['foreign.md']}", headers=own).status_code != 404:
+        raise RuntimeError("Cross-tenant document access allowed")
+
+    def search(query, category, mode=None):
+        body = {"query": query, "category": category, "tenant_id": "tenant_other"}
+        if mode:
+            body["retrieval_mode"] = mode
+        data = request("POST", "/rag/search", headers=own, json=body)
+        if data["retrieval_mode"] != (mode or "dense"):
+            raise RuntimeError("Wrong retrieval mode")
+        if any(r["tenant_id"] != "tenant_demo" or r["category"] != category for r in data["results"]):
+            raise RuntimeError("Search scope violation")
+        return data
+
+    dense = search(VPN_QUERY, "it")
+    hybrid = search(VPN_QUERY, "it", "hybrid")
+    if dense["top_k"] != 3 or {r["document_id"] for r in dense["results"]} != {ids['vpn.md'], ids['printer.md']}:
+        raise RuntimeError("Default Dense regression")
+    if hybrid["top_k"] != 14 or {r["document_id"] for r in hybrid["results"]} != {ids['vpn.md'], ids['printer.md']}:
+        raise RuntimeError("Hybrid candidate scope mismatch")
+    if hybrid["results"][0]["document_id"] != ids["vpn.md"]:
+        raise RuntimeError("VPN synthetic relevance check failed")
+    hr = search(HR_QUERY, "hr", "hybrid")
+    if [r["document_id"] for r in hr["results"]] != [ids["leave.md"]]:
+        raise RuntimeError("Category filtering failed")
+    request("DELETE", f"/documents/{ids['vpn.md']}", headers=own)
+    after_delete = search(VPN_QUERY, "it", "hybrid")
+    if [r["document_id"] for r in after_delete["results"]] != [ids["printer.md"]]:
+        raise RuntimeError("Deleted document still searchable")
+    request("DELETE", f"/documents/{ids['printer.md']}", headers=own)
+    if search(VPN_QUERY, "it", "hybrid")["results"]:
+        raise RuntimeError("Empty tenant/category scope was not empty")
+    logs = request("GET", "/agent-ops/retrieval-logs", headers=own)
+    hybrid_logs = [r for r in logs if isinstance(json.loads(r["scores_json"]), dict)]
+    if len(logs) != 5 or len(hybrid_logs) != 4 or any(r["top_distance"] is not None for r in hybrid_logs):
+        raise RuntimeError("Missing or incorrectly typed retrieval audit")
+    return {"status": "PASS", "checks": ["upload/index", "JWT tenant scope", "body tenant ignored",
+            "default Dense", "Hybrid VPN ranking", "category", "delete", "empty corpus", "typed audit"]}
+
+
+def execute(env_file):
+    from dotenv import dotenv_values
+    from openai import OpenAI
+    import chromadb
+    from fastapi.testclient import TestClient
+
+    if "database" in sys.modules:
+        raise RuntimeError("Execute in a fresh process to isolate the database")
+    config = dotenv_values(env_file)
+    embed_url = config.get("DASHSCOPE_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1")
+    rerank_url = config.get("DASHSCOPE_RERANK_BASE_URL", "")
+    rerank_host = urlsplit(rerank_url).hostname or ""
+    if (embed_url.rstrip("/") != "https://dashscope.aliyuncs.com/compatible-mode/v1"
+            or not rerank_host.startswith("ws-") or not rerank_host.endswith(".ap-southeast-1.maas.aliyuncs.com")
+            or urlsplit(rerank_url).scheme != "https" or urlsplit(rerank_url).path.rstrip("/") != "/compatible-api/v1"
+            or not config.get("DASHSCOPE_API_KEY") or not config.get("DASHSCOPE_RERANK_API_KEY")):
+        raise RuntimeError("Missing credentials or endpoints outside the approved Beijing/Singapore scope")
+    records = []
+    with tempfile.TemporaryDirectory(prefix="hybrid_smoke_") as temp, ExitStack() as stack:
+        os.environ.update(DATABASE_URL="sqlite:///" + str(Path(temp) / "smoke.db"), SQL_ECHO="false",
+            DOCUMENT_STORAGE_ROOT=str(Path(temp) / "documents"), AUTH_DEMO_USERS_JSON="",
+            JWT_SECRET=secrets.token_hex(32), RAG_HYBRID_SEARCH_ENABLED="true", ANONYMIZED_TELEMETRY="False",
+            EMBEDDING_MODEL="text-embedding-v4", PYTHON_DOTENV_DISABLED="1",
+            DASHSCOPE_API_KEY=config["DASHSCOPE_API_KEY"], DASHSCOPE_BASE_URL=embed_url)
+        import database
+        import main
+        import rag_runtime.build_rag_index as embeddings
+        import rag_runtime.hybrid_search as hybrid
+        import rag_runtime.query_chroma as dense
+        import services.document_service as documents
+        database.create_db_and_tables()
+        stack.callback(database.engine.dispose)
+        stack.callback(main.llm_client.close)
+        for name in ("httpx", "httpx2", "openai", "bm25s"):
+            logging.getLogger(name).setLevel(logging.WARNING)
+        chroma = chromadb.EphemeralClient(settings=chromadb.Settings(anonymized_telemetry=False))
+        collection = chroma.create_collection("smoke_" + uuid4().hex, embedding_function=None)
+        stack.callback(chroma.delete_collection, collection.name)
+
+        def model_client(kind, url, key):
+            transport = BoundedTransport(kind, url, records, httpx.HTTPTransport(retries=0))
+            return OpenAI(api_key=key, base_url=url, max_retries=0, timeout=30,
+                http_client=httpx.Client(transport=transport, trust_env=False, follow_redirects=False))
+
+        embedding_client = stack.enter_context(model_client("embedding", embed_url, config["DASHSCOPE_API_KEY"]))
+        stack.enter_context(patch.object(embeddings, "get_client", lambda: embedding_client))
+        stack.enter_context(patch.object(hybrid, "get_rerank_client", lambda: model_client(
+            "rerank", rerank_url, config["DASHSCOPE_RERANK_API_KEY"])))
+        for module, attr in ((dense, "get_collection"), (hybrid, "get_collection"), (documents, "get_document_chroma_collection")):
+            stack.enter_context(patch.object(module, attr, lambda **_: collection))
+        try:
+            with TestClient(main.app) as client:
+                result = run_flow(client)
+            result["calls"] = records
+            return result
+        finally:
+            print(json.dumps({"provider_attempts": records}, ensure_ascii=False))
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--execute", action="store_true", help="Real paid calls; requires prior human authorization, not granted by this flag")
+    parser.add_argument("--env-file", type=Path)
+    args = parser.parse_args()
+    if not args.execute:
+        print(json.dumps({"mode": "PLAN_ONLY", "limits": LIMITS, "max_total_request_bytes": MAX_TOTAL_BYTES,
+                          "documents": DOCS, "queries": sorted(QUERIES)}, ensure_ascii=False, indent=2))
+    elif args.env_file is None:
+        parser.error("--execute requires --env-file")
+    else:
+        try:
+            print(json.dumps(execute(args.env_file), ensure_ascii=False, indent=2))
+        except Exception as exc:
+            print(f"Smoke failed ({type(exc).__name__}); no automatic rerun. Provider bodies are not printed.", file=sys.stderr)
+            raise SystemExit(1) from None
+
+
+if __name__ == "__main__":
+    main()