fix: 问股 single-agent 增加 provider-aware trace (#1473)

* fix: add provider-aware trace for ask chat

* fix: drop mismatched provider trace attempts

docs/CHANGELOG.md

@@ -42,6 +42,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 - [修复] 修复问股会话切换和首页任务重连后可能残留 Agent/分析任务进行中状态的问题。
 - [新功能] 问股新增默认关闭的可见对话上下文压缩，支持 Web 开关、Agent 高级 preset、滚动摘要和最近轮次原文保护，降低长会话 token 消耗。
 - [改进] P2-min：LLM Prompt 注入市场阶段上下文。
+- [修复] 问股 single-agent 新增 provider-aware trace 分轨，跨轮保留 DeepSeek V4 thinking + tool-call 的 `reasoning_content` 与工具协议材料。
 
 ## [3.18.0] - 2026-05-21
 

docs/LLM_CONFIG_GUIDE.md

@@ -213,6 +213,8 @@ AGENT_CONTEXT_PROTECTED_TURNS=
 
 压缩只处理 `session_id` 下用户可见的 `user` / `assistant` 文本历史，不处理 provider trace、thinking blocks、tool calls 或 tool results，也不会改变同轮工具调用透传。三档 preset 分别是 `cost`（6000 tokens / 保护 2 轮）、`balanced`（12000 / 4）和 `long_context_raw_first`（24000 / 6）；trigger / protected 留空时跟随当前 profile，显式填写时覆盖 profile。
 
+问股 single-agent 路径会额外维护一条 provider-aware trace 分轨，用于 DeepSeek V4 thinking + tool-call 的跨轮协议回放：只有同一轮同时出现 `tool_calls` 与 `reasoning_content` 时才会按当前 `session_id + provider + model` 保存最近 3 条最小协议材料，并在下一轮按原始时序插回对应可见 assistant 回复之前。该 trace 只能原样保留或整段丢弃，不参与摘要、不写入 Web 会话消息、不新增 `.env` 配置；model/provider 不匹配、锚点已被 summary 覆盖或预算不足时会整段跳过。Claude extended thinking 本轮只覆盖 adapter/storage 级 opaque `thinking` / `redacted_thinking` / `signature` blocks plumbing 与离线 fixture，不声明生产端到端支持；multi-agent trace 注入仍是 follow-up。外部协议依据包括 DeepSeek thinking mode 文档（<https://api-docs.deepseek.com/guides/thinking_mode>）和 Anthropic Claude extended thinking 文档（<https://platform.claude.com/docs/en/docs/build-with-claude/extended-thinking>），LiteLLM 兼容窗口仍以 `requirements.txt` 的 `litellm>=1.80.10,!=1.82.7,!=1.82.8,<2.0.0` 为准。
+
 ### 严格 temperature 模型兼容说明
 
 - Moonshot 官方说明 Kimi API 兼容 OpenAI 接口，Base URL 使用 `https://api.moonshot.ai/v1`：<https://platform.kimi.ai/docs/guide/kimi-k2-6-quickstart>

docs/LLM_CONFIG_GUIDE_EN.md

@@ -192,6 +192,8 @@ LITELLM_MODEL=ollama/qwen3:8b
 - If the current environment has no valid Agent model path at all, the ask-stock page still returns a failure and now surfaces the backend's real configuration diagnosis. As soon as you restore any valid model source, the flow recovers without running any migration step.
 - The recommended forward path is still to configure `LITELLM_MODEL` / `AGENT_LITELLM_MODEL` explicitly or move to `LLM_CHANNELS`; legacy provider keys remain a compatibility fallback for older `.env` files, local macOS development, and existing deployments.
 
+For the single-agent ask-stock path, the backend also keeps a provider-aware trace track for DeepSeek V4 thinking + tool-call roundtrip. A trace is persisted only when the same run has both `tool_calls` and `reasoning_content`; the last 3 minimal protocol slices per `session_id + provider + model` are spliced back into the next request before the anchored visible assistant reply. Provider trace is either preserved exactly or dropped as a whole; it is never summarized, never returned by Web session-history APIs, and adds no `.env` setting. Model/provider mismatch, summarized anchors, or insufficient budget drop the whole trace. Claude extended thinking is limited in this PR to adapter/storage-level opaque `thinking` / `redacted_thinking` / `signature` block plumbing with offline fixtures; production end-to-end Claude and multi-agent trace injection remain follow-ups. Protocol references: DeepSeek thinking mode (<https://api-docs.deepseek.com/guides/thinking_mode>) and Anthropic Claude extended thinking (<https://platform.claude.com/docs/en/docs/build-with-claude/extended-thinking>). The LiteLLM compatibility window remains `litellm>=1.80.10,!=1.82.7,!=1.82.8,<2.0.0` from `requirements.txt`.
+
 ### Strict Temperature Model Compatibility Notes
 
 - Moonshot officially documents Kimi as an OpenAI-compatible API, with `https://api.moonshot.ai/v1` as the base URL: <https://platform.kimi.ai/docs/guide/kimi-k2-6-quickstart>

docs/full-guide.md

@@ -242,6 +242,8 @@ daily_stock_analysis/
 
 > *注：`ANSPIRE_API_KEYS`、`AIHUBMIX_KEY`、`GEMINI_API_KEY`、`ANTHROPIC_API_KEY`、`OPENAI_API_KEY` 或 `OLLAMA_API_BASE` 至少配置一个。`ANSPIRE_API_KEYS` 与 `AIHUBMIX_KEY` 无需配置 `OPENAI_BASE_URL`，系统自动适配。
 
+> 问股 single-agent 路径会在后台为 DeepSeek V4 thinking + tool-call 保存最近 3 条 provider trace，并按原时序回放 `reasoning_content` / tool 结果；该能力不新增配置项，不进入 Web 历史 API，Claude extended thinking 仅覆盖离线 plumbing，multi-agent trace 注入留作后续增强。
+
 ### 通知渠道配置
 
 更多通知配置基线、诊断和部署场景说明见 [通知专题文档](notifications.md)。

src/agent/chat_context.py

@@ -11,6 +11,15 @@ from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
 from src.config import (
     get_agent_context_compression_preset,
     get_effective_agent_primary_model,
+    get_effective_agent_models_to_try,
+)
+from src.agent.provider_trace import (
+    TRACE_MODEL_KEY,
+    TRACE_PROVIDER_KEY,
+    TraceDiagnostics,
+    resolved_provider_namespace,
+    strip_trace_metadata,
+    trace_model_matches,
 )
 from src.storage import get_db, persist_llm_usage
 
@@ -45,6 +54,23 @@ class VisibleMessage:
     created_at: Any = None
 
 
+@dataclass(frozen=True)
+class VisibleHistoryState:
+    """Id-aware visible history state used for summary and trace splicing."""
+
+    messages: List[Dict[str, Any]]
+    visible_ids: set[int]
+    visible_tokens: int
+
+
+@dataclass(frozen=True)
+class AgentChatContextBundle:
+    """Prepared context messages for a single-agent chat request."""
+
+    context_messages: List[Dict[str, Any]]
+    diagnostics: Dict[str, Any]
+
+
 def build_summary_message(summary_text: str) -> Dict[str, str]:
     """Build the synthetic summary message injected into chat history."""
     return {
@@ -95,13 +121,117 @@ def build_visible_chat_history(
     config: Any,
 ) -> List[Dict[str, str]]:
     """Return visible chat history according to the compression state table."""
+    state = _build_visible_history_state(session_id, llm_adapter, config)
+    return _strip_internal_message_ids(state.messages)
+
+
+def build_agent_chat_context_bundle(
+    session_id: str,
+    llm_adapter: Any,
+    config: Any,
+) -> AgentChatContextBundle:
+    """Return id-spliced visible history plus provider trace messages.
+
+    The bundle excludes the current user turn.  ``AgentExecutor.chat`` appends
+    factual context and the current user after these messages, preserving the
+    existing request assembly order.
+    """
+    state = _build_visible_history_state(session_id, llm_adapter, config)
+    diagnostics = TraceDiagnostics(visible_tokens=state.visible_tokens)
+    db = get_db()
+    candidate_models = get_effective_agent_models_to_try(config)
+    if not candidate_models:
+        candidate_models = [get_effective_agent_primary_model(config)]
+    candidate_trace_targets = _build_trace_match_targets(candidate_models, config)
+    turns = db.get_agent_provider_turns(session_id, must_roundtrip_only=True)
+    traces_by_anchor: Dict[int, List[Dict[str, Any]]] = {}
+    pending_trace_tokens = 0
+    pending_trace_count = 0
+
+    for turn in turns:
+        if not any(
+            trace_model_matches(
+                turn.get("provider"),
+                turn.get("model"),
+                model,
+                current_provider=provider,
+            )
+            for model, provider in candidate_trace_targets
+        ):
+            diagnostics.model_mismatch += 1
+            diagnostics.dropped_trace_count += 1
+            diagnostics.trace_dropped_reason = diagnostics.trace_dropped_reason or "model_mismatch"
+            continue
+        anchor_id = _coerce_int(turn.get("anchor_assistant_message_id"), default=0)
+        if anchor_id <= 0 or anchor_id not in state.visible_ids:
+            diagnostics.anchor_summarized += 1
+            diagnostics.dropped_trace_count += 1
+            diagnostics.trace_dropped_reason = diagnostics.trace_dropped_reason or "anchor_summarized"
+            continue
+        trace_messages = _restore_trace_metadata(
+            turn.get("messages") or [],
+            provider=turn.get("provider"),
+            model=turn.get("model"),
+        )
+        if not trace_messages:
+            continue
+        pending_trace_count += 1
+        pending_trace_tokens += _coerce_int(
+            turn.get("estimated_tokens"),
+            default=estimate_messages_tokens(trace_messages, config),
+        )
+        traces_by_anchor.setdefault(anchor_id, []).extend(trace_messages)
+
+    if traces_by_anchor:
+        preset = get_agent_context_compression_preset(
+            getattr(config, "agent_context_compression_profile", None)
+        )
+        history_budget = _coerce_int(
+            getattr(config, "agent_context_history_budget_tokens", preset.history_budget_tokens),
+            default=preset.history_budget_tokens,
+        )
+        remaining_budget = history_budget - state.visible_tokens
+        if remaining_budget < pending_trace_tokens:
+            diagnostics.budget_exceeded = True
+            diagnostics.trace_dropped_reason = "budget_exceeded"
+            diagnostics.dropped_trace_count += pending_trace_count
+            traces_by_anchor = {}
+        else:
+            diagnostics.trace_injected = True
+            diagnostics.trace_tokens = pending_trace_tokens
+
+    merged: List[Dict[str, Any]] = []
+    for msg in state.messages:
+        msg_id = _coerce_int(msg.get("_message_id"), default=0)
+        if msg_id and msg_id in traces_by_anchor:
+            merged.extend(traces_by_anchor[msg_id])
+        merged.append(msg)
+
+    return AgentChatContextBundle(
+        context_messages=_strip_internal_message_ids(merged),
+        diagnostics=diagnostics.to_dict(),
+    )
+
+
+def _build_visible_history_state(
+    session_id: str,
+    llm_adapter: Any,
+    config: Any,
+) -> VisibleHistoryState:
+    """Return visible history with private ``_message_id`` anchors."""
     db = get_db()
     if not getattr(config, "agent_context_compression_enabled", False):
-        return db.get_conversation_history(session_id, limit=20)
+        selected = _load_visible_messages(session_id, limit=20)
+        messages = _to_chat_messages(selected, include_ids=True)
+        return VisibleHistoryState(
+            messages=messages,
+            visible_ids={msg.id for msg in selected},
+            visible_tokens=estimate_messages_tokens(_strip_internal_message_ids(messages), config),
+        )
 
     visible_messages = _load_visible_messages(session_id)
     if not visible_messages:
-        return []
+        return VisibleHistoryState(messages=[], visible_ids=set(), visible_tokens=0)
 
     summary_record = db.get_conversation_summary(session_id)
     previous_summary = (summary_record or {}).get("summary") or ""
@@ -122,14 +252,18 @@ def build_visible_chat_history(
     protected_ids = {msg.id for msg in protected_tail}
     uncovered_messages = [msg for msg in visible_messages if msg.id > covered_message_id]
     candidate = (
-        [build_summary_message(previous_summary)] + _to_chat_messages(uncovered_messages)
+        [build_summary_message(previous_summary)] + _to_chat_messages(uncovered_messages, include_ids=True)
         if previous_summary
-        else _to_chat_messages(visible_messages)
+        else _to_chat_messages(visible_messages, include_ids=True)
     )
-    candidate_tokens = estimate_messages_tokens(candidate, config)
+    candidate_tokens = estimate_messages_tokens(_strip_internal_message_ids(candidate), config)
 
     if candidate_tokens <= trigger_tokens:
-        return candidate
+        return VisibleHistoryState(
+            messages=candidate,
+            visible_ids={msg.id for msg in visible_messages if msg.id > covered_message_id or not previous_summary},
+            visible_tokens=candidate_tokens,
+        )
 
     to_summarize = [
         msg
@@ -142,12 +276,22 @@ def build_visible_chat_history(
                 "Conversation context compression skipped for session %s: protected tail exceeds trigger",
                 session_id,
             )
-            return [build_summary_message(previous_summary)] + _to_chat_messages(protected_tail)
+            messages = [build_summary_message(previous_summary)] + _to_chat_messages(protected_tail, include_ids=True)
+            return VisibleHistoryState(
+                messages=messages,
+                visible_ids={msg.id for msg in protected_tail},
+                visible_tokens=estimate_messages_tokens(_strip_internal_message_ids(messages), config),
+            )
         logger.warning(
             "Conversation context compression skipped for session %s: all visible history is protected",
             session_id,
         )
-        return _to_chat_messages(visible_messages)
+        messages = _to_chat_messages(visible_messages, include_ids=True)
+        return VisibleHistoryState(
+            messages=messages,
+            visible_ids={msg.id for msg in visible_messages},
+            visible_tokens=estimate_messages_tokens(_strip_internal_message_ids(messages), config),
+        )
 
     logger.info(
         "Conversation context compression summarizing session %s: %d messages, candidate_tokens=%d, trigger=%d",
@@ -178,19 +322,34 @@ def build_visible_chat_history(
             getattr(response, "model", "") or get_effective_agent_primary_model(config) or "unknown",
             call_type="agent",
         )
-        return [build_summary_message(summary_text)] + _to_chat_messages(protected_tail)
+        messages = [build_summary_message(summary_text)] + _to_chat_messages(protected_tail, include_ids=True)
+        return VisibleHistoryState(
+            messages=messages,
+            visible_ids={msg.id for msg in protected_tail},
+            visible_tokens=estimate_messages_tokens(_strip_internal_message_ids(messages), config),
+        )
 
     logger.warning(
         "Conversation context compression failed for session %s; using state-table fallback",
         session_id,
     )
     if previous_summary:
-        return candidate
-    return db.get_conversation_history(session_id, limit=20)
+        return VisibleHistoryState(
+            messages=candidate,
+            visible_ids={msg.id for msg in visible_messages if msg.id > covered_message_id},
+            visible_tokens=candidate_tokens,
+        )
+    selected = visible_messages[-20:]
+    messages = _to_chat_messages(selected, include_ids=True)
+    return VisibleHistoryState(
+        messages=messages,
+        visible_ids={msg.id for msg in selected},
+        visible_tokens=estimate_messages_tokens(_strip_internal_message_ids(messages), config),
+    )
 
 
-def _load_visible_messages(session_id: str) -> List[VisibleMessage]:
-    rows = get_db().get_visible_conversation_messages(session_id)
+def _load_visible_messages(session_id: str, *, limit: Optional[int] = None) -> List[VisibleMessage]:
+    rows = get_db().get_visible_conversation_messages(session_id, limit=limit)
     messages = []
     for row in rows:
         role = str(row.get("role") or "")
@@ -253,8 +412,57 @@ def _generate_summary(
     return content, response
 
 
-def _to_chat_messages(messages: Iterable[VisibleMessage]) -> List[Dict[str, str]]:
-    return [{"role": msg.role, "content": msg.content} for msg in messages]
+def _to_chat_messages(
+    messages: Iterable[VisibleMessage],
+    *,
+    include_ids: bool = False,
+) -> List[Dict[str, Any]]:
+    result: List[Dict[str, Any]] = []
+    for msg in messages:
+        row: Dict[str, Any] = {"role": msg.role, "content": msg.content}
+        if include_ids:
+            row["_message_id"] = msg.id
+        result.append(row)
+    return result
+
+
+def _strip_internal_message_ids(messages: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    return [
+        {key: value for key, value in msg.items() if key != "_message_id"}
+        for msg in messages
+    ]
+
+
+def _restore_trace_metadata(
+    messages: Sequence[Any],
+    *,
+    provider: Any,
+    model: Any,
+) -> List[Dict[str, Any]]:
+    restored: List[Dict[str, Any]] = []
+    for msg in messages:
+        if not isinstance(msg, dict):
+            continue
+        clean = strip_trace_metadata(msg)
+        if clean.get("role") in {"assistant", "tool"}:
+            clean[TRACE_PROVIDER_KEY] = provider
+            clean[TRACE_MODEL_KEY] = model
+        restored.append(clean)
+    return restored
+
+
+def _build_trace_match_targets(
+    models: Sequence[str],
+    config: Any,
+) -> List[Tuple[str, str]]:
+    model_list = getattr(config, "llm_model_list", []) or []
+    targets: List[Tuple[str, str]] = []
+    for model in models:
+        normalized = str(model or "").strip()
+        if not normalized:
+            continue
+        targets.append((normalized, resolved_provider_namespace(normalized, model_list)))
+    return targets
 
 
 def _render_messages(messages: Sequence[Dict[str, Any]]) -> str:

src/agent/conversation.py

@@ -23,10 +23,11 @@ class ConversationSession:
     created_at: datetime = field(default_factory=datetime.now)
     last_active: datetime = field(default_factory=datetime.now)
 
-    def add_message(self, role: str, content: str):
+    def add_message(self, role: str, content: str) -> int:
         """Add a message to the session history."""
-        get_db().save_conversation_message(self.session_id, role, content)
+        message_id = get_db().save_conversation_message(self.session_id, role, content)
         self.last_active = datetime.now()
+        return message_id
 
     def update_context(self, key: str, value: Any):
         """Update session context."""
@@ -60,10 +61,10 @@ class ConversationManager:
 
             return self._sessions[session_id]
 
-    def add_message(self, session_id: str, role: str, content: str):
+    def add_message(self, session_id: str, role: str, content: str) -> int:
         """Add a message to a session."""
         session = self.get_or_create(session_id)
-        session.add_message(role, content)
+        return session.add_message(role, content)
 
     def get_history(self, session_id: str) -> List[Dict[str, Any]]:
         """Get message history for a session."""

src/agent/executor.py

@@ -16,13 +16,16 @@ same implementation.
 
 import json
 import logging
+import uuid
 from dataclasses import dataclass, field
 from typing import Any, Callable, Dict, List, Optional
 
 from src.config import get_config
-from src.agent.chat_context import build_visible_chat_history
+from src.agent.chat_context import build_agent_chat_context_bundle
 from src.agent.llm_adapter import LLMToolAdapter
+from src.agent.provider_trace import extract_provider_trace_turns
 from src.agent.runner import run_agent_loop, parse_dashboard_json
+from src.storage import get_db
 from src.agent.tools.registry import ToolRegistry
 from src.report_language import normalize_report_language
 from src.market_context import get_market_role, get_market_guidelines
@@ -47,6 +50,7 @@ class AgentResult:
     provider: str = ""
     model: str = ""                            # comma-separated models used (supports fallback)
     error: Optional[str] = None
+    messages: List[Dict[str, Any]] = field(default_factory=list)
 
 
 # ============================================================
@@ -560,13 +564,13 @@ class AgentExecutor:
         # Get conversation history
         conversation_manager.get_or_create(session_id)
         config = getattr(self.llm_adapter, "_config", None) or get_config()
-        history = build_visible_chat_history(session_id, self.llm_adapter, config)
+        bundle = build_agent_chat_context_bundle(session_id, self.llm_adapter, config)
 
         # Initialize conversation
         messages: List[Dict[str, Any]] = [
             {"role": "system", "content": system_prompt},
         ]
-        messages.extend(history)
+        messages.extend(bundle.context_messages)
 
         # Inject previous analysis context if provided (data reuse from report follow-up)
         if context:
@@ -593,21 +597,105 @@ class AgentExecutor:
                 messages.append({"role": "assistant", "content": "好的，我已了解该股票的历史分析数据。请告诉我你想了解什么？"})
 
         messages.append({"role": "user", "content": message})
+        baseline_len = len(messages)
+        run_id = str(uuid.uuid4())
 
         # Persist the user turn immediately so the session appears in history during processing
-        conversation_manager.add_message(session_id, "user", message)
+        user_message_id = conversation_manager.add_message(session_id, "user", message)
 
         result = self._run_loop(messages, tool_decls, parse_dashboard=False, progress_callback=progress_callback)
 
         # Persist assistant reply (or error note) for context continuity
         if result.success:
-            conversation_manager.add_message(session_id, "assistant", result.content)
+            assistant_message_id = conversation_manager.add_message(session_id, "assistant", result.content)
+            self._persist_provider_trace(
+                session_id=session_id,
+                run_id=run_id,
+                messages=result.messages,
+                baseline_len=baseline_len,
+                user_message_id=user_message_id,
+                assistant_message_id=assistant_message_id,
+            )
         else:
             error_note = f"[分析失败] {result.error or '未知错误'}"
             conversation_manager.add_message(session_id, "assistant", error_note)
 
         return result
 
+    def _persist_provider_trace(
+        self,
+        *,
+        session_id: str,
+        run_id: str,
+        messages: List[Dict[str, Any]],
+        baseline_len: int,
+        user_message_id: int,
+        assistant_message_id: int,
+    ) -> None:
+        try:
+            turns, diagnostics = extract_provider_trace_turns(
+                messages,
+                baseline_len=baseline_len,
+                run_id=run_id,
+                anchor_user_message_id=user_message_id,
+                anchor_assistant_message_id=assistant_message_id,
+            )
+        except Exception:
+            logger.warning(
+                "Provider trace extraction failed for session %s run %s",
+                session_id,
+                run_id,
+                exc_info=True,
+            )
+            return
+
+        if diagnostics.trace_dropped_reason:
+            logger.debug(
+                "Provider trace skipped for session %s run %s: %s",
+                session_id,
+                run_id,
+                diagnostics.trace_dropped_reason,
+            )
+        if not turns:
+            return
+
+        try:
+            db = get_db()
+        except Exception:
+            logger.warning(
+                "Provider trace storage unavailable for session %s run %s",
+                session_id,
+                run_id,
+                exc_info=True,
+            )
+            return
+
+        for turn in turns:
+            try:
+                db.save_agent_provider_turn(
+                    session_id=session_id,
+                    run_id=run_id,
+                    provider=turn.provider,
+                    model=turn.model,
+                    anchor_user_message_id=user_message_id,
+                    anchor_assistant_message_id=assistant_message_id,
+                    messages=turn.messages,
+                    contains_reasoning=turn.contains_reasoning,
+                    contains_tool_calls=turn.contains_tool_calls,
+                    contains_thinking_blocks=turn.contains_thinking_blocks,
+                    must_roundtrip=turn.must_roundtrip,
+                    estimated_tokens=turn.estimated_tokens,
+                )
+            except Exception:
+                logger.warning(
+                    "Provider trace persistence failed for session %s run %s provider=%s model=%s",
+                    session_id,
+                    run_id,
+                    turn.provider,
+                    turn.model,
+                    exc_info=True,
+                )
+
     def _run_loop(self, messages: List[Dict[str, Any]], tool_decls: List[Dict[str, Any]], parse_dashboard: bool, progress_callback: Optional[Callable] = None) -> AgentResult:
         """Delegate to the shared runner and adapt the result.
 
@@ -638,6 +726,7 @@ class AgentExecutor:
                 provider=loop_result.provider,
                 model=model_str,
                 error=None if dashboard else "Failed to parse dashboard JSON from agent response",
+                messages=loop_result.messages,
             )
 
         return AgentResult(
@@ -650,6 +739,7 @@ class AgentExecutor:
             provider=loop_result.provider,
             model=model_str,
             error=loop_result.error,
+            messages=loop_result.messages,
         )
 
     def _build_user_message(self, task: str, context: Optional[Dict[str, Any]] = None) -> str:

src/agent/llm_adapter.py

@@ -23,10 +23,15 @@ from src.config import (
     get_configured_llm_models,
     get_effective_agent_models_to_try,
     get_effective_agent_primary_model,
-    resolve_litellm_wire_model,
+)
+from src.agent.provider_trace import (
+    TRACE_MODEL_KEY,
+    TRACE_PROVIDER_KEY,
+    resolved_provider_namespace,
+    trace_model_matches,
 )
 from src.llm.errors import call_litellm_with_param_recovery
-from src.llm.generation_params import apply_litellm_generation_params
+from src.llm.generation_params import apply_litellm_generation_params, resolve_litellm_wire_model
 
 logger = logging.getLogger(__name__)
 
@@ -55,6 +60,7 @@ class ToolCall:
     name: str
     arguments: Dict[str, Any]
     thought_signature: Optional[str] = None
+    provider_specific_fields: Dict[str, Any] = field(default_factory=dict)
 
 
 @dataclass
@@ -63,6 +69,7 @@ class LLMResponse:
     content: Optional[str] = None          # text response (final answer)
     tool_calls: List[ToolCall] = field(default_factory=list)  # tool calls to execute
     reasoning_content: Optional[str] = None  # Chain-of-thought (CoT) from DeepSeek thinking mode; must be passed back in multi-turn assistant messages; None for other providers
+    provider_blocks: List[Dict[str, Any]] = field(default_factory=list)  # Opaque provider content blocks (e.g. Claude thinking/redacted_thinking)
     usage: Dict[str, Any] = field(default_factory=dict)       # token usage info
     provider: str = ""                     # which provider handled this call
     model: str = ""                        # full model name used (e.g. gemini/gemini-2.0-flash), for report meta
@@ -126,6 +133,87 @@ def _split_provider_model(model: str) -> Tuple[str, str]:
     return "openai", normalized
 
 
+def _object_to_dict(value: Any) -> Dict[str, Any]:
+    if isinstance(value, dict):
+        return dict(value)
+    if hasattr(value, "model_dump"):
+        try:
+            dumped = value.model_dump()
+            if isinstance(dumped, dict):
+                return dumped
+        except Exception:
+            pass
+    if hasattr(value, "dict"):
+        try:
+            dumped = value.dict()
+            if isinstance(dumped, dict):
+                return dumped
+        except Exception:
+            pass
+    result: Dict[str, Any] = {}
+    for key in ("type", "text", "content", "thinking", "signature", "data"):
+        if hasattr(value, key):
+            result[key] = getattr(value, key)
+    return result
+
+
+def _provider_specific_fields_from(value: Any) -> Dict[str, Any]:
+    if value is None:
+        return {}
+    if isinstance(value, dict):
+        return dict(value)
+    data = _object_to_dict(value)
+    return data if isinstance(data, dict) else {}
+
+
+def _extract_provider_blocks(choice: Any) -> Tuple[List[Dict[str, Any]], Optional[str]]:
+    """Return opaque provider blocks and joined text block content, if present."""
+    block_sources = []
+    message = getattr(choice, "message", None)
+    for owner in (message, choice):
+        if owner is None:
+            continue
+        for attr in ("content", "content_blocks", "provider_blocks", "thinking_blocks"):
+            value = getattr(owner, attr, None)
+            if isinstance(value, list):
+                block_sources.append(value)
+
+    blocks: List[Dict[str, Any]] = []
+    text_parts: List[str] = []
+    for source in block_sources:
+        for raw_block in source:
+            block = _object_to_dict(raw_block)
+            if not block:
+                continue
+            blocks.append(block)
+            block_type = str(block.get("type") or "")
+            text = block.get("text") or block.get("content")
+            if block_type == "text" and text:
+                text_parts.append(str(text))
+    return blocks, ("".join(text_parts).strip() or None)
+
+
+def _message_trace_matches_target(
+    message: Dict[str, Any],
+    target_model: Optional[str],
+    *,
+    target_provider: Optional[str] = None,
+) -> bool:
+    """Whether provider-specific fields in ``message`` can be sent to target."""
+    if not target_model:
+        return True
+    trace_provider = message.get(TRACE_PROVIDER_KEY)
+    trace_model = message.get(TRACE_MODEL_KEY)
+    if not trace_provider and not trace_model:
+        return True
+    return trace_model_matches(
+        trace_provider,
+        trace_model,
+        target_model,
+        current_provider=target_provider,
+    )
+
+
 def _model_matches(model: str, entries: List[str]) -> bool:
     """Check if model name matches any entry (exact or prefix with version suffix)."""
     if not model:
@@ -454,7 +542,7 @@ class LLMToolAdapter:
         timeout: Optional[float] = None,
     ) -> LLMResponse:
         """Call a specific litellm model with OpenAI-format messages and tools."""
-        openai_messages = self._convert_messages(messages)
+        openai_messages = self._convert_messages(messages, target_model=model)
 
         # Use short model name (without provider prefix) for thinking model lookup
         model_short = model.split("/")[-1] if "/" in model else model
@@ -536,10 +624,23 @@ class LLMToolAdapter:
         """Return the raw configured temperature before per-model normalization."""
         return float(self._config.llm_temperature)
 
-    def _convert_messages(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    def _convert_messages(
+        self,
+        messages: List[Dict[str, Any]],
+        *,
+        target_model: Optional[str] = None,
+    ) -> List[Dict[str, Any]]:
         """Convert internal message format to OpenAI-compatible format for litellm."""
         openai_messages: List[Dict[str, Any]] = []
+        target_provider = self._trace_provider_for_target(target_model)
         for msg in messages:
+            trace_matches_target = _message_trace_matches_target(
+                msg,
+                target_model,
+                target_provider=target_provider,
+            )
+            if not trace_matches_target:
+                continue
             if msg["role"] == "tool":
                 openai_messages.append({
                     "role": "tool",
@@ -557,13 +658,21 @@ class LLMToolAdapter:
                             "arguments": json.dumps(tc["arguments"]),
                         },
                     }
+                    provider_specific_fields = dict(tc.get("provider_specific_fields") or {})
                     sig = tc.get("thought_signature")
                     if sig is not None:
-                        tc_dict["provider_specific_fields"] = {"thought_signature": sig}
+                        provider_specific_fields.setdefault("thought_signature", sig)
+                    if provider_specific_fields:
+                        tc_dict["provider_specific_fields"] = provider_specific_fields
                     openai_tc.append(tc_dict)
+                content = (
+                    msg.get("provider_blocks")
+                    if msg.get("provider_blocks")
+                    else msg.get("content")
+                )
                 openai_msg: Dict[str, Any] = {
                     "role": "assistant",
-                    "content": msg.get("content"),
+                    "content": content,
                     "tool_calls": openai_tc,
                 }
                 if msg.get("reasoning_content") is not None:
@@ -576,35 +685,28 @@ class LLMToolAdapter:
                 })
         return openai_messages
 
+    def _trace_provider_for_target(self, target_model: Optional[str]) -> str:
+        if not target_model:
+            return ""
+        model_list = getattr(getattr(self, "_config", None), "llm_model_list", []) or []
+        return resolved_provider_namespace(target_model, model_list)
+
     def _parse_litellm_response(self, response: Any, model: str) -> LLMResponse:
         """Parse litellm OpenAI-compatible response into LLMResponse."""
         choice = response.choices[0]
         tool_calls: List[ToolCall] = []
 
+        provider_blocks, provider_text = _extract_provider_blocks(choice)
+
         # Handle MiniMax-specific content_blocks format
         # MiniMax-M2.7 may return content_blocks at choice level or inside message
         # Check both possible locations for content_blocks to ensure consistency
         # Concatenate ALL text blocks to avoid truncating multi-block responses
         text_content = choice.message.content
+        if isinstance(text_content, list):
+            text_content = provider_text
         if text_content is None:
-            content_blocks = None
-            if hasattr(choice, "content_blocks"):
-                content_blocks = choice.content_blocks
-            elif hasattr(choice.message, "content_blocks"):
-                content_blocks = choice.message.content_blocks
-
-            if content_blocks:
-                # MiniMax response format: content_blocks[].text
-                # Concatenate ALL text blocks to preserve complete response
-                text_parts = []
-                for block in content_blocks:
-                    if getattr(block, "type", None) == "text":
-                        text = getattr(block, "text", "") or ""
-                        if text:
-                            text_parts.append(text)
-                    elif hasattr(block, "content") and block.content:
-                        text_parts.append(block.content)
-                text_content = "".join(text_parts).strip()
+            text_content = provider_text
 
         # DeepSeek/Qwen thinking mode; not in standard OpenAI type, accessed via getattr
         reasoning_content = getattr(choice.message, "reasoning_content", None)
@@ -618,22 +720,24 @@ class LLMToolAdapter:
                     except json.JSONDecodeError:
                         args = {"raw": tc.function.arguments}
 
-                # Extract thought_signature: stored in provider_specific_fields (Gemini 3 via LiteLLM proxy)
-                psf = getattr(tc, "provider_specific_fields", None)
-                if psf is not None:
-                    sig = psf.get("thought_signature") if isinstance(psf, dict) else getattr(psf, "thought_signature", None)
-                else:
-                    func_psf = getattr(tc.function, "provider_specific_fields", None)
-                    if func_psf is not None:
-                        sig = func_psf.get("thought_signature") if isinstance(func_psf, dict) else getattr(func_psf, "thought_signature", None)
-                    else:
-                        sig = getattr(tc, "thought_signature", None)
+                provider_specific_fields = _provider_specific_fields_from(
+                    getattr(tc, "provider_specific_fields", None)
+                )
+                provider_specific_fields.update(
+                    _provider_specific_fields_from(
+                        getattr(tc.function, "provider_specific_fields", None)
+                    )
+                )
+                sig = provider_specific_fields.get("thought_signature")
+                if sig is None:
+                    sig = getattr(tc, "thought_signature", None)
 
                 tool_calls.append(ToolCall(
                     id=tc.id,
                     name=tc.function.name,
                     arguments=args,
                     thought_signature=sig,
+                    provider_specific_fields=provider_specific_fields,
                 ))
 
         usage: Dict[str, Any] = {}
@@ -644,11 +748,13 @@ class LLMToolAdapter:
                 "total_tokens": response.usage.total_tokens,
             }
 
-        provider_name = model.split("/")[0] if "/" in model else model
+        model_list = getattr(getattr(self, "_config", None), "llm_model_list", []) or []
+        provider_name = resolved_provider_namespace(model, model_list)
         return LLMResponse(
             content=text_content,
             tool_calls=tool_calls,
             reasoning_content=reasoning_content,
+            provider_blocks=provider_blocks,
             usage=usage,
             provider=provider_name,
             model=model,

src/agent/provider_trace.py

@@ -0,0 +1,267 @@
+# -*- coding: utf-8 -*-
+"""Provider-specific protocol trace helpers for Agent chat.
+
+These helpers keep opaque thinking/tool-call protocol material on a separate
+track from user-visible conversation history.  The persisted payload is the
+minimal provider protocol slice required for roundtrip:
+
+    assistant(tool_calls + reasoning/thinking metadata) -> tool ...
+
+Final assistant text is intentionally excluded because it is already stored in
+``conversation_messages`` and merged back by id anchor.
+"""
+
+from __future__ import annotations
+
+import json
+import math
+from dataclasses import dataclass, field
+from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
+
+from src.llm.generation_params import resolve_litellm_wire_model
+
+
+PROVIDER_TRACE_RETENTION_LIMIT = 3
+TRACE_PROVIDER_KEY = "_trace_provider"
+TRACE_MODEL_KEY = "_trace_model"
+
+THINKING_BLOCK_TYPES = {"thinking", "redacted_thinking", "signature"}
+
+
+@dataclass
+class ProviderTraceTurn:
+    """One persisted provider protocol turn for a completed chat run."""
+
+    session_id: str = ""
+    run_id: str = ""
+    provider: str = ""
+    model: str = ""
+    anchor_user_message_id: int = 0
+    anchor_assistant_message_id: int = 0
+    messages: List[Dict[str, Any]] = field(default_factory=list)
+    contains_reasoning: bool = False
+    contains_tool_calls: bool = False
+    contains_thinking_blocks: bool = False
+    must_roundtrip: bool = False
+    estimated_tokens: int = 0
+
+
+@dataclass
+class TraceDiagnostics:
+    """Structured trace context diagnostics for tests and server logs."""
+
+    trace_injected: bool = False
+    trace_dropped_reason: str = ""
+    trace_tokens: int = 0
+    visible_tokens: int = 0
+    retention_trimmed_count: int = 0
+    dropped_trace_count: int = 0
+    mixed_model_trace: bool = False
+    model_mismatch: int = 0
+    anchor_summarized: int = 0
+    budget_exceeded: bool = False
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "trace_injected": self.trace_injected,
+            "trace_dropped_reason": self.trace_dropped_reason,
+            "trace_tokens": self.trace_tokens,
+            "visible_tokens": self.visible_tokens,
+            "retention_trimmed_count": self.retention_trimmed_count,
+            "dropped_trace_count": self.dropped_trace_count,
+            "mixed_model_trace": self.mixed_model_trace,
+            "model_mismatch": self.model_mismatch,
+            "anchor_summarized": self.anchor_summarized,
+            "budget_exceeded": self.budget_exceeded,
+        }
+
+
+def normalize_model_name(model: Any) -> str:
+    """Normalize a model string for exact trace compatibility checks."""
+    return str(model or "").strip().lower()
+
+
+def provider_namespace(model: Any) -> str:
+    """Return the provider namespace used by LiteLLM-style model strings."""
+    normalized = normalize_model_name(model)
+    if not normalized:
+        return ""
+    if "/" in normalized:
+        return normalized.split("/", 1)[0]
+    return "openai"
+
+
+def resolved_provider_namespace(
+    model: Any,
+    model_list: Optional[Sequence[Dict[str, Any]]] = None,
+) -> str:
+    """Resolve router aliases before deriving the provider namespace."""
+    normalized = str(model or "").strip()
+    if not normalized:
+        return ""
+    wire_model = resolve_litellm_wire_model(normalized, list(model_list or []))
+    return provider_namespace(wire_model)
+
+
+def trace_model_matches(
+    trace_provider: Any,
+    trace_model: Any,
+    current_model: Any,
+    *,
+    current_provider: Any = None,
+) -> bool:
+    """Return True only when provider namespace and full model string match."""
+    trace_model_normalized = normalize_model_name(trace_model)
+    current_model_normalized = normalize_model_name(current_model)
+    if not trace_model_normalized or not current_model_normalized:
+        return False
+    if trace_model_normalized != current_model_normalized:
+        return False
+    provider_normalized = normalize_model_name(trace_provider)
+    expected_provider = normalize_model_name(current_provider) or provider_namespace(current_model_normalized)
+    return provider_normalized == expected_provider
+
+
+def estimate_protocol_tokens(messages: Sequence[Dict[str, Any]]) -> int:
+    """Cheap deterministic estimate used only for trace budget decisions."""
+    payload = json.dumps(messages, ensure_ascii=False, default=str)
+    return int(math.ceil(len(payload) / 3))
+
+
+def extract_provider_trace_turns(
+    messages: Sequence[Dict[str, Any]],
+    *,
+    baseline_len: int,
+    run_id: str = "",
+    anchor_user_message_id: int = 0,
+    anchor_assistant_message_id: int = 0,
+) -> Tuple[List[ProviderTraceTurn], TraceDiagnostics]:
+    """Extract this run's provider trace from ``messages[baseline_len:]``.
+
+    Only the current run's appended tool-loop protocol messages are considered.
+    Existing traces injected into the request live before ``baseline_len`` and
+    are therefore not persisted again.
+    """
+    diagnostics = TraceDiagnostics()
+    protocol_messages: List[Dict[str, Any]] = []
+    providers: set[str] = set()
+    models: set[str] = set()
+    contains_reasoning = False
+    contains_tool_calls = False
+    contains_thinking_blocks = False
+    contains_provider_specific_fields = False
+
+    for raw_msg in list(messages)[max(0, int(baseline_len)) :]:
+        role = raw_msg.get("role")
+        if role == "assistant" and raw_msg.get("tool_calls"):
+            provider = normalize_model_name(raw_msg.get(TRACE_PROVIDER_KEY))
+            model = normalize_model_name(raw_msg.get(TRACE_MODEL_KEY))
+            if provider:
+                providers.add(provider)
+            if model:
+                models.add(model)
+
+            contains_tool_calls = True
+            contains_reasoning = contains_reasoning or raw_msg.get("reasoning_content") is not None
+            contains_thinking_blocks = contains_thinking_blocks or message_contains_thinking_blocks(raw_msg)
+            contains_provider_specific_fields = (
+                contains_provider_specific_fields
+                or _tool_calls_have_provider_specific_fields(raw_msg.get("tool_calls") or [])
+            )
+            protocol_messages.append(strip_trace_metadata(raw_msg))
+            continue
+
+        if role == "tool" and protocol_messages:
+            protocol_messages.append(strip_trace_metadata(raw_msg))
+
+    if not protocol_messages or not contains_tool_calls:
+        return [], diagnostics
+
+    if len(providers) != 1 or len(models) != 1:
+        diagnostics.mixed_model_trace = True
+        diagnostics.trace_dropped_reason = "mixed_model_trace"
+        diagnostics.dropped_trace_count = 1
+        return [], diagnostics
+
+    provider = next(iter(providers))
+    model = next(iter(models))
+    if provider == "deepseek":
+        must_roundtrip = contains_tool_calls and contains_reasoning
+    elif provider == "anthropic":
+        must_roundtrip = contains_tool_calls and contains_thinking_blocks
+    else:
+        must_roundtrip = contains_tool_calls and (
+            contains_reasoning or contains_thinking_blocks or contains_provider_specific_fields
+        )
+    if not must_roundtrip:
+        diagnostics.trace_dropped_reason = "not_required"
+        diagnostics.dropped_trace_count = 1
+        return [], diagnostics
+
+    trace = ProviderTraceTurn(
+        run_id=run_id,
+        provider=provider,
+        model=model,
+        anchor_user_message_id=int(anchor_user_message_id or 0),
+        anchor_assistant_message_id=int(anchor_assistant_message_id or 0),
+        messages=protocol_messages,
+        contains_reasoning=contains_reasoning,
+        contains_tool_calls=contains_tool_calls,
+        contains_thinking_blocks=contains_thinking_blocks,
+        must_roundtrip=True,
+        estimated_tokens=estimate_protocol_tokens(protocol_messages),
+    )
+    diagnostics.trace_tokens = trace.estimated_tokens
+    return [trace], diagnostics
+
+
+def strip_trace_metadata(message: Dict[str, Any]) -> Dict[str, Any]:
+    """Return a JSON-safe message without internal trace routing metadata."""
+    return {
+        key: _strip_trace_metadata_value(value)
+        for key, value in message.items()
+        if not str(key).startswith("_trace_")
+    }
+
+
+def message_contains_thinking_blocks(message: Dict[str, Any]) -> bool:
+    """Detect Claude/Gemini opaque thinking blocks in known message locations."""
+    candidates: List[Any] = []
+    for key in ("provider_blocks", "content", "thinking_blocks"):
+        if key in message:
+            candidates.append(message.get(key))
+    return any(_contains_thinking_block(candidate) for candidate in candidates)
+
+
+def _contains_thinking_block(value: Any) -> bool:
+    if isinstance(value, dict):
+        block_type = str(value.get("type") or "").strip()
+        if block_type in THINKING_BLOCK_TYPES:
+            return True
+        return any(_contains_thinking_block(v) for v in value.values())
+    if isinstance(value, list):
+        return any(_contains_thinking_block(item) for item in value)
+    return False
+
+
+def _tool_calls_have_provider_specific_fields(tool_calls: Iterable[Dict[str, Any]]) -> bool:
+    for tool_call in tool_calls:
+        if not isinstance(tool_call, dict):
+            continue
+        if tool_call.get("provider_specific_fields"):
+            return True
+        if tool_call.get("thought_signature") is not None:
+            return True
+    return False
+
+
+def _strip_trace_metadata_value(value: Any) -> Any:
+    if isinstance(value, dict):
+        return {
+            key: _strip_trace_metadata_value(child)
+            for key, child in value.items()
+            if not str(key).startswith("_trace_")
+        }
+    if isinstance(value, list):
+        return [_strip_trace_metadata_value(item) for item in value]
+    return value

src/agent/runner.py

@@ -502,11 +502,14 @@ def run_agent_loop(
             assistant_msg: Dict[str, Any] = {
                 "role": "assistant",
                 "content": response.content,
+                "_trace_provider": response.provider,
+                "_trace_model": m,
                 "tool_calls": [
                     {
                         "id": tc.id,
                         "name": tc.name,
                         "arguments": tc.arguments,
+                        **({"provider_specific_fields": tc.provider_specific_fields} if tc.provider_specific_fields else {}),
                         **({"thought_signature": tc.thought_signature} if tc.thought_signature is not None else {}),
                     }
                     for tc in response.tool_calls
@@ -514,6 +517,8 @@ def run_agent_loop(
             }
             if response.reasoning_content is not None:
                 assistant_msg["reasoning_content"] = response.reasoning_content
+            if response.provider_blocks:
+                assistant_msg["provider_blocks"] = response.provider_blocks
             messages.append(assistant_msg)
 
             # Execute tools (parallel when > 1)

src/storage.py

@@ -52,6 +52,7 @@ from sqlalchemy.orm import (
 )
 from sqlalchemy.exc import IntegrityError, OperationalError
 
+from src.agent.provider_trace import PROVIDER_TRACE_RETENTION_LIMIT
 from src.config import get_config
 
 logger = logging.getLogger(__name__)
@@ -624,6 +625,31 @@ class ConversationSummary(Base):
     updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now, index=True)
 
 
+class AgentProviderTurn(Base):
+    """Provider protocol trace required for thinking/tool-call roundtrip."""
+
+    __tablename__ = 'agent_provider_turns'
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    session_id = Column(String(100), nullable=False, index=True)
+    run_id = Column(String(64), nullable=False, index=True)
+    provider = Column(String(64), nullable=False, index=True)
+    model = Column(String(160), nullable=False, index=True)
+    anchor_user_message_id = Column(Integer, nullable=False, index=True)
+    anchor_assistant_message_id = Column(Integer, nullable=False, index=True)
+    messages_json = Column(Text, nullable=False)
+    contains_reasoning = Column(Boolean, nullable=False, default=False)
+    contains_tool_calls = Column(Boolean, nullable=False, default=False)
+    contains_thinking_blocks = Column(Boolean, nullable=False, default=False)
+    must_roundtrip = Column(Boolean, nullable=False, default=False, index=True)
+    estimated_tokens = Column(Integer, nullable=False, default=0)
+    created_at = Column(DateTime, default=datetime.now, index=True)
+
+    __table_args__ = (
+        Index('ix_agent_provider_turn_bucket', 'session_id', 'provider', 'model', 'must_roundtrip'),
+    )
+
+
 class LLMUsage(Base):
     """One row per litellm.completion() call — token-usage audit log."""
 
@@ -2089,7 +2115,7 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
         digest = hashlib.md5(raw_key.encode("utf-8")).hexdigest()
         return f"no-url:{code}:{digest}"
 
-    def save_conversation_message(self, session_id: str, role: str, content: str) -> None:
+    def save_conversation_message(self, session_id: str, role: str, content: str) -> int:
         """
         保存 Agent 对话消息
         """
@@ -2100,6 +2126,8 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
                 content=content
             )
             session.add(msg)
+            session.flush()
+            return int(msg.id)
 
     def get_conversation_history(self, session_id: str, limit: int = 20) -> List[Dict[str, Any]]:
         """
@@ -2114,7 +2142,7 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
             # 倒序返回，保证时间顺序
             return [{"role": msg.role, "content": msg.content} for msg in reversed(messages)]
 
-    def get_visible_conversation_messages(self, session_id: str) -> List[Dict[str, Any]]:
+    def get_visible_conversation_messages(self, session_id: str, limit: Optional[int] = None) -> List[Dict[str, Any]]:
         """Return visible user/assistant conversation messages in chronological order."""
         with self.session_scope() as session:
             stmt = (
@@ -2127,7 +2155,15 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
                 )
                 .order_by(ConversationMessage.created_at, ConversationMessage.id)
             )
+            if limit is not None:
+                stmt = (
+                    stmt.order_by(None)
+                    .order_by(ConversationMessage.created_at.desc(), ConversationMessage.id.desc())
+                    .limit(limit)
+                )
             messages = session.execute(stmt).scalars().all()
+            if limit is not None:
+                messages = list(reversed(messages))
             return [
                 {
                     "id": msg.id,
@@ -2159,6 +2195,135 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
                 "updated_at": row.updated_at,
             }
 
+    def save_agent_provider_turn(
+        self,
+        *,
+        session_id: str,
+        run_id: str,
+        provider: str,
+        model: str,
+        anchor_user_message_id: int,
+        anchor_assistant_message_id: int,
+        messages: List[Dict[str, Any]],
+        contains_reasoning: bool,
+        contains_tool_calls: bool,
+        contains_thinking_blocks: bool,
+        must_roundtrip: bool,
+        estimated_tokens: int,
+    ) -> int:
+        """Persist one provider protocol trace and enforce per-model retention."""
+        with self.session_scope() as session:
+            row = AgentProviderTurn(
+                session_id=session_id,
+                run_id=run_id,
+                provider=provider,
+                model=model,
+                anchor_user_message_id=int(anchor_user_message_id or 0),
+                anchor_assistant_message_id=int(anchor_assistant_message_id or 0),
+                messages_json=json.dumps(messages or [], ensure_ascii=False, default=str),
+                contains_reasoning=bool(contains_reasoning),
+                contains_tool_calls=bool(contains_tool_calls),
+                contains_thinking_blocks=bool(contains_thinking_blocks),
+                must_roundtrip=bool(must_roundtrip),
+                estimated_tokens=int(estimated_tokens or 0),
+            )
+            session.add(row)
+            session.flush()
+            row_id = int(row.id)
+            if row.must_roundtrip:
+                self._trim_agent_provider_turns(
+                    session=session,
+                    session_id=session_id,
+                    provider=provider,
+                    model=model,
+                    keep=PROVIDER_TRACE_RETENTION_LIMIT,
+                )
+            return row_id
+
+    def get_agent_provider_turns(
+        self,
+        session_id: str,
+        *,
+        provider: Optional[str] = None,
+        model: Optional[str] = None,
+        must_roundtrip_only: bool = True,
+    ) -> List[Dict[str, Any]]:
+        """Return provider trace turns in chronological order."""
+        with self.session_scope() as session:
+            conditions = [AgentProviderTurn.session_id == session_id]
+            if provider:
+                conditions.append(AgentProviderTurn.provider == provider)
+            if model:
+                conditions.append(AgentProviderTurn.model == model)
+            if must_roundtrip_only:
+                conditions.append(AgentProviderTurn.must_roundtrip.is_(True))
+            stmt = (
+                select(AgentProviderTurn)
+                .where(and_(*conditions))
+                .order_by(AgentProviderTurn.created_at, AgentProviderTurn.id)
+            )
+            rows = session.execute(stmt).scalars().all()
+            result = []
+            for row in rows:
+                try:
+                    messages = json.loads(row.messages_json or "[]")
+                except json.JSONDecodeError as exc:
+                    logger.warning(
+                        "Invalid provider trace messages_json skipped for session %s turn %s: %s",
+                        row.session_id,
+                        row.id,
+                        exc,
+                    )
+                    messages = []
+                result.append({
+                    "id": row.id,
+                    "session_id": row.session_id,
+                    "run_id": row.run_id,
+                    "provider": row.provider,
+                    "model": row.model,
+                    "anchor_user_message_id": row.anchor_user_message_id,
+                    "anchor_assistant_message_id": row.anchor_assistant_message_id,
+                    "messages": messages if isinstance(messages, list) else [],
+                    "messages_json": row.messages_json,
+                    "contains_reasoning": row.contains_reasoning,
+                    "contains_tool_calls": row.contains_tool_calls,
+                    "contains_thinking_blocks": row.contains_thinking_blocks,
+                    "must_roundtrip": row.must_roundtrip,
+                    "estimated_tokens": row.estimated_tokens,
+                    "created_at": row.created_at,
+                })
+            return result
+
+    def _trim_agent_provider_turns(
+        self,
+        *,
+        session: Session,
+        session_id: str,
+        provider: str,
+        model: str,
+        keep: int,
+    ) -> int:
+        old_ids_stmt = (
+            select(AgentProviderTurn.id)
+            .where(
+                and_(
+                    AgentProviderTurn.session_id == session_id,
+                    AgentProviderTurn.provider == provider,
+                    AgentProviderTurn.model == model,
+                    AgentProviderTurn.must_roundtrip.is_(True),
+                )
+            )
+            .order_by(AgentProviderTurn.created_at.desc(), AgentProviderTurn.id.desc())
+            .offset(max(0, int(keep)))
+        )
+        old_ids = list(session.execute(old_ids_stmt).scalars().all())
+        if not old_ids:
+            return 0
+        result = session.execute(
+            delete(AgentProviderTurn).where(AgentProviderTurn.id.in_(old_ids))
+        )
+        return int(result.rowcount or 0)
+
     def upsert_conversation_summary(
         self,
         session_id: str,
@@ -2304,6 +2469,11 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
             删除的消息数
         """
         with self.session_scope() as session:
+            session.execute(
+                delete(AgentProviderTurn).where(
+                    AgentProviderTurn.session_id == session_id
+                )
+            )
             session.execute(
                 delete(ConversationSummary).where(
                     ConversationSummary.session_id == session_id

tests/test_agent_chat_api.py

@@ -0,0 +1,62 @@
+# -*- coding: utf-8 -*-
+"""Agent chat history API regressions."""
+
+from pathlib import Path
+from unittest.mock import patch
+
+from fastapi.testclient import TestClient
+
+from api.app import create_app
+from src.config import Config
+from src.storage import DatabaseManager
+
+
+def teardown_function() -> None:
+    DatabaseManager.reset_instance()
+    Config.reset_instance()
+
+
+def test_chat_session_messages_api_does_not_expose_provider_trace(tmp_path: Path) -> None:
+    DatabaseManager.reset_instance()
+    Config.reset_instance()
+    db = DatabaseManager(db_url=f"sqlite:///{tmp_path / 'trace.db'}")
+    session_id = "api-trace-hidden"
+    user_id = db.save_conversation_message(session_id, "user", "visible question")
+    assistant_id = db.save_conversation_message(session_id, "assistant", "visible answer")
+    db.save_agent_provider_turn(
+        session_id=session_id,
+        run_id="run-hidden",
+        provider="deepseek",
+        model="deepseek/deepseek-chat",
+        anchor_user_message_id=user_id,
+        anchor_assistant_message_id=assistant_id,
+        messages=[
+            {
+                "role": "assistant",
+                "content": "checking",
+                "reasoning_content": "SECRET_REASONING",
+                "tool_calls": [{"id": "call_1", "name": "echo", "arguments": {}}],
+            },
+            {"role": "tool", "tool_call_id": "call_1", "content": "SECRET_TOOL_RESULT"},
+        ],
+        contains_reasoning=True,
+        contains_tool_calls=True,
+        contains_thinking_blocks=False,
+        must_roundtrip=True,
+        estimated_tokens=10,
+    )
+
+    with patch("api.middlewares.auth.is_auth_enabled", return_value=False):
+        client = TestClient(create_app(static_dir=tmp_path / "static"))
+        response = client.get(f"/api/v1/agent/chat/sessions/{session_id}")
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["session_id"] == session_id
+    assert [(msg["role"], msg["content"]) for msg in payload["messages"]] == [
+        ("user", "visible question"),
+        ("assistant", "visible answer"),
+    ]
+    assert "SECRET_REASONING" not in response.text
+    assert "SECRET_TOOL_RESULT" not in response.text
+    assert "tool_calls" not in response.text

tests/test_agent_executor.py

@@ -17,6 +17,7 @@ import unittest
 import sys
 import os
 from dataclasses import dataclass
+from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
@@ -31,6 +32,8 @@ from src.agent.executor import AgentExecutor, AgentResult
 from src.agent.llm_adapter import LLMResponse, ToolCall
 from src.agent.runner import parse_dashboard_json, run_agent_loop, serialize_tool_result
 from src.agent.tools.registry import ToolRegistry, ToolDefinition, ToolParameter
+from src.config import Config
+from src.storage import DatabaseManager
 
 
 # ============================================================
@@ -104,7 +107,10 @@ class TestAgentExecutor(unittest.TestCase):
         ]
 
         with patch.object(executor, "_run_loop", side_effect=fake_run_loop):
-            with patch("src.agent.executor.build_visible_chat_history", return_value=compressed_history):
+            with patch(
+                "src.agent.executor.build_agent_chat_context_bundle",
+                return_value=SimpleNamespace(context_messages=compressed_history, diagnostics={}),
+            ):
                 with patch("src.agent.conversation.conversation_manager.get_or_create"):
                     with patch("src.agent.conversation.conversation_manager.add_message"):
                         executor.chat(
@@ -235,6 +241,149 @@ class TestAgentExecutor(unittest.TestCase):
         self.assertEqual(result.tool_calls_log[0]["tool"], "echo")
         self.assertTrue(result.tool_calls_log[0]["success"])
 
+    def test_run_agent_loop_replays_reasoning_and_provider_specific_fields_on_followup_call(self):
+        registry = _make_registry_with_echo()
+        adapter = _make_mock_adapter()
+        adapter.call_with_tools.side_effect = [
+            LLMResponse(
+                content="Checking.",
+                tool_calls=[
+                    ToolCall(
+                        id="call_reason",
+                        name="echo",
+                        arguments={"message": "hello"},
+                        thought_signature="sig-1",
+                        provider_specific_fields={"thought_signature": "sig-1", "extra": "keep"},
+                    )
+                ],
+                reasoning_content="deepseek reasoning",
+                usage={"total_tokens": 10},
+                provider="deepseek",
+                model="deepseek/deepseek-chat",
+            ),
+            LLMResponse(
+                content=json.dumps(SAMPLE_DASHBOARD, ensure_ascii=False),
+                tool_calls=[],
+                usage={"total_tokens": 20},
+                provider="deepseek",
+                model="deepseek/deepseek-chat",
+            ),
+        ]
+
+        result = run_agent_loop(
+            messages=[{"role": "user", "content": "Analyze"}],
+            tool_registry=registry,
+            llm_adapter=adapter,
+            max_steps=2,
+        )
+
+        self.assertTrue(result.success)
+        followup_messages = adapter.call_with_tools.call_args_list[1].args[0]
+        assistant_msg = followup_messages[-2]
+        tool_msg = followup_messages[-1]
+        self.assertEqual(assistant_msg["role"], "assistant")
+        self.assertEqual(assistant_msg["reasoning_content"], "deepseek reasoning")
+        self.assertEqual(assistant_msg["_trace_provider"], "deepseek")
+        self.assertEqual(assistant_msg["_trace_model"], "deepseek/deepseek-chat")
+        self.assertEqual(
+            assistant_msg["tool_calls"][0]["provider_specific_fields"],
+            {"thought_signature": "sig-1", "extra": "keep"},
+        )
+        self.assertEqual(assistant_msg["tool_calls"][0]["thought_signature"], "sig-1")
+        self.assertEqual(tool_msg["role"], "tool")
+        self.assertEqual(tool_msg["tool_call_id"], "call_reason")
+
+    def test_chat_persists_single_provider_trace_and_reinjects_without_duplication(self):
+        DatabaseManager.reset_instance()
+        Config.reset_instance()
+        db = DatabaseManager(db_url="sqlite:///:memory:")
+        registry = _make_registry_with_echo()
+        adapter = _make_mock_adapter()
+        adapter._config = SimpleNamespace(
+            agent_context_compression_enabled=False,
+            agent_context_compression_profile="balanced",
+            agent_context_compression_trigger_tokens=999999,
+            agent_context_protected_turns=1,
+            llm_model_list=[],
+            agent_litellm_model="deepseek/deepseek-chat",
+            litellm_model="deepseek/deepseek-chat",
+            litellm_fallback_models=[],
+        )
+        adapter.call_with_tools.side_effect = [
+            LLMResponse(
+                content="Checking.",
+                tool_calls=[ToolCall(id="call_1", name="echo", arguments={"message": "first"})],
+                reasoning_content="r1",
+                usage={"total_tokens": 10},
+                provider="deepseek",
+                model="deepseek/deepseek-chat",
+            ),
+            LLMResponse(
+                content="first final",
+                tool_calls=[],
+                usage={"total_tokens": 5},
+                provider="deepseek",
+                model="deepseek/deepseek-chat",
+            ),
+            LLMResponse(
+                content="second final",
+                tool_calls=[],
+                usage={"total_tokens": 5},
+                provider="deepseek",
+                model="deepseek/deepseek-chat",
+            ),
+        ]
+
+        executor = AgentExecutor(registry, adapter, max_steps=3)
+
+        first = executor.chat("first question", "executor-trace")
+        second = executor.chat("second question", "executor-trace")
+
+        self.assertTrue(first.success)
+        self.assertTrue(second.success)
+        self.assertEqual(len(db.get_agent_provider_turns("executor-trace")), 1)
+        second_request_messages = adapter.call_with_tools.call_args_list[2].args[0]
+        ordered_roles = [msg["role"] for msg in second_request_messages[-5:]]
+        self.assertEqual(ordered_roles, ["user", "assistant", "tool", "assistant", "user"])
+        self.assertEqual(second_request_messages[-4]["reasoning_content"], "r1")
+        self.assertEqual(second_request_messages[-3]["tool_call_id"], "call_1")
+        self.assertEqual(second_request_messages[-2]["content"], "first final")
+        self.assertEqual(second_request_messages[-1]["content"], "second question")
+
+        DatabaseManager.reset_instance()
+        Config.reset_instance()
+
+    def test_persist_provider_trace_logs_save_failure_without_failing_chat(self):
+        registry = _make_registry_with_echo()
+        adapter = _make_mock_adapter()
+        executor = AgentExecutor(registry, adapter, max_steps=2)
+        messages = [
+            {"role": "user", "content": "question"},
+            {
+                "role": "assistant",
+                "content": "checking",
+                "_trace_provider": "deepseek",
+                "_trace_model": "deepseek/deepseek-chat",
+                "reasoning_content": "r1",
+                "tool_calls": [{"id": "call_1", "name": "echo", "arguments": {"message": "x"}}],
+            },
+            {"role": "tool", "tool_call_id": "call_1", "content": "tool-result"},
+        ]
+        db = SimpleNamespace(save_agent_provider_turn=MagicMock(side_effect=RuntimeError("db down")))
+
+        with patch("src.agent.executor.get_db", return_value=db):
+            with self.assertLogs("src.agent.executor", level="WARNING") as logs:
+                executor._persist_provider_trace(
+                    session_id="executor-trace-fail-open",
+                    run_id="run-1",
+                    messages=messages,
+                    baseline_len=1,
+                    user_message_id=10,
+                    assistant_message_id=11,
+                )
+
+        self.assertIn("Provider trace persistence failed", "\n".join(logs.output))
+
     def test_multiple_tool_calls_in_one_step(self):
         """Agent requests multiple tool calls in a single response."""
         registry = _make_registry_with_echo()

tests/test_chat_context.py

@@ -11,9 +11,11 @@ from src.agent.chat_context import (  # noqa: E402
     SUMMARY_USER_PREFIX,
     VisibleMessage,
     _split_protected_tail,
+    build_agent_chat_context_bundle,
     build_visible_chat_history,
     estimate_text_tokens,
 )
+from src.agent.llm_adapter import LLMToolAdapter  # noqa: E402
 from src.config import Config  # noqa: E402
 from src.storage import DatabaseManager  # noqa: E402
 
@@ -99,6 +101,263 @@ def test_existing_summary_under_trigger_returns_summary_and_uncovered_messages()
     assert [msg["content"] for msg in history[1:]] == ["u2", "a2"]
 
 
+def test_bundle_splices_provider_trace_before_visible_final_assistant() -> None:
+    db = _reset_db()
+    session_id = "chat-trace-splice"
+    user_id = db.save_conversation_message(session_id, "user", "u1")
+    assistant_id = db.save_conversation_message(session_id, "assistant", "a1-final")
+    db.save_agent_provider_turn(
+        session_id=session_id,
+        run_id="run-1",
+        provider="openai",
+        model="openai/test-model",
+        anchor_user_message_id=user_id,
+        anchor_assistant_message_id=assistant_id,
+        messages=[
+            {
+                "role": "assistant",
+                "content": "checking",
+                "reasoning_content": "r1",
+                "tool_calls": [{"id": "call_1", "name": "echo", "arguments": {"message": "x"}}],
+            },
+            {"role": "tool", "tool_call_id": "call_1", "content": "tool-result"},
+        ],
+        contains_reasoning=True,
+        contains_tool_calls=True,
+        contains_thinking_blocks=False,
+        must_roundtrip=True,
+        estimated_tokens=10,
+    )
+
+    bundle = build_agent_chat_context_bundle(session_id, MagicMock(), _config(enabled=False))
+
+    assert [msg["role"] for msg in bundle.context_messages] == ["user", "assistant", "tool", "assistant"]
+    assert bundle.context_messages[0]["content"] == "u1"
+    assert bundle.context_messages[1]["reasoning_content"] == "r1"
+    assert bundle.context_messages[2]["tool_call_id"] == "call_1"
+    assert bundle.context_messages[3]["content"] == "a1-final"
+    assert sum(1 for msg in bundle.context_messages if msg.get("content") == "a1-final") == 1
+    assert bundle.diagnostics["trace_injected"] is True
+
+
+def test_bundle_drops_trace_on_model_mismatch_budget_and_summarized_anchor() -> None:
+    db = _reset_db()
+    mismatch_session = "chat-trace-mismatch"
+    user_id = db.save_conversation_message(mismatch_session, "user", "u1")
+    assistant_id = db.save_conversation_message(mismatch_session, "assistant", "a1")
+    db.save_agent_provider_turn(
+        session_id=mismatch_session,
+        run_id="run-mismatch",
+        provider="deepseek",
+        model="deepseek/deepseek-chat",
+        anchor_user_message_id=user_id,
+        anchor_assistant_message_id=assistant_id,
+        messages=[{"role": "assistant", "reasoning_content": "r", "tool_calls": [{"id": "c", "name": "echo", "arguments": {}}]}],
+        contains_reasoning=True,
+        contains_tool_calls=True,
+        contains_thinking_blocks=False,
+        must_roundtrip=True,
+        estimated_tokens=10,
+    )
+
+    mismatch = build_agent_chat_context_bundle(mismatch_session, MagicMock(), _config(enabled=False))
+
+    assert mismatch.diagnostics["model_mismatch"] == 1
+    assert mismatch.diagnostics["trace_injected"] is False
+    assert all("reasoning_content" not in msg for msg in mismatch.context_messages)
+
+    budget_session = "chat-trace-budget"
+    user_id = db.save_conversation_message(budget_session, "user", "u1")
+    assistant_id = db.save_conversation_message(budget_session, "assistant", "a1")
+    db.save_agent_provider_turn(
+        session_id=budget_session,
+        run_id="run-budget",
+        provider="openai",
+        model="openai/test-model",
+        anchor_user_message_id=user_id,
+        anchor_assistant_message_id=assistant_id,
+        messages=[{"role": "assistant", "reasoning_content": "r", "tool_calls": [{"id": "c", "name": "echo", "arguments": {}}]}],
+        contains_reasoning=True,
+        contains_tool_calls=True,
+        contains_thinking_blocks=False,
+        must_roundtrip=True,
+        estimated_tokens=999999,
+    )
+
+    budget = build_agent_chat_context_bundle(budget_session, MagicMock(), _config(enabled=False))
+
+    assert budget.diagnostics["budget_exceeded"] is True
+    assert budget.diagnostics["trace_injected"] is False
+    assert all("reasoning_content" not in msg for msg in budget.context_messages)
+
+    summarized_session = "chat-trace-summary-anchor"
+    user_id = db.save_conversation_message(summarized_session, "user", "u1")
+    assistant_id = db.save_conversation_message(summarized_session, "assistant", "a1")
+    db.save_conversation_message(summarized_session, "user", "u2")
+    db.upsert_conversation_summary(summarized_session, "old summary", assistant_id, 2, 10)
+    db.save_agent_provider_turn(
+        session_id=summarized_session,
+        run_id="run-summary",
+        provider="openai",
+        model="openai/test-model",
+        anchor_user_message_id=user_id,
+        anchor_assistant_message_id=assistant_id,
+        messages=[{"role": "assistant", "reasoning_content": "r", "tool_calls": [{"id": "c", "name": "echo", "arguments": {}}]}],
+        contains_reasoning=True,
+        contains_tool_calls=True,
+        contains_thinking_blocks=False,
+        must_roundtrip=True,
+        estimated_tokens=10,
+    )
+
+    summarized = build_agent_chat_context_bundle(summarized_session, MagicMock(), _config(trigger=999999))
+
+    assert summarized.diagnostics["anchor_summarized"] == 1
+    assert summarized.diagnostics["trace_injected"] is False
+    assert all("reasoning_content" not in msg for msg in summarized.context_messages)
+
+
+def test_bundle_injects_trace_for_configured_fallback_model_with_trace_metadata() -> None:
+    db = _reset_db()
+    session_id = "chat-trace-fallback-model"
+    user_id = db.save_conversation_message(session_id, "user", "u1")
+    assistant_id = db.save_conversation_message(session_id, "assistant", "a1-final")
+    db.save_agent_provider_turn(
+        session_id=session_id,
+        run_id="run-fallback",
+        provider="deepseek",
+        model="deepseek/deepseek-chat",
+        anchor_user_message_id=user_id,
+        anchor_assistant_message_id=assistant_id,
+        messages=[
+            {
+                "role": "assistant",
+                "content": "checking",
+                "reasoning_content": "r",
+                "tool_calls": [{"id": "call_1", "name": "echo", "arguments": {}}],
+            },
+            {"role": "tool", "tool_call_id": "call_1", "content": "tool-result"},
+        ],
+        contains_reasoning=True,
+        contains_tool_calls=True,
+        contains_thinking_blocks=False,
+        must_roundtrip=True,
+        estimated_tokens=10,
+    )
+    config = _config(enabled=False)
+    config.agent_litellm_model = "openai/test-model"
+    config.litellm_model = "openai/test-model"
+    config.litellm_fallback_models = ["deepseek/deepseek-chat"]
+
+    bundle = build_agent_chat_context_bundle(session_id, MagicMock(), config)
+
+    assert bundle.diagnostics["trace_injected"] is True
+    assistant_trace = bundle.context_messages[1]
+    assert assistant_trace["role"] == "assistant"
+    assert assistant_trace["reasoning_content"] == "r"
+    assert assistant_trace["_trace_provider"] == "deepseek"
+    assert assistant_trace["_trace_model"] == "deepseek/deepseek-chat"
+    tool_trace = bundle.context_messages[2]
+    assert tool_trace["role"] == "tool"
+    assert tool_trace["_trace_provider"] == "deepseek"
+    assert tool_trace["_trace_model"] == "deepseek/deepseek-chat"
+
+
+def test_bundle_trace_is_replayed_only_for_matching_fallback_attempt() -> None:
+    db = _reset_db()
+    session_id = "chat-trace-fallback-attempt"
+    user_id = db.save_conversation_message(session_id, "user", "u1")
+    assistant_id = db.save_conversation_message(session_id, "assistant", "a1-final")
+    db.save_agent_provider_turn(
+        session_id=session_id,
+        run_id="run-fallback-attempt",
+        provider="deepseek",
+        model="deepseek/deepseek-chat",
+        anchor_user_message_id=user_id,
+        anchor_assistant_message_id=assistant_id,
+        messages=[
+            {
+                "role": "assistant",
+                "content": "checking",
+                "reasoning_content": "r",
+                "tool_calls": [{"id": "call_1", "name": "echo", "arguments": {}}],
+            },
+            {"role": "tool", "tool_call_id": "call_1", "content": "tool-result"},
+        ],
+        contains_reasoning=True,
+        contains_tool_calls=True,
+        contains_thinking_blocks=False,
+        must_roundtrip=True,
+        estimated_tokens=10,
+    )
+    config = _config(enabled=False)
+    config.agent_litellm_model = "openai/test-model"
+    config.litellm_model = "openai/test-model"
+    config.litellm_fallback_models = ["deepseek/deepseek-chat"]
+    adapter = LLMToolAdapter.__new__(LLMToolAdapter)
+    adapter._config = config
+
+    bundle = build_agent_chat_context_bundle(session_id, MagicMock(), config)
+    primary_messages = adapter._convert_messages(bundle.context_messages, target_model="openai/test-model")
+    fallback_messages = adapter._convert_messages(bundle.context_messages, target_model="deepseek/deepseek-chat")
+
+    assert bundle.diagnostics["trace_injected"] is True
+    assert [msg["role"] for msg in primary_messages] == ["user", "assistant"]
+    assert primary_messages[-1]["content"] == "a1-final"
+    assert all(msg.get("tool_call_id") != "call_1" for msg in primary_messages)
+    assert [msg["role"] for msg in fallback_messages] == ["user", "assistant", "tool", "assistant"]
+    assert fallback_messages[1]["reasoning_content"] == "r"
+    assert fallback_messages[2]["tool_call_id"] == "call_1"
+    assert fallback_messages[-1]["content"] == "a1-final"
+
+
+def test_bundle_matches_slashless_router_alias_fallback_by_resolved_provider() -> None:
+    db = _reset_db()
+    session_id = "chat-trace-router-alias"
+    user_id = db.save_conversation_message(session_id, "user", "u1")
+    assistant_id = db.save_conversation_message(session_id, "assistant", "a1-final")
+    db.save_agent_provider_turn(
+        session_id=session_id,
+        run_id="run-router-alias",
+        provider="openai",
+        model="gpt4o",
+        anchor_user_message_id=user_id,
+        anchor_assistant_message_id=assistant_id,
+        messages=[
+            {
+                "role": "assistant",
+                "content": "checking",
+                "reasoning_content": "r",
+                "tool_calls": [{"id": "call_1", "name": "echo", "arguments": {}}],
+            },
+            {"role": "tool", "tool_call_id": "call_1", "content": "tool-result"},
+        ],
+        contains_reasoning=True,
+        contains_tool_calls=True,
+        contains_thinking_blocks=False,
+        must_roundtrip=True,
+        estimated_tokens=10,
+    )
+    config = _config(enabled=False)
+    config.agent_litellm_model = "anthropic/claude-test"
+    config.litellm_model = "anthropic/claude-test"
+    config.litellm_fallback_models = ["gpt4o"]
+    config.llm_model_list = [
+        {
+            "model_name": "gpt4o",
+            "litellm_params": {"model": "openai/gpt-4o-mini"},
+        }
+    ]
+
+    bundle = build_agent_chat_context_bundle(session_id, MagicMock(), config)
+
+    assert bundle.diagnostics["trace_injected"] is True
+    assert bundle.diagnostics["model_mismatch"] == 0
+    assistant_trace = bundle.context_messages[1]
+    assert assistant_trace["_trace_provider"] == "openai"
+    assert assistant_trace["_trace_model"] == "gpt4o"
+
+
 def test_over_trigger_generates_summary_and_updates_covered_message_id() -> None:
     db = _reset_db()
     session_id = "chat-summarize"

tests/test_llm_adapter_provider_trace.py

@@ -0,0 +1,229 @@
+# -*- coding: utf-8 -*-
+import os
+import sys
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+try:
+    import litellm  # noqa: F401
+except ModuleNotFoundError:
+    sys.modules["litellm"] = MagicMock()
+
+from src.agent.llm_adapter import LLMToolAdapter  # noqa: E402
+
+
+def test_convert_messages_preserves_reasoning_blocks_and_provider_specific_fields() -> None:
+    adapter = LLMToolAdapter.__new__(LLMToolAdapter)
+    messages = [
+        {
+            "role": "assistant",
+            "content": "checking",
+            "_trace_provider": "anthropic",
+            "_trace_model": "anthropic/claude-test",
+            "provider_blocks": [
+                {"type": "thinking", "thinking": "opaque"},
+                {"type": "redacted_thinking", "data": "redacted"},
+                {"type": "text", "text": "checking"},
+            ],
+            "reasoning_content": "reasoning",
+            "tool_calls": [
+                {
+                    "id": "call_1",
+                    "name": "echo",
+                    "arguments": {"message": "hello"},
+                    "thought_signature": "sig-1",
+                    "provider_specific_fields": {"thought_signature": "sig-1", "extra": "keep"},
+                }
+            ],
+        }
+    ]
+
+    converted = adapter._convert_messages(messages)
+
+    assert converted[0]["role"] == "assistant"
+    assert converted[0]["content"][0]["type"] == "thinking"
+    assert converted[0]["reasoning_content"] == "reasoning"
+    assert converted[0]["tool_calls"][0]["provider_specific_fields"] == {
+        "thought_signature": "sig-1",
+        "extra": "keep",
+    }
+    assert "_trace_provider" not in converted[0]
+
+
+def test_convert_messages_only_sends_provider_trace_to_matching_target_model() -> None:
+    adapter = LLMToolAdapter.__new__(LLMToolAdapter)
+    messages = [
+        {
+            "role": "assistant",
+            "content": "checking",
+            "_trace_provider": "anthropic",
+            "_trace_model": "anthropic/claude-test",
+            "provider_blocks": [{"type": "thinking", "thinking": "opaque"}],
+            "reasoning_content": "provider-only",
+            "tool_calls": [
+                {
+                    "id": "call_1",
+                    "name": "echo",
+                    "arguments": {"message": "hello"},
+                    "thought_signature": "sig-1",
+                    "provider_specific_fields": {"thought_signature": "sig-1"},
+                }
+            ],
+        }
+    ]
+
+    matching = adapter._convert_messages(messages, target_model="anthropic/claude-test")
+    mismatched = adapter._convert_messages(messages, target_model="openai/gpt-4o-mini")
+
+    assert matching[0]["content"] == [{"type": "thinking", "thinking": "opaque"}]
+    assert matching[0]["reasoning_content"] == "provider-only"
+    assert matching[0]["tool_calls"][0]["provider_specific_fields"] == {"thought_signature": "sig-1"}
+
+    assert mismatched == []
+
+
+def test_convert_messages_skips_entire_trace_segment_for_mismatched_attempt() -> None:
+    adapter = LLMToolAdapter.__new__(LLMToolAdapter)
+    messages = [
+        {"role": "user", "content": "u1"},
+        {
+            "role": "assistant",
+            "content": "checking",
+            "_trace_provider": "deepseek",
+            "_trace_model": "deepseek/deepseek-chat",
+            "reasoning_content": "provider-only",
+            "tool_calls": [
+                {
+                    "id": "call_1",
+                    "name": "echo",
+                    "arguments": {"message": "hello"},
+                    "provider_specific_fields": {"thought_signature": "sig-1"},
+                }
+            ],
+        },
+        {
+            "role": "tool",
+            "tool_call_id": "call_1",
+            "content": "tool-result",
+            "_trace_provider": "deepseek",
+            "_trace_model": "deepseek/deepseek-chat",
+        },
+        {"role": "assistant", "content": "a1-final"},
+    ]
+
+    primary = adapter._convert_messages(messages, target_model="openai/gpt-4o-mini")
+    fallback = adapter._convert_messages(messages, target_model="deepseek/deepseek-chat")
+
+    assert [msg["role"] for msg in primary] == ["user", "assistant"]
+    assert primary[-1]["content"] == "a1-final"
+    assert all(msg.get("tool_call_id") != "call_1" for msg in primary)
+
+    assert [msg["role"] for msg in fallback] == ["user", "assistant", "tool", "assistant"]
+    assert fallback[1]["reasoning_content"] == "provider-only"
+    assert fallback[1]["tool_calls"][0]["provider_specific_fields"] == {"thought_signature": "sig-1"}
+    assert fallback[2]["tool_call_id"] == "call_1"
+
+
+def test_convert_messages_matches_slashless_openai_target_without_provider_leakage() -> None:
+    adapter = LLMToolAdapter.__new__(LLMToolAdapter)
+    messages = [
+        {
+            "role": "assistant",
+            "content": "checking",
+            "_trace_provider": "openai",
+            "_trace_model": "gpt-4o-mini",
+            "reasoning_content": "provider-only",
+            "tool_calls": [
+                {
+                    "id": "call_1",
+                    "name": "echo",
+                    "arguments": {},
+                    "provider_specific_fields": {"thought_signature": "sig-1"},
+                }
+            ],
+        }
+    ]
+
+    matching = adapter._convert_messages(messages, target_model="gpt-4o-mini")
+    mismatched = adapter._convert_messages(messages, target_model="claude-router")
+
+    assert matching[0]["reasoning_content"] == "provider-only"
+    assert matching[0]["tool_calls"][0]["provider_specific_fields"] == {"thought_signature": "sig-1"}
+    assert mismatched == []
+
+
+def test_parse_litellm_response_extracts_claude_blocks_and_tool_provider_fields() -> None:
+    adapter = LLMToolAdapter.__new__(LLMToolAdapter)
+    blocks = [
+        {"type": "thinking", "thinking": "opaque"},
+        {"type": "redacted_thinking", "data": "hidden"},
+        {"type": "text", "text": "Need data"},
+    ]
+    response = SimpleNamespace(
+        choices=[
+            SimpleNamespace(
+                message=SimpleNamespace(
+                    content=blocks,
+                    reasoning_content=None,
+                    tool_calls=[
+                        SimpleNamespace(
+                            id="call_1",
+                            function=SimpleNamespace(
+                                name="echo",
+                                arguments='{"message": "hello"}',
+                                provider_specific_fields=None,
+                            ),
+                            provider_specific_fields={"thought_signature": "sig-1", "extra": "keep"},
+                        )
+                    ],
+                )
+            )
+        ],
+        usage=SimpleNamespace(prompt_tokens=1, completion_tokens=2, total_tokens=3),
+    )
+
+    parsed = adapter._parse_litellm_response(response, "anthropic/claude-test")
+
+    assert parsed.content == "Need data"
+    assert parsed.provider_blocks == blocks
+    assert parsed.provider == "anthropic"
+    assert parsed.model == "anthropic/claude-test"
+    assert parsed.tool_calls[0].thought_signature == "sig-1"
+    assert parsed.tool_calls[0].provider_specific_fields == {
+        "thought_signature": "sig-1",
+        "extra": "keep",
+    }
+
+
+def test_parse_litellm_response_resolves_provider_for_slashless_router_alias() -> None:
+    adapter = LLMToolAdapter.__new__(LLMToolAdapter)
+    adapter._config = SimpleNamespace(
+        llm_model_list=[
+            {
+                "model_name": "claude-router",
+                "litellm_params": {"model": "anthropic/claude-sonnet-test"},
+            }
+        ]
+    )
+    response = SimpleNamespace(
+        choices=[
+            SimpleNamespace(
+                message=SimpleNamespace(
+                    content="ok",
+                    reasoning_content=None,
+                    tool_calls=[],
+                )
+            )
+        ],
+        usage=SimpleNamespace(prompt_tokens=1, completion_tokens=2, total_tokens=3),
+    )
+
+    parsed_alias = adapter._parse_litellm_response(response, "claude-router")
+    parsed_bare_openai = adapter._parse_litellm_response(response, "gpt-4o-mini")
+
+    assert parsed_alias.provider == "anthropic"
+    assert parsed_alias.model == "claude-router"
+    assert parsed_bare_openai.provider == "openai"
+    assert parsed_bare_openai.model == "gpt-4o-mini"

tests/test_multi_agent.py

@@ -35,7 +35,8 @@ from src.agent.protocols import (
     StageResult,
     StageStatus,
 )
-from src.config import AGENT_MAX_STEPS_DEFAULT
+from src.config import AGENT_MAX_STEPS_DEFAULT, Config
+from src.storage import DatabaseManager
 
 
 # ============================================================
@@ -913,6 +914,54 @@ class TestOrchestratorExecution(unittest.TestCase):
         self.assertEqual(build_history.call_args.args[0], "session-1")
         self.assertIs(build_history.call_args.args[1], orch.llm_adapter)
 
+    def test_chat_does_not_read_or_write_provider_trace(self):
+        from src.agent.orchestrator import OrchestratorResult
+
+        DatabaseManager.reset_instance()
+        Config.reset_instance()
+        db = DatabaseManager(db_url="sqlite:///:memory:")
+        session_id = "multi-agent-trace-boundary"
+        user_id = db.save_conversation_message(session_id, "user", "previous question")
+        assistant_id = db.save_conversation_message(session_id, "assistant", "previous answer")
+        db.save_agent_provider_turn(
+            session_id=session_id,
+            run_id="run-existing",
+            provider="deepseek",
+            model="deepseek/deepseek-chat",
+            anchor_user_message_id=user_id,
+            anchor_assistant_message_id=assistant_id,
+            messages=[
+                {
+                    "role": "assistant",
+                    "reasoning_content": "reasoning",
+                    "tool_calls": [{"id": "call_1", "name": "echo", "arguments": {}}],
+                },
+                {"role": "tool", "tool_call_id": "call_1", "content": "tool-result"},
+            ],
+            contains_reasoning=True,
+            contains_tool_calls=True,
+            contains_thinking_blocks=False,
+            must_roundtrip=True,
+            estimated_tokens=10,
+        )
+
+        orch = self._make_orchestrator()
+        try:
+            with patch.object(orch, "_execute_pipeline", return_value=OrchestratorResult(success=True, content="ok")):
+                with patch("src.agent.orchestrator.build_visible_chat_history", return_value=[]) as build_history:
+                    with patch.object(db, "get_agent_provider_turns", wraps=db.get_agent_provider_turns) as get_turns:
+                        result = orch.chat("hello", session_id)
+
+            self.assertTrue(result.success)
+            build_history.assert_called_once()
+            get_turns.assert_not_called()
+            rows = db.get_agent_provider_turns(session_id)
+            self.assertEqual(len(rows), 1)
+            self.assertEqual(rows[0]["run_id"], "run-existing")
+        finally:
+            DatabaseManager.reset_instance()
+            Config.reset_instance()
+
     def test_chat_persists_user_and_assistant_messages(self):
         from src.agent.orchestrator import OrchestratorResult
 

tests/test_provider_trace.py

@@ -0,0 +1,154 @@
+# -*- coding: utf-8 -*-
+import os
+import sys
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+from src.agent.provider_trace import (  # noqa: E402
+    extract_provider_trace_turns,
+    provider_namespace,
+    resolved_provider_namespace,
+    trace_model_matches,
+)
+
+
+def test_trace_model_matches_slashless_openai_namespace_without_widening_model_match() -> None:
+    assert provider_namespace("gpt-4o-mini") == "openai"
+    assert trace_model_matches("openai", "gpt-4o-mini", "gpt-4o-mini") is True
+    assert trace_model_matches("anthropic", "gpt-4o-mini", "gpt-4o-mini") is False
+    assert trace_model_matches("openai", "gpt-4o-mini", "openai/gpt-4o-mini") is False
+    assert trace_model_matches(
+        "anthropic",
+        "claude-router",
+        "claude-router",
+        current_provider="anthropic",
+    ) is True
+
+
+def test_resolved_provider_namespace_uses_router_alias_before_slashless_default() -> None:
+    model_list = [
+        {
+            "model_name": "claude-router",
+            "litellm_params": {"model": "anthropic/claude-sonnet-test"},
+        }
+    ]
+
+    assert resolved_provider_namespace("claude-router", model_list) == "anthropic"
+    assert resolved_provider_namespace("gpt-4o-mini", model_list) == "openai"
+
+
+def test_extract_trace_scans_only_current_run_and_keeps_multi_step_tool_loop() -> None:
+    messages = [
+        {
+            "role": "assistant",
+            "_trace_provider": "deepseek",
+            "_trace_model": "deepseek/deepseek-chat",
+            "reasoning_content": "old",
+            "tool_calls": [{"id": "old", "name": "echo", "arguments": {}}],
+        },
+        {"role": "tool", "tool_call_id": "old", "content": "old-result"},
+        {"role": "user", "content": "current"},
+        {
+            "role": "assistant",
+            "_trace_provider": "deepseek",
+            "_trace_model": "deepseek/deepseek-chat",
+            "content": "step1",
+            "reasoning_content": "r1",
+            "tool_calls": [{"id": "c1", "name": "echo", "arguments": {}}],
+        },
+        {"role": "tool", "tool_call_id": "c1", "content": "result1"},
+        {
+            "role": "assistant",
+            "_trace_provider": "deepseek",
+            "_trace_model": "deepseek/deepseek-chat",
+            "content": "step2",
+            "reasoning_content": "r2",
+            "tool_calls": [{"id": "c2", "name": "echo", "arguments": {}}],
+        },
+        {"role": "tool", "tool_call_id": "c2", "content": "result2"},
+    ]
+
+    turns, diagnostics = extract_provider_trace_turns(
+        messages,
+        baseline_len=3,
+        run_id="run-1",
+        anchor_user_message_id=10,
+        anchor_assistant_message_id=11,
+    )
+
+    assert diagnostics.trace_dropped_reason == ""
+    assert len(turns) == 1
+    assert [msg["role"] for msg in turns[0].messages] == ["assistant", "tool", "assistant", "tool"]
+    assert turns[0].messages[0]["reasoning_content"] == "r1"
+    assert turns[0].messages[2]["reasoning_content"] == "r2"
+    assert "_trace_provider" not in turns[0].messages[0]
+    assert turns[0].must_roundtrip is True
+
+
+def test_extract_trace_drops_deepseek_without_required_tool_reasoning_pair_and_mixed_model() -> None:
+    no_tool_turns, no_tool_diag = extract_provider_trace_turns(
+        [
+            {"role": "user", "content": "u"},
+            {
+                "role": "assistant",
+                "_trace_provider": "deepseek",
+                "_trace_model": "deepseek/deepseek-chat",
+                "reasoning_content": "reasoning-only",
+                "content": "final",
+            },
+        ],
+        baseline_len=1,
+    )
+
+    assert no_tool_turns == []
+    assert no_tool_diag.trace_dropped_reason == ""
+
+    no_reasoning_turns, no_reasoning_diag = extract_provider_trace_turns(
+        [
+            {"role": "user", "content": "u"},
+            {
+                "role": "assistant",
+                "_trace_provider": "deepseek",
+                "_trace_model": "deepseek/deepseek-chat",
+                "tool_calls": [
+                    {
+                        "id": "c0",
+                        "name": "echo",
+                        "arguments": {},
+                        "provider_specific_fields": {"extra": "not-enough"},
+                    }
+                ],
+            },
+            {"role": "tool", "tool_call_id": "c0", "content": "result"},
+        ],
+        baseline_len=1,
+    )
+
+    assert no_reasoning_turns == []
+    assert no_reasoning_diag.trace_dropped_reason == "not_required"
+
+    mixed_turns, mixed_diag = extract_provider_trace_turns(
+        [
+            {"role": "user", "content": "u"},
+            {
+                "role": "assistant",
+                "_trace_provider": "deepseek",
+                "_trace_model": "deepseek/deepseek-chat",
+                "reasoning_content": "r1",
+                "tool_calls": [{"id": "c1", "name": "echo", "arguments": {}}],
+            },
+            {"role": "tool", "tool_call_id": "c1", "content": "result1"},
+            {
+                "role": "assistant",
+                "_trace_provider": "deepseek",
+                "_trace_model": "deepseek/deepseek-reasoner",
+                "reasoning_content": "r2",
+                "tool_calls": [{"id": "c2", "name": "echo", "arguments": {}}],
+            },
+        ],
+        baseline_len=1,
+    )
+
+    assert mixed_turns == []
+    assert mixed_diag.mixed_model_trace is True
+    assert mixed_diag.trace_dropped_reason == "mixed_model_trace"

tests/test_storage.py

@@ -133,6 +133,159 @@ class TestStorage(unittest.TestCase):
 
         DatabaseManager.reset_instance()
 
+    def test_conversation_message_save_returns_id(self):
+        DatabaseManager.reset_instance()
+        db = DatabaseManager(db_url="sqlite:///:memory:")
+
+        message_id = db.save_conversation_message("message-id-session", "user", "hello")
+
+        self.assertIsInstance(message_id, int)
+        self.assertGreater(message_id, 0)
+
+        DatabaseManager.reset_instance()
+
+    def test_provider_turn_round_trip_preserves_protocol_fields_and_flags(self):
+        DatabaseManager.reset_instance()
+        db = DatabaseManager(db_url="sqlite:///:memory:")
+        user_id = db.save_conversation_message("trace-session", "user", "question")
+        assistant_id = db.save_conversation_message("trace-session", "assistant", "final")
+        trace_messages = [
+            {
+                "role": "assistant",
+                "content": "checking",
+                "reasoning_content": "reasoning",
+                "tool_calls": [
+                    {
+                        "id": "call_1",
+                        "name": "echo",
+                        "arguments": {"message": "hello"},
+                        "provider_specific_fields": {"thought_signature": "sig"},
+                    }
+                ],
+            },
+            {"role": "tool", "tool_call_id": "call_1", "content": "{\"ok\": true}"},
+        ]
+
+        turn_id = db.save_agent_provider_turn(
+            session_id="trace-session",
+            run_id="run-1",
+            provider="deepseek",
+            model="deepseek/deepseek-chat",
+            anchor_user_message_id=user_id,
+            anchor_assistant_message_id=assistant_id,
+            messages=trace_messages,
+            contains_reasoning=True,
+            contains_tool_calls=True,
+            contains_thinking_blocks=False,
+            must_roundtrip=True,
+            estimated_tokens=42,
+        )
+        rows = db.get_agent_provider_turns("trace-session")
+
+        self.assertIsInstance(turn_id, int)
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(rows[0]["messages"], trace_messages)
+        self.assertTrue(rows[0]["contains_reasoning"])
+        self.assertTrue(rows[0]["contains_tool_calls"])
+        self.assertTrue(rows[0]["must_roundtrip"])
+        self.assertEqual(rows[0]["estimated_tokens"], 42)
+
+        DatabaseManager.reset_instance()
+
+    def test_provider_turns_do_not_appear_in_visible_or_web_messages_and_delete_with_session(self):
+        DatabaseManager.reset_instance()
+        db = DatabaseManager(db_url="sqlite:///:memory:")
+        user_id = db.save_conversation_message("trace-hidden", "user", "visible question")
+        assistant_id = db.save_conversation_message("trace-hidden", "assistant", "visible answer")
+        db.save_agent_provider_turn(
+            session_id="trace-hidden",
+            run_id="run-hidden",
+            provider="deepseek",
+            model="deepseek/deepseek-chat",
+            anchor_user_message_id=user_id,
+            anchor_assistant_message_id=assistant_id,
+            messages=[{"role": "assistant", "reasoning_content": "SECRET_REASONING", "tool_calls": []}],
+            contains_reasoning=True,
+            contains_tool_calls=True,
+            contains_thinking_blocks=False,
+            must_roundtrip=True,
+            estimated_tokens=5,
+        )
+
+        self.assertEqual(
+            [(m["role"], m["content"]) for m in db.get_visible_conversation_messages("trace-hidden")],
+            [("user", "visible question"), ("assistant", "visible answer")],
+        )
+        self.assertEqual(
+            [(m["role"], m["content"]) for m in db.get_conversation_history("trace-hidden")],
+            [("user", "visible question"), ("assistant", "visible answer")],
+        )
+        self.assertEqual(
+            [(m["role"], m["content"]) for m in db.get_conversation_messages("trace-hidden")],
+            [("user", "visible question"), ("assistant", "visible answer")],
+        )
+
+        deleted = db.delete_conversation_session("trace-hidden")
+
+        self.assertEqual(deleted, 2)
+        self.assertEqual(db.get_agent_provider_turns("trace-hidden"), [])
+
+        DatabaseManager.reset_instance()
+
+    def test_provider_turn_retention_is_bucketed_by_session_provider_model(self):
+        DatabaseManager.reset_instance()
+        db = DatabaseManager(db_url="sqlite:///:memory:")
+        for idx in range(5):
+            user_id = db.save_conversation_message("retention", "user", f"q{idx}")
+            assistant_id = db.save_conversation_message("retention", "assistant", f"a{idx}")
+            db.save_agent_provider_turn(
+                session_id="retention",
+                run_id=f"run-{idx}",
+                provider="deepseek",
+                model="deepseek/deepseek-chat",
+                anchor_user_message_id=user_id,
+                anchor_assistant_message_id=assistant_id,
+                messages=[{"role": "assistant", "reasoning_content": f"r{idx}", "tool_calls": [{"id": f"c{idx}", "name": "echo", "arguments": {}}]}],
+                contains_reasoning=True,
+                contains_tool_calls=True,
+                contains_thinking_blocks=False,
+                must_roundtrip=True,
+                estimated_tokens=idx + 1,
+            )
+        user_id = db.save_conversation_message("retention", "user", "other")
+        assistant_id = db.save_conversation_message("retention", "assistant", "other")
+        db.save_agent_provider_turn(
+            session_id="retention",
+            run_id="run-other",
+            provider="anthropic",
+            model="anthropic/claude-test",
+            anchor_user_message_id=user_id,
+            anchor_assistant_message_id=assistant_id,
+            messages=[{"role": "assistant", "provider_blocks": [{"type": "thinking"}], "tool_calls": [{"id": "c-other", "name": "echo", "arguments": {}}]}],
+            contains_reasoning=False,
+            contains_tool_calls=True,
+            contains_thinking_blocks=True,
+            must_roundtrip=True,
+            estimated_tokens=1,
+        )
+
+        deepseek_rows = db.get_agent_provider_turns(
+            "retention",
+            provider="deepseek",
+            model="deepseek/deepseek-chat",
+        )
+        anthropic_rows = db.get_agent_provider_turns(
+            "retention",
+            provider="anthropic",
+            model="anthropic/claude-test",
+        )
+
+        self.assertEqual(len(deepseek_rows), 3)
+        self.assertEqual([row["run_id"] for row in deepseek_rows], ["run-2", "run-3", "run-4"])
+        self.assertEqual(len(anthropic_rows), 1)
+
+        DatabaseManager.reset_instance()
+
     def test_get_visible_conversation_messages_returns_ordered_visible_content(self):
         DatabaseManager.reset_instance()
         db = DatabaseManager(db_url="sqlite:///:memory:")
@@ -151,6 +304,21 @@ class TestStorage(unittest.TestCase):
 
         DatabaseManager.reset_instance()
 
+    def test_get_visible_conversation_messages_limit_returns_ordered_tail(self):
+        DatabaseManager.reset_instance()
+        db = DatabaseManager(db_url="sqlite:///:memory:")
+
+        for idx in range(25):
+            db.save_conversation_message("visible-limit", "user", f"msg-{idx}")
+
+        messages = db.get_visible_conversation_messages("visible-limit", limit=20)
+
+        self.assertEqual(len(messages), 20)
+        self.assertEqual(messages[0]["content"], "msg-5")
+        self.assertEqual(messages[-1]["content"], "msg-24")
+
+        DatabaseManager.reset_instance()
+
     def test_file_sqlite_enables_wal_and_busy_timeout(self):
         temp_dir = tempfile.TemporaryDirectory()
         db_path = os.path.join(temp_dir.name, "sqlite_pragmas.db")