feat(session-routing): 新增基于 Session 标题前缀的可配置 Vendor 自动绑定;

当 Session 标题以配置前缀（默认「# 目标」）开头时，自动将该 Session
绑定到指定 vendor（默认 zhipu），不走默认 vendor 路由。前缀与目标
vendor 均通过 YAML 配置（session_policies.title_vendor_bindings），
支持多条规则、按顺序匹配首次命中生效、置空列表即禁用。

实现复用现有 SessionPolicyResolver.upsert() 运行时绑定机制，在标题
提取后、tier 解析前注入策略，仅对新 Session 首请求生效以避免覆盖手动
绑定。新增启动期 vendor 存在性校验（未知 vendor 告警而非硬失败）与
prefix/vendor 非空约束（杜绝空前缀全量误绑定）。

- config/session_policy.py: 新增 TitleVendorBinding 模型 + 配置字段
- config/config.default.yaml: 默认「# 目标」→ zhipu 绑定
- routing/executor.py: _apply_title_based_policy 方法、启动校验、双路径注入
- routing/router.py + server/app.py: 配置透传
- tests: 新增 17 个用例（前缀匹配/边缘/校验/流式与非流式端到端）

🤖 Generated with [Claude Code](https://github.com/claude), [CodeX](https://openai.com), [Gemini](https://github.com/apps/gemini-code-assist)
Co-Authored-By: Aurelius Huang<threefish.ai@gmail.com>

Author: ThreeFish-AI

src/coding/proxy/config/config.default.yaml

@@ -685,4 +685,11 @@ native_api:
 #       tiers: ["copilot", "anthropic", "zhipu"]
 #
 # 未配置时（默认），所有 Session 使用全局 tiers 顺序。
-session_policies: []
+session_policies:
+  policies: []
+  # 标题前缀 → 供应商自动绑定。
+  # 当 Session 标题以指定前缀开头时，自动将该 Session 绑定到对应供应商。
+  # 匹配规则按列表顺序求值，首次匹配生效。
+  title_vendor_bindings:
+    - prefix: "# 目标"
+      vendor: "zhipu"

src/coding/proxy/config/schema.py

@@ -44,7 +44,7 @@
 
 # ── 子模块 re-export ────────────────────────────────────────────
 from .server import DatabaseConfig, LoggingConfig, ServerConfig  # noqa: F401
-from .session_policy import SessionPoliciesConfig  # noqa: F401
+from .session_policy import SessionPoliciesConfig, TitleVendorBinding  # noqa: F401
 from .vendors import (  # noqa: F401
     AlibabaConfig,
     AnthropicConfig,
@@ -350,4 +350,5 @@ def compat_state_path(self) -> Path:
     "NativeApiConfig",
     # session policy
     "SessionPoliciesConfig",
+    "TitleVendorBinding",
 ]

src/coding/proxy/config/session_policy.py

@@ -50,10 +50,34 @@ class SessionPolicy(BaseModel):
     )
 
 
+class TitleVendorBinding(BaseModel):
+    """标题前缀 → 供应商自动绑定规则."""
+
+    prefix: str = Field(
+        min_length=1,
+        description=(
+            "标题前缀匹配模式（大小写敏感的 startswith 匹配）。"
+            "禁止空字符串——空前缀会匹配所有标题,导致全量误绑定。"
+        ),
+    )
+    vendor: str = Field(
+        min_length=1,
+        description="匹配后绑定的目标供应商名称",
+    )
+
+
 class SessionPoliciesConfig(BaseModel):
     """顶层 Session 策略配置容器."""
 
     policies: list[SessionPolicy] = Field(
         default_factory=list,
         description="Session 路由策略列表，按定义顺序求值，首次匹配生效",
     )
+    title_vendor_bindings: list[TitleVendorBinding] = Field(
+        default_factory=list,
+        description=(
+            "标题前缀 → 供应商自动绑定规则。"
+            "当 Session 标题以指定前缀开头时，自动绑定到对应供应商。"
+            "匹配规则按列表顺序求值，首次匹配生效。"
+        ),
+    )

src/coding/proxy/routing/executor.py

@@ -11,10 +11,13 @@
 import re
 import time
 from collections.abc import AsyncIterator
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import httpx
 
