feat(scheduler): 会话粘性——对话进行中不切换凭证

OpenAI 协议无会话概念，但同对话下一轮的 messages 必然以本轮完整
messages 为前缀。据此用增量前缀指纹（掺入用户名防跨用户串号）做
粘性匹配：命中则固定复用原凭证，不再按到期积分/健康度重排，避免
对话中途换号触发上游风控、丢掉上游提示词缓存。

- 粘住的凭证出错仍走正常轮换（≤3 次），成功后重新粘到实际服务的凭证
- 凭证被删/冷却/禁用/属其他上游时自动回退常规调度，不会永久失粘
- CONVERSATION_STICKY_SECONDS（默认 3600s）控制 TTL，≤0 关闭
- 纯内存实现，重启丢粘性只影响一轮选号

Author: RobbsLuo

README.en.md

@@ -13,6 +13,7 @@ upstream channels, with a shared credential pool, unified scheduling, and per-us
 - **OpenAI-compatible surface**: `/v1/chat/completions` (streaming and non-streaming), `/v1/models`, `/v1/user/balance` (DeepSeek-compatible balance query)
 - **Two upstreams, one model namespace**: flat model names auto-route — credits expiring within the window first, then health; `model@provider` pins an upstream
 - **Expiry-aware scheduling**: sums the credits expiring within `QUOTA_EXPIRY_WINDOW_SECONDS` (default 36h) per credential and burns the largest first, so near-expiry quota is not wasted
+- **Conversation stickiness**: multi-turn conversations keep the same credential (matched by message-prefix fingerprint) and only rotate on errors, avoiding upstream risk control and lost prompt cache
 - **Three-state health + tiered cooldowns**: quota exhausted 12h, rate limited 60s, consecutive errors 10m, dead session disabled
 - **Shared credential pool**: admins maintain credentials, everyone shares them; usage is tracked per user
 - **Encrypted credentials at rest**: Fernet (AES-128-CBC + HMAC), key from `APP_SECRET`

README.md

@@ -9,7 +9,7 @@
 ## 特性
 
 - **OpenAI 兼容出口**：`/v1/chat/completions`（流式 + 非流式）、`/v1/models`、`/v1/user/balance`（DeepSeek 兼容余额查询）
-- **统一调度**：扁平模型名按健康度自动选号，`模型@渠道` 强制指定；三态健康度 + 分级冷却自动避开坏号；积分 36h 内即将到期多者优先（先用掉，避免过期浪费；管理台凭证列表直接显示每个账号的到期积分）
+- **统一调度**：扁平模型名按健康度自动选号，`模型@渠道` 强制指定；三态健康度 + 分级冷却自动避开坏号；积分 36h 内即将到期多者优先（先用掉，避免过期浪费；管理台凭证列表直接显示每个账号的到期积分）；同一对话多轮请求粘住同一凭证（对话进行中不换号，出错才轮换）
 - **公共凭证池**：admin 集中维护、全员共享；按人统计用量
 - **完整凭证运维**：设备码登录、多账号切换、额度探测、每日签到、token 预刷新；凭证加密入库（`APP_SECRET`）
 - **脱敏统计**：不存对话内容；明细 90 天、小时汇总永久；按人/渠道/模型可视化
@@ -103,6 +103,7 @@ curl http://127.0.0.1:8000/v1/user/balance -H "Authorization: Bearer sk-你的ke
 | `DATA_DIR` | `./data` | SQLite 与运行数据目录 |
 | `QUOTA_PROBE_MINUTES` | `60` | 额度探测周期 |
 | `QUOTA_EXPIRY_WINDOW_SECONDS` | `129600` | 到期排序窗口：把距到期 ≤ 该秒数的积分加总，多的账号先用（避免积分过期浪费）；`≤0` 关闭，退回纯健康度排序 |
+| `CONVERSATION_STICKY_SECONDS` | `3600` | 会话粘性 TTL：同一对话（消息前缀延续）多轮请求固定用同一凭证；凭证出错仍会轮换，成功后重新粘定；`≤0` 关闭 |
 | `MODEL_BLOCKLIST` | `custom_model_*,*sub*agent*,summary,browser_use_*` | 模型列表黑名单（仅影响列表展示） |
 | `ALLOWED_HOSTS` | 空 | Host 白名单，防 DNS rebinding |
 

