diff --git a/deeptutor/agents/chat/agent_loop.py b/deeptutor/agents/chat/agent_loop.py
index 12f28336cd..3ead79ec28 100644
--- a/deeptutor/agents/chat/agent_loop.py
+++ b/deeptutor/agents/chat/agent_loop.py
@@ -256,6 +256,7 @@ async def _run_loop(
                     trace_role="explore",
                     max_tokens=self.pipeline.loop_max_tokens,
                     tool_schemas=self.tool_schemas,
+                    tool_choice=(self.pipeline.initial_tool_choice if _round == 0 else None),
                 )
             except Exception as exc:
                 # A mid-loop LLM failure (timeout / transient network) must not
@@ -444,6 +445,7 @@ async def _call_llm(
         trace_role: str,
         max_tokens: int,
         tool_schemas: list[dict[str, Any]] | None = None,
+        tool_choice: str | None = None,
     ) -> LLMCallResult:
         await self.pipeline._guard_context_window(messages, self.stream)
         stage = LOOP_STAGE
@@ -477,7 +479,20 @@ async def _call_llm(
             kwargs["stream_options"] = {"include_usage": True}
         if tool_schemas:
             kwargs["tools"] = tool_schemas
-            kwargs["tool_choice"] = "auto"
+            available_tools = {
+                str((schema.get("function") or {}).get("name") or "")
+                for schema in tool_schemas
+                if isinstance(schema, dict)
+            }
+            kwargs["tool_choice"] = (
+                {
+                    "type": "function",
+                    "function": {"name": tool_choice},
+                }
+                if tool_choice and tool_choice in available_tools
+                else "auto"
+            )
+        forced_tool_choice = isinstance(kwargs.get("tool_choice"), dict)
         # What this request actually carried, pinned now: the loop keeps
         # appending to ``messages`` and the deferred loader keeps appending to
         # ``tool_schemas``, so the turn's context budget is read off the last
@@ -558,10 +573,15 @@ async def _emit_segments(segments: list[tuple[str, str]]) -> None:
                     # channel so the content stream stays user-facing.
                     if not dsml_stream_active and has_dsml_tool_calls("".join(text_parts)):
                         dsml_stream_active = True
-                    segments = think_filter.feed(content)
-                    if dsml_stream_active:
-                        segments = [("thinking", seg) for _, seg in segments]
-                    await _emit_segments(segments)
+                    # Buffer prose during a forced-tool round. If the model
+                    # obeys the choice, the card is the first user-facing
+                    # artefact; if a provider ignores it, the buffered prose
+                    # is released below as a graceful fallback.
+                    if not forced_tool_choice:
+                        segments = think_filter.feed(content)
+                        if dsml_stream_active:
+                            segments = [("thinking", seg) for _, seg in segments]
+                        await _emit_segments(segments)
 
                 for tc_delta in getattr(delta, "tool_calls", None) or []:
                     index = int(getattr(tc_delta, "index", 0) or 0)
@@ -586,10 +606,11 @@ async def _emit_segments(segments: list[tuple[str, str]]) -> None:
                 with suppress(Exception):
                     await close()
 
-        flushed = think_filter.flush()
-        if dsml_stream_active:
-            flushed = [("thinking", seg) for _, seg in flushed]
-        await _emit_segments(flushed)
+        if not forced_tool_choice:
+            flushed = think_filter.flush()
+            if dsml_stream_active:
+                flushed = [("thinking", seg) for _, seg in flushed]
+            await _emit_segments(flushed)
         text = "".join(text_parts)
         record_streamed_usage(
             self.pipeline.usage,
@@ -619,6 +640,12 @@ async def _emit_segments(segments: list[tuple[str, str]]) -> None:
                 tool_calls = dsml_calls
                 text = cleaned_text
 
+        if forced_tool_choice and not tool_calls and text:
+            # Some compatibility providers accept ``tool_choice`` but ignore
+            # it. Do not lose their answer merely because it was buffered.
+            fallback_filter = InlineThinkFilter()
+            await _emit_segments(fallback_filter.feed(text) + fallback_filter.flush())
+
         await self.stream.progress(
             "",
             source="chat",
diff --git a/deeptutor/agents/chat/agentic_pipeline.py b/deeptutor/agents/chat/agentic_pipeline.py
index 0adffb955a..c042b29176 100644
--- a/deeptutor/agents/chat/agentic_pipeline.py
+++ b/deeptutor/agents/chat/agentic_pipeline.py
@@ -192,6 +192,7 @@ def __init__(
         max_rounds: int | None = None,
         temperature: float | None = None,
         max_tokens: int | None = None,
+        initial_tool_choice: str | None = None,
     ) -> None:
         self.language = "zh" if language.lower().startswith("zh") else "en"
         self.llm_config = get_llm_config()
@@ -211,6 +212,9 @@ def __init__(
         self._deferred_pool: list[Any] = []
         self._exec_enabled = False
         self._kb_manifests: list[KbManifest] = []
+        # A selected capability may require one specific tool on the first
+        # internal loop round. Later rounds return to model-directed selection.
+        self.initial_tool_choice = (initial_tool_choice or "").strip() or None
         # The blocks the turn's system prompt was rendered from, kept for the
         # context-budget breakdown (see ``measure_context_budget``).
         self._last_prompt_blocks: list[PromptBlock] = []
diff --git a/deeptutor/capabilities/ask_questions/__init__.py b/deeptutor/capabilities/ask_questions/__init__.py
new file mode 100644
index 0000000000..8534aa92f4
--- /dev/null
+++ b/deeptutor/capabilities/ask_questions/__init__.py
@@ -0,0 +1,5 @@
+"""Ask Questions capability exports."""
+
+from deeptutor.capabilities.ask_questions.loop import AskQuestionsLoopCapability
+
+__all__ = ["AskQuestionsLoopCapability"]
diff --git a/deeptutor/capabilities/ask_questions/capability.py b/deeptutor/capabilities/ask_questions/capability.py
new file mode 100644
index 0000000000..f856f1e26f
--- /dev/null
+++ b/deeptutor/capabilities/ask_questions/capability.py
@@ -0,0 +1,39 @@
+"""Ask Questions capability — an explicit user-selected interview mode."""
+
+from __future__ import annotations
+
+from deeptutor.agents.chat.agentic_pipeline import AgenticChatPipeline
+from deeptutor.core.capability_protocol import BaseCapability, CapabilityManifest
+from deeptutor.core.context import UnifiedContext
+from deeptutor.core.stream_bus import StreamBus
+from deeptutor.runtime.request_contracts import get_capability_request_schema
+
+
+class AskQuestionsCapability(BaseCapability):
+    """Start the selected turn with a context-aware question card."""
+
+    manifest = CapabilityManifest(
+        name="ask_questions",
+        description=(
+            "Ask the user high-value questions to fill in missing context, "
+            "then complete the original request with their answers."
+        ),
+        stages=["responding"],
+        tools_used=["ask_user"],
+        cli_aliases=["ask"],
+        request_schema=get_capability_request_schema("chat"),
+    )
+
+    async def run(self, context: UnifiedContext, stream: StreamBus) -> None:
+        context.metadata["ask_questions_mode"] = True
+        # This is the first *agent-loop round of the selected turn*, not the
+        # first turn in the conversation.  The prompt still receives the full
+        # history, so a turn selected much later asks a new, contextual question.
+        pipeline = AgenticChatPipeline(
+            language=context.language,
+            initial_tool_choice="ask_user",
+        )
+        await pipeline.run(context, stream)
+
+
+__all__ = ["AskQuestionsCapability"]
diff --git a/deeptutor/capabilities/ask_questions/loop.py b/deeptutor/capabilities/ask_questions/loop.py
new file mode 100644
index 0000000000..c10b291339
--- /dev/null
+++ b/deeptutor/capabilities/ask_questions/loop.py
@@ -0,0 +1,53 @@
+"""Chat-loop hooks for the Ask Questions capability."""
+
+from __future__ import annotations
+
+from importlib import resources
+from typing import Any
+
+from deeptutor.capabilities.protocol import PromptBlock
+from deeptutor.core.context import UnifiedContext
+
+
+class AskQuestionsLoopCapability:
+    """Contribute an adaptive questioning policy to the normal chat loop."""
+
+    name = "ask_questions"
+    owned_tools: tuple[str, ...] = ()
+
+    def is_active(self, context: UnifiedContext) -> bool:
+        return bool(context.metadata.get("ask_questions_mode"))
+
+    def system_block(
+        self,
+        context: UnifiedContext,
+        *,
+        language: str,
+        prompts: dict[str, Any],
+    ) -> PromptBlock | None:
+        _ = prompts
+        if not self.is_active(context):
+            return None
+        return PromptBlock("ask_questions", _load_system_prompt(language))
+
+    def augment_kwargs(
+        self,
+        tool_name: str,
+        kwargs: dict[str, Any],
+        context: UnifiedContext,
+    ) -> dict[str, Any]:
+        _ = tool_name, context
+        return kwargs
+
+    def pre_loop_seed(self, context: UnifiedContext) -> str:
+        _ = context
+        return ""
+
+
+def _load_system_prompt(language: str) -> str:
+    lang = "zh" if language.lower().startswith("zh") else "en"
+    prompt = resources.files(__package__).joinpath("prompts", lang, "system.md")
+    return prompt.read_text(encoding="utf-8").strip()
+
+
+__all__ = ["AskQuestionsLoopCapability"]
diff --git a/deeptutor/capabilities/ask_questions/prompts/en/system.md b/deeptutor/capabilities/ask_questions/prompts/en/system.md
new file mode 100644
index 0000000000..252e176475
--- /dev/null
+++ b/deeptutor/capabilities/ask_questions/prompts/en/system.md
@@ -0,0 +1,9 @@
+# Ask Questions mode
+
+The user explicitly selected Ask Questions for this turn. Begin this selected turn by calling `ask_user` exactly once with a useful question card. Do not write any answer, preamble, explanation, or narration before the tool call. "Begin this turn" refers to the current selected turn, which may be the second, third, tenth, or any later turn in the conversation—not the beginning of the whole conversation.
+
+Before choosing the question, study all available context: the current request, the full prior conversation, earlier `ask_user` questions and answers, memory, persona, attachments, selected sources, and knowledge-base context. Ask about the most valuable remaining unknown that only the user can supply. Even when substantial context already exists, use the card to refine a relevant goal, constraint, priority, difficulty, or preference instead of skipping the question.
+
+Before calling the tool, check the clarification history and never repeat a question the user already answered or a fact they already supplied. If new evidence conflicts with an older answer, ask only what changed and explain why an update is needed. Do not ask generic filler or request confirmation merely to delay action. Ask 1–4 specific, high-information questions in one call. Use concise, meaningful options only when options genuinely help; otherwise allow free text. Useful targets include the user's real goal, existing knowledge, constraints, prior attempts, point of confusion, audience, and preferred depth or output.
+
+After the user answers, continue the original request in the same turn using the new context; do not end with a bare acknowledgment. Ask again later only if an answer or subsequent tool result exposes another material information gap.
diff --git a/deeptutor/capabilities/ask_questions/prompts/zh/system.md b/deeptutor/capabilities/ask_questions/prompts/zh/system.md
new file mode 100644
index 0000000000..bd9612829f
--- /dev/null
+++ b/deeptutor/capabilities/ask_questions/prompts/zh/system.md
@@ -0,0 +1,9 @@
+# 主动提问模式
+
+用户为当前 turn 明确选择了 Ask Questions。这个被选中的 turn 必须以一次 `ask_user` 调用开始，生成一张有价值的问题卡片。工具调用之前不要输出答案、开场白、解释或旁白。“当前 turn 开始”指本次选中模式后的内部首轮；它可以发生在整段对话的第 2、3、10 轮或更晚，而不是只指整段对话的开局。
+
+选择问题前，先审视全部可用上下文：当前请求、此前所有对话、较早的 `ask_user` 问答、记忆、persona、附件、选中的来源和知识库内容。询问当前最有价值、且只能由用户补充的未知信息。即使已有 context 很丰富，也不要跳过卡片；应进一步澄清与当前请求相关的目标、限制、优先级、难点或偏好。
+
+调用前必须检查历史澄清记录，不要重复询问用户已经回答或明确说明的事实。若新情况与旧答案冲突，只询问发生变化的部分，并明确指出需要更新的原因。不要问空泛套话，不要为了拖延而要求确认。问题应少而具体，一次集中提出 1–4 个高信息量问题；只有在选项确实有帮助时才提供简洁且有意义的选项，否则允许自由输入。可关注真实目标、已有知识、约束条件、此前尝试、具体卡点、目标受众，以及期望的深度或输出形式。
+
+用户回答后，在同一个 turn 中根据新增上下文继续原始请求，不要只回复一句确认收到。只有当回答或后续工具结果又暴露了新的实质性信息缺口时，才在之后再次提问。
diff --git a/deeptutor/capabilities/registry.py b/deeptutor/capabilities/registry.py
index bec40a4e19..7d056bd1e7 100644
--- a/deeptutor/capabilities/registry.py
+++ b/deeptutor/capabilities/registry.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from deeptutor.capabilities.ask_questions import AskQuestionsLoopCapability
 from deeptutor.capabilities.explore_context import ExploreContextCapability
 from deeptutor.capabilities.mastery import MasteryLoopCapability
 from deeptutor.capabilities.obsidian import ObsidianCapability
@@ -11,6 +12,7 @@
 from deeptutor.core.context import UnifiedContext
 
 LOOP_CAPABILITIES: tuple[LoopCapability, ...] = (
+    AskQuestionsLoopCapability(),
     MasteryLoopCapability(),
     SolveLoopCapability(),
     ObsidianCapability(),
diff --git a/deeptutor/runtime/bootstrap/builtin_capabilities.py b/deeptutor/runtime/bootstrap/builtin_capabilities.py
index 462d3188bc..e822177b40 100644
--- a/deeptutor/runtime/bootstrap/builtin_capabilities.py
+++ b/deeptutor/runtime/bootstrap/builtin_capabilities.py
@@ -2,6 +2,7 @@
 
 BUILTIN_CAPABILITY_CLASSES: dict[str, str] = {
     "chat": "deeptutor.agents.chat.capability:ChatCapability",
+    "ask_questions": ("deeptutor.capabilities.ask_questions.capability:AskQuestionsCapability"),
     "deep_solve": "deeptutor.capabilities.solve.capability:DeepSolveCapability",
     "deep_question": "deeptutor.agents.question.capability:DeepQuestionCapability",
     "deep_research": "deeptutor.agents.research.capability:DeepResearchCapability",
diff --git a/deeptutor/services/session/ask_user_trace.py b/deeptutor/services/session/ask_user_trace.py
new file mode 100644
index 0000000000..af580ba51f
--- /dev/null
+++ b/deeptutor/services/session/ask_user_trace.py
@@ -0,0 +1,111 @@
+"""Recover resolved ``ask_user`` exchanges from persisted assistant rows.
+
+Card replies resume the same backend turn, so they are stored in the assistant
+message's event trace rather than as standalone user messages. Rehydrate them
+into future model context or later turns would forget the answers and could ask
+the same questions again.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+#: Raw-text probe run before parsing a stored trace. An assistant row can hold
+#: a thousand streamed content deltas, and only the rare row that paused on
+#: ``ask_user`` is worth the JSON parse.
+_RESOLVED_MARKER = '"ask_user_resolved"'
+
+
+def _ask_user_payload(metadata: dict[str, Any]) -> dict[str, Any] | None:
+    tool_metadata = metadata.get("tool_metadata")
+    payload = (
+        tool_metadata.get("ask_user") if isinstance(tool_metadata, dict) else None
+    ) or metadata.get("ask_user")
+    return payload if isinstance(payload, dict) else None
+
+
+def _is_ask_user_event(event: dict[str, Any]) -> bool:
+    metadata = event.get("metadata")
+    if not isinstance(metadata, dict):
+        return False
+    if metadata.get("ask_user_resolved"):
+        return True
+    return event.get("type") == "tool_result" and _ask_user_payload(metadata) is not None
+
+
+def filter_ask_user_events(events: Any) -> list[dict[str, Any]]:
+    """Keep only the ask_user exchanges of an already-parsed event trace.
+
+    Context building never needs the streamed deltas, so the caller keeps only
+    these events instead of holding a full trace per message in memory.
+    """
+    if not isinstance(events, list):
+        return []
+    return [event for event in events if isinstance(event, dict) and _is_ask_user_event(event)]
+
+
+def select_ask_user_events(raw_events: str | None) -> list[dict[str, Any]]:
+    """Filter a stored ``events_json`` blob down to its ask_user exchanges."""
+    if not raw_events or _RESOLVED_MARKER not in raw_events:
+        return []
+    try:
+        return filter_ask_user_events(json.loads(raw_events))
+    except (TypeError, ValueError):
+        return []
+
+
+def extract_ask_user_clarifications(message: dict[str, Any]) -> str:
+    """Render a message's resolved ask_user exchanges as plain context text."""
+
+    pending_questions: dict[str, str] = {}
+    exchanges: list[tuple[str, str]] = []
+    for event in message.get("events") or []:
+        if not isinstance(event, dict):
+            continue
+        metadata = event.get("metadata") or {}
+        if not isinstance(metadata, dict):
+            continue
+        if event.get("type") == "tool_result":
+            ask_user = _ask_user_payload(metadata)
+            if ask_user is None:
+                continue
+            pending_questions = {
+                str(question.get("id") or ""): str(question.get("prompt") or "").strip()
+                for question in ask_user.get("questions") or []
+                if isinstance(question, dict) and str(question.get("prompt") or "").strip()
+            }
+            continue
+        if not metadata.get("ask_user_resolved"):
+            continue
+        answers = metadata.get("answers") or []
+        resolved = False
+        for answer in answers:
+            if not isinstance(answer, dict):
+                continue
+            question_id = str(answer.get("questionId") or answer.get("question_id") or "")
+            answer_text = str(answer.get("text") or "").strip()
+            question_text = pending_questions.get(question_id, question_id).strip()
+            if question_text and answer_text:
+                exchanges.append((question_text, answer_text))
+                resolved = True
+        if not resolved:
+            preview = str(metadata.get("reply_preview") or "").strip()
+            if preview:
+                question_text = next(iter(pending_questions.values()), "User clarification")
+                exchanges.append((question_text, preview))
+        pending_questions = {}
+
+    if not exchanges:
+        return ""
+    lines = ["[Earlier ask_user clarification — treat these answers as user-provided context]"]
+    for question, answer in exchanges:
+        lines.extend((f"- Question: {question}", f"  User answer: {answer}"))
+    return "\n".join(lines)
+
+
+__all__ = [
+    "extract_ask_user_clarifications",
+    "filter_ask_user_events",
+    "select_ask_user_events",
+]
diff --git a/deeptutor/services/session/context_builder.py b/deeptutor/services/session/context_builder.py
index 7aaed07239..8a7ebda29e 100644
--- a/deeptutor/services/session/context_builder.py
+++ b/deeptutor/services/session/context_builder.py
@@ -13,6 +13,7 @@
 from deeptutor.services.llm.config import LLMConfig
 from deeptutor.services.llm.context_window import resolve_effective_context_window
 
+from .ask_user_trace import extract_ask_user_clarifications
 from .protocol import SessionStoreProtocol
 
 #: When the summarizer's output lands within this fraction of its hard token
@@ -53,11 +54,15 @@ def format_messages_as_transcript(messages: list[dict[str, Any]]) -> str:
         "system": "System",
     }
     for item in messages:
+        # User clarifications answered an ask_user card *before* the assistant
+        # answer was produced, so render them before the row's own content.
+        clarification = extract_ask_user_clarifications(item)
+        if clarification:
+            lines.append(f"User: {clarification}")
         content = str(item.get("content", "") or "").strip()
-        if not content:
-            continue
-        role = role_map.get(str(item.get("role", "user")), "User")
-        lines.append(f"{role}: {content}")
+        if content:
+            role = role_map.get(str(item.get("role", "user")), "User")
+            lines.append(f"{role}: {content}")
     return "\n\n".join(lines)
 
 
@@ -141,15 +146,19 @@ def _build_history(self, summary: str, messages: list[dict[str, Any]]) -> list[d
         cleaned_summary = summary.strip()
         if cleaned_summary:
             history.append({"role": "system", "content": cleaned_summary})
-        history.extend(
-            {
-                "role": item.get("role", "user"),
-                "content": str(item.get("content", "") or ""),
-            }
-            for item in messages
-            if item.get("role") in {"user", "assistant"}
-            and str(item.get("content", "") or "").strip()
-        )
+        for item in messages:
+            role = item.get("role")
+            content = str(item.get("content", "") or "")
+            # Resolved ask_user clarifications happened *before* the assistant
+            # produced this row's answer, so they must precede the row's own
+            # content in history. Otherwise later turns see the answer first
+            # and the user's clarification second, and the model wrongly
+            # assumes the answer came before the user supplied the context.
+            clarification = extract_ask_user_clarifications(item)
+            if clarification:
+                history.append({"role": "user", "content": clarification})
+            if role in {"user", "assistant"} and content.strip():
+                history.append({"role": role, "content": content})
         return history
 
     async def _append_event(
@@ -171,7 +180,8 @@ def _select_recent_messages(
         total = 0
         for item in reversed(messages):
             content = str(item.get("content", "") or "")
-            tokens = count_tokens(content)
+            clarification = extract_ask_user_clarifications(item)
+            tokens = count_tokens(f"{content}\n{clarification}" if clarification else content)
             if selected and total + tokens > recent_budget:
                 break
             selected.insert(0, item)
@@ -474,5 +484,6 @@ async def build(
     "build_history_text",
     "count_tokens",
     "format_messages_as_transcript",
+    "extract_ask_user_clarifications",
     "trim_incomplete_tail",
 ]
diff --git a/deeptutor/services/session/pocketbase_store.py b/deeptutor/services/session/pocketbase_store.py
index 42e11902b6..ae1b53a6e8 100644
--- a/deeptutor/services/session/pocketbase_store.py
+++ b/deeptutor/services/session/pocketbase_store.py
@@ -30,6 +30,8 @@
 from typing import Any
 import uuid
 
+from .ask_user_trace import filter_ask_user_events
+
 logger = logging.getLogger(__name__)
 
 _VALID_ID = re.compile(r"^[a-zA-Z0-9_-]+$")
@@ -436,7 +438,12 @@ async def get_messages_for_context(
         _ = leaf_message_id
         messages = await self.get_messages(session_id)
         return [
-            {"id": m["id"], "role": m["role"], "content": m["content"] or ""}
+            {
+                "id": m["id"],
+                "role": m["role"],
+                "content": m["content"] or "",
+                "events": filter_ask_user_events(m.get("events")),
+            }
             for m in messages
             if m["role"] in ("user", "assistant", "system")
         ]
diff --git a/deeptutor/services/session/sqlite_store.py b/deeptutor/services/session/sqlite_store.py
index 8ef2ebab86..709fcaeb38 100644
--- a/deeptutor/services/session/sqlite_store.py
+++ b/deeptutor/services/session/sqlite_store.py
@@ -19,6 +19,8 @@
 
 from deeptutor.services.path_service import get_path_service
 
+from .ask_user_trace import select_ask_user_events
+
 
 def _json_dumps(value: Any) -> str:
     # default=str: a single non-serializable object inside an event payload
@@ -1285,7 +1287,7 @@ def _get_messages_for_context_sync(
             if leaf_message_id is None:
                 rows = conn.execute(
                     """
-                    SELECT id, role, content
+                    SELECT id, role, content, events_json
                     FROM messages
                     WHERE session_id = ?
                       AND role IN ('user', 'assistant', 'system')
@@ -1298,6 +1300,7 @@ def _get_messages_for_context_sync(
                         "id": row["id"],
                         "role": row["role"],
                         "content": row["content"] or "",
+                        "events": select_ask_user_events(row["events_json"]),
                     }
                     for row in rows
                 ]
@@ -1309,7 +1312,7 @@ def _get_messages_for_context_sync(
             while current is not None and safety > 0:
                 row = conn.execute(
                     """
-                    SELECT id, role, content, parent_message_id
+                    SELECT id, role, content, events_json, parent_message_id
                     FROM messages
                     WHERE id = ? AND session_id = ?
                       AND role IN ('user', 'assistant', 'system')
@@ -1323,6 +1326,7 @@ def _get_messages_for_context_sync(
                         "id": row["id"],
                         "role": row["role"],
                         "content": row["content"] or "",
+                        "events": select_ask_user_events(row["events_json"]),
                     }
                 )
                 parent = row["parent_message_id"]
diff --git a/tests/agents/chat/test_agent_loop.py b/tests/agents/chat/test_agent_loop.py
index ab035d53c1..863fe9e950 100644
--- a/tests/agents/chat/test_agent_loop.py
+++ b/tests/agents/chat/test_agent_loop.py
@@ -708,6 +708,64 @@ async def test_ask_user_available_every_round(monkeypatch: pytest.MonkeyPatch) -
     assert loop_tools == {"web_search", "ask_user"}
 
 
+@pytest.mark.asyncio
+async def test_initial_tool_choice_only_forces_first_round_and_hides_preamble(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class _PausingRegistry(_Registry):
+        async def execute(self, name: str, **kwargs):
+            self.executed.append({"name": name, "kwargs": kwargs})
+            return ToolResult(
+                content="Asked the user.",
+                success=True,
+                pause_for_user={"questions": [{"id": "q1", "prompt": "What matters most?"}]},
+            )
+
+    registry = _PausingRegistry()
+    client = _ScriptedChatClient(
+        [
+            [
+                _llm_chunk(content="Let me ask one thing first."),
+                _llm_chunk(
+                    tool_calls=[
+                        {
+                            "id": "call-1",
+                            "name": "ask_user",
+                            "arguments": json.dumps(
+                                {"questions": [{"id": "q1", "prompt": "What matters most?"}]}
+                            ),
+                        }
+                    ]
+                ),
+            ],
+            [_llm_chunk(content="Completed with the added context.")],
+        ]
+    )
+    pipeline = AgenticChatPipeline(language="en", initial_tool_choice="ask_user")
+    pipeline.registry = registry
+    monkeypatch.setattr(pipeline, "_compose_enabled_tools", lambda _context: ["ask_user"])
+    monkeypatch.setattr(pipeline, "_build_openai_client", lambda: client)
+
+    async def _waiter():
+        return {"text": "Accuracy"}
+
+    events = await _run(
+        pipeline,
+        UnifiedContext(
+            session_id="s1",
+            user_message="Help with this task",
+            metadata={"wait_for_user_reply": _waiter},
+        ),
+    )
+
+    assert client.calls[0]["tool_choice"] == {
+        "type": "function",
+        "function": {"name": "ask_user"},
+    }
+    assert client.calls[1]["tool_choice"] == "auto"
+    assert _contents(events) == ["Completed with the added context."]
+
+
 @pytest.mark.asyncio
 async def test_ask_user_pause_resumes_and_streams_interleaved(
     monkeypatch: pytest.MonkeyPatch,
diff --git a/tests/agents/chat/test_language_prompts.py b/tests/agents/chat/test_language_prompts.py
index 113458bd9a..21d1436348 100644
--- a/tests/agents/chat/test_language_prompts.py
+++ b/tests/agents/chat/test_language_prompts.py
@@ -77,6 +77,35 @@ def build_prompt_text(self, *_args, **_kwargs) -> str:
     assert "Mastery Tutor mode" in en_prompt
 
 
+def test_ask_questions_plugin_system_prompt_uses_localized_fallback(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class FakeRegistry:
+        def build_prompt_text(self, *_args, **_kwargs) -> str:
+            return "- tool"
+
+    monkeypatch.setattr(
+        "deeptutor.agents.chat.agentic_pipeline.get_tool_registry",
+        lambda: FakeRegistry(),
+    )
+
+    from deeptutor.core.context import UnifiedContext
+
+    ctx = UnifiedContext(metadata={"ask_questions_mode": True})
+    zh_prompt = AgenticChatPipeline(language="zh")._build_system_prompt([], ctx)
+    en_prompt = AgenticChatPipeline(language="en")._build_system_prompt([], ctx)
+
+    assert "## ask_questions" in zh_prompt
+    assert "主动提问模式" in zh_prompt
+    assert "必须以一次 `ask_user` 调用开始" in zh_prompt
+    assert "第 2、3、10 轮" in zh_prompt
+    assert "此前所有对话" in zh_prompt
+    assert "## ask_questions" in en_prompt
+    assert "Ask Questions mode" in en_prompt
+    assert "second, third, tenth" in en_prompt
+    assert "calling `ask_user` exactly once" in en_prompt
+
+
 def test_legacy_chat_agent_system_prompt_uses_selected_language() -> None:
     zh_messages = ChatAgent(language="zh", config={}).build_messages(
         message="解释梯度下降",
diff --git a/tests/cli/test_chat_cli.py b/tests/cli/test_chat_cli.py
index f341eecbac..d5fe21a890 100644
--- a/tests/cli/test_chat_cli.py
+++ b/tests/cli/test_chat_cli.py
@@ -78,6 +78,7 @@ def test_builtin_capability_aliases_resolve_to_canonical_names() -> None:
     runtime = DeepTutorApp()
 
     assert runtime.resolve_capability("solve") == "deep_solve"
+    assert runtime.resolve_capability("ask") == "ask_questions"
     assert runtime.resolve_capability("quiz") == "deep_question"
     assert runtime.resolve_capability("research") == "deep_research"
     assert runtime.resolve_capability("viz") == "visualize"
diff --git a/tests/core/test_capabilities_runtime.py b/tests/core/test_capabilities_runtime.py
index 5a876eb038..e50ed7cbdd 100644
--- a/tests/core/test_capabilities_runtime.py
+++ b/tests/core/test_capabilities_runtime.py
@@ -15,6 +15,7 @@
 from deeptutor.agents.research.capability import DeepResearchCapability
 from deeptutor.agents.visualize.capability import VisualizeCapability
 import deeptutor.agents.visualize.pipeline as visualize_pipeline
+from deeptutor.capabilities.ask_questions.capability import AskQuestionsCapability
 from deeptutor.capabilities.solve.capability import DeepSolveCapability
 from deeptutor.core.context import Attachment, UnifiedContext
 from deeptutor.core.stream import StreamEvent, StreamEventType
@@ -69,6 +70,7 @@ async def _consume() -> None:
 def test_builtin_capability_registry_covers_documented_capabilities() -> None:
     assert set(BUILTIN_CAPABILITY_CLASSES) == {
         "chat",
+        "ask_questions",
         "deep_solve",
         "deep_question",
         "deep_research",
@@ -78,6 +80,43 @@ def test_builtin_capability_registry_covers_documented_capabilities() -> None:
     }
 
 
+@pytest.mark.asyncio
+async def test_ask_questions_capability_forces_card_on_selected_turn(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    captured: dict[str, Any] = {}
+
+    class FakePipeline:
+        def __init__(
+            self,
+            *,
+            language: str = "en",
+            initial_tool_choice: str | None = None,
+        ) -> None:
+            captured["language"] = language
+            captured["initial_tool_choice"] = initial_tool_choice
+
+        async def run(self, context: UnifiedContext, stream: StreamBus) -> None:
+            captured["ask_questions_mode"] = context.metadata.get("ask_questions_mode")
+            await stream.content("question", source="chat", stage="responding")
+
+    monkeypatch.setattr(
+        "deeptutor.capabilities.ask_questions.capability.AgenticChatPipeline",
+        FakePipeline,
+    )
+
+    context = UnifiedContext(user_message="Help me plan", language="zh")
+    capability = AskQuestionsCapability()
+    events = await _collect_events(lambda bus: capability.run(context, bus))
+
+    assert captured == {
+        "language": "zh",
+        "initial_tool_choice": "ask_user",
+        "ask_questions_mode": True,
+    }
+    assert any(event.type == StreamEventType.CONTENT for event in events)
+
+
 @pytest.mark.asyncio
 async def test_chat_capability_streams_content_and_geogebra_context(
     monkeypatch: pytest.MonkeyPatch,
diff --git a/tests/services/session/test_context_builder.py b/tests/services/session/test_context_builder.py
index 9b4f8ca83c..81fb55cb39 100644
--- a/tests/services/session/test_context_builder.py
+++ b/tests/services/session/test_context_builder.py
@@ -12,6 +12,7 @@
     ContextBuildResult,
     build_history_text,
     count_tokens,
+    extract_ask_user_clarifications,
     format_messages_as_transcript,
     trim_incomplete_tail,
 )
@@ -222,6 +223,47 @@ def test_empty_content_filtered(self) -> None:
         )
         assert len(history) == 1
 
+    def test_resolved_ask_user_answers_are_rehydrated_as_user_context(self) -> None:
+        message = {
+            "role": "assistant",
+            "content": "I adapted the explanation.",
+            "events": [
+                {
+                    "type": "tool_result",
+                    "metadata": {
+                        "tool_metadata": {
+                            "ask_user": {
+                                "questions": [
+                                    {"id": "level", "prompt": "What have you studied?"},
+                                    {"id": "goal", "prompt": "What is your goal?"},
+                                ]
+                            }
+                        }
+                    },
+                },
+                {
+                    "type": "progress",
+                    "metadata": {
+                        "ask_user_resolved": True,
+                        "answers": [
+                            {"questionId": "level", "text": "High-school calculus"},
+                            {"questionId": "goal", "text": "Understand the intuition"},
+                        ],
+                    },
+                },
+            ],
+        }
+
+        clarification = extract_ask_user_clarifications(message)
+        assert "What have you studied?" in clarification
+        assert "High-school calculus" in clarification
+
+        history = ContextBuilder(store=MagicMock())._build_history("", [message])
+        assert history == [
+            {"role": "user", "content": clarification},
+            {"role": "assistant", "content": "I adapted the explanation."},
+        ]
+
 
 # ---------------------------------------------------------------------------
 # ContextBuilder._select_recent_messages
diff --git a/tests/services/session/test_sqlite_store.py b/tests/services/session/test_sqlite_store.py
index 80984fd3f8..2cd7ac0a0b 100644
--- a/tests/services/session/test_sqlite_store.py
+++ b/tests/services/session/test_sqlite_store.py
@@ -265,3 +265,53 @@ def test_category_cascade_on_entry_delete(store: SQLiteSessionStore) -> None:
     asyncio.run(store.delete_notebook_entry(eid))
     cats = asyncio.run(store.list_categories())
     assert cats[0]["entry_count"] == 0
+
+
+# ── Context messages ──────────────────────────────────────────────
+
+
+_ASK_USER_EVENTS = [
+    {"type": "content", "content": "streamed delta", "metadata": {}},
+    {
+        "type": "tool_result",
+        "metadata": {
+            "tool_metadata": {"ask_user": {"questions": [{"id": "level", "prompt": "Your level?"}]}}
+        },
+    },
+    {
+        "type": "progress",
+        "metadata": {
+            "ask_user_resolved": True,
+            "answers": [{"questionId": "level", "text": "Beginner"}],
+        },
+    },
+]
+
+
+def _add_ask_user_turn(store: SQLiteSessionStore, session_id: str) -> None:
+    asyncio.run(store.add_message(session_id, "user", "Plan my study"))
+    asyncio.run(
+        store.add_message(session_id, "assistant", "Here is a plan", events=_ASK_USER_EVENTS)
+    )
+
+
+def test_context_messages_carry_ask_user_events(store: SQLiteSessionStore) -> None:
+    session = asyncio.run(store.create_session())
+    _add_ask_user_turn(store, session["id"])
+
+    messages = asyncio.run(store.get_messages_for_context(session["id"]))
+
+    assert [m["role"] for m in messages] == ["user", "assistant"]
+    # Streamed deltas are dropped; only the ask_user exchange survives, so a
+    # later turn can see which questions the learner already answered.
+    assert [e["type"] for e in messages[1]["events"]] == ["tool_result", "progress"]
+
+
+def test_branch_context_messages_carry_ask_user_events(store: SQLiteSessionStore) -> None:
+    session = asyncio.run(store.create_session())
+    _add_ask_user_turn(store, session["id"])
+    leaf = asyncio.run(store.add_message(session["id"], "user", "Still not right"))
+
+    messages = asyncio.run(store.get_messages_for_context(session["id"], leaf_message_id=leaf))
+
+    assert [e["type"] for e in messages[1]["events"]] == ["tool_result", "progress"]
diff --git a/web/app/(workspace)/home/[[...sessionId]]/page.tsx b/web/app/(workspace)/home/[[...sessionId]]/page.tsx
index 45451eeb41..ba27fefbe2 100644
--- a/web/app/(workspace)/home/[[...sessionId]]/page.tsx
+++ b/web/app/(workspace)/home/[[...sessionId]]/page.tsx
@@ -14,6 +14,7 @@ import { useParams, useRouter } from "next/navigation";
 import {
   BarChart3,
   BrainCircuit,
+  CircleHelp,
   Clapperboard,
   Code2,
   Compass,
@@ -240,6 +241,24 @@ const CAPABILITIES: CapabilityDef[] = [
     defaultTools: ["web_search", "code_execution", "reason"],
     loopEngine: true,
   },
+  {
+    value: "ask_questions",
+    label: "Ask Questions",
+    description: "Let the model ask you questions to fill in missing context",
+    icon: CircleHelp,
+    allowedTools: [
+      "brainstorm",
+      "geogebra_analysis",
+      "web_search",
+      "code_execution",
+      "reason",
+      "paper_search",
+      "imagegen",
+      "videogen",
+    ],
+    defaultTools: [],
+    loopEngine: true,
+  },
   {
     value: "deep_question",
     label: "Quiz",
diff --git a/web/components/chat/home/ChatMessages.tsx b/web/components/chat/home/ChatMessages.tsx
index 19fd8da35b..aa62331453 100644
--- a/web/components/chat/home/ChatMessages.tsx
+++ b/web/components/chat/home/ChatMessages.tsx
@@ -107,6 +107,7 @@ interface NotebookReferenceGroup {
 // the same wording the bubble carries.
 export function getModeBadgeLabel(capability?: string | null): string {
   if (!capability || capability === "chat") return "Chat";
+  if (capability === "ask_questions") return "Ask Questions";
   if (capability === "deep_solve") return "Deep Solve";
   if (capability === "deep_question") return "Quiz Generation";
   if (capability === "deep_research") return "Deep Research";
diff --git a/web/locales/en/app.json b/web/locales/en/app.json
index 1fd8706db3..d9058b0b3c 100644
--- a/web/locales/en/app.json
+++ b/web/locales/en/app.json
@@ -283,6 +283,8 @@
   "Solve": "Solve",
   "More Capabilities": "More Capabilities",
   "Agent-loop driven modes": "Agent-loop driven modes",
+  "Ask Questions": "Ask Questions",
+  "Let the model ask you questions to fill in missing context": "Let the model ask you questions to fill in missing context",
   "Question": "Question",
   "No notebooks yet": "No notebooks yet",
   "Create your first notebook": "Create your first notebook",
diff --git a/web/locales/zh/app.json b/web/locales/zh/app.json
index 826cf9b57a..5af33f2ce9 100644
--- a/web/locales/zh/app.json
+++ b/web/locales/zh/app.json
@@ -283,6 +283,8 @@
   "Solve": "解题",
   "More Capabilities": "更多能力",
   "Agent-loop driven modes": "由对话引擎驱动的模式",
+  "Ask Questions": "主动提问",
+  "Let the model ask you questions to fill in missing context": "让模型向你提问，补充它尚不了解的上下文",
   "Question": "题目",
   "No notebooks yet": "暂无笔记本",
   "Create your first notebook": "创建您的第一个笔记本",