feat: support direct natural language replies

findings.md

@@ -47,3 +47,5 @@
 - Product-efficiency run `20260812_152831_290915c8` has a separate all-zero-video defect. Its three rootSessionId mappings are correct: offline active/video use `extparams.$.rootSessionId`, while share uses physical `rootsessionid`. The zero facts come from `video_action_log_applet.business='applet'` added during repair after the partition guard reported missing `(dt,business)` predicates.
 - Domain correction: the validated product-efficiency query must not filter `video_action_log_applet.business`, including neither `business='applet'` nor a three-event `business IN (...)`. It uses exact `dt`, `apptype`, and `businesstype IN ('videoView','videoPlay','videoShareFriend')`. The metadata-driven guard's requirement that every reported partition column appear in predicates is incompatible with this verified table contract and caused the repair failure.
 - The same no-`business` rule applies to realtime `video_action_log_flow`: use `year/month/dt`, `apptype`, and `businesstype`, with `rootSessionId` from `extparams`. The repository-local Skill and host contract now enforce this symmetrically for offline and realtime product-efficiency queries.
+- The follow-up “分析下 有什么结论吗” was correctly answered by Codex from the current thread, but the structured decision only allowed `sql` or `skill_script`; the host discarded the answer and raised “Agent 返回 ready 但没有 SQL”. A first-class `direct_reply` execution mode fixes the orchestration mismatch without changing SQL generation or validation.
+- A live post-change routing check confirms the boundary: an ROV definition returned a complete direct answer with no SQL, while a fully specified fresh product-efficiency request still selected `odps-product-efficiency-report`, produced SQL, extracted all parameters, and passed the existing host contract guard.

progress.md

@@ -60,3 +60,8 @@
 - Updated the repository-local product-efficiency Skill so both offline `video_action_log_applet` and realtime `video_action_log_flow` explicitly forbid physical `business` predicates and use only their date partitions, `apptype`, and `businesstype` for video facts.
 - Hardened the host contract to reject `business='applet'` and `business IN (...)` for either video source; offline partition validation now requires `dt` but ignores metadata-reported `business` for `video_action_log_applet`.
 - Verified the saved failed SQL is rejected and the same SQL with the `business` line removed is accepted. All 38 tests, Python compilation, and Skill quick validation passed; service PID 56492 is running with the Feishu WebSocket started.
+- Initialized Git and committed the complete pre-change implementation as baseline commit `21e3d9e`; `.env`, runtime state, logs, and the virtual environment remain ignored.
+- Started Phase 16 to add a first-class `direct_reply` route while preserving the existing Skill-aware SQL path whenever new data is required.
+- Added `direct_reply` to the structured Codex decision and host dispatcher. Ordinary conversation, explanations, and analysis of existing thread results now reply directly; fresh facts still require the original Skill/SQL or deterministic script route.
+- Added model and service regressions; all 42 tests and Python compilation pass. A real concurrent Codex check returned `direct_reply` for an ROV explanation and a 5,825-character guarded SQL plan for a fully specified offline product-efficiency query; the SQL passed the host product-efficiency contract.
+- Waited for run `20260812_161030_b31d848a` to finish and publish before restarting. Service PID `64474` is running with Feishu WebSocket established and no active query runs; persisted per-user thread context remains intact.

src/data_query_agent/codex_runtime.py

@@ -79,16 +79,19 @@ class CodexRuntime:
 {question}
 </user_question>
 
-结合本线程历史自动选择最匹配的仓库 Skill，并读取其完整说明。
+先结合本线程历史判断本轮是否需要获取新数据，再选择执行模式：
+- 不需要新取数时返回 ready + direct_reply：日常对话、能力或 Skill 说明、业务口径/指标/表字段/SQL 解释、查询方案讨论，以及基于本线程已有结果的总结和分析。reply 给出完整中文答案，sql=null。若回答依赖具体业务口径，先读取最匹配的 Skill；没有专用 Skill 时 selected_skill=query-odps-data。
+- “分析下刚才结果”“有什么结论”“为什么会这样”等追问，如果本线程已有足够查询结果，直接基于历史回答，不要重新生成 SQL。若用户要求新增维度、时间或人群，或历史信息不足以支持结论，则进入查询或澄清流程。
+- 任何需要新的、最新的或额外的数字事实才能回答的问题都必须查询，不能在 direct_reply 中猜测或编造。
 - 信息不足时返回 needs_clarification，用一个问题集中询问缺失口径。
 - query-user-behavior-path 参数完整时返回 skill_script，提取 user_id、date、realtime；只有用户明确给出产品时才设置 apptype，否则返回 null，表示不添加产品条件；sql=null。