docker-compose.yml

@@ -18,6 +18,8 @@ services:
       QUOTA_PROBE_MINUTES: ${QUOTA_PROBE_MINUTES:-60}
       # 到期排序窗口（秒）：把距到期 ≤ 该值的积分加总，多的账号先用；≤0 关闭
       QUOTA_EXPIRY_WINDOW_SECONDS: ${QUOTA_EXPIRY_WINDOW_SECONDS:-129600}
+      # 会话粘性 TTL（秒）：同一对话多轮请求固定用同一凭证；≤0 关闭
+      CONVERSATION_STICKY_SECONDS: ${CONVERSATION_STICKY_SECONDS:-3600}
       LOG_LEVEL: ${LOG_LEVEL:-INFO}
     ports:
       - "${PORT:-8000}:8000"

src/config.py

@@ -39,6 +39,10 @@ class Settings(BaseSettings):
     # 到期排序窗口：把「距到期 ≤ 该秒数」的积分加总，作为选号第一排序指标（多者先用）；
     # CodeBuddy 是每日 100 积分 × N 的小包；≤0 关闭该指标，退回纯健康度排序
     quota_expiry_window_seconds: int = 36 * 3600
+    # 会话粘性 TTL（秒）：同一对话（消息前缀延续）的多轮请求固定用同一凭证，
+    # 对话进行中不换号（避免上游风控与丢失上游提示词缓存）；凭证出错仍会
+    # 正常轮换，成功后重新粘定。≤0 关闭粘性
+    conversation_sticky_seconds: int = 3600
 
     # 后台任务
     quota_probe_minutes: int = 60

src/engine/affinity.py

