diff --git a/src/coding/proxy/config/config.default.yaml b/src/coding/proxy/config/config.default.yaml index f67c542..52950e0 100644 --- a/src/coding/proxy/config/config.default.yaml +++ b/src/coding/proxy/config/config.default.yaml @@ -704,3 +704,10 @@ session_policies: title_vendor_bindings: - prefix: "# 目标" vendor: "zhipu" + # Session 标题豁免前缀:user 输入(经噪声剥离清洗后)以任一前缀开头则跳过, + # 继续向后查找 title 候选。用于过滤注入式 Prompt(如示例语言指令), + # 避免其被误用作 Session 标题。大小写敏感 startswith。 + # ⚠️ 自定义 yaml 中该字段为整体替换(loader 的 list 合并约定,非追加合并); + # 写空列表即关闭豁免。加载时会自动 strip + 去空 + 去重。 + title_exempt_prefixes: + - "Write the title in the language" diff --git a/src/coding/proxy/config/session_policy.py b/src/coding/proxy/config/session_policy.py index c25da7a..7217f73 100644 --- a/src/coding/proxy/config/session_policy.py +++ b/src/coding/proxy/config/session_policy.py @@ -2,7 +2,7 @@ from __future__ import annotations -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, field_validator class SessionPolicyMatch(BaseModel): @@ -81,3 +81,34 @@ class SessionPoliciesConfig(BaseModel): "匹配规则按列表顺序求值,首次匹配生效。" ), ) + title_exempt_prefixes: list[str] = Field( + default_factory=list, + description=( + "Session 标题豁免前缀名单。当首条 user TEXT 输入(经噪声剥离清洗后)" + "以任一前缀开头时,跳过该输入、继续向后查找合适的 title 候选。" + "用于过滤注入式 Prompt(如示例语言指令),避免其被误用作 Session 标题。" + "大小写敏感的 startswith 匹配,与 title_vendor_bindings 语义一致。" + "加载时会自动 strip + 去空 + 去重;空字符串前缀会被丢弃" + "(防空串 startswith 恒真导致全量误豁免)。" + ), + ) + + @field_validator("title_exempt_prefixes", mode="after") + @classmethod + def _normalize_exempt_prefixes(cls, value: list[str]) -> list[str]: + """归一化豁免前缀:strip + 去空 + 去重保序. + + 空字符串前缀必须丢弃——``"".startswith`` 恒真会豁免一切用户输入, + 导致 Level 1 标题提取永久失效。 + """ + seen: set[str] = set() + result: list[str] = [] + for item in value: + if not isinstance(item, str): + continue + cleaned = item.strip() + if not cleaned or cleaned in seen: + continue + seen.add(cleaned) + result.append(cleaned) + return result diff --git a/src/coding/proxy/routing/executor.py b/src/coding/proxy/routing/executor.py index 9ade9dd..0569126 100644 --- a/src/coding/proxy/routing/executor.py +++ b/src/coding/proxy/routing/executor.py @@ -148,14 +148,30 @@ def _sanitize_user_text(raw: str) -> str: # ───────────────────────────────────────────────────────────────── -def _extract_title_from_user_text(messages: list[CanonicalMessagePart]) -> str: - """Level 1: 从 user TEXT 部分提取经噪声剥离后的首条非空文本.""" +def _extract_title_from_user_text( + messages: list[CanonicalMessagePart], + exempt_prefixes: list[str] | None = None, +) -> str: + """Level 1: 从 user TEXT 部分提取经噪声剥离后的首条非空文本. + + ``exempt_prefixes`` 非空时,清洗后文本以任一前缀开头则跳过该输入、继续向后 + 查找——用于过滤注入式 Prompt,避免其被误用作 Session 标题。匹配作用于 + ``_sanitize_user_text`` 清洗后的纯文本(而非带 XML 标签的原始 raw)。 + """ for part in messages: if part.role != "user" or part.type != CanonicalPartType.TEXT: continue cleaned = _sanitize_user_text(part.text) - if cleaned: - return cleaned[:_SESSION_TITLE_MAX_LEN] + if not cleaned: + continue + if exempt_prefixes and any(cleaned.startswith(p) for p in exempt_prefixes): + logger.debug( + "Title candidate skipped by exempt prefix %r (snippet=%r)", + next(p for p in exempt_prefixes if cleaned.startswith(p)), + cleaned[:60], + ) + continue + return cleaned[:_SESSION_TITLE_MAX_LEN] return "" @@ -193,15 +209,24 @@ def _extract_title_from_metadata(request: CanonicalRequest) -> str: return "" -def _extract_session_title(request: CanonicalRequest) -> str: +def _extract_session_title( + request: CanonicalRequest, + exempt_prefixes: list[str] | None = None, +) -> str: """从规范化请求中提取 session 标题 — 多层级回退策略。 依次尝试: user TEXT 噪声剥离 → TOOL_RESULT 摘要 → IMAGE 计数 → 元数据兜底。 任意级别命中即返回,确保 Dashboard 尽可能展示有辨识度的标题。 + + ``exempt_prefixes`` 仅作用于 Level 1(用户直接输入的 TEXT);Level 2/3/4 + 为非用户直接输入的回退来源,不参与豁免。默认 ``None`` 时行为与历史一致。 """ messages = request.messages + # Level 1 需要豁免前缀;单独调用以传入参数,避免污染 L2/L3 的统一签名。 + title = _extract_title_from_user_text(messages, exempt_prefixes) + if title: + return title[:_SESSION_TITLE_MAX_LEN] for extractor in ( - _extract_title_from_user_text, _extract_title_from_tool_results, _extract_title_from_images, ): @@ -614,6 +639,7 @@ def __init__( reauth_coordinator: Any | None = None, session_policy_resolver: SessionPolicyResolver | None = None, title_vendor_bindings: list[TitleVendorBinding] | None = None, + title_exempt_prefixes: list[str] | None = None, ) -> None: self._router = router self._tiers = tiers @@ -622,6 +648,13 @@ def __init__( self._reauth_coordinator = reauth_coordinator self._policy_resolver = session_policy_resolver or SessionPolicyResolver() self._title_vendor_bindings = title_vendor_bindings or [] + # 豁免前缀名单:构造期二次归一化(strip + 去空),双重防御配置层 field_validator。 + # 即便绕过配置层直接构造 executor,也能避免空串 startswith 恒真与纯空白失配。 + self._title_exempt_prefixes: list[str] = [ + stripped + for p in (title_exempt_prefixes or []) + if isinstance(p, str) and (stripped := p.strip()) + ] self._validate_title_vendor_bindings() # Tier 名称 → OAuth provider 名称的映射 @@ -676,6 +709,17 @@ def _resolve_effective_tiers(self, session_key: str) -> list[VendorTier]: seen.add(tier.name) return ordered + def _extract_session_title(self, request: CanonicalRequest) -> str: + """封装模块级标题提取,注入实例持有的豁免前缀名单. + + 将 ``self._title_exempt_prefixes`` 透传给模块级 ``_extract_session_title``, + 使 Level 1 能跳过以豁免前缀开头的注入式 Prompt。4 处标题提取调用点统一 + 经此方法,避免重复透传(零透传污染)。 + """ + return _extract_session_title( + request, exempt_prefixes=self._title_exempt_prefixes + ) + def _apply_title_based_policy(self, session_key: str, title: str) -> None: """根据 Session 标题前缀自动绑定供应商. @@ -790,7 +834,7 @@ async def execute_stream( canonical_request.trace_id, ) if is_new_session: - title = _extract_session_title(canonical_request) + title = self._extract_session_title(canonical_request) if title: await self._recorder.set_session_title( canonical_request.session_key, title @@ -798,7 +842,7 @@ async def execute_stream( self._apply_title_based_policy(canonical_request.session_key, title) else: # 延迟标题补写: 若 session 尚无标题,尝试从当前请求中提取并回写。 - title = _extract_session_title(canonical_request) + title = self._extract_session_title(canonical_request) if title: await self._recorder.update_empty_session_title( canonical_request.session_key, title @@ -977,7 +1021,7 @@ async def execute_message( canonical_request.trace_id, ) if is_new_session: - title = _extract_session_title(canonical_request) + title = self._extract_session_title(canonical_request) if title: await self._recorder.set_session_title( canonical_request.session_key, title @@ -985,7 +1029,7 @@ async def execute_message( self._apply_title_based_policy(canonical_request.session_key, title) else: # 延迟标题补写: 若 session 尚无标题,尝试从当前请求中提取并回写。 - title = _extract_session_title(canonical_request) + title = self._extract_session_title(canonical_request) if title: await self._recorder.update_empty_session_title( canonical_request.session_key, title diff --git a/src/coding/proxy/routing/router.py b/src/coding/proxy/routing/router.py index a9b5379..3ea162c 100644 --- a/src/coding/proxy/routing/router.py +++ b/src/coding/proxy/routing/router.py @@ -40,6 +40,7 @@ def __init__( compat_session_store: CompatSessionStore | None = None, session_policy_resolver: SessionPolicyResolver | None = None, title_vendor_bindings: list[TitleVendorBinding] | None = None, + title_exempt_prefixes: list[str] | None = None, ) -> None: if not tiers: raise ValueError("至少需要一个供应商层级") @@ -59,6 +60,7 @@ def __init__( reauth_coordinator=reauth_coordinator, session_policy_resolver=session_policy_resolver, title_vendor_bindings=title_vendor_bindings, + title_exempt_prefixes=title_exempt_prefixes, ) def set_pricing_table(self, table: PricingTable) -> None: diff --git a/src/coding/proxy/server/app.py b/src/coding/proxy/server/app.py index 396b89f..cf8aad4 100644 --- a/src/coding/proxy/server/app.py +++ b/src/coding/proxy/server/app.py @@ -162,6 +162,7 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI: compat_session_store, session_policy_resolver=SessionPolicyResolver(config.session_policies.policies), title_vendor_bindings=config.session_policies.title_vendor_bindings, + title_exempt_prefixes=config.session_policies.title_exempt_prefixes, ) app = FastAPI(title="coding-proxy", version=__version__, lifespan=lifespan) diff --git a/tests/test_router_executor.py b/tests/test_router_executor.py index 2e5c13e..4077271 100644 --- a/tests/test_router_executor.py +++ b/tests/test_router_executor.py @@ -2409,6 +2409,80 @@ def test_skips_assistant_role(self): req = self._build_request(messages) assert _extract_session_title(req) == "新的用户问题" + # ── 豁免前缀端到端(exempt_prefixes 透传至 Level 1)── + + def test_exempt_prefix_falls_through_to_next_user_text(self): + """端到端:L1 首条命中豁免 → 取第二条 user text 作为标题.""" + messages = [ + {"role": "user", "content": "Write the title in the language ..."}, + {"role": "user", "content": "端到端业务标题"}, + ] + req = self._build_request(messages) + assert ( + _extract_session_title( + req, exempt_prefixes=["Write the title in the language"] + ) + == "端到端业务标题" + ) + + def test_exempt_l1_falls_back_to_l2_tool_result(self): + """L1 命中豁免且无其他 user text → 回退 Level 2 tool_result.""" + messages = [ + {"role": "user", "content": "Write the title in the language ..."}, + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "tu_1", + "content": [{"type": "text", "text": "文件内容摘要"}], + } + ], + }, + ] + req = self._build_request(messages) + title = _extract_session_title( + req, exempt_prefixes=["Write the title in the language"] + ) + assert title == "[Tool output] 文件内容摘要" + + def test_exempt_l1_falls_back_to_l4_metadata(self): + """L1 命中豁免且无 L2/L3 → 回退 Level 4 元数据兜底.""" + messages = [{"role": "user", "content": "Write the title in the language ..."}] + req = self._build_request(messages) + assert ( + _extract_session_title( + req, exempt_prefixes=["Write the title in the language"] + ) + == "[Session] test" + ) + + def test_exempt_does_not_affect_l2(self): + """豁免仅作用于 L1:tool_result 文本以豁免前缀开头,L2 仍正常提取.""" + messages = [ + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "tu_1", + "content": [ + { + "type": "text", + "text": "Write the title in the language ...", + } + ], + } + ], + }, + ] + req = self._build_request(messages) + title = _extract_session_title( + req, exempt_prefixes=["Write the title in the language"] + ) + # L2 不受豁免影响,正常提取并加 [Tool output] 前缀 + assert title == "[Tool output] Write the title in the language ..." + # ═══════════════════════════════════════════════════════════════════ # 多层级回退标题提取测试 @@ -2453,6 +2527,141 @@ def test_returns_empty_for_noise_only(self): ] assert _extract_title_from_user_text(msgs) == "" + # ── 豁免前缀(exempt_prefixes)—— 过滤注入式 Prompt ── + + def test_exempt_prefix_skips_first_falls_through_to_second(self): + """豁免前缀命中首条 user text → 跳过、返回第二条.""" + from coding.proxy.model.compat import CanonicalMessagePart, CanonicalPartType + + msgs = [ + CanonicalMessagePart( + type=CanonicalPartType.TEXT, + role="user", + text="Write the title in the language the user wrote in, regardless.", + ), + CanonicalMessagePart( + type=CanonicalPartType.TEXT, role="user", text="帮我重构 executor" + ), + ] + assert ( + _extract_title_from_user_text( + msgs, exempt_prefixes=["Write the title in the language"] + ) + == "帮我重构 executor" + ) + + def test_exempt_prefix_case_sensitive(self): + """大小写敏感:小写前缀不命中大写开头的输入.""" + from coding.proxy.model.compat import CanonicalMessagePart, CanonicalPartType + + msgs = [ + CanonicalMessagePart( + type=CanonicalPartType.TEXT, + role="user", + text="Write the title in the language the user wrote in.", + ), + ] + # 前缀小写、输入首字母大写 → 不命中,正常返回 + assert ( + _extract_title_from_user_text(msgs, exempt_prefixes=["write the title"]) + == "Write the title in the language the user wrote in." + ) + + def test_exempt_no_param_backward_compatible(self): + """不传 exempt_prefixes → 行为与历史一致(首条非空即返回).""" + from coding.proxy.model.compat import CanonicalMessagePart, CanonicalPartType + + msgs = [ + CanonicalMessagePart( + type=CanonicalPartType.TEXT, + role="user", + text="Write the title in the language the user wrote in.", + ), + ] + assert _extract_title_from_user_text(msgs) == ( + "Write the title in the language the user wrote in." + ) + + def test_exempt_empty_list_no_op(self): + """传空列表 → 等价于不豁免.""" + from coding.proxy.model.compat import CanonicalMessagePart, CanonicalPartType + + msgs = [ + CanonicalMessagePart( + type=CanonicalPartType.TEXT, + role="user", + text="Write the title in the language the user wrote in.", + ), + ] + assert _extract_title_from_user_text(msgs, exempt_prefixes=[]) == ( + "Write the title in the language the user wrote in." + ) + + def test_exempt_all_user_inputs_skipped_returns_empty(self): + """全部 user text 命中豁免 → 返回空字符串.""" + from coding.proxy.model.compat import CanonicalMessagePart, CanonicalPartType + + msgs = [ + CanonicalMessagePart( + type=CanonicalPartType.TEXT, + role="user", + text="Write the title in the language variant A", + ), + CanonicalMessagePart( + type=CanonicalPartType.TEXT, + role="user", + text="Write the title in the language variant B", + ), + ] + assert ( + _extract_title_from_user_text( + msgs, exempt_prefixes=["Write the title in the language"] + ) + == "" + ) + + def test_exempt_matches_cleaned_not_raw(self): + """豁免判断作用于清洗后文本:raw 含 system-reminder 包裹、清洗后命中前缀 → 跳过.""" + from coding.proxy.model.compat import CanonicalMessagePart, CanonicalPartType + + raw = ( + "注入的系统上下文" + "Write the title in the language the user wrote in." + ) + msgs = [ + CanonicalMessagePart(type=CanonicalPartType.TEXT, role="user", text=raw), + CanonicalMessagePart( + type=CanonicalPartType.TEXT, role="user", text="真实业务问题" + ), + ] + assert ( + _extract_title_from_user_text( + msgs, exempt_prefixes=["Write the title in the language"] + ) + == "真实业务问题" + ) + + def test_exempt_multi_prefix_any_match(self): + """多前缀:命中任一即跳过.""" + from coding.proxy.model.compat import CanonicalMessagePart, CanonicalPartType + + msgs = [ + CanonicalMessagePart( + type=CanonicalPartType.TEXT, + role="user", + text="IGNORE_PREFIX_A something", + ), + CanonicalMessagePart( + type=CanonicalPartType.TEXT, role="user", text="第二输入" + ), + ] + assert ( + _extract_title_from_user_text( + msgs, exempt_prefixes=["IGNORE_PREFIX_A", "IGNORE_PREFIX_B"] + ) + == "第二输入" + ) + class TestExtractTitleFromToolResults: """Level 2 辅助函数 ``_extract_title_from_tool_results``.""" @@ -2946,3 +3155,39 @@ def test_known_vendor_no_startup_warning(self, caplog): r for r in caplog.records if "title_vendor_bindings" in r.message ] assert not binding_warnings + + +class TestExemptPrefixInjection: + """``_RouteExecutor._extract_session_title`` 实例方法注入豁免前缀名单.""" + + @staticmethod + def _build_request(messages: list[dict]): + return build_canonical_request({"model": "test", "messages": messages}, {}) + + def test_executor_injects_exempt_prefixes_to_l1(self): + """executor 持有的豁免前缀经实例方法注入,L1 跳过注入式 Prompt.""" + req = self._build_request( + [ + {"role": "user", "content": "Write the title in the language ..."}, + {"role": "user", "content": "注入后的真实标题"}, + ] + ) + executor = _executor(title_exempt_prefixes=["Write the title in the language"]) + assert executor._extract_session_title(req) == "注入后的真实标题" + + def test_executor_without_exempt_prefixes_backward_compatible(self): + """未配置豁免前缀 → 实例方法行为与模块级默认一致(不跳过).""" + req = self._build_request( + [{"role": "user", "content": "Write the title in the language ..."}] + ) + executor = _executor() # 不传 title_exempt_prefixes + assert executor._extract_session_title(req) == ( + "Write the title in the language ..." + ) + + def test_executor_filters_empty_string_prefix_at_construction(self): + """构造期过滤空串:即便传入空串/纯空白也不会豁免一切.""" + req = self._build_request([{"role": "user", "content": "正常标题"}]) + executor = _executor(title_exempt_prefixes=["", " ", "\t"]) + assert executor._title_exempt_prefixes == [] + assert executor._extract_session_title(req) == "正常标题" diff --git a/tests/test_session_aware.py b/tests/test_session_aware.py index 29518e5..9f3441a 100644 --- a/tests/test_session_aware.py +++ b/tests/test_session_aware.py @@ -462,6 +462,30 @@ def test_config_parse(): assert config.policies[0].tiers == ["anthropic", "copilot"] +def test_config_exempt_prefixes_default_empty(): + """未配置时默认空列表.""" + config = SessionPoliciesConfig() + assert config.title_exempt_prefixes == [] + + +def test_config_exempt_prefixes_strips_whitespace(): + """加载时 strip 每个前缀的前后空白.""" + config = SessionPoliciesConfig(title_exempt_prefixes=[" Write ", "\tRead\t"]) + assert config.title_exempt_prefixes == ["Write", "Read"] + + +def test_config_exempt_prefixes_drops_empty_string(): + """空字符串/纯空白前缀必须丢弃——防空串 startswith 恒真导致全量误豁免.""" + config = SessionPoliciesConfig(title_exempt_prefixes=["", " ", "Write"]) + assert config.title_exempt_prefixes == ["Write"] + + +def test_config_exempt_prefixes_dedupes_preserves_order(): + """去重并保留首次出现顺序.""" + config = SessionPoliciesConfig(title_exempt_prefixes=["a", "b", "a", "b", "c"]) + assert config.title_exempt_prefixes == ["a", "b", "c"] + + # ── 9. SessionPolicyResolver 运行时可变性 ────────────────────────