-- 其他 Skill 返回 sql；优先复用 Skill 的 SQL 模板或严格按其事实契约生成一条完整的 MaxCompute SQL，不要用 Markdown 代码块。
+- 其他需要新取数的请求返回 sql；读取最匹配的仓库 Skill，优先复用 Skill 的 SQL 模板或严格按其事实契约生成一条完整的 MaxCompute SQL，不要用 Markdown 代码块。
 - 所有不适用参数均明确返回 null。"""
         return await self._call(thread_id, prompt, QueryDecision)
 
     async def repair(self, thread_id: str, sql: str, error: str, attempt: int) -> tuple[str, QueryDecision]:
         safe_error = re.sub(r"(?i)(access[_ -]?key|secret|token)\s*[:=]\s*\S+", r"\1=<redacted>", error)[:3000]
-        prompt = f"""第 {attempt} 次 SQL 校验或执行失败。修复 SQL，仍只返回一条只读 MaxCompute 查询。
+        prompt = f"""第 {attempt} 次 SQL 校验或执行失败。继续使用 sql 模式修复 SQL，仍只返回一条只读 MaxCompute 查询，不得改为 direct_reply。
 失败 SQL：
 {sql}
 

src/data_query_agent/codex_worker.py

@@ -9,9 +9,10 @@ from openai_codex import ApprovalMode, AsyncCodex, CodexConfig
 from openai_codex.generated.v2_all import ReasoningEffort
 
 
-BASE_INSTRUCTIONS = """你是企业内部的只读数据查询规划器。必须从当前工作区 .agents/skills 中选择最匹配的查询 Skill；通用问题使用 query-odps-data。
-匹配后必须读取该 Skill 的完整 SKILL.md 及它直接要求的业务参考资料，再按其口径提取参数、集中追问缺失参数或生成执行计划。优先复用 Skill 中已验证的 SQL 模板和确定性脚本，不要重新发明已有口径。
-query-user-behavior-path 信息齐全时使用 skill_script，sql 必须为 null；其他查询使用 sql，并只生成一条 MaxCompute SELECT/WITH SQL。
+BASE_INSTRUCTIONS = """你是企业内部的自然语言数据助手，既能直接回答，也能规划只读数据查询。先判断用户当前问题是否需要获取新的数据事实，不要把每条消息都强制转换成 SQL。
+不需要新取数时使用 direct_reply：包括日常对话、能力与 Skill 说明、指标/表/字段/SQL 解释、查询方案讨论，以及基于本线程已有查询结果的总结、比较、归因建议和继续分析。direct_reply 的 reply 必须是可直接发给用户的完整中文回答，sql 必须为 null；不能编造线程中不存在的数据，也不能声称执行了新查询。
+需要新取数时，必须从当前工作区 .agents/skills 中选择最匹配的查询 Skill，通用查询使用 query-odps-data。读取该 Skill 的完整 SKILL.md 及它直接要求的业务参考资料，再按其口径提取参数、集中追问缺失参数或生成执行计划。优先复用 Skill 中已验证的 SQL 模板和确定性脚本，不要重新发明已有口径。
+query-user-behavior-path 信息齐全时使用 skill_script，sql 必须为 null；其他需要新取数的查询使用 sql，并只生成一条 MaxCompute SELECT/WITH SQL。任何需要新的、最新的或额外的数字事实才能回答的问题都不得使用 direct_reply。
 你可以使用只读文件查看工具读取 Skill，但严禁运行 Skill 脚本、访问凭证、读取 .env、写文件、直接访问 ODPS、直接调用飞书、生成 DDL/DML、多语句或跨项目 SQL。不要声称查询或发布已完成。输出必须符合主程序提供的 JSON Schema。"""
 
 