@@ -0,0 +1,89 @@
+"""会话粘性：同一对话的多轮请求固定使用同一凭证。
+
+OpenAI Chat Completions 本身无会话概念，但客户端（agent/CLI）的「对话」
+表现为消息数组的前缀延续：下一轮的 messages 以上一轮的完整 messages
+为前缀再追加。据此以增量前缀指纹做匹配，命中则请求固定复用原凭证，
+不再按到期积分/健康度重排——对话进行中换号会触发上游风控并丢掉
+上游侧的提示词缓存。
+
+指纹链里掺入用户名，防止不同 API 用户的相同消息数组串到同一凭证。
+TTL 内没有后续轮次即视为对话结束，条目过期；条目数有上限，超量先淘汰
+过期条目、再按最旧淘汰。纯内存即可：重启后丢粘性只影响一轮选号。
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import time
+
+
+def _message_digest(message: object) -> bytes:
+    """单条消息的规范哈希：键序归一，杜绝 JSON 键序差异造成假失配。"""
+    text = json.dumps(message, ensure_ascii=False, sort_keys=True,
+                      separators=(",", ":"), default=str)
+    return hashlib.sha256(text.encode("utf-8")).digest()
+
+
+class ConversationAffinity:
+    """前缀指纹 → 凭证的粘性表（单事件循环内使用，无需加锁）。"""
+
+    def __init__(self, *, ttl_seconds: int, max_entries: int = 512) -> None:
+        self.ttl_seconds = ttl_seconds
+        self.max_entries = max_entries
+        # fingerprint_hex → (credential_id, expires_at)；dict 保持插入序，供淘汰
+        self._entries: dict[str, tuple[str, float]] = {}
+
+    def pin_for(self, messages: list[dict], username: str, *,
+                now: float | None = None) -> str | None:
+        """最长前缀匹配的粘性凭证；无匹配（含禁用/过期）返回 None。
+
+        逐条累积指纹并在每一步查表：上一轮请求存的指纹正是本轮某个前缀
+        的指纹，最长匹配者即本对话最近一轮实际使用的凭证。
+        """
+        if self.ttl_seconds <= 0 or not messages:
+            return None
+        current = now if now is not None else time.monotonic()
+        pinned: str | None = None
+        chain = hashlib.sha256(username.encode("utf-8")).digest()
+        for message in messages:
+            chain = hashlib.sha256(chain + _message_digest(message)).digest()
+            entry = self._entries.get(chain.hex())
+            if entry is None:
+                continue
+            credential_id, expires_at = entry
+            if expires_at <= current:
+                del self._entries[chain.hex()]
+                continue
+            pinned = credential_id
+            self._touch(chain.hex(), credential_id, current)
+        return pinned
+
+    def remember(self, messages: list[dict], username: str, credential_id: str, *,
+                 now: float | None = None) -> None:
+        """记录本轮请求的完整指纹 → 实际服务的凭证（成功后调用）。"""
+        if self.ttl_seconds <= 0 or not messages:
+            return
+        current = now if now is not None else time.monotonic()
+        chain = hashlib.sha256(username.encode("utf-8")).digest()
+        for message in messages:
+            chain = hashlib.sha256(chain + _message_digest(message)).digest()
+        self._touch(chain.hex(), credential_id, current)
+        self._evict(current)
+
+    def _touch(self, key: str, credential_id: str, now: float) -> None:
+        """写入/刷新条目并移到末尾（最近使用优先保留）。"""
+        self._entries.pop(key, None)
+        self._entries[key] = (credential_id, now + self.ttl_seconds)
+
+    def _evict(self, now: float) -> None:
+        """超上限时先清过期条目，仍超则按最旧淘汰。"""
+        if len(self._entries) <= self.max_entries:
+            return
+        for key in [k for k, (_, exp) in self._entries.items() if exp <= now]:
+            del self._entries[key]
+        while len(self._entries) > self.max_entries:
+            self._entries.pop(next(iter(self._entries)))
+
+    def __len__(self) -> int:
+        return len(self._entries)

src/engine/executor.py

@@ -46,6 +46,9 @@ class ExecutorDeps:
     # provider → {小写模型名: 上游原始 id}；api/models.list_models 拉取后就地更新。
     # 用于把独有模型的候选上游收窄到真正登记了它的上游，避免白打一次请求
     model_aliases: dict[str, dict[str, str]] | None = None
+    # 会话粘性（ConversationAffinity）；None 表示关闭。对话进行中固定用原
+    # 凭证，出错才轮换，成功后重新粘定实际服务的凭证
+    affinity: Any | None = None
 
     def record(self, **fields: Any) -> None:
         """统计写入失败绝不能影响聊天响应。"""
@@ -136,6 +139,8 @@ async def stream(self, request: ChatRequest, *, username: str = "unknown"
         target = self.resolve_target(request)
         state = _StreamState(translator=StreamTranslator(target.model),
                              started=time.monotonic(), username=username)
+        if self._deps.affinity is not None:
+            state.affinity_id = self._deps.affinity.pin_for(request.messages, username)
         try:
             async for frame in self._stream_loop(request, target, state):
                 yield frame
@@ -159,7 +164,7 @@ async def _stream_loop(self, request: ChatRequest, target: ModelTarget,
         last_kind: ErrKind | None = None
 
         while True:
-            pick = self._pick(target, tried)
+            pick = self._pick(target, tried, state.affinity_id)
             if pick is None:
                 if last_kind is ErrKind.INVALID:
                     # 所有候选上游都拒绝了该模型：400 语义而非 503
@@ -212,6 +217,7 @@ async def _stream_loop(self, request: ChatRequest, target: ModelTarget,
                         yield frame
                 else:
                     self._deps.credentials.save_success(credential_id)
+                    self._remember(request, state.username, credential_id)
                     self._record_success(target, state, provider_id, credential_id)
                     for frame in state.translator.finish():
                         yield frame
@@ -291,9 +297,12 @@ async def complete(self, request: ChatRequest, *, username: str = "unknown"
         last_credential: str | None = None
         # 首事件时刻：上游响应的第一个事件（TTFE，非流式的首字延迟）；跨重试只记最早一次
         first_event_at: float | None = None
+        # 会话粘性：对话上一轮用过哪个凭证，本轮优先复用
+        affinity_id = (self._deps.affinity.pin_for(request.messages, username)
+                       if self._deps.affinity is not None else None)
 
         while True:
-            pick = self._pick(target, tried)
+            pick = self._pick(target, tried, affinity_id)
             if pick is None:
                 if last_kind is ErrKind.INVALID:
                     # 所有候选上游都拒绝了该模型：400 而非 503
@@ -356,6 +365,7 @@ async def complete(self, request: ChatRequest, *, username: str = "unknown"
                         last_error = error
                 else:
                     self._deps.credentials.save_success(credential_id)
+                    self._remember(request, username, credential_id)
                     usage = result.get("usage") or {}
                     self._deps.record(
                         username=username, provider=provider_id,
@@ -386,7 +396,13 @@ async def complete(self, request: ChatRequest, *, username: str = "unknown"
 
     # -------------------------------------------------------------- 内部
 
-    def _pick(self, target: ModelTarget, tried: set[str]):
+    def _remember(self, request: ChatRequest, username: str, credential_id: str) -> None:
+        """成功后把本对话粘到实际服务的凭证（轮换降级后随之换粘）。"""
+        if self._deps.affinity is not None:
+            self._deps.affinity.remember(request.messages, username, credential_id)
+
+    def _pick(self, target: ModelTarget, tried: set[str],
+              affinity_id: str | None = None):
         # 区分两种情况：模型所属 provider 完全没注册（400）vs 注册了但没有可用凭证（503）
         registered = [pid for pid in self._narrow_providers(target)
                       if pid in self._deps.providers]
@@ -395,15 +411,31 @@ def _pick(self, target: ModelTarget, tried: set[str]):
         candidates = self._deps.credentials.candidates(registered)
         if not candidates:
             return None
-        credential_id = self._deps.scheduler.select(candidates, tried, int(time.time()))
+        credential_id = (self._sticky(candidates, tried, affinity_id)
+                         or self._deps.scheduler.select(candidates, tried, int(time.time())))
         if credential_id is None:
             return None
         credential_data = self._deps.credentials.credential_data(credential_id)
         if credential_data is None:  # 并发删除
             tried.add(credential_id)
-            return self._pick(target, tried)
+            return self._pick(target, tried, affinity_id)
         return credential_id, credential_data
 
+    def _sticky(self, candidates: list, tried: set[str],
+                affinity_id: str | None) -> str | None:
+        """会话粘性优先：指纹命中的凭证仍可选时直接复用，不参与排序。
+
+        只校验「在候选池里且未冷却/禁用/本请求已轮换过」；其余情况
+        （凭证被删、冷却中、用户强制了别的上游）回退常规调度，成功后
+        重新粘定，对话不会因此永久失粘。
+        """
+        if affinity_id is None or affinity_id in tried:
+            return None
+        match = next((c for c in candidates if c.credential_id == affinity_id), None)
+        if match is None or not match.is_selectable(int(time.time())):
+            return None
+        return affinity_id
+
     def _narrow_providers(self, target: ModelTarget) -> tuple[str, ...]:
         """模型目录能证明归属时，把候选上游收窄到登记了该模型的上游。
 
@@ -443,6 +475,7 @@ class _StreamState:
     username: str
     provider: str = "-"
     credential_id: str | None = None
+    affinity_id: str | None = None
     _first_byte_at: float | None = None
     # 已记账标记：断开分支防与正常成功路径重复记账
     recorded: bool = False

src/main.py

@@ -34,6 +34,7 @@
 from .db.crypto import CredentialCipher, CredentialDecryptError
 from .db.migrate import apply_schema
 from .db.repo import ApiKeyRepository, CredentialRepository
+from .engine.affinity import ConversationAffinity
 from .engine.executor import Executor, ExecutorDeps, NoHealthyCredential, NoProviderForModel
 from .engine.model_resolver import UnknownModelError
 from .engine.scheduler import Scheduler
@@ -207,6 +208,8 @@ def build_app(settings: Settings | None = None, *, providers: dict | None = None
                                          expiry_window=config.quota_expiry_window_seconds),
                                      default_model=config.default_model,
                                      stats=StatsCollector(db),
+                                     affinity=ConversationAffinity(
+                                         ttl_seconds=config.conversation_sticky_seconds),
                                      upstream_model_name=lambda provider_id, model_name: (
                                          model_aliases.get(provider_id, {}).get(
                                              model_name.lower(), model_name)