+if TYPE_CHECKING:
+    from ..config.session_policy import TitleVendorBinding
+
 from ..vendors.base import (
     NoCompatibleVendorError,
     RequestCapabilities,
@@ -610,20 +613,43 @@ def __init__(
         session_manager: RouteSessionManager,
         reauth_coordinator: Any | None = None,
         session_policy_resolver: SessionPolicyResolver | None = None,
+        title_vendor_bindings: list[TitleVendorBinding] | None = None,
     ) -> None:
         self._router = router
         self._tiers = tiers
         self._recorder = usage_recorder
         self._session_mgr = session_manager
         self._reauth_coordinator = reauth_coordinator
         self._policy_resolver = session_policy_resolver or SessionPolicyResolver()
+        self._title_vendor_bindings = title_vendor_bindings or []
+        self._validate_title_vendor_bindings()
 
         # Tier 名称 → OAuth provider 名称的映射
         self._tier_provider_map: dict[str, str] = {
             "copilot": "github",
             "antigravity": "google",
         }
 
+    def _validate_title_vendor_bindings(self) -> None:
+        """启动期校验标题绑定引用的 vendor 均存在,缺失则告警.
+
+        与手动绑定 API（拒绝未知 vendor）的语义对齐：此处不硬失败，
+        仅记录警告——避免单条误配置阻断整个代理启动；运行时
+        `_resolve_effective_tiers` 会静默跳过未知 vendor 回退默认顺序。
+        """
+        if not self._title_vendor_bindings:
+            return
+        valid = {t.name for t in self._tiers}
+        for binding in self._title_vendor_bindings:
+            if binding.vendor not in valid:
+                logger.warning(
+                    "title_vendor_bindings 引用了未知 vendor %r（前缀 %r）；"
+                    "可用 vendor: %s。该绑定将在运行时被静默跳过。",
+                    binding.vendor,
+                    binding.prefix,
+                    sorted(valid),
+                )
+
     # ── 公开执行入口 ──────────────────────────────────────
 
     def _resolve_effective_tiers(self, session_key: str) -> list[VendorTier]:
@@ -650,6 +676,27 @@ def _resolve_effective_tiers(self, session_key: str) -> list[VendorTier]:
                 seen.add(tier.name)
         return ordered
 
+    def _apply_title_based_policy(self, session_key: str, title: str) -> None:
+        """根据 Session 标题前缀自动绑定供应商.
+
+        当标题以预配置的前缀开头时，通过 SessionPolicyResolver.upsert()
+        将该 Session 绑定到指定供应商，后续请求无需再走默认路由。
+
+        仅在新 Session 首次提取标题时调用，避免覆盖手动绑定的策略。
+        """
+        if not title or not self._title_vendor_bindings:
+            return
+        for binding in self._title_vendor_bindings:
+            if title.startswith(binding.prefix):
+                self._policy_resolver.upsert(session_key, [binding.vendor])
+                logger.info(
+                    "Session title prefix %r matched → auto-bind to %s (session=%s)",
+                    binding.prefix,
+                    binding.vendor,
+                    session_key[:12],
+                )
+                return
+
     def _prepare_body_for_tier(
         self,
         body: dict[str, Any],
@@ -748,6 +795,7 @@ async def execute_stream(
                 await self._recorder.set_session_title(
                     canonical_request.session_key, title
                 )
+                self._apply_title_based_policy(canonical_request.session_key, title)
         else:
             # 延迟标题补写: 若 session 尚无标题,尝试从当前请求中提取并回写。
             title = _extract_session_title(canonical_request)
@@ -934,6 +982,7 @@ async def execute_message(
                 await self._recorder.set_session_title(
                     canonical_request.session_key, title
                 )
+                self._apply_title_based_policy(canonical_request.session_key, title)
         else:
             # 延迟标题补写: 若 session 尚无标题,尝试从当前请求中提取并回写。
             title = _extract_session_title(canonical_request)

src/coding/proxy/routing/router.py

@@ -14,6 +14,7 @@
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
+    from ..config.session_policy import TitleVendorBinding
     from ..pricing import PricingTable
 
 from .executor import _RouteExecutor
@@ -38,6 +39,7 @@ def __init__(
         reauth_coordinator: Any | None = None,
         compat_session_store: CompatSessionStore | None = None,
         session_policy_resolver: SessionPolicyResolver | None = None,
+        title_vendor_bindings: list[TitleVendorBinding] | None = None,
     ) -> None:
         if not tiers:
             raise ValueError("至少需要一个供应商层级")
@@ -56,6 +58,7 @@ def __init__(
             session_manager=self._session_mgr,
             reauth_coordinator=reauth_coordinator,
             session_policy_resolver=session_policy_resolver,
+            title_vendor_bindings=title_vendor_bindings,
         )
 
     def set_pricing_table(self, table: PricingTable) -> None:

src/coding/proxy/server/app.py

@@ -161,6 +161,7 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
         reauth_coordinator,
         compat_session_store,
         session_policy_resolver=SessionPolicyResolver(config.session_policies.policies),
+        title_vendor_bindings=config.session_policies.title_vendor_bindings,
     )
 
     app = FastAPI(title="coding-proxy", version=__version__, lifespan=lifespan)