feat(chat): add AskQuestions mode for contextual clarification

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",

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] = []

deeptutor/capabilities/ask_questions/__init__.py

@@ -0,0 +1,5 @@
+"""Ask Questions capability exports."""
+
+from deeptutor.capabilities.ask_questions.loop import AskQuestionsLoopCapability
+
+__all__ = ["AskQuestionsLoopCapability"]

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"]

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"]

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.

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 中根据新增上下文继续原始请求，不要只回复一句确认收到。只有当回答或后续工具结果又暴露了新的实质性信息缺口时，才在之后再次提问。

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(),

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",

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",
+]