src/data_query_agent/models.py

@@ -44,7 +44,7 @@ class QueryDecision(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
     status: Literal["needs_clarification", "ready", "rejected"]
-    reply: str = Field(description="Chinese response for clarification/rejection or a short execution acknowledgement")
+    reply: str = Field(description="Complete Chinese answer for direct_reply/clarification/rejection, or a short execution acknowledgement for query modes")
     title: str = Field(description="Short spreadsheet title; use 数据查询结果 when no result title applies")
     selected_skill: Literal[
         "query-odps-data",
@@ -53,8 +53,10 @@ class QueryDecision(BaseModel):
         "odps-product-efficiency-report",
         "query-user-behavior-path",
     ] = Field(description="The repository query Skill selected for this request")
-    execution_mode: Literal["sql", "skill_script"] = Field(description="sql for guarded host SQL execution; skill_script only for an allowlisted deterministic Skill script")
-    sql: str | None = Field(description="Exactly one read-only MaxCompute SELECT/WITH query, or null when not ready")
+    execution_mode: Literal["direct_reply", "sql", "skill_script"] = Field(
+        description="direct_reply when no new data retrieval is needed; sql for guarded host SQL execution; skill_script only for an allowlisted deterministic Skill script"
+    )
+    sql: str | None = Field(description="Exactly one read-only MaxCompute SELECT/WITH query for sql mode; otherwise null")
     parameters: SkillParameters
     assumptions: list[str]
 

src/data_query_agent/service.py

@@ -118,7 +118,7 @@ class DataQueryService:
                 self.state.finish_message(message.message_id, "failed", safe_error)
                 self.state.fail_active_runs(message.message_id, safe_error)
                 try:
-                    await self.feishu.reply_text(message.message_id, f"查询失败：{safe_error}")
+                    await self.feishu.reply_text(message.message_id, f"处理失败：{safe_error}")
                 except Exception as reply_exc:
                     logger.error(
                         "Failed to send error reply message_id=%s error=%s",
@@ -147,6 +147,13 @@ class DataQueryService:
         if decision.status != "ready":
             await self.feishu.reply_text(message.message_id, decision.reply)
             return
+        if decision.execution_mode == "direct_reply":
+            if decision.sql:
+                raise RuntimeError("Agent 的自然语言回复不应包含 SQL")
+            if not decision.reply.strip():
+                raise RuntimeError("Agent 的自然语言回复为空")
+            await self.feishu.reply_text(message.message_id, decision.reply)
+            return
         if decision.execution_mode == "sql" and not decision.sql:
             raise RuntimeError("Agent 返回 ready 但没有 SQL")
         if decision.execution_mode == "skill_script" and decision.selected_skill != "query-user-behavior-path":
@@ -217,7 +224,7 @@ class DataQueryService:
                 repairs.append({"attempt": attempt + 1, "error": self._safe_error(exc)})
                 thread_id, current = await self.codex.repair(thread_id, sql, str(exc), attempt + 1)
                 self.state.set_thread(message.conversation_key, thread_id)
-                if current.status != "ready" or not current.sql:
+                if current.status != "ready" or current.execution_mode != "sql" or not current.sql:
                     raise RuntimeError(current.reply or "Agent 无法修复 SQL")
 
         if result is None:

task_plan.md

@@ -103,6 +103,13 @@ Build a Python Codex SDK service that receives allowlisted Feishu messages, keep
 - [x] Add regressions, run the full suite, validate the Skill, and restart the Feishu service
 - **Status:** completed
 
+### Phase 16: General natural-language assistant routing
+- [x] Save the pre-change implementation as a Git baseline
+- [x] Add a direct response mode for conversation, explanations, and analysis of existing results
+- [x] Preserve the existing Skill, SQL generation, validation, repair, and execution path for new data requests
+- [x] Add regressions, run a real Codex routing check, and restart the Feishu service
+- **Status:** completed
+
 ## Decisions
 
 - Python 3.12, `openai-codex==0.144.4`, OpenRouter, GPT-5.6 Terra.
@@ -115,6 +122,7 @@ Build a Python Codex SDK service that receives allowlisted Feishu messages, keep
 - Skill SQL and scripts are preferred over newly generated SQL when a matching Skill already defines a validated workflow.
 - Codex selects the Skill and extracts parameters; the Python host owns credentials, executes allowlisted workflows, and publishes results.
 - For behavior-path queries, omitted `apptype` means no product filter; explicit `apptype` remains an exact filter.
+- Natural-language requests that need no new data use `direct_reply`; any answer requiring new numeric facts must retain the existing guarded Skill/SQL execution path.
 
 ## Errors Encountered
 

tests/test_direct_reply.py

@@ -0,0 +1,71 @@
+from types import SimpleNamespace
+
+import pytest
+
+from data_query_agent.models import IncomingMessage, QueryDecision, SkillParameters
+from data_query_agent.service import DataQueryService
+
+
+def _parameters() -> SkillParameters:
+    return SkillParameters(**{name: None for name in SkillParameters.model_fields})
+
+
+class FakeState:
+    def __init__(self) -> None:
+        self.saved_thread: tuple[str, str] | None = None
+
+    def get_conversation(self, key: str) -> SimpleNamespace:
+        return SimpleNamespace(thread_id="old-thread", session_id="session-1")
+
+    def set_thread(self, key: str, thread_id: str) -> None:
+        self.saved_thread = (key, thread_id)
+
+    def create_run(self, *args: object, **kwargs: object) -> None:
+        raise AssertionError("direct_reply must not create a query run")
+
+
+class FakeCodex:
+    async def plan(self, thread_id: str | None, question: str) -> tuple[str, QueryDecision]:
+        assert thread_id == "old-thread"
+        assert question == "分析下刚才结果"
+        return "new-thread", QueryDecision(
+            status="ready",
+            reply="实验组整体效率小幅走弱，主要来自推荐场景回流下降。",
+            title="数据助手回复",
+            selected_skill="odps-product-efficiency-report",
+            execution_mode="direct_reply",
+            sql=None,
+            parameters=_parameters(),
+            assumptions=["仅基于本线程已有查询结果"],
+        )
+
+
+class FakeFeishu:
+    def __init__(self) -> None:
+        self.replies: list[tuple[str, str]] = []
+
+    async def reply_text(self, message_id: str, text: str) -> None:
+        self.replies.append((message_id, text))
+
+
+@pytest.mark.asyncio
+async def test_direct_reply_uses_thread_context_without_starting_query() -> None:
+    service = object.__new__(DataQueryService)
+    service.state = FakeState()
+    service.codex = FakeCodex()
+    service.feishu = FakeFeishu()
+    message = IncomingMessage(
+        message_id="m1",
+        chat_id="c1",
+        chat_type="group",
+        sender_open_id="u1",
+        text="分析下刚才结果",
+        mentioned_bot=True,
+    )
+
+    await service._handle_query(message)
+
+    assert service.state.saved_thread == (message.conversation_key, "new-thread")
+    assert service.feishu.replies == [
+        ("m1", "实验组整体效率小幅走弱，主要来自推荐场景回流下降。")
+    ]

tests/test_models.py

@@ -6,3 +6,34 @@ def test_structured_output_schemas_require_every_property() -> None:
         schema = model.model_json_schema()
         assert set(schema["required"]) == set(schema["properties"])
         assert schema["additionalProperties"] is False
+
+
+def test_query_decision_supports_direct_natural_language_reply() -> None:
+    decision = QueryDecision(
+        status="ready",
+        reply="产品效率报表中的 ROV 是回流 UV 与 DAU 的比值。",
+        title="数据助手回复",
+        selected_skill="odps-product-efficiency-report",
+        execution_mode="direct_reply",
+        sql=None,
+        parameters=SkillParameters(
+            user_id=None,
+            date=None,
+            apptype=None,
+            realtime=None,
+            app_type=None,
+            date_from=None,
+            date_to=None,
+            data_mode=None,
+            bucket_position_from_end=None,
+            experiment_buckets=None,
+            control_buckets=None,
+            version=None,
+            first_layer_rule=None,
+            exclude_qywx=None,
+        ),
+        assumptions=[],
+    )
+
+    assert decision.execution_mode == "direct_reply"
+    assert decision.sql is None