feat: add legacy message stability audit (#1716)

docs/CHANGELOG.md

@@ -25,6 +25,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 - [修复] #1390 收紧建议动作 legacy fallback：英文 `not to ...` 与 `avoid selling/reducing/trimming ...` 等否定/回避表达不再误判为买卖动作，Web 旧记录不再把中文金融上下文、`buy or sell`、多 guard 歧义文本或 `buyback` / `buy-back` / `buy back` / `selloff` / `sell-off` / `sell off` 等英文复合词渲染成 action badge，并在有结构化 `action` 时让回测/历史趋势等入口按界面语言显示 action 标签。
 - [改进] 完善运行时日志上下文，补充 logger name、触发来源、市场统计与实时行情预取链路状态，便于排查调度、API、Bot 和数据源降级路径。
 - [改进] #1595 P0a 新增 legacy LLM usage provider/cache telemetry 与 message HMAC 诊断字段，保持 prompt、provider 请求参数、fallback 与用量 summary API 行为不变。
+- [改进] #1595 P0.5a 为普通个股分析 legacy `[system, user]` message 新增内部稳定性审计字段，记录低敏路由上下文、skill 配置 HMAC、动态 marker 位置和 common-prefix 粗估，不改变 prompt、provider 参数或公开 Usage API。
 - [新功能] 新增分析任务与历史报告运行流快照 API，提供 lanes、nodes、edges、events、summary 等统一契约，并从任务队列、运行诊断和 AnalysisContextPack overview 构建脱敏数据流/信息流。
 - [新功能] Web 端为活跃任务、历史报告和大盘复盘报告补充运行流视图入口，支持查看运行摘要、拓扑节点、事件流和基础排障详情。
 - [修复] 修复历史报告运行流快照在混合时区事件时间戳下返回 500 的问题。

docs/LLM_CONFIG_GUIDE.md

@@ -284,6 +284,21 @@ LLM_USAGE_HMAC_KEY_VERSION=local-v1
 - 轮换密钥时同步更新 `LLM_USAGE_HMAC_KEY_VERSION`，避免不同密钥生成的 HMAC 被误比较。
 - 不要复用登录 session secret，也不要把真实密钥提交到版本控制或暴露在 issue、日志、截图中。
 
+### Legacy message stability audit（P0.5a）
+
+P0.5a 在普通个股分析路径为 legacy `[system, user]` message 追加内部稳定性审计字段，继续写入本地 `llm_usage`。它复用上面的 message HMAC，不修改 prompt 内容、message 顺序、provider 请求参数、cache hint、模型输出、fallback 顺序，也不扩展公开 Usage API 或 Web 页面。
+
+新增字段只用于维护者诊断：
+
+- `language`、`market_group`、`analysis_mode`、`legacy_prompt_mode`、`provider`、`transport`、`message_count` 描述本次普通个股分析调用的低敏路由上下文。
+- `skill_config_hmac` 是基于已解析 skill prompt 片段、默认 skill 策略和 legacy prompt 模式生成的 HMAC-SHA256，用于判断 system message 是否随 skill configuration 变化；不会保存 skill 原文。
+- `known_dynamic_marker_positions` 是 JSON string，只记录 `marker_name`、`message_role`、`char_offset`；不会保存股票代码、股票名称、日期、新闻正文、行情值、headers、response text 或 prompt 片段。
+- `estimated_total_prompt_tokens`、`approx_common_prefix_chars`、`approx_common_prefix_tokens` 基于项目内稳定 canonical render 估算：按 message 顺序拼接 `role + "\n" + content`，并用固定分隔符连接。该口径不声称等同 provider 真实 wire bytes。
+- `char_offset` 是 marker 在对应 message `content` 内的位置；`approx_common_prefix_chars` 是 canonical render 起点到第一个已知动态 marker 之前的字符数。没有 marker 时 common-prefix 字段为 `NULL`。
+- token 估算使用 `ceil(chars / 3)`，只作 diagnostics，不替代 provider usage，也不参与 cache threshold 判定；中文场景可能偏低。
+
+P0.5a 不引入 PromptBlock IR、`block_id`、`stability_class`、`static_prefix_hash` 或 `dynamic_context_hash`。Agent、research 与 market review 路径暂不接入该审计。
+
 ---
 
 ## 方式三：YAML 高级配置（适合老手自定义）

docs/LLM_CONFIG_GUIDE_EN.md

@@ -283,6 +283,21 @@ LLM_USAGE_HMAC_KEY_VERSION=local-v1
 - When rotating the secret, update `LLM_USAGE_HMAC_KEY_VERSION` so old and new fingerprints are not compared as if they used the same key.
 - Do not reuse the login session secret and do not commit or expose the real secret in version control, issues, logs, or screenshots.
 
+### Legacy message stability audit (P0.5a)
+
+P0.5a adds internal stability-audit fields for the ordinary stock-analysis legacy `[system, user]` message path. The fields are written only to local `llm_usage` records. They reuse the message HMAC pipeline above and do not change prompt text, message order, provider request parameters, cache hints, model output, fallback order, the public Usage API, or Web pages.
+
+The added fields are for maintainer diagnostics only:
+
+- `language`, `market_group`, `analysis_mode`, `legacy_prompt_mode`, `provider`, `transport`, and `message_count` describe low-sensitivity routing context for the stock-analysis call.
+- `skill_config_hmac` is an HMAC-SHA256 over the resolved skill prompt fragments, default skill policy, and legacy prompt mode. It lets maintainers tell whether the system message changes with skill configuration without storing raw skill text.
+- `known_dynamic_marker_positions` is a JSON string. Each entry stores only `marker_name`, `message_role`, and `char_offset`; it does not store stock codes, stock names, dates, news body text, quote values, headers, response text, or prompt snippets.
+- `estimated_total_prompt_tokens`, `approx_common_prefix_chars`, and `approx_common_prefix_tokens` use the repository's stable canonical render: messages are concatenated in order as `role + "\n" + content` with a fixed separator. This is not claimed to match provider wire bytes.
+- `char_offset` is measured inside the matching message `content`. `approx_common_prefix_chars` is the character count from canonical-render start to the first known dynamic marker. When no marker is found, common-prefix fields stay `NULL`.
+- Token estimates use `ceil(chars / 3)`. They are diagnostics only, do not replace provider usage, and are not used for cache-threshold decisions; Chinese text can be underestimated.
+
+P0.5a does not introduce PromptBlock IR, `block_id`, `stability_class`, `static_prefix_hash`, or `dynamic_context_hash`. Agent, research, and market-review paths are not wired into this audit yet.
+
 ### GitHub Actions Notes
 
 The bundled `00-daily-analysis.yml` explicitly passes the common LLM runtime fields to the job environment:

src/analyzer.py

@@ -42,6 +42,7 @@ from src.config import (
 from src.llm.generation_params import apply_litellm_generation_params
 from src.llm.errors import call_litellm_with_param_recovery
 from src.llm.usage import (
+    attach_legacy_message_stability_audit,
     attach_message_hmacs,
     extract_usage_payload,
     normalize_litellm_usage,
@@ -62,7 +63,7 @@ from src.report_language import (
 )
 from src.schemas.decision_action import build_action_fields
 from src.schemas.report_schema import AnalysisReportSchema
-from src.market_context import get_market_role, get_market_guidelines
+from src.market_context import detect_market, get_market_role, get_market_guidelines
 from src.services.daily_market_context import format_daily_market_context_prompt_section
 from src.market_phase_prompt import format_market_phase_prompt_section
 
@@ -164,6 +165,50 @@ def _should_hide_regular_session_ohlc(context: Dict[str, Any]) -> bool:
     )
 
 
+def _legacy_market_group(stock_code: Any) -> str:
+    code = str(stock_code or "").strip()
+    if not code or code.lower() == "unknown":
+        return "unknown"
+    market = detect_market(code)
+    return market if market in {"cn", "hk", "us"} else "unknown"
+
+
+def _legacy_audit_marker_specs(
+    context: Dict[str, Any],
+    *,
+    code: str,
+    stock_name: str,
+    report_language: str,
+    news_context: Optional[str],
+    analysis_context_pack_summary: Optional[str],
+) -> List[Dict[str, Any]]:
+    markers: List[Dict[str, Any]] = []
+
+    def add(marker_name: str, value: Any) -> None:
+        if value is None:
+            return
+        text = str(value).strip()
+        if not text:
+            return
+        markers.append(
+            {
+                "marker_name": marker_name,
+                "message_role": "user",
+                "text": text,
+            }
+        )
+
+    add("stock_code", code)
+    add("stock_name", stock_name)
+    add("analysis_date", context.get("date"))
+    add("market_phase", "## Market Phase Context" if report_language == "en" else "## 市场阶段上下文")
+    add("daily_market_context", "## Daily Market Context" if report_language == "en" else "## 大盘环境摘要")
+    add("analysis_context_pack", analysis_context_pack_summary)
+    add("quote", "## 📈 技术面数据")
+    add("news_context", "## 📰 舆情情报" if news_context else None)
+    return markers
+
+
 class _LiteLLMStreamError(RuntimeError):
     """Internal error wrapper that records whether any text was streamed."""
 
@@ -2498,6 +2543,7 @@ class GeminiAnalyzer:
         stream: bool = False,
         stream_progress_callback: Optional[Callable[[int], None]] = None,
         response_validator: Optional[Callable[[str], None]] = None,
+        audit_context: Optional[Dict[str, Any]] = None,
     ) -> Tuple[str, str, Dict[str, Any]]:
         """Call LLM via litellm with fallback across configured models.
 
@@ -2546,6 +2592,23 @@ class GeminiAnalyzer:
             usage_model, usage_provider = resolved_model_provider_identity(model, recovery_model_list)
 
             try:
+                def _attach_usage_audit(
+                    usage: Dict[str, Any],
+                    messages: List[Dict[str, Any]],
+                ) -> Dict[str, Any]:
+                    if audit_context is None:
+                        return attach_message_hmacs(usage, messages)
+                    effective_audit_context = dict(audit_context)
+                    effective_audit_context["provider"] = usage_provider
+                    effective_audit_context["transport"] = (
+                        effective_audit_context.get("transport") or "litellm"
+                    )
+                    return attach_legacy_message_stability_audit(
+                        usage,
+                        messages,
+                        effective_audit_context,
+                    )
+
                 model_short = model.split("/")[-1] if "/" in model else model
                 extra = get_thinking_extra_body(model_short)
                 call_kwargs: Dict[str, Any] = {
@@ -2630,7 +2693,7 @@ class GeminiAnalyzer:
                 if _stream_text is not None:
                     last_response_text = _stream_text
                     last_model = model
-                    _stream_usage = attach_message_hmacs(_stream_usage, call_kwargs["messages"])
+                    _stream_usage = _attach_usage_audit(_stream_usage, call_kwargs["messages"])
                     last_usage = _stream_usage
                     if response_validator is not None:
                         response_validator(_stream_text)
@@ -2652,12 +2715,15 @@ class GeminiAnalyzer:
 
                 content = self._extract_completion_text(response)
                 if content:
+                    usage_messages = None if audit_context is not None else call_kwargs["messages"]
                     usage = self._normalize_usage(
                         extract_usage_payload(response),
                         model=usage_model or model,
                         provider=usage_provider,
-                        messages=call_kwargs["messages"],
+                        messages=usage_messages,
                     )
+                    if audit_context is not None:
+                        usage = _attach_usage_audit(usage, call_kwargs["messages"])
                     last_response_text = content
                     last_model = model
                     last_usage = usage
@@ -2748,6 +2814,7 @@ class GeminiAnalyzer:
         config = self._get_runtime_config()
         report_language = normalize_report_language(getattr(config, "report_language", "zh"))
         system_prompt = self._get_analysis_system_prompt(report_language, stock_code=code)
+        skill_instructions, default_skill_policy, use_legacy_default_prompt = self._get_skill_prompt_sections()
         
         # 请求前增加延时（防止连续请求触发限流）
         request_delay = config.gemini_request_delay
@@ -2792,6 +2859,26 @@ class GeminiAnalyzer:
                 report_language=report_language,
                 analysis_context_pack_summary=analysis_context_pack_summary,
             )
+            legacy_audit_context = {
+                "language": report_language,
+                "market_group": _legacy_market_group(code),
+                "analysis_mode": "stock_analysis",
+                "legacy_prompt_mode": "legacy_default" if use_legacy_default_prompt else "skill_aware",
+                "skill_config": {
+                    "skill_instructions": skill_instructions,
+                    "default_skill_policy": default_skill_policy,
+                    "use_legacy_default_prompt": use_legacy_default_prompt,
+                },
+                "transport": "litellm",
+                "dynamic_markers": _legacy_audit_marker_specs(
+                    context,
+                    code=code,
+                    stock_name=name,
+                    report_language=report_language,
+                    news_context=news_context,
+                    analysis_context_pack_summary=analysis_context_pack_summary,
+                ),
+            }
             
             config = self._get_runtime_config()
             model_name = config.litellm_model or "unknown"
@@ -2829,6 +2916,7 @@ class GeminiAnalyzer:
                         stream=True,
                         stream_progress_callback=stream_progress_callback,
                         response_validator=self._validate_json_response,
+                        audit_context=legacy_audit_context,
                     )
                 except _AllModelsFailedError as exc:
                     if exc.last_response_text is not None:

src/llm/usage.py

@@ -3,10 +3,12 @@
 
 from __future__ import annotations
 
+from collections.abc import Iterable as IterableABC
 import hashlib
 import hmac
 import json
 import logging
+import math
 import os
 import re
 import secrets
@@ -23,6 +25,20 @@ PROVIDER_USAGE_MAX_SIZE_BYTES = 4096
 DEFAULT_HMAC_DOMAIN = "prompt_message"
 DEFAULT_HASH_SCOPE = "deployment"
 DEFAULT_HMAC_KEY_VERSION = "local-v1"
+LEGACY_AUDIT_MESSAGE_SEPARATOR = "\n\n---legacy-message---\n\n"
+
+_LEGACY_AUDIT_MARKER_NAMES = frozenset(
+    {
+        "stock_code",
+        "stock_name",
+        "analysis_date",
+        "market_phase",
+        "daily_market_context",
+        "analysis_context_pack",
+        "quote",
+        "news_context",
+    }
+)
 
 _HMAC_SECRET_CACHE: Optional[bytes] = None
 _DROP_RAW_USAGE_VALUE = object()
@@ -456,6 +472,68 @@ def attach_message_hmacs(
     return result
 
 
+def attach_legacy_message_stability_audit(
+    usage: Dict[str, Any],
+    messages: Optional[Sequence[Mapping[str, Any]]],
+    audit_context: Optional[Mapping[str, Any]] = None,
+    *,
+    hash_scope: str = DEFAULT_HASH_SCOPE,
+) -> Dict[str, Any]:
+    """Attach P0.5a legacy message stability diagnostics to usage telemetry.
+
+    The audit reuses message HMACs and records only stable metadata plus marker
+    offsets. Marker search values are never persisted.
+    """
+    result = attach_message_hmacs(usage, messages, hash_scope=hash_scope)
+    context = audit_context or {}
+    message_list = list(messages or [])
+    marker_specs = context.get("known_dynamic_markers")
+    if marker_specs is None:
+        marker_specs = context.get("dynamic_markers")
+    if marker_specs is None:
+        marker_specs = context.get("markers")
+
+    canonical_render, content_starts = _render_legacy_audit_messages(message_list)
+    marker_positions, first_marker_render_offset = _legacy_marker_positions(
+        message_list,
+        content_starts,
+        marker_specs,
+    )
+
+    approx_common_prefix_chars: Optional[int] = first_marker_render_offset
+    approx_common_prefix_tokens = (
+        _estimate_chars_as_tokens(approx_common_prefix_chars)
+        if approx_common_prefix_chars is not None
+        else None
+    )
+    estimated_total_prompt_tokens = _estimate_chars_as_tokens(len(canonical_render))
+
+    result.update(
+        {
+            "language": _audit_scalar(context.get("language"), max_len=16),
+            "market_group": _audit_scalar(context.get("market_group"), max_len=16),
+            "analysis_mode": _audit_scalar(context.get("analysis_mode"), max_len=64),
+            "legacy_prompt_mode": _audit_scalar(context.get("legacy_prompt_mode"), max_len=32),
+            "skill_config_hmac": _legacy_skill_config_hmac(context),
+            "provider": _audit_scalar(context.get("provider"), max_len=64),
+            "transport": _audit_scalar(context.get("transport"), max_len=64),
+            "message_count": len(message_list),
+            "estimated_total_prompt_tokens": estimated_total_prompt_tokens,
+            "approx_common_prefix_chars": approx_common_prefix_chars,
+            "approx_common_prefix_tokens": approx_common_prefix_tokens,
+            "known_dynamic_marker_positions": json.dumps(
+                marker_positions,
+                ensure_ascii=False,
+                separators=(",", ":"),
+            ),
+        }
+    )
+
+    if result.get("eligibility_confidence") in (None, "", "unknown"):
+        result["eligibility_confidence"] = "estimated"
+    return result
+
+
 def build_message_hmacs(
     messages: Optional[Sequence[Mapping[str, Any]]],
     *,
@@ -507,6 +585,142 @@ def _message_for_hmac(message: Mapping[str, Any]) -> Dict[str, Any]:
     return normalized
 
 
+def _render_legacy_audit_messages(
+    messages: Sequence[Mapping[str, Any]],
+) -> tuple[str, Dict[int, int]]:
+    parts = []
+    content_starts: Dict[int, int] = {}
+    cursor = 0
+    for index, message in enumerate(messages):
+        if index:
+            parts.append(LEGACY_AUDIT_MESSAGE_SEPARATOR)
+            cursor += len(LEGACY_AUDIT_MESSAGE_SEPARATOR)
+        role = str(message.get("role") or "")
+        content = _legacy_audit_text(message.get("content"))
+        prefix = f"{role}\n"
+        parts.append(prefix)
+        cursor += len(prefix)
+        content_starts[index] = cursor
+        parts.append(content)
+        cursor += len(content)
+    return "".join(parts), content_starts
+
+
+def _legacy_marker_positions(
+    messages: Sequence[Mapping[str, Any]],
+    content_starts: Mapping[int, int],
+    marker_specs: Any,
+) -> tuple[list[Dict[str, Any]], Optional[int]]:
+    positions_with_render_offset: list[tuple[int, Dict[str, Any]]] = []
+
+    for spec in _iter_marker_specs(marker_specs):
+        marker_name = _audit_scalar(
+            spec.get("marker_name") or spec.get("name"),
+            max_len=96,
+        )
+        if marker_name not in _LEGACY_AUDIT_MARKER_NAMES:
+            continue
+        requested_role = _audit_scalar(
+            spec.get("message_role") or spec.get("role") or "user",
+            max_len=32,
+        )
+        for candidate_text in _iter_marker_texts(spec):
+            found = _find_marker_in_messages(messages, requested_role, candidate_text)
+            if found is None:
+                continue
+            message_index, message_role, char_offset = found
+            render_offset = content_starts.get(message_index, 0) + char_offset
+            positions_with_render_offset.append(
+                (
+                    render_offset,
+                    {
+                        "marker_name": marker_name,
+                        "message_role": message_role,
+                        "char_offset": char_offset,
+                    },
+                )
+            )
+            break
+
+    positions_with_render_offset.sort(key=lambda item: item[0])
+    marker_positions = [position for _, position in positions_with_render_offset]
+    first_render_offset = positions_with_render_offset[0][0] if positions_with_render_offset else None
+    return marker_positions, first_render_offset
+
+
+def _legacy_skill_config_hmac(context: Mapping[str, Any]) -> Optional[str]:
+    skill_config = context.get("skill_config")
+    if not isinstance(skill_config, Mapping):
+        return None
+    secret = _load_usage_hmac_secret()
+    if not secret:
+        return None
+    payload = {
+        "domain": "legacy_skill_config",
+        "skill_instructions": _legacy_audit_text(skill_config.get("skill_instructions")),
+        "default_skill_policy": _legacy_audit_text(skill_config.get("default_skill_policy")),
+        "use_legacy_default_prompt": bool(skill_config.get("use_legacy_default_prompt")),
+    }
+    return _hmac_json(secret, payload)
+
+
+def _iter_marker_specs(marker_specs: Any) -> Iterable[Mapping[str, Any]]:
+    if not isinstance(marker_specs, IterableABC) or isinstance(marker_specs, (str, bytes, Mapping)):
+        return
+    for spec in marker_specs:
+        if isinstance(spec, Mapping):
+            yield spec
+
+
+def _iter_marker_texts(spec: Mapping[str, Any]) -> Iterable[str]:
+    text = _legacy_audit_text(spec.get("text"))
+    if text:
+        yield text
+
+
+def _find_marker_in_messages(
+    messages: Sequence[Mapping[str, Any]],
+    requested_role: Optional[str],
+    candidate_text: str,
+) -> Optional[tuple[int, str, int]]:
+    for index, message in enumerate(messages):
+        role = str(message.get("role") or "")
+        if requested_role and role != requested_role:
+            continue
+        content = _legacy_audit_text(message.get("content"))
+        offset = content.find(candidate_text)
+        if offset >= 0:
+            return index, role, offset
+    return None
+
+
+def _legacy_audit_text(value: Any) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, str):
+        return value
+    plain = _plain_value(value)
+    if isinstance(plain, str):
+        return plain
+    try:
+        return json.dumps(plain, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    except (TypeError, ValueError):
+        return str(plain)
+
+
+def _estimate_chars_as_tokens(chars: int) -> int:
+    return int(math.ceil(max(chars, 0) / 3))
+
+
+def _audit_scalar(value: Any, *, max_len: int) -> Optional[str]:
+    if value is None:
+        return None
+    text = str(value).strip()
+    if not text:
+        return None
+    return text[:max_len]
+
+
 def _load_usage_hmac_secret() -> Optional[bytes]:
     global _HMAC_SECRET_CACHE
     env_secret = os.getenv("LLM_USAGE_HMAC_SECRET")

src/storage.py

@@ -747,6 +747,7 @@ class LLMUsage(Base):
     call_type = Column(String(32), nullable=False, index=True)
     model = Column(String(128), nullable=False)
     stock_code = Column(String(16), nullable=True)
+    provider = Column(String(64), nullable=True)
     prompt_tokens = Column(Integer, nullable=False, default=0)
     completion_tokens = Column(Integer, nullable=False, default=0)
     total_tokens = Column(Integer, nullable=False, default=0)
@@ -792,11 +793,26 @@ class LLMUsage(Base):
     hmac_key_version = Column(String(64), nullable=True)
     hmac_domain = Column(String(32), nullable=True)
     hash_scope = Column(String(32), nullable=True)
+
+    # P0.5a internal legacy message stability audit. These diagnostics are
+    # stored locally only and are not returned by public usage APIs.
+    language = Column(String(16), nullable=True)
+    market_group = Column(String(16), nullable=True)
+    analysis_mode = Column(String(64), nullable=True)
+    legacy_prompt_mode = Column(String(32), nullable=True)
+    skill_config_hmac = Column(String(64), nullable=True)
+    transport = Column(String(64), nullable=True)
+    message_count = Column(Integer, nullable=True)
+    estimated_total_prompt_tokens = Column(Integer, nullable=True)
+    approx_common_prefix_chars = Column(Integer, nullable=True)
+    approx_common_prefix_tokens = Column(Integer, nullable=True)
+    known_dynamic_marker_positions = Column(Text, nullable=True)
     called_at = Column(DateTime, default=datetime.now, index=True)
 
 
 _LLM_USAGE_TELEMETRY_COLUMN_SQL: Dict[str, str] = {
     "provider_usage_json": "TEXT",
+    "provider": "VARCHAR(64)",
     "provider_usage_schema_name": "VARCHAR(64)",
     "provider_usage_schema_version": "VARCHAR(32)",
     "provider_usage_observed_at": "VARCHAR(32)",
@@ -826,6 +842,17 @@ _LLM_USAGE_TELEMETRY_COLUMN_SQL: Dict[str, str] = {
     "hmac_key_version": "VARCHAR(64)",
     "hmac_domain": "VARCHAR(32)",
     "hash_scope": "VARCHAR(32)",
+    "language": "VARCHAR(16)",
+    "market_group": "VARCHAR(16)",
+    "analysis_mode": "VARCHAR(64)",
+    "legacy_prompt_mode": "VARCHAR(32)",
+    "skill_config_hmac": "VARCHAR(64)",
+    "transport": "VARCHAR(64)",
+    "message_count": "INTEGER",
+    "estimated_total_prompt_tokens": "INTEGER",
+    "approx_common_prefix_chars": "INTEGER",
+    "approx_common_prefix_tokens": "INTEGER",
+    "known_dynamic_marker_positions": "TEXT",
 }
 _LLM_USAGE_INTEGER_TELEMETRY_COLUMNS = {
     column

tests/test_llm_usage.py

@@ -29,6 +29,7 @@ except ModuleNotFoundError:
     from litellm.types.utils import Usage
 
 from src.llm.usage import (
+    attach_legacy_message_stability_audit,
     attach_message_hmacs,
     build_message_hmacs,
     extract_usage_payload,
@@ -1176,6 +1177,228 @@ class TestLLMUsageHMAC(unittest.TestCase):
         )
 
 
+class TestLegacyMessageStabilityAudit(unittest.TestCase):
+    def setUp(self):
+        self._hmac_secret_patch = patch.dict(
+            os.environ,
+            {"LLM_USAGE_HMAC_SECRET": "audit-secret"},
+            clear=False,
+        )
+        self._hmac_secret_patch.start()
+        _reset_usage_hmac_secret_cache_for_tests()
+
+    def tearDown(self):
+        self._hmac_secret_patch.stop()
+        _reset_usage_hmac_secret_cache_for_tests()
+        DatabaseManager.reset_instance()
+
+    def _messages(self):
+        return [
+            {"role": "system", "content": "system policy for zh stock analysis"},
+            {
+                "role": "user",
+                "content": (
+                    "# 决策仪表盘分析请求\n\n"
+                    "## 📊 股票基础信息\n"
+                    "| 股票代码 | **600519** |\n"
+                    "| 股票名称 | **贵州茅台** |\n"
+                    "| 分析日期 | 2026-06-19 |\n\n"
+                    "## 📈 技术面数据\n"
+                    "收盘价 1500 元\n\n"
+                    "## 📰 舆情情报\n"
+                    "IMPORTANT_NEWS_TEXT\n"
+                ),
+            },
+        ]
+
+    def _audit_context(self):
+        return {
+            "language": "zh",
+            "market_group": "cn",
+            "analysis_mode": "stock_analysis",
+            "legacy_prompt_mode": "skill_aware",
+            "skill_config": {
+                "skill_instructions": "RSI breakout skill raw instructions",
+                "default_skill_policy": "Default skill policy raw text",
+                "use_legacy_default_prompt": False,
+            },
+            "provider": "gemini",
+            "transport": "litellm",
+            "dynamic_markers": [
+                {"marker_name": "stock_code", "message_role": "user", "text": "600519"},
+                {"marker_name": "stock_name", "message_role": "user", "text": "贵州茅台"},
+                {"marker_name": "analysis_date", "message_role": "user", "text": "2026-06-19"},
+                {"marker_name": "quote", "message_role": "user", "text": "## 📈 技术面数据"},
+                {"marker_name": "news_context", "message_role": "user", "text": "IMPORTANT_NEWS_TEXT"},
+                {"marker_name": "raw-header", "message_role": "user", "text": "Authorization: Bearer token"},
+            ],
+        }
+
+    def test_attaches_hmac_and_internal_audit_fields_without_raw_marker_values(self):
+        usage = attach_legacy_message_stability_audit(
+            {},
+            self._messages(),
+            self._audit_context(),
+        )
+
+        self.assertEqual(usage["language"], "zh")
+        self.assertEqual(usage["market_group"], "cn")
+        self.assertEqual(usage["analysis_mode"], "stock_analysis")
+        self.assertEqual(usage["legacy_prompt_mode"], "skill_aware")
+        self.assertEqual(len(usage["skill_config_hmac"]), 64)
+        self.assertEqual(usage["provider"], "gemini")
+        self.assertEqual(usage["transport"], "litellm")
+        self.assertEqual(usage["message_count"], 2)
+        self.assertGreater(usage["estimated_total_prompt_tokens"], 0)
+        self.assertIsNotNone(usage["approx_common_prefix_chars"])
+        self.assertIsNotNone(usage["approx_common_prefix_tokens"])
+        self.assertEqual(usage["eligibility_confidence"], "estimated")
+        self.assertEqual(len(usage["messages_hmac"]), 64)
+        self.assertEqual(len(usage["system_message_hmac"]), 64)
+        self.assertEqual(len(usage["user_message_hmac"]), 64)
+
+        marker_json = usage["known_dynamic_marker_positions"]
+        self.assertIsInstance(marker_json, str)
+        markers = json.loads(marker_json)
+        self.assertEqual(
+            {tuple(marker.keys()) for marker in markers},
+            {("marker_name", "message_role", "char_offset")},
+        )
+        self.assertEqual(markers[0]["marker_name"], "stock_code")
+        self.assertEqual(markers[0]["message_role"], "user")
+        self.assertIsInstance(markers[0]["char_offset"], int)
+
+        serialized = json.dumps(usage, ensure_ascii=False, sort_keys=True)
+        self.assertNotIn("600519", marker_json)
+        self.assertNotIn("贵州茅台", marker_json)
+        self.assertNotIn("2026-06-19", marker_json)
+        self.assertNotIn("IMPORTANT_NEWS_TEXT", marker_json)
+        self.assertNotIn("RSI breakout skill raw instructions", serialized)
+        self.assertNotIn("Default skill policy raw text", serialized)
+        self.assertNotIn("Authorization", serialized)
+        self.assertNotIn("Bearer", serialized)
+
+    def test_skill_config_hmac_changes_when_resolved_skill_config_changes(self):
+        base_context = self._audit_context()
+        changed_context = dict(base_context)
+        changed_context["skill_config"] = dict(base_context["skill_config"])
+        changed_context["skill_config"]["skill_instructions"] = "Different resolved skill instructions"
+
+        first = attach_legacy_message_stability_audit(
+            {},
+            self._messages(),
+            base_context,
+        )
+        second = attach_legacy_message_stability_audit(
+            {},
+            self._messages(),
+            changed_context,
+        )
+
+        self.assertEqual(first["messages_hmac"], second["messages_hmac"])
+        self.assertNotEqual(first["skill_config_hmac"], second["skill_config_hmac"])
+
+    def test_common_prefix_estimate_uses_canonical_render_before_first_marker(self):
+        messages = self._messages()
+        usage = attach_legacy_message_stability_audit(
+            {},
+            messages,
+            self._audit_context(),
+        )
+        user_content = messages[1]["content"]
+        first_marker_offset = user_content.index("600519")
+
+        self.assertGreater(usage["approx_common_prefix_chars"], first_marker_offset)
+        self.assertEqual(
+            usage["approx_common_prefix_tokens"],
+            (usage["approx_common_prefix_chars"] + 2) // 3,
+        )
+
+    def test_empty_preferred_marker_list_does_not_fall_back_to_other_marker_keys(self):
+        context = dict(self._audit_context())
+        context["known_dynamic_markers"] = []
+        context["markers"] = [
+            {"marker_name": "stock_code", "message_role": "user", "text": "600519"},
+        ]
+
+        usage = attach_legacy_message_stability_audit(
+            {},
+            self._messages(),
+            context,
+        )
+
+        self.assertEqual(json.loads(usage["known_dynamic_marker_positions"]), [])
+        self.assertIsNone(usage["approx_common_prefix_chars"])
+        self.assertIsNone(usage["approx_common_prefix_tokens"])
+
+    def test_preserves_exact_and_invalid_provider_usage_confidence(self):
+        messages = self._messages()
+        exact = attach_legacy_message_stability_audit(
+            normalize_litellm_usage({"prompt_tokens": 9}, model="openai/gpt-4o"),
+            messages,
+            self._audit_context(),
+        )
+        invalid = attach_legacy_message_stability_audit(
+            normalize_litellm_usage({"prompt_tokens": -1}, model="openai/gpt-4o"),
+            messages,
+            self._audit_context(),
+        )
+
+        self.assertEqual(exact["eligibility_confidence"], "exact")
+        self.assertEqual(invalid["eligibility_confidence"], "invalid")
+
+    def test_persisted_marker_positions_remain_json_string(self):
+        usage = attach_legacy_message_stability_audit(
+            {},
+            self._messages(),
+            self._audit_context(),
+        )
+        db = _fresh_db()
+        persist_llm_usage(usage, "gemini/gemini-test", call_type="analysis", stock_code="600519")
+
+        with db.session_scope() as session:
+            row = session.query(LLMUsage).one()
+            persisted = {
+                "language": row.language,
+                "market_group": row.market_group,
+                "analysis_mode": row.analysis_mode,
+                "legacy_prompt_mode": row.legacy_prompt_mode,
+                "skill_config_hmac": row.skill_config_hmac,
+                "provider": row.provider,
+                "transport": row.transport,
+                "message_count": row.message_count,
+                "known_dynamic_marker_positions": row.known_dynamic_marker_positions,
+            }
+
+        self.assertEqual(persisted["language"], "zh")
+        self.assertEqual(persisted["market_group"], "cn")
+        self.assertEqual(persisted["analysis_mode"], "stock_analysis")
+        self.assertEqual(persisted["legacy_prompt_mode"], "skill_aware")
+        self.assertEqual(len(persisted["skill_config_hmac"]), 64)
+        self.assertEqual(persisted["provider"], "gemini")
+        self.assertEqual(persisted["transport"], "litellm")
+        self.assertEqual(persisted["message_count"], 2)
+        self.assertIsInstance(persisted["known_dynamic_marker_positions"], str)
+        parsed = json.loads(persisted["known_dynamic_marker_positions"])
+        self.assertEqual(parsed[0]["marker_name"], "stock_code")
+        self.assertNotIn("600519", persisted["known_dynamic_marker_positions"])
+
+    def test_does_not_emit_block_level_p05b_fields(self):
+        usage = attach_legacy_message_stability_audit(
+            {},
+            self._messages(),
+            self._audit_context(),
+        )
+
+        for field in (
+            "block_id",
+            "stability_class",
+            "static_prefix_hash",
+            "dynamic_context_hash",
+        ):
+            self.assertNotIn(field, usage)
+
+
 class TestGetLLMUsageSummary(unittest.TestCase):
     def setUp(self):
         self.db = _fresh_db()

tests/test_market_analyzer_generate_text.py

@@ -8,6 +8,7 @@ Covers:
 - Any provider configuration (Gemini / Anthropic / OpenAI / LLM_CHANNELS)
   does NOT trigger AttributeError (regression guard for the old bypass bug)
 """
+import json
 import sys
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
@@ -108,6 +109,54 @@ class TestAnalyzerGenerateText:
             analyzer._router = None
             return analyzer
 
+    def test_legacy_market_group_normalizes_supported_markets(self):
+        from src.analyzer import _legacy_market_group
+
+        assert _legacy_market_group("") == "unknown"
+        assert _legacy_market_group("unknown") == "unknown"
+        assert _legacy_market_group("600519") == "cn"
+        assert _legacy_market_group("hk00700") == "hk"
+        assert _legacy_market_group("AAPL") == "us"
+
+    def test_legacy_audit_marker_specs_use_language_and_optional_context(self):
+        from src.analyzer import _legacy_audit_marker_specs
+
+        zh_markers = _legacy_audit_marker_specs(
+            {"date": "2026-06-19"},
+            code="600519",
+            stock_name="贵州茅台",
+            report_language="zh",
+            news_context="news",
+            analysis_context_pack_summary="pack summary",
+        )
+        zh_by_name = {marker["marker_name"]: marker for marker in zh_markers}
+
+        assert zh_by_name["stock_code"]["text"] == "600519"
+        assert zh_by_name["stock_name"]["text"] == "贵州茅台"
+        assert zh_by_name["analysis_date"]["text"] == "2026-06-19"
+        assert zh_by_name["market_phase"]["text"] == "## 市场阶段上下文"
+        assert zh_by_name["daily_market_context"]["text"] == "## 大盘环境摘要"
+        assert zh_by_name["analysis_context_pack"]["text"] == "pack summary"
+        assert zh_by_name["quote"]["text"] == "## 📈 技术面数据"
+        assert zh_by_name["news_context"]["text"] == "## 📰 舆情情报"
+        assert {marker["message_role"] for marker in zh_markers} == {"user"}
+
+        en_markers = _legacy_audit_marker_specs(
+            {"date": ""},
+            code="AAPL",
+            stock_name="Apple",
+            report_language="en",
+            news_context=None,
+            analysis_context_pack_summary=None,
+        )
+        en_by_name = {marker["marker_name"]: marker for marker in en_markers}
+
+        assert en_by_name["market_phase"]["text"] == "## Market Phase Context"
+        assert en_by_name["daily_market_context"]["text"] == "## Daily Market Context"
+        assert "analysis_date" not in en_by_name
+        assert "analysis_context_pack" not in en_by_name
+        assert "news_context" not in en_by_name
+
     def test_generate_text_returns_llm_response(self):
         analyzer = self._make_analyzer()
         with patch.object(analyzer, "_call_litellm", return_value="市场分析报告") as mock_call:
@@ -198,6 +247,52 @@ class TestAnalyzerGenerateText:
         assert model == "openai/gpt-4o-mini"
         _assert_usage_contains(usage, {"prompt_tokens": 11, "completion_tokens": 2, "total_tokens": 13})
 
+    def test_call_litellm_stream_records_legacy_message_audit_for_actual_messages(self):
+        analyzer = self._make_analyzer()
+        analyzer._config_override = SimpleNamespace(
+            litellm_model="gemini/gemini-2.0-flash",
+            litellm_fallback_models=[],
+            llm_model_list=[],
+        )
+
+        def stream_response():
+            yield SimpleNamespace(
+                choices=[SimpleNamespace(delta=SimpleNamespace(content="ok"))],
+                usage=SimpleNamespace(prompt_tokens=8, completion_tokens=1, total_tokens=9),
+            )
+
+        audit_context = {
+            "language": "zh",
+            "market_group": "cn",
+            "analysis_mode": "stock_analysis",
+            "dynamic_markers": [
+                {"marker_name": "stock_code", "message_role": "user", "text": "600519"},
+                {"marker_name": "quote", "message_role": "user", "text": "## 📈 技术面数据"},
+            ],
+        }
+
+        with patch.object(analyzer, "_dispatch_litellm_completion", return_value=stream_response()):
+            text, model, usage = analyzer._call_litellm(
+                "## 📊 股票基础信息\n| 股票代码 | **600519** |\n\n## 📈 技术面数据\n",
+                {"max_tokens": 128, "temperature": 0.2},
+                system_prompt="system prompt",
+                stream=True,
+                audit_context=audit_context,
+            )
+
+        assert text == "ok"
+        assert model == "gemini/gemini-2.0-flash"
+        _assert_usage_contains(usage, {"prompt_tokens": 8, "completion_tokens": 1, "total_tokens": 9})
+        assert usage["language"] == "zh"
+        assert usage["market_group"] == "cn"
+        assert usage["analysis_mode"] == "stock_analysis"
+        assert usage["provider"] == "gemini"
+        assert usage["transport"] == "litellm"
+        assert usage["message_count"] == 2
+        markers = json.loads(usage["known_dynamic_marker_positions"])
+        assert [marker["marker_name"] for marker in markers] == ["stock_code", "quote"]
+        assert "600519" not in usage["known_dynamic_marker_positions"]
+
     def test_call_litellm_legacy_path_uses_legacy_model_list_for_param_recovery(self):
         with patch("src.analyzer.get_config") as mock_cfg:
             cfg = MagicMock()
@@ -428,6 +523,8 @@ class TestAnalyzerGenerateText:
         assert text == "message response"
         assert model_used == "openai/deepseek-chat"
         _assert_no_provider_usage_hmac_only(usage)
+        assert "message_count" not in usage
+        assert "known_dynamic_marker_positions" not in usage
 
     def test_call_litellm_normalizes_kimi_k26_temperature(self):
         analyzer = self._make_analyzer()
@@ -453,6 +550,107 @@ class TestAnalyzerGenerateText:
         call_kwargs = mock_dispatch.call_args.args[1]
         assert call_kwargs["temperature"] == 1.0
 
+    def test_call_litellm_non_stream_records_legacy_message_audit_for_actual_messages(self):
+        analyzer = self._make_analyzer()
+        analyzer._config_override = SimpleNamespace(
+            litellm_model="openai/gpt-4o-mini",
+            litellm_fallback_models=[],
+            llm_model_list=[],
+        )
+        prompt = (
+            "# 决策仪表盘分析请求\n"
+            "| 股票代码 | **600519** |\n"
+            "| 股票名称 | **贵州茅台** |\n"
+            "| 分析日期 | 2026-06-19 |\n\n"
+            "## ✅ 分析任务\n"
+            "请输出 JSON。"
+        )
+        fixed_rules_offset = prompt.index("## ✅ 分析任务")
+        audit_context = {
+            "language": "zh",
+            "market_group": "cn",
+            "analysis_mode": "stock_analysis",
+            "dynamic_markers": [
+                {"marker_name": "stock_code", "message_role": "user", "text": "600519"},
+                {"marker_name": "stock_name", "message_role": "user", "text": "贵州茅台"},
+                {"marker_name": "analysis_date", "message_role": "user", "text": "2026-06-19"},
+            ],
+        }
+        response = SimpleNamespace(
+            choices=[SimpleNamespace(message=SimpleNamespace(content="ok"))],
+            usage=SimpleNamespace(prompt_tokens=10, completion_tokens=1, total_tokens=11),
+        )
+
+        with patch.object(analyzer, "_dispatch_litellm_completion", return_value=response):
+            text, model_used, usage = analyzer._call_litellm(
+                prompt,
+                {"max_tokens": 128, "temperature": 0.2},
+                system_prompt="system prompt",
+                audit_context=audit_context,
+            )
+
+        assert text == "ok"
+        assert model_used == "openai/gpt-4o-mini"
+        _assert_usage_contains(usage, {"prompt_tokens": 10, "completion_tokens": 1, "total_tokens": 11})
+        assert usage["provider"] == "openai"
+        assert usage["message_count"] == 2
+        markers = {
+            marker["marker_name"]: marker
+            for marker in json.loads(usage["known_dynamic_marker_positions"])
+        }
+        for marker_name in ("stock_code", "stock_name", "analysis_date"):
+            assert markers[marker_name]["message_role"] == "user"
+            assert markers[marker_name]["char_offset"] < fixed_rules_offset
+        assert "600519" not in usage["known_dynamic_marker_positions"]
+        assert "贵州茅台" not in usage["known_dynamic_marker_positions"]
+        assert "2026-06-19" not in usage["known_dynamic_marker_positions"]
+
+    def test_call_litellm_system_hmac_distinguishes_language_and_market_prompt(self):
+        analyzer = self._make_analyzer()
+        analyzer._config_override = SimpleNamespace(
+            litellm_model="openai/gpt-4o-mini",
+            litellm_fallback_models=[],
+            llm_model_list=[],
+        )
+        response = SimpleNamespace(
+            choices=[SimpleNamespace(message=SimpleNamespace(content="ok"))],
+            usage=SimpleNamespace(prompt_tokens=10, completion_tokens=1, total_tokens=11),
+        )
+
+        with patch.object(analyzer, "_dispatch_litellm_completion", return_value=response):
+            _, _, zh_usage = analyzer._call_litellm(
+                "same user prompt 600519",
+                {"max_tokens": 128, "temperature": 0.2},
+                system_prompt="system prompt zh cn",
+                audit_context={
+                    "language": "zh",
+                    "market_group": "cn",
+                    "analysis_mode": "stock_analysis",
+                    "dynamic_markers": [
+                        {"marker_name": "stock_code", "message_role": "user", "text": "600519"},
+                    ],
+                },
+            )
+            _, _, en_usage = analyzer._call_litellm(
+                "same user prompt 600519",
+                {"max_tokens": 128, "temperature": 0.2},
+                system_prompt="system prompt en us",
+                audit_context={
+                    "language": "en",
+                    "market_group": "us",
+                    "analysis_mode": "stock_analysis",
+                    "dynamic_markers": [
+                        {"marker_name": "stock_code", "message_role": "user", "text": "600519"},
+                    ],
+                },
+            )
+
+        assert zh_usage["system_message_hmac"] != en_usage["system_message_hmac"]
+        assert zh_usage["messages_hmac"] != en_usage["messages_hmac"]
+        assert zh_usage["user_message_hmac"] == en_usage["user_message_hmac"]
+        assert zh_usage["market_group"] == "cn"
+        assert en_usage["market_group"] == "us"
+
     def test_call_litellm_normalizes_kimi_k26_temperature_for_yaml_alias(self):
         analyzer = self._make_analyzer()
         analyzer._config_override = SimpleNamespace(
@@ -953,6 +1151,7 @@ class TestAnalyzerGenerateText:
 
         with patch.object(analyzer, "is_available", return_value=True), \
              patch.object(analyzer, "_get_analysis_system_prompt", return_value="system"), \
+             patch.object(analyzer, "_get_skill_prompt_sections", return_value=("RSI skill raw", "Default skill policy", False)), \
              patch.object(analyzer, "_format_prompt", return_value="prompt"), \
              patch.object(analyzer, "_validate_json_response"), \
              patch.object(analyzer, "_dispatch_litellm_completion", return_value=stream_response()), \
@@ -966,8 +1165,114 @@ class TestAnalyzerGenerateText:
         usage_arg, model_arg = mock_usage.call_args[0]
         assert model_arg == "openai/gpt-4o-mini"
         _assert_usage_contains(usage_arg, {"prompt_tokens": 11, "completion_tokens": 2, "total_tokens": 13})
+        assert usage_arg["language"] == "zh"
+        assert usage_arg["market_group"] == "cn"
+        assert usage_arg["analysis_mode"] == "stock_analysis"
+        assert usage_arg["legacy_prompt_mode"] == "skill_aware"
+        assert usage_arg["skill_config_hmac"] and len(usage_arg["skill_config_hmac"]) == 64
+        assert usage_arg["provider"] == "openai"
+        assert usage_arg["transport"] == "litellm"
+        assert usage_arg["message_count"] == 2
+        assert json.loads(usage_arg["known_dynamic_marker_positions"]) == []
         assert mock_usage.call_args.kwargs == {"call_type": "analysis", "stock_code": "600519"}
 
+    def test_analyze_records_marker_positions_from_real_prompt_format(self):
+        analyzer = self._make_analyzer()
+        analyzer._config_override = SimpleNamespace(
+            gemini_request_delay=0,
+            report_language="zh",
+            litellm_model="openai/gpt-4o-mini",
+            litellm_fallback_models=[],
+            llm_model_list=[],
+            llm_temperature=0.2,
+            report_integrity_enabled=False,
+            report_integrity_retry=0,
+            news_max_age_days=3,
+            news_strategy_profile="short",
+        )
+
+        from src.analyzer import AnalysisResult
+
+        parsed_result = AnalysisResult(
+            code="600519",
+            name="贵州茅台",
+            sentiment_score=80,
+            trend_prediction="看多",
+            operation_advice="持有",
+            analysis_summary="分析结果",
+        )
+
+        def stream_response():
+            yield SimpleNamespace(
+                choices=[SimpleNamespace(delta=SimpleNamespace(content='{"sentiment_score":80}'))],
+                usage=SimpleNamespace(prompt_tokens=42, completion_tokens=3, total_tokens=45),
+            )
+
+        context = {
+            "code": "600519",
+            "stock_name": "贵州茅台",
+            "date": "2026-06-19",
+            "today": {
+                "close": 1500,
+                "open": 1490,
+                "high": 1510,
+                "low": 1480,
+                "pct_chg": 1.2,
+                "volume": 100000,
+                "amount": 150000000,
+            },
+            "market_phase_context": {
+                "phase": "intraday",
+                "is_partial_bar": False,
+            },
+            "daily_market_context": {
+                "summary": "市场偏谨慎，等待量能确认。",
+                "region": "cn",
+                "trade_date": "2026-06-19",
+            },
+        }
+
+        with patch.object(analyzer, "is_available", return_value=True), \
+             patch.object(analyzer, "_get_analysis_system_prompt", return_value="system"), \
+             patch.object(analyzer, "_get_skill_prompt_sections", return_value=("RSI skill raw", "Default skill policy", False)), \
+             patch.object(analyzer, "_validate_json_response"), \
+             patch.object(analyzer, "_dispatch_litellm_completion", return_value=stream_response()), \
+             patch.object(analyzer, "_parse_response", return_value=parsed_result), \
+             patch.object(analyzer, "_build_market_snapshot", return_value={}), \
+             patch("src.analyzer.persist_llm_usage") as mock_usage:
+            result = analyzer.analyze(
+                context,
+                news_context="2026-06-18 贵州茅台发布经营公告。",
+                analysis_context_pack_summary="## 分析上下文包\n- 估值处于中性区间。",
+            )
+
+        assert result.analysis_summary == "分析结果"
+        mock_usage.assert_called_once()
+        usage_arg, _ = mock_usage.call_args[0]
+        markers = {
+            marker["marker_name"]: marker
+            for marker in json.loads(usage_arg["known_dynamic_marker_positions"])
+        }
+        for marker_name in (
+            "stock_code",
+            "stock_name",
+            "analysis_date",
+            "market_phase",
+            "daily_market_context",
+            "analysis_context_pack",
+            "quote",
+            "news_context",
+        ):
+            assert marker_name in markers
+            assert markers[marker_name]["message_role"] == "user"
+            assert isinstance(markers[marker_name]["char_offset"], int)
+            assert markers[marker_name]["char_offset"] >= 0
+        assert usage_arg["legacy_prompt_mode"] == "skill_aware"
+        assert usage_arg["skill_config_hmac"] and len(usage_arg["skill_config_hmac"]) == 64
+        assert "600519" not in usage_arg["known_dynamic_marker_positions"]
+        assert "贵州茅台" not in usage_arg["known_dynamic_marker_positions"]
+        assert "2026-06-19" not in usage_arg["known_dynamic_marker_positions"]
+
     def test_parse_response_non_json_returns_failure(self):
         """_parse_response must return success=False when LLM output is not valid JSON."""
         analyzer = self._make_analyzer()

tests/test_usage_api.py

@@ -48,6 +48,18 @@ class FakeUsageDbManager:
                 "call_type": "analysis",
                 "model": "openai/gpt-test",
                 "stock_code": "600519",
+                "provider": "openai",
+                "language": "zh",
+                "market_group": "cn",
+                "analysis_mode": "stock_analysis",
+                "legacy_prompt_mode": "skill_aware",
+                "skill_config_hmac": "a" * 64,
+                "transport": "litellm",
+                "message_count": 2,
+                "estimated_total_prompt_tokens": 2000,
+                "approx_common_prefix_chars": 120,
+                "approx_common_prefix_tokens": 40,
+                "known_dynamic_marker_positions": '[{"marker_name":"stock_code","message_role":"user","char_offset":12}]',
                 "prompt_tokens": 10,
                 "completion_tokens": 50,
                 "total_tokens": 60,
@@ -74,7 +86,21 @@ class UsageDashboardApiTestCase(unittest.TestCase):
         self.assertNotIn("context_window", body["by_model"][0])
         self.assertNotIn("context_usage_ratio", body["by_model"][0])
         self.assertEqual(body["recent_calls"][0]["stock_code"], "600519")
-        self.assertNotIn("provider", body["recent_calls"][0])
+        p05a_internal_fields = {
+            "provider",
+            "language",
+            "market_group",
+            "analysis_mode",
+            "legacy_prompt_mode",
+            "skill_config_hmac",
+            "transport",
+            "message_count",
+            "estimated_total_prompt_tokens",
+            "approx_common_prefix_chars",
+            "approx_common_prefix_tokens",
+            "known_dynamic_marker_positions",
+        }
+        self.assertTrue(p05a_internal_fields.isdisjoint(body["recent_calls"][0]))
         self.assertNotIn("context_window", body["recent_calls"][0])
         self.assertNotIn("context_usage_ratio", body["recent_calls"][0])