Phase 4：补齐通用记忆闭环（验收 9/9 通过）

按 P1-Pipecat Flows Agent化与记忆改造交接.md 的 Phase 4 完成。
新增 p1/phase4_acceptance.py：跑真实服务端验 7 条，每条之间查数据库。

实测结果（2026-09-16）
1. "记住我不吃香菜"        → confirmed 落库 ✅
2. **新连接** "我不吃什么？" → "你不吃香菜。" ✅ 连接开始时加载档案生效
3. "我今天吃了香菜"        → 不新增 confirmed，模型还追问了是否口味变了 ✅
4. "其实我现在吃香菜了"    → 同键仍只一条 confirmed，值已更新 ✅
5. "忘掉我的饮食偏好"      → 已删除 ✅
6/7. 访客与跨用户隔离       → 由单测覆盖（客户端不支持切身份）

实现：会话开始时加载稳定档案
- load_stable_memories()：取已确认、未过期记忆（身份未确认返回空）
- _conversation_node 改用 build_system_prompt，与 legacy 共用同一套
  人格与记忆注入格式；记忆进**同一条** system 消息，仍不用 role_message

修掉一个真实缺陷：Flows 路径误用了语音的 max_tokens
- 症状：一句"忘掉我的饮食偏好"，服务端日志
  "Failed to parse function call arguments: {"query": "饮食"（JSON 被截断），
  回复为空、记忆没删掉，且**没有任何报错**
- 根因：Flows 路径沿用了语音的 max_tokens=200（为"回复要短"而设）。
  原始 API 对照实测：max_tokens=200 时模型 finish=stop、根本不调工具；
  600 时正常产生 tool_calls 且参数 JSON 完整
- 修法：新增 AgentConfig.max_tokens（AI_AGENT_MAX_TOKENS，默认 600），
  Flows 路径用它，语音路径仍用 200

新增 client/test_client.py 能力（为验收所需）
- RTVIReplies：收服务端 RTVI 文本消息（bot-output / bot-transcription）
- 打印 BOT_REPLY 供验收脚本断言

回归：test_agent_tools 45/45、test_flows_runtime 24/24、
test_memory 30/30、test_identity 19/19

服务器已改回 AI_AGENT_RUNTIME=legacy（改默认是 Phase 7 的事）

app/agent/__init__.py

@@ -0,0 +1,35 @@
+"""Agent 层：工具、记忆策略、Flows 装配。
+
+Phase 2 起建立。**两条运行路径共用这一层**：
+
+  - legacy：`ConversationEngine` 的手写工具循环
+  - flows：Pipecat Flows（Phase 3 接入，用 `AI_AGENT_RUNTIME` 切换）
+
+分层意图：
+
+  tools.py          工具 schema 与执行器（身份由 ToolContext 注入）
+  memory_policy.py  记忆写入/删除的裁决（模型只提议，策略决定）
+  flows_runtime.py  FlowManager 装配（Phase 3）
+
+策略层是本层最重要的部分：交接文档要求
+"不要仅依赖'请不要误记'的 system prompt"，
+所以这些判断必须是可测试的代码，而不是提示词里的请求。
+"""
+
+from .memory_policy import (
+    MemoryDecision,
+    PolicyResult,
+    decide_forget,
+    decide_remember,
+)
+from .tools import (
+    ToolContext,
+    ToolOutcome,
+    available_schemas,
+    execute,
+)
+
+__all__ = [
+    "ToolContext", "ToolOutcome", "execute", "available_schemas",
+    "MemoryDecision", "PolicyResult", "decide_remember", "decide_forget",
+]

app/agent/flows_runtime.py

@@ -0,0 +1,378 @@
+"""Pipecat Flows 运行时装配（Phase 3）。
+
+**只接管网页文字路径。** 语音仍走 legacy 的 `EngineProcessor`——交接文档
+明确要求 "语音路径暂时保持 legacy"，等 Phase 3、4 稳定后再迁。
+
+管线形态：
+
+    transport.input()
+      → EngineProcessor(owns_text=False)   语音仍由它处理；文字帧往下传
+      → flows 的 context aggregator.user()
+      → DeepSeekLLMService                 Flows 驱动
+      → FlowsBridge                        LLM 文本 → RTVI 消息 + 分句 TTS
+      → flows 的 context aggregator.assistant()
+      → transport.output()
+
+两个关键设计：
+
+  * **只用 `task_messages`，不用 `role_message`。** Phase 1 spike 实测：
+    两者会各产生一条 system 消息进请求。只用一种。这里把人格与任务都写进
+    `task_messages`，与 legacy"每轮重建单条 system"的做法一致。
+  * **工具执行复用 `app/agent/tools.py`**，不另写一套。身份由服务端注入，
+    模型不得传。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import UTC, datetime
+from typing import Any, AsyncIterator, Callable
+
+import numpy as np
+
+from pipecat.frames.frames import (
+    Frame,
+    LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
+    TextFrame,
+)
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import (
+    LLMContextAggregatorPair,
+)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+
+from ..config import AppConfig
+from ..memory.base import MemoryStore
+from ..prompts import PERSONA_RULES
+from ..voice.tts import split_sentences
+from . import tools as agent_tools
+
+log = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------- 本轮用户原话
+
+#: 本轮用户原话。策略层要它区分"用户明确要求记住"和"模型自己推断"，
+#: 而 Flows 的工具 handler 只看得到工具参数，看不到原话。
+#: 用单元素列表做可变闭包——handler 是闭包，拿不到外层的非局部变量。
+#:
+#: 注意：这是**单用户在用的原型**。多用户并发会话时这里会互相覆盖，
+#: 要支持并发得改成按 session 分键（Phase 5 迁移语音路径时一并处理）。
+_LAST_USER_TEXT: list[str] = [""]
+
+
+def set_last_user_text(text: str) -> None:
+    _LAST_USER_TEXT[0] = text or ""
+
+
+# ---------------------------------------------------------------- 节点
+
+#: 会话控制状态。**不要把长期记忆复制进来**——Flows 的 state 不是数据库，
+#: 长期数据一律走 MemoryStore（交接文档 §七）。
+def _conversation_node(*, identity_decision, memories: list,
+                       tool_schemas: list[dict],
+                       ctx_factory: Callable[[], agent_tools.ToolContext]) -> dict:
+    """默认节点：普通陪伴对话 + 可用工具。
+
+    人格与记忆都放进**同一条** system 消息（`task_messages`）——
+    **不用 role_message**，避免两条 system 消息（见模块文档）。
+
+    这里复用 legacy 的 `build_system_prompt`，保证两条路径：
+      - 人设与规则完全一致
+      - 记忆注入格式一致
+      - 身份未确认时的"按访客对待"口径一致
+    """
+    from ..prompts import build_system_prompt
+    persona = build_system_prompt(identity_decision, memories)
+
+    from pipecat.flows import FlowsFunctionSchema
+
+    functions = []
+    for schema in tool_schemas:
+        fn = schema["function"]
+        functions.append(FlowsFunctionSchema(
+            name=fn["name"],
+            description=fn["description"],
+            properties=fn["parameters"].get("properties", {}),
+            required=fn["parameters"].get("required", []),
+            handler=_make_handler(fn["name"], ctx_factory),
+        ))
+
+    return {
+        "name": "conversation",
+        "task_messages": [
+            {"role": "system", "content": persona},
+        ],
+        "functions": functions,
+        # 进入节点不自动开口：文字路径要等用户先说话。
+        # （spike 实测默认 True 会在 initialize 时立刻触发一轮回复，
+        #   那条回复与用户输入无关，会污染归因，也会让网页莫名多一段话。）
+        "respond_immediately": False,
+    }
+
+
+def _make_handler(name: str, ctx_factory: Callable[[], agent_tools.ToolContext]):
+    """把工具层的执行器包成 Flows 的 handler。
+
+    **用闭包捕获 ctx_factory，不用模块级全局**——全局在并发连接时会串
+    （两个会话各自的 owner_user_id 会互相覆盖）。
+    """
+    async def handler(args, flow_manager=None):
+        outcome = await agent_tools.execute(name, dict(args or {}), ctx_factory())
+        return outcome.content
+    return handler
+
+
+# ---------------------------------------------------------------- 输入记录
+
+class UserTextRecorder(FrameProcessor):
+    """记录本轮用户原话，供策略层判定语义等级。
+
+    策略层要区分"用户明确要求记住"和"模型自己推断"，依据是**用户原话**，
+    而 Flows 的工具 handler 只看得到工具参数。所以这里把原话截下来。
+    """
+
+    def __init__(self, extractor):
+        super().__init__()
+        self._extract = extractor
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        from pipecat.frames.frames import LLMMessagesAppendFrame
+        await super().process_frame(frame, direction)
+        if isinstance(frame, LLMMessagesAppendFrame):
+            try:
+                text = self._extract(frame.messages)
+                if text:
+                    set_last_user_text(text)
+            except Exception:  # noqa: BLE001 — 记录失败不该影响对话
+                log.debug("记录用户原话失败", exc_info=True)
+        await self.push_frame(frame, direction)
+
+
+# ---------------------------------------------------------------- 输出桥
+
+class FlowsBridge(FrameProcessor):
+    """把 Flows/LLM 的文本帧转成 RTVI 消息 + 分句 TTS。
+
+    产出**与 legacy 的 `_run_turn` 保持一致**（网页依赖这些消息渲染）：
+    bot-llm-started → 逐 token 的 bot-llm-text → bot-llm-stopped
+    → bot-transcription / bot-output（完整文本）
+    """
+
+    def __init__(self, *, rtvi, tts, cfg: AppConfig, audio_enabled: bool = True):
+        super().__init__()
+        self.rtvi = rtvi
+        self.tts = tts
+        self.cfg = cfg
+        self.audio_enabled = audio_enabled
+        self._buffer = ""
+        self._reply: list[str] = []
+        self._started = False
+        self._speak_tasks: list[asyncio.Task] = []
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, LLMFullResponseStartFrame):
+            self._started = True
+            self._buffer = ""
+            self._reply = []
+            await self.push_frame(frame, direction)
+            await self.rtvi.push_transport_message(_rtvi_bot_started())
+
+        elif isinstance(frame, TextFrame) and frame.text:
+            self._buffer += frame.text
+            self._reply.append(frame.text)
+            # 逐 token 给网页（Playground 的 Conversation 视图需要）
+            await self.rtvi.push_transport_message(_rtvi_bot_text(frame.text))
+            await self.push_frame(frame, direction)
+            # 攒到句子边界就送去合成——这是提前出声的唯一手段
+            segs = split_sentences(self._buffer)
+            if len(segs) > 1:
+                *ready, self._buffer = segs
+                for seg in ready:
+                    await self._speak(seg)
+
+        elif isinstance(frame, LLMFullResponseEndFrame):
+            if self._buffer.strip():
+                await self._speak(self._buffer)
+                self._buffer = ""
+            reply = "".join(self._reply).strip()
+            if reply:
+                # 完整文本单发一次：Playground 1.7 的 Conversation 视图
+                # 主要消费完整转写/bot-output，只给 token 级不会稳定渲染。
+                await self.rtvi.push_transport_message(_rtvi_bot_transcription(reply))
+                await self.rtvi.push_transport_message(
+                    _rtvi_bot_output(reply, spoken=self.audio_enabled))
+            if self._started:
+                await self.rtvi.push_transport_message(_rtvi_bot_stopped())
+                self._started = False
+            await self.push_frame(frame, direction)
+
+        else:
+            await self.push_frame(frame, direction)
+
+    async def _speak(self, text: str) -> None:
+        text = (text or "").strip()
+        if not text or not self.audio_enabled:
+            return
+        try:
+            audio = await asyncio.to_thread(self.tts.synthesize, text)
+        except Exception as e:  # noqa: BLE001 — TTS 失败不该中断文字显示
+            log.warning("Flows 路径 TTS 失败（文字仍显示）：%s", e)
+            return
+        if audio is None or not len(audio):
+            return
+        await self._push_audio(np.asarray(audio, dtype=np.float32),
+                               getattr(self.tts, "sample_rate", 24000))
+
+    async def _push_audio(self, samples: np.ndarray, sample_rate: int) -> None:
+        from pipecat.frames.frames import OutputAudioRawFrame
+        out_rate = self.cfg.voice.output_sample_rate if hasattr(
+            self.cfg.voice, "output_sample_rate") else 24000
+        if sample_rate != out_rate:
+            n = int(len(samples) * out_rate / sample_rate)
+            samples = np.interp(np.linspace(0, len(samples) - 1, n),
+                                np.arange(len(samples)), samples).astype(np.float32)
+        pcm = (np.clip(samples, -1, 1) * 32767).astype(np.int16)
+        step = out_rate // 50          # 20ms 一块
+        for i in range(0, len(pcm), step):
+            await self.push_frame(OutputAudioRawFrame(
+                audio=pcm[i:i + step].tobytes(), sample_rate=out_rate,
+                num_channels=1))
+
+
+# ---------------------------------------------------------------- RTVI 消息
+
+def _rtvi_bot_started():
+    import pipecat.processors.frameworks.rtvi.models as R
+    return R.BotLLMStartedMessage()
+
+
+def _rtvi_bot_stopped():
+    import pipecat.processors.frameworks.rtvi.models as R
+    return R.BotLLMStoppedMessage()
+
+
+def _rtvi_bot_text(text: str):
+    import pipecat.processors.frameworks.rtvi.models as R
+    return R.BotLLMTextMessage(data=R.TextMessageData(text=text))
+
+
+def _rtvi_bot_transcription(text: str):
+    import pipecat.processors.frameworks.rtvi.models as R
+    return R.BotTranscriptionMessage(data=R.TextMessageData(text=text))
+
+
+def _rtvi_bot_output(text: str, *, spoken: bool):
+    import pipecat.processors.frameworks.rtvi.models as R
+    return R.BotOutputMessage(data=R.BotOutputMessageData(
+        text=text, aggregated_by="sentence", spoken=spoken))
+
+
+# ---------------------------------------------------------------- 装配
+
+async def build_flows_parts(
+    *,
+    cfg: AppConfig,
+    memory: MemoryStore,
+    identity,
+    session_id: str,
+    tts,
+    rtvi,
+    audio_enabled: bool = True,
+):
+    """造出 Flows 路径需要的**零件**，不含 FlowManager。
+
+    为什么不在这里建 FlowManager：它要求传 `worker`（PipelineWorker），
+    而 worker 又必须在管线建好之后才能建，是个先后依赖。
+    所以顺序是：零件 → 管线 → worker → FlowManager → initialize。
+
+    返回 (llm, aggregator, bridge, ctx_factory, tool_schemas)
+    """
+    from .llm_service import create_llm_service
+
+    # 用 agent 的 max_tokens，不是语音的 200——带工具需要余量
+    llm = create_llm_service(cfg.llm, max_tokens=cfg.agent.max_tokens)
+    if llm is None:
+        raise RuntimeError(
+            f"Flows 路径不支持后端 {cfg.llm.name!r}（只支持 deepseek）。"
+            f"请把 AI_AGENT_RUNTIME 改回 legacy。")
+
+    context = LLMContext()
+    aggregator = LLMContextAggregatorPair(context)
+
+    def ctx_factory() -> agent_tools.ToolContext:
+        """每次工具调用现算身份——用户可能在会话中途切换。"""
+        d = identity.current()
+        return agent_tools.ToolContext(
+            owner_user_id=d.user_id,          # ← 服务端注入，模型不得传
+            session_id=session_id,
+            turn_id=f"t_{datetime.now(UTC).strftime('%H%M%S%f')}",
+            identity_confirmed=d.confirmed,
+            user_text=_LAST_USER_TEXT[0],
+            store=memory,
+            # Phase 3 先不暴露 web_search：搜索仍走 legacy 路径，
+            # 等 flows 文字路径验收通过后再接（交接文档要求分阶段）。
+            search=None,
+            memory_limit=cfg.context.max_memories,
+            memory_timeout_s=cfg.context.memory_timeout_s,
+        )
+
+    bridge = FlowsBridge(rtvi=rtvi, tts=tts, cfg=cfg, audio_enabled=audio_enabled)
+    tool_schemas = agent_tools.available_schemas(ctx_factory())
+    return llm, aggregator, bridge, ctx_factory, tool_schemas
+
+
+def make_flow_manager(*, worker, llm, aggregator):
+    """worker 建好之后再建 FlowManager。"""
+    from pipecat.flows import FlowManager
+    return FlowManager(worker=worker, llm=llm, context_aggregator=aggregator)
+
+
+async def load_stable_memories(memory, identity, *, limit: int = 8) -> list:
+    """加载"稳定档案"：已确认、未过期的记忆。
+
+    方案 §四要求"稳定档案（姓名、称呼、语言、明确的稳定偏好）
+    **连接开始时加载**"，而不是等模型自己想起来查。
+
+    身份未确认 → 返回空。访客不加载任何私人记忆（验收第 6 条）。
+    """
+    d = identity.current()
+    if not d.confirmed or memory is None:
+        return []
+    try:
+        rows = await memory.list_memories(d.user_id, include_candidates=False)
+    except Exception as e:  # noqa: BLE001 — 加载失败不阻塞会话
+        log.warning("加载稳定记忆失败（继续）：%s", e)
+        return []
+    # 最近更新的排前面
+    rows.sort(key=lambda m: getattr(m, "updated_at", 0), reverse=True)
+    return rows[:limit]
+
+
+async def initialize_conversation_node(
+    flow, *, identity_decision, memories: list, tool_schemas: list[dict],
+    ctx_factory: Callable[[], agent_tools.ToolContext],
+) -> None:
+    """设置初始节点。
+
+    **必须 await。** `FlowManager.initialize` 是协程，但签名显示 `-> None`，
+    漏 await 不报错，只是人格/任务/工具**全部静默失效**——
+    Phase 1 spike 在这里栽过一次，表现极具误导性：
+    模型会正常回复（因为 LLM 照样被调用），但没有人格、没有工具。
+    """
+    node = _conversation_node(identity_decision=identity_decision,
+                              memories=memories,
+                              tool_schemas=tool_schemas,
+                              ctx_factory=ctx_factory)
+    await flow.initialize(node)
+
+
+__all__ = [
+    "FlowsBridge", "UserTextRecorder", "build_flows_parts", "make_flow_manager",
+    "initialize_conversation_node", "load_stable_memories",
+    "set_last_user_text", "PERSONA_RULES",
+]

app/agent/llm_service.py

@@ -0,0 +1,61 @@
+"""Pipecat LLMService 工厂。
+
+Phase 1 spike 的选型结论固化在这里：**复用 Pipecat 自带的 `DeepSeekLLMService`**，
+不自写 `FallbackProvider → LLMService` 适配器。
+
+理由（详见笔记库 `P1-Flows兼容性Spike结论.md`）：
+
+  1. 它自带 `DeepSeekLLMAdapter`，处理"思考模式下必须把 reasoning_content
+     原样回传"这个约束——**正是本项目 P0 阶段踩过的坑**，框架已经做了。
+  2. `thinking` 默认就是 disabled，与本项目既有结论一致。
+  3. 支持自定义 `base_url`，将来要接本地 vLLM 也能用。
+  4. Flows 硬要求 `LLMService` 实例，而 `pipecat.flows` 的 `LLMAdapter`
+     **只管摘要生成**，不是 provider 适配器——想保留 `FallbackProvider`
+     必须自己写 `LLMService` 子类，成本不划算。
+
+**代价**：失去 `FallbackProvider` 的本地 vLLM 降级。当前本机 vLLM 已停用，
+fallback 实际没有可用目标，所以先只用 DeepSeek。将来若需双后端，
+Pipecat 有 `LLMSwitcher` 可用，不必自写。
+"""
+
+from __future__ import annotations
+
+import logging
+
+from ..config import LLMConfig
+
+log = logging.getLogger(__name__)
+
+
+def create_llm_service(cfg: LLMConfig, *, max_tokens: int | None = None):
+    """按配置造 Pipecat LLMService。
+
+    目前只支持 DeepSeek（OpenAI 兼容）。其它后端返回 None，调用方应当
+    回退到 legacy 路径，而不是带着一个坏 service 继续。
+    """
+    if cfg.name != "deepseek":
+        log.warning("Flows 路径暂不支持后端 %r，请继续使用 legacy", cfg.name)
+        return None
+
+    from pipecat.services.deepseek.llm import DeepSeekLLMService
+
+    # **必须显式覆盖模型名。**
+    # Pipecat 的默认是 `deepseek-v4-flash`，该模型不存在，用默认值会 400。
+    # 见 P0-DeepSeek验证报告.md：真实清单只有 deepseek-flash 与 deepseek-v4-pro。
+    # max_tokens 由调用方指定（Flows 路径用 AgentConfig.max_tokens，
+    # 不是语音的 200）。见 create_llm_service 的 max_tokens 参数。
+    settings = DeepSeekLLMService.Settings(
+        model=cfg.model,
+        # 显式关闭思考。Pipecat 的默认恰好也是 disabled，但写出来更安全：
+        # 将来默认值变化时不会悄悄改变行为，也不至于把思考内容送进 TTS。
+        thinking=DeepSeekLLMService.ThinkingConfig(type="disabled"),
+        max_tokens=max_tokens or cfg.max_tokens,
+    )
+    return DeepSeekLLMService(
+        api_key=cfg.api_key,
+        base_url=cfg.base_url,
+        settings=settings,
+    )
+
+
+__all__ = ["create_llm_service"]

app/agent/memory_policy.py

@@ -0,0 +1,269 @@
+"""记忆写入/删除的策略层。
+
+**为什么需要独立的一层**：交接文档明确要求
+"不要仅依赖'请不要误记'的 system prompt"。
+
+模型可以提议动作，**这一层拥有最终决定权**。理由：
+
+  - system prompt 是软的，模型会漂移；策略是硬的，可测试、可回归
+  - "我今天吃了香菜"不能被记成"用户不吃香菜"这种稳定偏好——
+    这不是措辞问题，是**语义等级**问题，必须在代码里判定
+  - 身份未确认时不得碰任何私人记忆，这也不能交给模型自觉
+
+输入：用户原话、工具请求参数、身份确认状态、已有同键记忆。
+输出：allow_confirmed / allow_candidate / needs_confirmation / reject。
+
+**本模块不依赖 LLM、不依赖网络、不依赖真实存储**，可以纯单元测试。
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from enum import Enum
+
+from ..memory.base import ConfirmStatus, Memory, MemoryKind
+
+
+class MemoryDecision(str, Enum):
+    """策略层的裁决。"""
+    ALLOW_CONFIRMED = "allow_confirmed"      # 用户明确要求 → 写 confirmed
+    ALLOW_CANDIDATE = "allow_candidate"      # 只是观察/推断 → 写 candidate
+    NEEDS_CONFIRMATION = "needs_confirmation"  # 目标不明确 → 转入确认节点
+    REJECT = "reject"                        # 不允许（越权、身份未确认等）
+
+
+@dataclass
+class PolicyResult:
+    decision: MemoryDecision
+    reason: str
+    #: 规范化后的写入内容。策略层可以改写模型给的 kind/key/value。
+    kind: MemoryKind = MemoryKind.FACT
+    key: str = ""
+    value: str = ""
+    #: NEEDS_CONFIRMATION 时，候选目标（供确认节点展示）
+    candidates: list[Memory] = field(default_factory=list)
+    #: REJECT/NEEDS_CONFIRMATION 时给模型看的话（不冒充已执行）
+    message: str = ""
+
+    @property
+    def confirm_status(self) -> ConfirmStatus:
+        return (ConfirmStatus.CONFIRMED
+                if self.decision is MemoryDecision.ALLOW_CONFIRMED
+                else ConfirmStatus.CANDIDATE)
+
+
+# ---------------------------------------------------------------- 显式意图
+
+#: 用户明确要求"记住"的说法。命中才允许写 confirmed。
+#: 刻意保守：宁可多判成 candidate（不参与召回），也不要误记成稳定事实。
+EXPLICIT_REMEMBER_PATTERNS = (
+    "记住", "记一下", "记下", "帮我记", "帮我记住",
+    "别忘了", "别忘记", "不要忘",
+    "以后都", "以后要", "以后别", "今后都",
+    "我的偏好是", "我习惯", "我通常", "我总是",
+)
+
+#: 明确表示"这是这一次的事"的说法。命中则**降级**为 candidate，
+#: 即使句中同时出现"记住"之类的词。
+#: 例："记住我今天吃了香菜"——语义是"记下今天这件事"，不是稳定偏好。
+#: 注意不要收太宽的词。比如"先"看着像临时，但"我习惯先喝汤"是稳定偏好，
+#: 收进来会造成漏判（把偏好降级成候选）。宁可少收。
+EPISODIC_MARKERS = (
+    "今天", "昨天", "刚才", "刚刚", "这次", "这一回", "暂时", "目前",
+    "这会儿", "眼下",
+)
+
+#: 主语是"今天/这次"这类时间词时，更可能是事件而非偏好
+_TIME_LEAD_RE = re.compile(r"^(今天|昨天|刚才|刚刚|这次|这会|这回|现在|目前)")
+
+
+def has_explicit_remember_intent(user_text: str) -> bool:
+    """用户是否明确要求记住（而不是模型自己觉得值得记）。
+
+    注意：本函数调用下面定义的 `looks_episodic`（Python 晚绑定，没问题）。
+    """
+    t = (user_text or "").strip()
+    if not t:
+        return False
+    if not any(p in t for p in EXPLICIT_REMEMBER_PATTERNS):
+        return False
+    # 明确是"这一次"的事 → 不算稳定偏好意图。
+    # **必须复用 looks_episodic**：早先这里只检查句首时间词，
+    # 而"记住我今天吃了香菜"的时间词在中间，导致漏判（测试逮到的）。
+    if looks_episodic(t):
+        return False
+    return True
+
+
+def looks_episodic(user_text: str) -> bool:
+    """是否是"这一次发生的事"而非稳定事实。"""
+    t = (user_text or "").strip()
+    if not t:
+        return False
+    if _TIME_LEAD_RE.search(t):
+        return True
+    return any(m in t for m in EPISODIC_MARKERS)
+
+
+# ---------------------------------------------------------------- 越权检查
+
+#: 模型**不得**传入的字段——身份与授权范围必须由服务端注入。
+FORBIDDEN_ARG_KEYS = (
+    "owner_user_id", "user_id", "owner", "uid",
+    "scope", "confirm_status", "trust_level",
+    "session_id", "turn_id",
+)
+
+
+def find_forbidden_args(args: dict) -> list[str]:
+    """返回模型试图传入的越权字段名。"""
+    if not isinstance(args, dict):
+        return []
+    lower = {str(k).lower() for k in args.keys()}
+    return [k for k in FORBIDDEN_ARG_KEYS if k.lower() in lower]
+
+
+# ---------------------------------------------------------------- 写入裁决
+
+def decide_remember(
+    *,
+    user_text: str,
+    args: dict,
+    identity_confirmed: bool,
+    model_claims_explicit: bool = False,
+) -> PolicyResult:
+    """裁决一次 memory_remember 请求。
+
+    参数：
+        user_text:         本轮用户原话（策略的主要依据）
+        args:              模型给的工具参数（含 kind/key/value）
+        identity_confirmed: 当前身份是否已确认
+        model_claims_explicit: 模型自称"用户明确要求记住"。
+            仅作参考，**不采信**——必须由 user_text 判定。
+    """
+    # 1. 身份未确认：一律拒绝，不碰任何私人记忆
+    if not identity_confirmed:
+        return PolicyResult(
+            decision=MemoryDecision.REJECT,
+            reason="身份未确认",
+            message="现在还不能记录私人信息。请先确认身份。",
+        )
+
+    # 2. 越权字段：拒绝（不静默忽略，避免模型以为生效了）
+    forbidden = find_forbidden_args(args)
+    if forbidden:
+        return PolicyResult(
+            decision=MemoryDecision.REJECT,
+            reason=f"模型传入越权字段 {forbidden}",
+            message=f"参数不合法：不接受 {', '.join(forbidden)}。",
+        )
+
+    value = str(args.get("value") or "").strip()
+    if not value:
+        return PolicyResult(
+            decision=MemoryDecision.REJECT,
+            reason="value 为空",
+            message="没有可记录的内容。",
+        )
+
+    kind = _coerce_kind(args.get("kind"))
+    key = str(args.get("key") or "").strip()
+
+    # 3. 判定语义等级。
+    #    注意：以 user_text 为准，不采信 model_claims_explicit。
+    explicit = has_explicit_remember_intent(user_text)
+    episodic = looks_episodic(user_text)
+
+    if explicit and not episodic:
+        return PolicyResult(
+            decision=MemoryDecision.ALLOW_CONFIRMED,
+            reason="用户明确要求记住",
+            kind=kind, key=key, value=value,
+        )
+
+    # 4. 其余一律 candidate：模型推断、总结、或"这一次"的事件
+    reason = "属于一次性事件" if episodic else "非用户明确要求，仅作观察"
+    return PolicyResult(
+        decision=MemoryDecision.ALLOW_CANDIDATE,
+        reason=reason,
+        kind=kind, key=key, value=value,
+    )
+
+
+def _coerce_kind(raw) -> MemoryKind:
+    """把模型给的 kind 收敛到枚举；不认识的一律当 FACT。"""
+    try:
+        return MemoryKind(str(raw))
+    except (ValueError, TypeError):
+        return MemoryKind.FACT
+
+
+# ---------------------------------------------------------------- 删除裁决
+
+#: 模糊删除的说法——命中则必须先确认，不得猜目标
+AMBIGUOUS_FORGET_MARKERS = (
+    "所有", "全部", "都删", "清空", "关于我的", "我的资料", "我的信息",
+    "这些", "那些", "乱七八糟",
+)
+
+
+def decide_forget(
+    *,
+    user_text: str,
+    matches: list[Memory],
+    identity_confirmed: bool,
+) -> PolicyResult:
+    """裁决一次 memory_forget 请求。
+
+    规则（交接文档 §八）：
+      - 精确匹配一条 → 返回待删除摘要，可执行
+      - 多条或模糊 → 进入确认节点，**不得猜测删除目标**
+      - 删除必须同时匹配 owner（由存储层保证，这里只判数量与措辞）
+    """
+    if not identity_confirmed:
+        return PolicyResult(
+            decision=MemoryDecision.REJECT,
+            reason="身份未确认",
+            message="现在不能操作私人记忆。",
+        )
+
+    if not matches:
+        return PolicyResult(
+            decision=MemoryDecision.REJECT,
+            reason="没有匹配的记忆",
+            message="没有找到相关的记忆。",
+        )
+
+    t = (user_text or "").strip()
+    ambiguous_wording = any(m in t for m in AMBIGUOUS_FORGET_MARKERS)
+
+    if len(matches) == 1 and not ambiguous_wording:
+        m = matches[0]
+        return PolicyResult(
+            decision=MemoryDecision.ALLOW_CONFIRMED,
+            reason="精确匹配一条",
+            kind=m.kind, key=m.key, value=m.value,
+            candidates=[m],
+            message=f"将删除「{m.text}」。",
+        )
+
+    # 多条 / 模糊 → 必须确认
+    return PolicyResult(
+        decision=MemoryDecision.NEEDS_CONFIRMATION,
+        reason=f"匹配到 {len(matches)} 条" + ("，且措辞模糊" if ambiguous_wording else ""),
+        candidates=list(matches),
+        message=(
+            f"找到 {len(matches)} 条相关记忆，需要你确认删哪一条：\n"
+            + "\n".join(f"  {i+1}. {m.text}" for i, m in enumerate(matches[:5]))
+        ),
+    )
+
+
+__all__ = [
+    "MemoryDecision", "PolicyResult",
+    "decide_remember", "decide_forget",
+    "has_explicit_remember_intent", "looks_episodic",
+    "find_forbidden_args",
+    "EXPLICIT_REMEMBER_PATTERNS", "EPISODIC_MARKERS", "FORBIDDEN_ARG_KEYS",
+]

app/agent/tools.py

@@ -0,0 +1,415 @@
+"""工具层。
+
+**两条路径共用**：legacy 的 `ConversationEngine` 手写循环和 Flows 都用这里的
+同一套 schema 与执行器。交接文档 Phase 2 的完成标准就是这个：
+"工具层可被 legacy 和 flows 两条路径复用"。
+
+设计要点：
+
+  * **身份由服务端注入，模型不得传。** `ToolContext.owner_user_id` 来自会话，
+    不在工具参数里。模型若试图传 `owner_user_id`/`scope`，策略层直接拒绝。
+  * **记忆写入必须过策略层**（`memory_policy`），策略拥有最终决定权。
+    模型只是提议者。
+  * **工具失败不拖垮对话。** 任何异常都转成给模型的文本结果，本轮继续。
+  * **不把"请求已收到"说成"已经做到"。** 删除类操作要如实返回
+    已删除 / 没有找到 / 需要确认。
+
+依赖注入：`ToolContext` 带 `store` 与 `search`，所以单元测试可以传 fake store，
+不需要真实 SQLite 或网络。
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from dataclasses import dataclass, field
+from typing import Any, Awaitable, Callable
+
+from ..memory.base import ConfirmStatus, Memory, MemoryKind, MemoryStore, Scope
+from ..search import WEB_SEARCH_TOOL, SearchError, SearXNGSearchClient
+from .memory_policy import (
+    MemoryDecision,
+    decide_forget,
+    decide_remember,
+    find_forbidden_args,
+)
+
+log = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------- 上下文
+
+@dataclass
+class ToolContext:
+    """一次工具调用所需的全部环境。**身份在这里，不在工具参数里。**"""
+
+    owner_user_id: str
+    session_id: str
+    turn_id: str
+    identity_confirmed: bool
+    user_text: str = ""
+    store: MemoryStore | None = None
+    search: SearXNGSearchClient | None = None
+    #: 记忆检索的条数与超时上限（沿用全局配置）
+    memory_limit: int = 5
+    memory_timeout_s: float = 0.3
+
+
+@dataclass
+class ToolOutcome:
+    """工具执行结果。`content` 是回灌给模型的文本。"""
+    content: str
+    ok: bool = True
+    meta: dict[str, Any] = field(default_factory=dict)
+
+
+# ---------------------------------------------------------------- 工具 schema
+
+MEMORY_SEARCH_TOOL = {
+    "type": "function",
+    "function": {
+        "name": "memory_search",
+        "description": (
+            "在本用户已确认的长期记忆中检索。当用户问起自己的偏好、习惯或"
+            "过去说过的事，而你手上没有这些信息时使用。只能查到已确认的记忆。"
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "query": {"type": "string", "description": "要查的内容"},
+            },
+            "required": ["query"],
+            "additionalProperties": False,
+        },
+    },
+}
+
+MEMORY_REMEMBER_TOOL = {
+    "type": "function",
+    "function": {
+        "name": "memory_remember",
+        "description": (
+            "长期记住一条关于用户的稳定信息（偏好、习惯、固定事实）。"
+            "**只在用户明确要求记住时调用**；你自行推断出的信息会被降级为"
+            "候选，不会立刻生效。"
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "kind": {
+                    "type": "string",
+                    "enum": ["preference", "fact", "task"],
+                    "description": "preference=稳定偏好，fact=事实，task=临时事项",
+                },
+                "key": {
+                    "type": "string",
+                    "description": "简短键名，如“饮食偏好”。同一键的新值会覆盖旧值",
+                },
+                "value": {"type": "string", "description": "要记住的内容"},
+            },
+            "required": ["kind", "value"],
+            "additionalProperties": False,
+        },
+    },
+}
+
+MEMORY_FORGET_TOOL = {
+    "type": "function",
+    "function": {
+        "name": "memory_forget",
+        "description": (
+            "删除关于用户的记忆。目标不明确时会被要求先确认，"
+            "**不要猜测用户想删哪一条**。"
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "query": {"type": "string", "description": "要删除的记忆内容或关键词"},
+            },
+            "required": ["query"],
+            "additionalProperties": False,
+        },
+    },
+}
+
+MEMORY_UPDATE_TOOL = {
+    "type": "function",
+    "function": {
+        "name": "memory_update",
+        "description": (
+            "更正已记住的信息。用户说“其实……”“不对，是……”这类纠正时使用。"
+            "会保留来源与更新时间。"
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "key": {"type": "string", "description": "要更正的记忆键名"},
+                "value": {"type": "string", "description": "更正后的内容"},
+            },
+            "required": ["key", "value"],
+            "additionalProperties": False,
+        },
+    },
+}
+
+ALL_MEMORY_TOOLS = [
+    MEMORY_SEARCH_TOOL,
+    MEMORY_REMEMBER_TOOL,
+    MEMORY_FORGET_TOOL,
+    MEMORY_UPDATE_TOOL,
+]
+
+
+def available_schemas(ctx: ToolContext) -> list[dict]:
+    """当前上下文下可用的工具。
+
+    身份未确认时**不暴露任何记忆工具**——比暴露后再拒绝更干净：
+    模型根本看不到这些能力，就不会尝试。
+    """
+    schemas: list[dict] = []
+    if ctx.search is not None:
+        schemas.append(WEB_SEARCH_TOOL)
+    if ctx.identity_confirmed and ctx.store is not None:
+        schemas.extend(ALL_MEMORY_TOOLS)
+    return schemas
+
+
+# ---------------------------------------------------------------- 执行入口
+
+async def execute(name: str, arguments: dict, ctx: ToolContext) -> ToolOutcome:
+    """按名字分发。未知工具/参数错误都返回可读文本，不抛异常。"""
+    fn = _HANDLERS.get(name)
+    if fn is None:
+        return ToolOutcome(
+            content=f"没有名为 {name} 的工具。请不要声称已经执行。",
+            ok=False,
+        )
+    if not isinstance(arguments, dict):
+        return ToolOutcome(content="工具参数不是合法的对象。", ok=False)
+
+    # 越权字段在这里统一拦一次（记忆工具内部还会再过策略层）
+    forbidden = find_forbidden_args(arguments)
+    if forbidden:
+        return ToolOutcome(
+            content=f"参数不合法：不接受 {', '.join(forbidden)}。",
+            ok=False,
+        )
+
+    try:
+        return await fn(arguments, ctx)
+    except Exception as e:  # noqa: BLE001 — 工具失败不能拖垮对话
+        log.exception("工具 %s 执行异常", name)
+        return ToolOutcome(content=f"工具执行失败：{type(e).__name__}。", ok=False)
+
+
+# ---------------------------------------------------------------- web_search
+
+async def _web_search(args: dict, ctx: ToolContext) -> ToolOutcome:
+    if ctx.search is None:
+        return ToolOutcome(content="搜索功能当前不可用。", ok=False)
+    query = args.get("query")
+    if not isinstance(query, str) or not query.strip():
+        return ToolOutcome(content="搜索失败：缺少有效的 query 参数。", ok=False)
+
+    t0 = time.perf_counter()
+    try:
+        results = await ctx.search.search(query)
+    except SearchError as exc:
+        log.warning("联网搜索失败：%s", exc)
+        return ToolOutcome(
+            content=f"搜索失败：{exc}。请明确告知用户暂时无法取得实时资料。",
+            ok=False,
+        )
+    ms = (time.perf_counter() - t0) * 1000
+    log.info("联网搜索 %r：%d 条，%.0fms", query[:60], len(results), ms)
+    return ToolOutcome(
+        content=ctx.search.for_model(query, results),
+        meta={"query": query, "count": len(results), "ms": ms,
+              "urls": [getattr(r, "url", "") for r in results]},
+    )
+
+
+# ---------------------------------------------------------------- 记忆工具
+
+async def _memory_search(args: dict, ctx: ToolContext) -> ToolOutcome:
+    if not ctx.identity_confirmed or ctx.store is None:
+        # 交接文档：不确认身份时返回不可用，不搜索任何私人记忆
+        return ToolOutcome(
+            content="现在还不能查询私人记忆。身份未确认。", ok=False)
+
+    query = str(args.get("query") or "").strip()
+    if not query:
+        return ToolOutcome(content="查询内容为空。", ok=False)
+
+    try:
+        # 只返回 confirmed、未过期、owner 匹配的记忆（存储层保证后两项）
+        memories = await ctx.store.retrieve(
+            ctx.owner_user_id, query,
+            limit=ctx.memory_limit,
+            include_candidates=False,
+            scopes=[Scope.PRIVATE, Scope.SHARED],
+        )
+    except Exception as e:  # noqa: BLE001
+        log.warning("记忆检索失败：%s", e)
+        return ToolOutcome(content="记忆检索暂时不可用。", ok=False)
+
+    if not memories:
+        return ToolOutcome(content="没有找到相关记忆。", meta={"count": 0})
+    lines = "\n".join(f"- {m.text}" for m in memories)
+    return ToolOutcome(
+        content=f"找到 {len(memories)} 条：\n{lines}",
+        meta={"count": len(memories)},
+    )
+
+
+async def _memory_remember(args: dict, ctx: ToolContext) -> ToolOutcome:
+    if ctx.store is None:
+        return ToolOutcome(content="记忆功能当前不可用。", ok=False)
+
+    result = decide_remember(
+        user_text=ctx.user_text,
+        args=args,
+        identity_confirmed=ctx.identity_confirmed,
+    )
+
+    if result.decision is MemoryDecision.REJECT:
+        return ToolOutcome(content=result.message or "不允许写入。", ok=False)
+
+    mem = Memory(
+        owner_user_id=ctx.owner_user_id,          # ← 服务端注入，非模型传入
+        scope=Scope.PRIVATE,                       # ← 默认 private，模型不得扩大
+        kind=result.kind,
+        key=result.key,
+        value=result.value,
+        confirm_status=result.confirm_status,
+        source_session_id=ctx.session_id,
+        source_turn_id=ctx.turn_id,
+    )
+    try:
+        memory_id = await ctx.store.remember(mem)
+    except Exception as e:  # noqa: BLE001
+        log.warning("写入记忆失败：%s", e)
+        return ToolOutcome(content="记忆写入暂时失败。", ok=False)
+
+    if result.decision is MemoryDecision.ALLOW_CONFIRMED:
+        return ToolOutcome(
+            content=f"已记住：{result.value}",
+            meta={"memory_id": memory_id, "status": "confirmed"},
+        )
+    # 候选：**如实告诉模型它没有立刻生效**，否则模型会对用户说"记住了"
+    return ToolOutcome(
+        content=(
+            f"已作为候选记录（{result.reason}），尚未生效。"
+            f"如果用户确实要求长期记住，请让用户明确说一次。"
+        ),
+        meta={"memory_id": memory_id, "status": "candidate"},
+    )
+
+
+async def _memory_forget(args: dict, ctx: ToolContext) -> ToolOutcome:
+    if not ctx.identity_confirmed or ctx.store is None:
+        return ToolOutcome(content="现在不能操作私人记忆。", ok=False)
+
+    query = str(args.get("query") or "").strip()
+    if not query:
+        return ToolOutcome(content="没有指定要删除的内容。", ok=False)
+
+    try:
+        matches = await ctx.store.retrieve(
+            ctx.owner_user_id, query, limit=6, include_candidates=True,
+            scopes=[Scope.PRIVATE, Scope.SHARED])
+    except Exception as e:  # noqa: BLE001
+        log.warning("删除前检索失败：%s", e)
+        return ToolOutcome(content="记忆检索暂时不可用，未执行删除。", ok=False)
+
+    result = decide_forget(
+        user_text=ctx.user_text, matches=matches,
+        identity_confirmed=ctx.identity_confirmed)
+
+    if result.decision is MemoryDecision.REJECT:
+        return ToolOutcome(content=result.message, ok=False)
+
+    if result.decision is MemoryDecision.NEEDS_CONFIRMATION:
+        # 不猜目标，交给确认流程
+        return ToolOutcome(
+            content=result.message,
+            meta={"needs_confirmation": True,
+                  "candidates": [m.memory_id for m in result.candidates]},
+        )
+
+    target = result.candidates[0]
+    if target.memory_id is None:
+        return ToolOutcome(content="无法定位要删除的记忆。", ok=False)
+    try:
+        deleted = await ctx.store.forget(ctx.owner_user_id, target.memory_id)
+    except Exception as e:  # noqa: BLE001
+        log.warning("删除记忆失败：%s", e)
+        return ToolOutcome(content="删除失败。", ok=False)
+
+    # 如实反馈，不把"请求已收到"说成"已经删除"
+    return ToolOutcome(
+        content=f"已删除：{target.text}" if deleted else "没有找到要删除的记忆。",
+        meta={"deleted": bool(deleted), "memory_id": target.memory_id},
+    )
+
+
+async def _memory_update(args: dict, ctx: ToolContext) -> ToolOutcome:
+    if not ctx.identity_confirmed or ctx.store is None:
+        return ToolOutcome(content="现在不能修改私人记忆。", ok=False)
+
+    key = str(args.get("key") or "").strip()
+    value = str(args.get("value") or "").strip()
+    if not key or not value:
+        return ToolOutcome(content="更正需要同时给出 key 和 value。", ok=False)
+
+    try:
+        existing = await ctx.store.retrieve(
+            ctx.owner_user_id, key, limit=5, include_candidates=True,
+            scopes=[Scope.PRIVATE, Scope.SHARED])
+    except Exception as e:  # noqa: BLE001
+        log.warning("更正前检索失败：%s", e)
+        return ToolOutcome(content="记忆检索暂时不可用。", ok=False)
+
+    target = next((m for m in existing if m.key == key), None)
+    if target is None:
+        return ToolOutcome(
+            content=f"没有找到键为「{key}」的记忆，无法更正。", ok=False)
+
+    # 同 key 的 confirmed 写入是幂等的：存储层按 owner+kind+key 覆盖旧值
+    mem = Memory(
+        owner_user_id=ctx.owner_user_id,
+        scope=target.scope,
+        kind=target.kind,
+        key=key,
+        value=value,
+        confirm_status=ConfirmStatus.CONFIRMED,
+        source_session_id=ctx.session_id,
+        source_turn_id=ctx.turn_id,
+    )
+    try:
+        await ctx.store.remember(mem)
+    except Exception as e:  # noqa: BLE001
+        log.warning("更正记忆失败：%s", e)
+        return ToolOutcome(content="更正失败。", ok=False)
+
+    return ToolOutcome(
+        content=f"已更新「{key}」：{target.value} → {value}",
+        meta={"updated": True, "key": key},
+    )
+
+
+_HANDLERS: dict[str, Callable[[dict, ToolContext], Awaitable[ToolOutcome]]] = {
+    "web_search": _web_search,
+    "memory_search": _memory_search,
+    "memory_remember": _memory_remember,
+    "memory_forget": _memory_forget,
+    "memory_update": _memory_update,
+}
+
+
+__all__ = [
+    "ToolContext", "ToolOutcome", "execute", "available_schemas",
+    "WEB_SEARCH_TOOL", "ALL_MEMORY_TOOLS",
+    "MEMORY_SEARCH_TOOL", "MEMORY_REMEMBER_TOOL",
+    "MEMORY_FORGET_TOOL", "MEMORY_UPDATE_TOOL",
+]

app/config.py

@@ -212,6 +212,31 @@ class MemoryConfig:
 
 # ---------------------------------------------------------------- 汇总
 
+@dataclass
+class AgentConfig:
+    """Agent 运行时的选择。
+
+    Phase 3 起有两条路径，**默认必须是 legacy**——交接文档要求
+    "flows 默认值只有在线上验收后才能从 legacy 改为 flows"。
+    """
+
+    #: "legacy" = ConversationEngine 手写工具循环（当前线上行为）
+    #: "flows"  = Pipecat Flows 驱动文字路径（Phase 3 新路径）
+    runtime: str = field(default_factory=lambda: _env("AI_AGENT_RUNTIME", "legacy"))
+
+    #: Flows 路径的 max_tokens。
+    #: **不能沿用语音的 200**：实测同一句"忘掉我的饮食偏好"，
+    #: max_tokens=200 时模型直接 finish=stop、根本不调工具；
+    #: 600 时才正常产生 tool_calls 且参数 JSON 完整。
+    #: 200 是为"语音回复要短"设的，而带工具的路径需要额外余量。
+    max_tokens: int = field(
+        default_factory=lambda: _env_int("AI_AGENT_MAX_TOKENS", 600))
+
+    @property
+    def use_flows(self) -> bool:
+        return self.runtime.strip().lower() == "flows"
+
+
 @dataclass
 class AppConfig:
     # 单用户原型的稳定内部档案 ID。它只用于数据隔离，绝不是用户可见姓名。
@@ -228,6 +253,7 @@ class AppConfig:
     search: SearchConfig = field(default_factory=SearchConfig)
     voice: VoiceConfig = field(default_factory=VoiceConfig)
     memory: MemoryConfig = field(default_factory=MemoryConfig)
+    agent: AgentConfig = field(default_factory=AgentConfig)
 
     @classmethod
     def load(cls) -> "AppConfig":

app/engine.py

@@ -37,6 +37,7 @@ import numpy as np
 
 from .config import AppConfig
 from .identity.base import IdentityDecision, IdentityProvider
+from .agent.tools import ToolContext, execute as execute_tool
 from .llm.base import ChunkKind, LLMProvider, ToolCall
 from .memory.base import (
     ConfirmStatus, Memory, MemoryKind, MemoryStore, Scope, Turn,
@@ -45,7 +46,6 @@ from .prompts import build_system_prompt, trim_history
 from .search import (
     WEB_SEARCH_PROMPT,
     WEB_SEARCH_TOOL,
-    SearchError,
     SearXNGSearchClient,
 )
 from .voice.stt import StreamingSTT
@@ -250,6 +250,9 @@ class ConversationEngine:
         buffer = ""
         speech_done = False
         speaks = 0
+        # 策略层要拿用户原话来判定"明确要求记住"还是"模型推断"，
+        # 所以本轮原话要挂到实例上，供 _execute_tool 构造 ToolContext。
+        self._current_user_text = user_text
         async for ev in self._stream_and_speak(messages, metrics):
             if ev.kind == EventKind.RESPONSE_DELTA:
                 reply_parts.append(ev.text)
@@ -462,27 +465,30 @@ class ConversationEngine:
         metrics.llm_backend = getattr(self.llm, "last_backend", self.llm.name)
 
     async def _execute_tool(self, call: ToolCall, metrics: TurnMetrics) -> str:
-        """执行只读搜索工具；错误作为工具结果回灌，不让整轮对话中断。"""
+        """执行工具。**实现已抽到 `app/agent/tools.py`，两条路径共用。**
-        if call.name != "web_search" or self.search is None:
-            return "工具不可用。请不要声称已经执行。"
-        query = call.parsed_arguments().get("query")
-        if not isinstance(query, str) or not query.strip():
-            return "搜索失败：缺少有效的 query 参数。"
 
+        legacy 与 Flows 必须用同一套 schema 和执行器，否则容易两处行为漂移
+        （交接文档 Phase 2 的完成标准就是"工具层可被两条路径复用"）。
+        这里只负责把引擎的状态转成 ToolContext。
+        """
+        identity = self.identity.current()
+        ctx = ToolContext(
+            owner_user_id=identity.user_id,      # ← 服务端注入，模型不得传
+            session_id=self._session_id,
+            turn_id=f"t_{uuid.uuid4().hex[:10]}",
+            identity_confirmed=identity.confirmed,
+            user_text=getattr(self, "_current_user_text", ""),
+            store=self.memory,
+            search=self.search,
+            memory_limit=self.cfg.context.max_memories,
+            memory_timeout_s=self.cfg.context.memory_timeout_s,
+        )
         started = time.perf_counter()
+        outcome = await execute_tool(call.name, call.parsed_arguments(), ctx)
+        if call.name == "web_search":
             metrics.searches += 1
-        try:
-            results = await self.search.search(query)
-            log.info("联网搜索 %r：返回 %d 条", query[:80], len(results))
-            return self.search.for_model(query, results)
-        except SearchError as exc:
-            log.warning("联网搜索失败：%s", exc)
-            return f"搜索失败：{exc}。请明确告知用户暂时无法取得实时资料。"
-        except Exception as exc:  # noqa: BLE001 — 工具失败不能拖垮对话
-            log.exception("联网搜索异常：%s", exc)
-            return "搜索服务发生异常。请明确告知用户暂时无法取得实时资料。"
-        finally:
             metrics.search_ms += (time.perf_counter() - started) * 1000
+        return outcome.content
 
     async def _speak(self, text: str, metrics: TurnMetrics) -> AsyncIterator[Event]:
         """合成一段并产出可播放音频。"""

app/tests/test_agent_tools.py

@@ -0,0 +1,347 @@
+#!/usr/bin/env python3
+"""Phase 2 工具层与记忆策略测试。
+
+**完全不依赖 LLM、网络或真实 SQLite**——用 fake store。
+这是交接文档 Phase 2 的完成标准之一（"策略测试不依赖真实 LLM"）。
+
+覆盖交接文档 §十 的单元测试清单：
+
+  - 工具参数 schema 与非法参数
+  - 模型不能传 owner_user_id/scope 越权
+  - 未确认身份拒绝 private memory
+  - 明确"记住"写 confirmed
+  - 推断信息只写 candidate
+  - 同 key 更新幂等
+  - 模糊忘记返回 needs_confirmation
+  - 精确忘记只删除当前 owner
+
+用法：
+    ~/venvs/ai_assistant/bin/python app/tests/test_agent_tools.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(ROOT))
+
+from app.agent.memory_policy import (  # noqa: E402
+    MemoryDecision, decide_forget, decide_remember,
+    find_forbidden_args, has_explicit_remember_intent, looks_episodic,
+)
+from app.agent.tools import (  # noqa: E402
+    ToolContext, available_schemas, execute,
+)
+from app.memory.base import (  # noqa: E402
+    ConfirmStatus, Memory, MemoryKind, MemoryStore, Scope,
+)
+
+OK, FAIL = [], []
+
+
+def check(name: str, cond: bool, detail: str = "") -> None:
+    (OK if cond else FAIL).append(name)
+    print(f"  [{'PASS' if cond else 'FAIL'}] {name}" + (f" — {detail}" if detail else ""))
+
+
+# ---------------------------------------------------------------- fake store
+
+class FakeStore(MemoryStore):
+    """内存实现。只实现测试用得到的方法，其余 raise。"""
+
+    def __init__(self):
+        self.rows: list[Memory] = []
+        self._next = 1
+        self.calls: list[tuple[str, tuple]] = []
+
+    async def open(self): pass
+    async def close(self): pass
+    async def ensure_user(self, user_id, display_name="", trust_level="normal"): pass
+    async def set_user_display_name(self, user_id, display_name): pass
+    async def get_user_display_name(self, user_id): return ""
+    async def start_session(self, session_id, user_id): pass
+    async def end_session(self, session_id): pass
+    async def add_turn(self, turn): pass
+    async def purge_user(self, owner_user_id): return 0
+
+    async def remember(self, mem: Memory) -> int:
+        self.calls.append(("remember", (mem.owner_user_id, mem.key, mem.value,
+                                        mem.confirm_status.value)))
+        # 复刻存储层的幂等语义：同 owner+kind+key 的 confirmed 覆盖旧值
+        if mem.confirm_status is ConfirmStatus.CONFIRMED and mem.key:
+            for r in self.rows:
+                if (r.owner_user_id == mem.owner_user_id and r.kind == mem.kind
+                        and r.key == mem.key
+                        and r.confirm_status is ConfirmStatus.CONFIRMED):
+                    r.value = mem.value
+                    r.updated_at = time.time()
+                    return r.memory_id or 0
+        m = Memory(**{**mem.__dict__, "memory_id": self._next})
+        self._next += 1
+        self.rows.append(m)
+        return m.memory_id
+
+    async def forget(self, owner_user_id: str, memory_id: int) -> bool:
+        self.calls.append(("forget", (owner_user_id, memory_id)))
+        before = len(self.rows)
+        # 必须同时匹配 owner——只凭 id 删是越权
+        self.rows = [r for r in self.rows
+                     if not (r.memory_id == memory_id
+                             and r.owner_user_id == owner_user_id)]
+        return len(self.rows) < before
+
+    async def retrieve(self, owner_user_id, query, *, limit=5,
+                       include_candidates=False, scopes=None):
+        self.calls.append(("retrieve", (owner_user_id,)))
+        out = []
+        for r in self.rows:
+            if r.owner_user_id != owner_user_id:
+                continue
+            if not include_candidates and r.confirm_status is not ConfirmStatus.CONFIRMED:
+                continue
+            if query and self.relevance(query, r.text) <= 0:
+                continue
+            out.append(r)
+        return out[:limit]
+
+    async def list_memories(self, owner_user_id, *, include_candidates=True):
+        return [r for r in self.rows if r.owner_user_id == owner_user_id]
+
+
+def ctx(store=None, *, confirmed=True, user_text="", uid="u_a"):
+    return ToolContext(owner_user_id=uid, session_id="s1", turn_id="t1",
+                       identity_confirmed=confirmed, user_text=user_text,
+                       store=store)
+
+
+# ---------------------------------------------------------------- 策略：写入
+
+def test_policy_remember():
+    print("\n=== 1. 写入裁决（策略层，无 LLM）===")
+
+    # 明确要求 → confirmed
+    r = decide_remember(user_text="记住我不吃香菜", identity_confirmed=True,
+                        args={"kind": "preference", "key": "饮食偏好", "value": "不吃香菜"})
+    check("明确“记住” → confirmed", r.decision is MemoryDecision.ALLOW_CONFIRMED, r.reason)
+
+    # 模型推断 → candidate
+    r = decide_remember(user_text="我今天点了份牛肉面", identity_confirmed=True,
+                        args={"kind": "preference", "key": "饮食偏好", "value": "喜欢牛肉面"})
+    check("无明确意图 → candidate", r.decision is MemoryDecision.ALLOW_CANDIDATE, r.reason)
+
+    # 模型自称"用户明确要求"不采信
+    r = decide_remember(user_text="我今天点了份牛肉面", identity_confirmed=True,
+                        args={"kind": "fact", "value": "喜欢牛肉面"},
+                        model_claims_explicit=True)
+    check("不采信模型自称的显式意图", r.decision is MemoryDecision.ALLOW_CANDIDATE, r.reason)
+
+    # 身份未确认 → reject
+    r = decide_remember(user_text="记住我不吃香菜", identity_confirmed=False,
+                        args={"kind": "preference", "value": "不吃香菜"})
+    check("身份未确认 → reject", r.decision is MemoryDecision.REJECT, r.reason)
+
+    # 越权字段 → reject
+    r = decide_remember(user_text="记住我不吃香菜", identity_confirmed=True,
+                        args={"kind": "preference", "value": "x",
+                              "owner_user_id": "u_b"})
+    check("模型传 owner_user_id → reject", r.decision is MemoryDecision.REJECT, r.reason)
+    r = decide_remember(user_text="记住我不吃香菜", identity_confirmed=True,
+                        args={"kind": "preference", "value": "x", "scope": "shared"})
+    check("模型传 scope → reject", r.decision is MemoryDecision.REJECT, r.reason)
+
+    # 空 value
+    r = decide_remember(user_text="记住", identity_confirmed=True,
+                        args={"kind": "fact", "value": "  "})
+    check("空 value → reject", r.decision is MemoryDecision.REJECT, r.reason)
+
+
+def test_episodic():
+    print("\n=== 2. 事件 vs 稳定偏好（交接文档的关键验收）===")
+    check("“记住我不吃香菜”是明确意图",
+          has_explicit_remember_intent("记住我不吃香菜"))
+    check("“记住我今天吃了香菜”不算稳定意图（时间词在前）",
+          not has_explicit_remember_intent("记住我今天吃了香菜"))
+    check("“我今天吃了香菜”判为事件", looks_episodic("我今天吃了香菜"))
+    check("“我不吃香菜”不是事件", not looks_episodic("我不吃香菜"))
+
+    # 交接文档 Phase 4 验收第 3 条：不能把一次性事件变成稳定偏好
+    r = decide_remember(user_text="记住我今天吃了香菜", identity_confirmed=True,
+                        args={"kind": "preference", "key": "饮食偏好",
+                              "value": "吃香菜"})
+    check("一次性事件降级为 candidate",
+          r.decision is MemoryDecision.ALLOW_CANDIDATE, r.reason)
+
+
+def test_forbidden_args():
+    print("\n=== 3. 越权字段识别 ===")
+    check("识别 owner_user_id", find_forbidden_args({"owner_user_id": "x"}) == ["owner_user_id"])
+    check("识别 scope", find_forbidden_args({"scope": "shared"}) == ["scope"])
+    check("大小写不敏感", find_forbidden_args({"Owner_User_ID": "x"}) != [])
+    check("正常参数不误报", find_forbidden_args({"query": "x", "value": "y"}) == [])
+
+
+# ---------------------------------------------------------------- 策略：删除
+
+def test_policy_forget():
+    print("\n=== 4. 删除裁决 ===")
+    m1 = Memory(memory_id=1, owner_user_id="u_a", kind=MemoryKind.PREFERENCE,
+                key="饮食偏好", value="不吃香菜", confirm_status=ConfirmStatus.CONFIRMED)
+    m2 = Memory(memory_id=2, owner_user_id="u_a", kind=MemoryKind.PREFERENCE,
+                key="饮品偏好", value="喜欢美式", confirm_status=ConfirmStatus.CONFIRMED)
+
+    r = decide_forget(user_text="忘掉我的饮食偏好", matches=[m1], identity_confirmed=True)
+    check("精确匹配一条 → 可执行",
+          r.decision is MemoryDecision.ALLOW_CONFIRMED, r.reason)
+
+    r = decide_forget(user_text="忘掉我的偏好", matches=[m1, m2], identity_confirmed=True)
+    check("多条 → needs_confirmation",
+          r.decision is MemoryDecision.NEEDS_CONFIRMATION, r.reason)
+    check("needs_confirmation 带候选", len(r.candidates) == 2)
+
+    r = decide_forget(user_text="忘掉我所有的记忆", matches=[m1], identity_confirmed=True)
+    check("模糊措辞 → needs_confirmation",
+          r.decision is MemoryDecision.NEEDS_CONFIRMATION, r.reason)
+
+    r = decide_forget(user_text="忘掉我的饮食偏好", matches=[], identity_confirmed=True)
+    check("无匹配 → reject", r.decision is MemoryDecision.REJECT, r.reason)
+
+    r = decide_forget(user_text="忘掉我的饮食偏好", matches=[m1], identity_confirmed=False)
+    check("身份未确认 → reject", r.decision is MemoryDecision.REJECT, r.reason)
+
+
+# ---------------------------------------------------------------- 工具执行
+
+async def test_tools():
+    print("\n=== 5. 工具执行（fake store）===")
+    store = FakeStore()
+
+    # 可用工具随身份变化
+    s_off = available_schemas(ctx(store, confirmed=False))
+    names_off = {t["function"]["name"] for t in s_off}
+    check("身份未确认时不暴露记忆工具", "memory_search" not in names_off, str(names_off))
+    s_on = available_schemas(ctx(store, confirmed=True))
+    names_on = {t["function"]["name"] for t in s_on}
+    check("身份确认后暴露记忆工具",
+          {"memory_search", "memory_remember", "memory_forget", "memory_update"}
+          <= names_on, str(names_on))
+
+    # 未知工具
+    out = await execute("no_such_tool", {}, ctx(store))
+    check("未知工具返回可读文本", not out.ok and "没有名为" in out.content)
+
+    # 越权参数在入口就被拦
+    out = await execute("memory_remember",
+                        {"kind": "fact", "value": "x", "owner_user_id": "u_b"},
+                        ctx(store))
+    check("入口拦越权参数", not out.ok and "不合法" in out.content)
+
+    # 身份未确认时记忆工具不可用
+    out = await execute("memory_search", {"query": "香菜"},
+                        ctx(store, confirmed=False))
+    check("未确认身份拒绝 memory_search", not out.ok and "身份未确认" in out.content)
+
+    # 明确记住 → confirmed，且 owner 来自 ctx 而非模型
+    out = await execute("memory_remember",
+                        {"kind": "preference", "key": "饮食偏好", "value": "不吃香菜"},
+                        ctx(store, user_text="记住我不吃香菜", uid="u_a"))
+    check("明确记住写入成功", out.ok and out.meta.get("status") == "confirmed", out.content)
+    check("owner 用 ctx 的 u_a",
+          any(c[1][0] == "u_a" for c in store.calls if c[0] == "remember"))
+
+    # 推断 → candidate，且如实告知模型未生效
+    out = await execute("memory_remember",
+                        {"kind": "fact", "value": "喜欢牛肉面"},
+                        ctx(store, user_text="我今天吃了牛肉面"))
+    check("推断写入为 candidate", out.meta.get("status") == "candidate", out.content)
+    check("如实告知未生效", "尚未生效" in out.content, out.content)
+
+    # 同 key 幂等更新：不应产生第二条 confirmed
+    await execute("memory_remember",
+                  {"kind": "preference", "key": "饮食偏好", "value": "现在吃香菜了"},
+                  ctx(store, user_text="记住我饮食偏好改了"))
+    confirmed = [r for r in store.rows
+                 if r.owner_user_id == "u_a" and r.key == "饮食偏好"
+                 and r.confirm_status is ConfirmStatus.CONFIRMED]
+    check("同 key 只保留一条 confirmed", len(confirmed) == 1, f"{len(confirmed)} 条")
+    check("值已更新", confirmed[0].value == "现在吃香菜了", confirmed[0].value)
+
+    # 检索：candidate 默认不返回
+    out = await execute("memory_search", {"query": "牛肉面"}, ctx(store))
+    check("candidate 不出现在检索结果", "牛肉面" not in out.content, out.content)
+
+    # 更新已有键
+    out = await execute("memory_update",
+                        {"key": "饮食偏好", "value": "其实我现在吃香菜了"},
+                        ctx(store, user_text="其实我现在吃香菜了"))
+    check("更正成功", out.ok and out.meta.get("updated"), out.content)
+    check("更正后值变了",
+          any(r.key == "饮食偏好" and r.value == "其实我现在吃香菜了"
+              for r in store.rows))
+
+    # 更正不存在的键
+    out = await execute("memory_update", {"key": "不存在的键", "value": "x"}, ctx(store))
+    check("更正不存在的键 → 拒绝", not out.ok, out.content)
+
+    # 精确删除
+    out = await execute("memory_forget", {"query": "饮食偏好"},
+                        ctx(store, user_text="忘掉我的饮食偏好"))
+    check("精确删除执行", out.ok and out.meta.get("deleted"), out.content)
+    check("删除话术如实（已删除）", "已删除" in out.content, out.content)
+
+    # 删除不存在
+    out = await execute("memory_forget", {"query": "完全不存在的东西"},
+                        ctx(store, user_text="忘掉完全不存在的东西"))
+    check("删除不存在的记忆 → 不谎报", not out.ok and "没有找到" in out.content, out.content)
+
+    # 模糊删除 → 需要确认
+    store2 = FakeStore()
+    for k, v in (("饮食偏好", "不吃香菜"), ("饮品偏好", "喜欢美式")):
+        await execute("memory_remember",
+                      {"kind": "preference", "key": k, "value": v},
+                      ctx(store2, user_text=f"记住我的{k}是{v}"))
+    out = await execute("memory_forget", {"query": "偏好"},
+                        ctx(store2, user_text="忘掉我的偏好"))
+    check("模糊删除 → 不猜目标", out.meta.get("needs_confirmation") is True, out.content)
+    check("模糊删除不实际删除", len(store2.rows) == 2, f"{len(store2.rows)} 条")
+
+
+async def test_owner_isolation():
+    print("\n=== 6. 跨用户隔离（交接文档 Phase 4 验收第 7 条）===")
+    store = FakeStore()
+    await execute("memory_remember",
+                  {"kind": "preference", "key": "饮食偏好", "value": "不吃香菜"},
+                  ctx(store, user_text="记住我不吃香菜", uid="u_a"))
+
+    out = await execute("memory_search", {"query": "香菜"}, ctx(store, uid="u_b"))
+    check("用户 B 查不到用户 A 的记忆", "香菜" not in out.content, out.content)
+
+    # B 试图删 A 的 id：存储层必须同时匹配 owner
+    a_id = store.rows[0].memory_id
+    deleted = await store.forget("u_b", a_id)
+    check("跨 owner 删除失败", deleted is False)
+    check("A 的记忆还在", len(store.rows) == 1)
+
+
+# ---------------------------------------------------------------- 主流程
+
+def main() -> int:
+    test_policy_remember()
+    test_episodic()
+    test_forbidden_args()
+    test_policy_forget()
+    asyncio.run(test_tools())
+    asyncio.run(test_owner_isolation())
+
+    print("\n" + "=" * 62)
+    print(f"通过 {len(OK)} / {len(OK) + len(FAIL)}")
+    for n in FAIL:
+        print(f"  未通过: {n}")
+    return 0 if not FAIL else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

app/tests/test_flows_runtime.py

@@ -0,0 +1,232 @@
+#!/usr/bin/env python3
+"""Phase 3 Flows 运行时装配测试。
+
+**不依赖网络、不启动 WebRTC**。验证的是装配正确性：
+
+  - 节点配置形态（只用 task_messages、不自动开口）
+  - 工具 schema 随身份变化
+  - handler 桥接到工具层，且身份由服务端注入
+  - 运行时开关默认是 legacy
+
+真正跑起来的行为（RTVI 消息、TTS、延迟）由现场测试和真人网页验收覆盖。
+
+用法：
+    ~/venvs/ai_assistant/bin/python app/tests/test_flows_runtime.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(ROOT))
+
+from app.agent import flows_runtime as FR  # noqa: E402
+from app.agent.tools import ToolContext  # noqa: E402
+from app.config import AgentConfig, AppConfig  # noqa: E402
+from app.identity import ManualIdentityProvider  # noqa: E402
+from app.memory.base import ConfirmStatus, Memory, MemoryKind  # noqa: E402
+
+OK, FAIL = [], []
+
+
+def check(name: str, cond: bool, detail: str = "") -> None:
+    (OK if cond else FAIL).append(name)
+    print(f"  [{'PASS' if cond else 'FAIL'}] {name}" + (f" — {detail}" if detail else ""))
+
+
+class FakeStore:
+    """最小实现，只够装配与 handler 调用。"""
+    def __init__(self):
+        self.rows: list[Memory] = []
+        self._n = 1
+
+    async def retrieve(self, owner_user_id, query, **kw):
+        return [r for r in self.rows if r.owner_user_id == owner_user_id][:5]
+
+    async def remember(self, mem: Memory) -> int:
+        m = Memory(**{**mem.__dict__, "memory_id": self._n})
+        self._n += 1
+        self.rows.append(m)
+        return m.memory_id
+
+    async def forget(self, owner_user_id, memory_id):
+        before = len(self.rows)
+        self.rows = [r for r in self.rows
+                     if not (r.memory_id == memory_id
+                             and r.owner_user_id == owner_user_id)]
+        return len(self.rows) < before
+
+
+class FakeTTS:
+    sample_rate = 24000
+    def synthesize(self, text):
+        import numpy as np
+        return np.zeros(2400, dtype=np.float32)
+
+
+class FakeRTVI:
+    def __init__(self): self.msgs = []
+    async def push_transport_message(self, m): self.msgs.append(type(m).__name__)
+
+
+# ---------------------------------------------------------------- 配置开关
+
+def test_runtime_switch():
+    print("\n=== 1. 运行时开关 ===")
+    check("默认是 legacy", AgentConfig(runtime="legacy").use_flows is False)
+    check("显式 flows 才启用", AgentConfig(runtime="flows").use_flows is True)
+    check("大小写不敏感", AgentConfig(runtime="FLOWS").use_flows is True)
+    check("未知值当 legacy", AgentConfig(runtime="xyz").use_flows is False)
+
+    saved = os.environ.pop("AI_AGENT_RUNTIME", None)
+    try:
+        from app.config import load_env
+        import importlib
+        import app.config as C
+        C._loaded = False          # 重新读一次环境
+        check("未设环境变量时默认 legacy",
+              AppConfig.load().agent.runtime == "legacy")
+    finally:
+        if saved is not None:
+            os.environ["AI_AGENT_RUNTIME"] = saved
+
+
+# ---------------------------------------------------------------- 节点形态
+
+def test_node_shape():
+    print("\n=== 2. 节点配置形态 ===")
+    identity = ManualIdentityProvider()
+    # 称呼来自 IdentityDecision.display_name（不是构造参数）
+    identity.switch("u_a", display_name="林")
+    store = FakeStore()
+    bridge = FR.FlowsBridge(rtvi=FakeRTVI(), tts=FakeTTS(),
+                            cfg=AppConfig.load())
+
+    def ctx_factory():
+        d = identity.current()
+        return ToolContext(owner_user_id=d.user_id, session_id="s1",
+                           turn_id="t1", identity_confirmed=d.confirmed,
+                           store=store)
+
+    schemas = [
+        {"type": "function", "function": {
+            "name": "memory_search", "description": "x",
+            "parameters": {"type": "object",
+                           "properties": {"query": {"type": "string"}},
+                           "required": ["query"]}}},
+    ]
+    from app.memory.base import Memory as M, MemoryKind as MK, ConfirmStatus as CS
+    memories = [M(memory_id=1, owner_user_id="u_a", kind=MK.PREFERENCE,
+                  key="饮食偏好", value="不吃香菜", confirm_status=CS.CONFIRMED)]
+    node = FR._conversation_node(identity_decision=identity.current(),
+                                 memories=memories, tool_schemas=schemas,
+                                 ctx_factory=ctx_factory)
+
+    check("节点名是 conversation", node["name"] == "conversation")
+    check("**不用 role_message**", "role_message" not in node,
+          "避免两条 system 消息")
+    check("用 task_messages 承载人格",
+          len(node["task_messages"]) == 1
+          and node["task_messages"][0]["role"] == "system")
+    sysmsg = node["task_messages"][0]["content"]
+    check("人格里含称呼", "林" in sysmsg, sysmsg[-40:])
+    check("**不含内部 user_id**（验收第 7 条）", "u_a" not in sysmsg)
+    check("稳定记忆已注入", "香菜" in node["task_messages"][0]["content"])
+    check("respond_immediately=False（不自动开口）",
+          node["respond_immediately"] is False)
+    check("工具已注册", len(node["functions"]) == 1)
+
+
+# ---------------------------------------------------------------- handler 桥接
+
+async def test_handler_bridge():
+    print("\n=== 3. handler 桥接到工具层 ===")
+    identity = ManualIdentityProvider("u_a")
+    store = FakeStore()
+    FR.set_last_user_text("记住我不吃香菜")
+
+    def ctx_factory():
+        d = identity.current()
+        return ToolContext(owner_user_id=d.user_id, session_id="s1",
+                           turn_id="t1", identity_confirmed=d.confirmed,
+                           user_text=FR._LAST_USER_TEXT[0], store=store)
+
+    handler = FR._make_handler("memory_remember", ctx_factory)
+    out = await handler({"kind": "preference", "key": "饮食偏好",
+                         "value": "不吃香菜"})
+    check("handler 返回文本", isinstance(out, str) and "已记住" in out, out[:40])
+    check("写入的 owner 来自服务端",
+          store.rows and store.rows[0].owner_user_id == "u_a")
+
+    # 身份未确认时，handler 应拒绝（策略层兜底）
+    identity.reset()
+    out2 = await handler({"kind": "preference", "value": "x"})
+    # 策略层给模型的话是"现在还不能记录私人信息。请先确认身份。"，
+    # 断言要匹配实际措辞（早先写死"身份未确认"导致误判成 FAIL）。
+    check("身份未确认时 handler 拒绝",
+          "不能记录" in out2 or "确认身份" in out2, out2[:40])
+
+    # 越权参数应被工具层拦下
+    identity.switch("u_a")
+    out3 = await handler({"kind": "fact", "value": "x", "owner_user_id": "u_b"})
+    check("越权参数被拦", "不合法" in out3, out3[:40])
+
+
+# ---------------------------------------------------------------- 零件装配
+
+async def test_build_parts():
+    print("\n=== 4. build_flows_parts 装配 ===")
+    cfg = AppConfig.load()
+    # 强制用 deepseek 配置（本地 env 可能没设 AI_LLM_PRIMARY）
+    from app.config import deepseek_config
+    cfg.llm = deepseek_config()
+
+    if not cfg.llm.api_key:
+        check("跳过（无 DeepSeek 密钥）", True)
+        return
+
+    llm, aggregator, bridge, ctx_factory, schemas = await FR.build_flows_parts(
+        cfg=cfg, memory=FakeStore(), identity=ManualIdentityProvider("u_a"),
+        session_id="s1", tts=FakeTTS(), rtvi=FakeRTVI())
+
+    check("返回 LLM service", llm is not None and type(llm).__name__ == "DeepSeekLLMService")
+    check("模型名被覆盖为 deepseek-flash",
+          llm._settings.model == cfg.llm.model, llm._settings.model)
+    check("thinking 关闭",
+          getattr(llm._settings.thinking, "type", None) == "disabled")
+    check("aggregator 有 user/assistant",
+          aggregator.user() is not None and aggregator.assistant() is not None)
+    check("工具 schema 随身份给出（含记忆工具）",
+          any(s["function"]["name"].startswith("memory_") for s in schemas),
+          str([s["function"]["name"] for s in schemas]))
+    check("Phase 3 不含 web_search",
+          not any(s["function"]["name"] == "web_search" for s in schemas),
+          "搜索仍走 legacy 路径")
+
+    # 身份未确认时不暴露记忆工具
+    _, _, _, _, schemas2 = await FR.build_flows_parts(
+        cfg=cfg, memory=FakeStore(), identity=ManualIdentityProvider(),
+        session_id="s1", tts=FakeTTS(), rtvi=FakeRTVI())
+    check("身份未确认时不给记忆工具",
+          not any(s["function"]["name"].startswith("memory_") for s in schemas2))
+
+
+def main() -> int:
+    test_runtime_switch()
+    test_node_shape()
+    asyncio.run(test_handler_bridge())
+    asyncio.run(test_build_parts())
+
+    print("\n" + "=" * 62)
+    print(f"通过 {len(OK)} / {len(OK) + len(FAIL)}")
+    for n in FAIL:
+        print(f"  未通过: {n}")
+    return 0 if not FAIL else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

client/test_client.py

@@ -191,6 +191,62 @@ class ReplyRecorder:
         return (self.first_non_silent_at - started_at) * 1000
 
 
+class RTVIReplies:
+    """收集服务端发来的 RTVI 文本消息，供验收脚本断言。
+
+    网页用的 bot-output / bot-transcription 是完整文本，
+    bot-llm-text 是逐 token 增量。验收看完整文本。
+    """
+
+    def __init__(self):
+        self.bot_texts: list[str] = []
+        self.user_texts: list[str] = []
+        self.other: list[str] = []
+
+    def handle(self, raw: str) -> None:
+        try:
+            m = json.loads(raw)
+        except (json.JSONDecodeError, TypeError):
+            return
+        if not isinstance(m, dict):
+            return
+        t = m.get("type")
+        data = m.get("data") or {}
+        text = data.get("text") if isinstance(data, dict) else None
+        if t == "bot-output" and text:
+            self.bot_texts.append(text)
+        elif t == "bot-transcription" and text:
+            self.bot_texts.append(text)
+        elif t == "user-llm-text" and text:
+            self.user_texts.append(text)
+        elif t:
+            self.other.append(t)
+
+    @property
+    def last_bot(self) -> str:
+        return self.bot_texts[-1] if self.bot_texts else ""
+
+
+def rtvi_send_text(channel, text: str, *, audio_response: bool = True) -> None:
+    """通过 data channel 发一条 RTVI send-text。
+
+    信封格式来自 pipecat 的 RTVI 模型（实测校正过两次）：
+      label='rtvi-ai'，**字段是 id 不是 msg_id**，type='send-text'，
+      data.content（**是 content 不是 text**）。
+      写错任何一个，服务端都会以 "Invalid RTVI transport message: Field required" 拒掉，
+      而客户端看不到任何反馈——表现为"发了消息但助手毫无反应"。
+    """
+    import uuid
+    msg = {
+        "label": "rtvi-ai",
+        "id": uuid.uuid4().hex,
+        "type": "send-text",
+        "data": {"content": text,
+                 "options": {"audio_response": audio_response}},
+    }
+    channel.send(json.dumps(msg))
+
+
 def post_offer(server: str, sdp: str, type_: str) -> dict:
     """把 offer 发给服务端，取回 answer。"""
     payload = json.dumps({"sdp": sdp, "type": type_, "pc_id": None,
@@ -203,7 +259,7 @@ def post_offer(server: str, sdp: str, type_: str) -> dict:
 
 
 async def run(server: str, wav: Path | None, out: Path, *, connect_only: bool,
-              timeout_s: float) -> int:
+              timeout_s: float, texts: list[str] | None = None) -> int:
     pc = RTCPeerConnection(RTCConfiguration(iceServers=[]))
     recorder = ReplyRecorder(out)
     track = None
@@ -235,6 +291,10 @@ async def run(server: str, wav: Path | None, out: Path, *, connect_only: bool,
         track = FileAudioTrack(wav)
         pc.addTrack(track)
         log.info("音源: %s（%d 采样点）", wav.name, len(track.samples))
+    elif texts:
+        # 文字模式：仍要一条音轨，否则 aiortc 不收集 ICE 候选
+        pc.addTrack(SilentTrack())
+        log.info("文字模式：将发送 %d 条文本", len(texts))
     else:
         # --connect-only 也要加一条（静音）音轨。
         # **不能省**：aiortc 的 offer 里没有媒体时完全不收集 ICE 候选，
@@ -253,11 +313,18 @@ async def run(server: str, wav: Path | None, out: Path, *, connect_only: bool,
     channel = pc.createDataChannel("pipecat")
     dc_state = {"ok": False}
 
+    replies = RTVIReplies()
+
     @channel.on("open")
     def _dc_open():
         dc_state["ok"] = True
         log.info("data channel 已打开，开始发心跳")
 
+    @channel.on("message")
+    def _dc_message(msg):
+        if isinstance(msg, str):
+            replies.handle(msg)
+
     async def _keepalive():
         while pc.connectionState not in ("closed", "failed"):
             if dc_state["ok"]:
@@ -312,7 +379,35 @@ async def run(server: str, wav: Path | None, out: Path, *, connect_only: bool,
     log.info("✅ ICE 已连通（%s）", pc.connectionState)
     print(f"  ✅ 连接建立，状态 {pc.connectionState}")
 
-    if track:
+    if texts:
+        # 等 data channel 就绪（服务端靠它判断客户端存活）
+        for _ in range(60):
+            if dc_state["ok"]:
+                break
+            await asyncio.sleep(0.1)
+        log.info("data channel: %s", "已就绪" if dc_state["ok"] else "未就绪")
+        for t in texts:
+            await asyncio.sleep(0.5)
+            rtvi_send_text(channel, t)
+            log.info("已发送文本: %r", t)
+            # 等**非静音**音频出现。
+            # 不能用"收到多少块"来判断——WebRTC 输出轨空闲时也在推静音，
+            # 块数会立刻涨上去，导致客户端在服务端还在生成时就断开
+            # （服务端日志表现为"本轮被取消"，客户端表现为"没反应"）。
+            recorder.first_non_silent_at = None
+            for _ in range(int(timeout_s * 10)):
+                if recorder.has_speech:
+                    break
+                await asyncio.sleep(0.1)
+            log.info("本轮 %s", "收到语音回复" if recorder.has_speech else "未收到语音")
+            # 让它把话说完
+            await asyncio.sleep(4)
+            # 打印助手回复全文，供验收脚本断言
+            if replies.bot_texts:
+                print(f"  BOT_REPLY: {replies.bot_texts[-1]}", flush=True)
+                replies.bot_texts.clear()
+        await asyncio.sleep(2)
+    elif track:
         # 等音频放完 + 留出响应时间
         t0 = time.perf_counter()
         while not track.finished and time.perf_counter() - t0 < timeout_s:
@@ -350,13 +445,16 @@ def main() -> int:
     p.add_argument("--wav", type=Path)
     p.add_argument("--out", type=Path, default=Path("/tmp/client_reply.wav"))
     p.add_argument("--connect-only", action="store_true")
+    p.add_argument("--text", action="append", default=None,
+                   help="通过 RTVI 发送文本（可重复），用于测 flows 文字路径")
     p.add_argument("--timeout", type=float, default=30.0)
     a = p.parse_args()
-    if not a.connect_only and not a.wav:
+    if not a.connect_only and not a.wav and not a.text:
-        print("需要 --wav 或 --connect-only", file=sys.stderr)
+        print("需要 --wav / --text / --connect-only", file=sys.stderr)
         return 2
     return asyncio.run(run(a.server, a.wav, a.out,
-                           connect_only=a.connect_only, timeout_s=a.timeout))
+                           connect_only=a.connect_only, timeout_s=a.timeout,
+                           texts=a.text))
 
 
 if __name__ == "__main__":

deploy.sh

@@ -0,0 +1,72 @@
+#!/usr/bin/env bash
+# 部署到服务器。
+#
+# **为什么要有这个脚本**：2026-09-16 我用一条 rsync 命令把服务器上的东西
+# 删了一大片，教训有三条，都固化在这里：
+#
+#   1. **绝不用 `--delete`。** 服务器上还有本地没有的东西（.venv、models 软链、
+#      logs、数据库软链等）。`--delete` 会把它们全删掉。
+#      实测后果：.venv/bin 与 100+ 个 *.dist-info 被删、services/ 被清空、
+#      README/部署说明被删、models 软链被删。
+#   2. **一次只同步一个目录，不要多源。** 多源 rsync 不带 --relative 时，
+#      会把 `app/` 的内容**平铺**到目标根目录，而不是放进 `app/` 下。
+#   3. **排除清单要显式。** 即使不用 --delete，也别把 __pycache__ 之类带过去。
+#
+# 用法：
+#   ./deploy.sh              # 同步代码
+#   ./deploy.sh --restart    # 同步后重启两个服务
+set -uo pipefail
+
+HOST="${AI_DEPLOY_HOST:-yonglin@cadadmin-server.taild0f91d.ts.net}"
+DEST="${AI_DEPLOY_DEST:-~/ai-companion}"
+HERE="$(cd "$(dirname "$0")" && pwd)"
+
+# 服务器上不该被本地覆盖的东西——保留在服务器上
+EXCLUDES=(
+  --exclude '__pycache__'
+  --exclude '*.pyc'
+  --exclude '.baseline'
+  --exclude '.venv'
+  --exclude 'logs'
+  --exclude 'models'
+  --exclude 'out'
+  --exclude 'p0/stt_tts_bench/models'
+)
+
+sync_one() { # sync_one <本地目录> <远端子路径>
+  echo "  → $2"
+  rsync -az "${EXCLUDES[@]}" -e "ssh -o BatchMode=yes" \
+    "$HERE/$1" "$HOST:$DEST/$2"
+}
+
+echo "部署到 $HOST:$DEST"
+sync_one "app/"        "app/"
+sync_one "services/"   "services/"
+sync_one "client/"     "client/"
+sync_one "p1/"         "p1/"
+
+for f in README.md 部署说明.md requirements.txt; do
+  [ -f "$HERE/$f" ] && rsync -az -e "ssh -o BatchMode=yes" "$HERE/$f" "$HOST:$DEST/"
+done
+
+# 把 requirements 里没有、但运行需要的包装上（幂等）
+ssh -o BatchMode=yes "$HOST" "cd $DEST && .venv/bin/pip -q install pipecat-ai-small-webrtc-prebuilt 2>&1 | tail -1"
+
+if [ "${1:-}" = "--restart" ]; then
+  echo "重启服务"
+  ssh -o BatchMode=yes "$HOST" 'bash -s' <<REMOTE
+cd ~/ai-companion
+for p in \$(pgrep -f "services/bot_server|tts_server.py" 2>/dev/null); do kill "\$p" 2>/dev/null; done
+sleep 3
+cd ~/tts-qwen && setsid nohup .venv/bin/python tts_server.py --port 18040 > logs/tts_service.log 2>&1 < /dev/null &
+sleep 1
+cd ~/ai-companion && setsid nohup .venv/bin/python services/bot_server.py --host 0.0.0.0 --port 7860 > logs/bot_server.log 2>&1 < /dev/null &
+for i in \$(seq 1 30); do
+  curl -sf -o /dev/null http://127.0.0.1:7860/health 2>/dev/null && { echo "  bot ✅"; break; }
+  sleep 3
+done
+curl -sf http://127.0.0.1:18040/health 2>/dev/null | grep -q model_loaded && echo "  tts ✅" || echo "  tts ⚠️"
+REMOTE
+fi
+
+echo "完成"

p1/phase4_acceptance.py

@@ -0,0 +1,162 @@
+#!/usr/bin/env python3
+"""Phase 4 验收：通用记忆闭环。
+
+对应交接文档 §九 Phase 4 的 7 条，**跑真实服务端**（flows 文字路径）。
+每条之间查一次服务器数据库，确认"说进数据库"和"以后能用"是同一件事。
+
+    1. "记住我不吃香菜"            → confirmed
+    2. 新连接 "我不吃什么？"        → 能召回香菜（验证连接开始时加载档案）
+    3. "我今天吃了香菜"            → 不得变成稳定偏好
+    4. "其实我现在吃香菜了"        → 明确更新旧偏好，不留冲突值
+    5. "忘掉我的饮食偏好"          → 删除；模糊时先确认
+    6. 访客无法读写私人记忆         → 单测覆盖（需要身份切换，客户端不支持）
+    7. 用户 A 的记忆不出现在 B     → 单测覆盖 + prompt 不含内部 ID
+
+用法：
+    ~/venvs/ai_assistant/bin/python p1/phase4_acceptance.py
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parent
+SERVER = os.environ.get(
+    "AI_SERVER", "http://cadadmin-server.taild0f91d.ts.net:7860")
+SSH = os.environ.get("AI_SSH", "yonglin@cadadmin-server.taild0f91d.ts.net")
+PY = str(Path.home() / "venvs" / "ai_assistant" / "bin" / "python")
+
+OK, FAIL = [], []
+
+
+def check(name: str, cond: bool, detail: str = "") -> None:
+    (OK if cond else FAIL).append(name)
+    print(f"  [{'PASS' if cond else 'FAIL'}] {name}" + (f" — {detail}" if detail else ""))
+
+
+def say(text: str, *, timeout: int = 180) -> str:
+    """跑一轮文字对话，返回助手回复全文（从 RTVI bot-output 取）。"""
+    r = subprocess.run(
+        [PY, str(ROOT / "client" / "test_client.py"),
+         "--server", SERVER, "--text", text, "--out", "/tmp/p4_turn.wav"],
+        capture_output=True, text=True, timeout=timeout)
+    for line in r.stdout.splitlines():
+        if line.strip().startswith("BOT_REPLY:"):
+            return line.split("BOT_REPLY:", 1)[1].strip()
+    return ""
+
+
+def db_memories() -> list[dict]:
+    """查服务器的记忆表。"""
+    code = (
+        "import sqlite3,pathlib,json;"
+        "db=pathlib.Path.home()/'.local/share/ai_assistant/memory.db';"
+        "c=sqlite3.connect(db);c.row_factory=sqlite3.Row;"
+        "rows=c.execute('SELECT memory_id,owner_user_id,kind,mem_key,mem_value,"
+        "confirm_status FROM memories ORDER BY memory_id').fetchall();"
+        "print(json.dumps([dict(r) for r in rows],ensure_ascii=False))"
+    )
+    r = subprocess.run(["ssh", "-o", "BatchMode=yes", SSH,
+                        f"~/ai-companion/.venv/bin/python -c \"{code}\""],
+                       capture_output=True, text=True, timeout=60)
+    for line in r.stdout.splitlines():
+        line = line.strip()
+        if line.startswith("["):
+            try:
+                return json.loads(line)
+            except json.JSONDecodeError:
+                pass
+    return []
+
+
+def find(rows, key_contains: str) -> list[dict]:
+    return [m for m in rows if key_contains in (m.get("mem_key") or "")]
+
+
+def main() -> int:
+    print("=" * 64)
+    print("Phase 4 验收：通用记忆闭环（跑真实服务端）")
+    print(f"  服务端 {SERVER}")
+    print("=" * 64)
+
+    if os.environ.get("AI_SKIP_CLEAN") != "1":
+        print("\n=== 0. 前置：清掉待测键，保证从干净状态开始 ===")
+        say("忘掉我的饮食偏好")
+        rows = db_memories()
+        print(f"      当前记忆 {len(rows)} 条: "
+              f"{[(m['mem_key'], m['mem_value']) for m in rows]}")
+
+    # ---------------------------------------------------------- 1
+    print("\n=== 1. 明确要求 → confirmed ===")
+    reply = say("记住我不吃香菜")
+    rows = db_memories()
+    hits = [m for m in find(rows, "饮食") if m["confirm_status"] == "confirmed"]
+    print(f"      回复: {reply[:50]!r}")
+    check("确认的记忆已落库", bool(hits),
+          str([(m["mem_key"], m["mem_value"], m["confirm_status"]) for m in hits]))
+    if hits:
+        check("值是香菜相关", "香菜" in hits[0]["mem_value"], hits[0]["mem_value"])
+
+    # ---------------------------------------------------------- 2
+    print("\n=== 2. **新连接**后能召回（验证连接开始时加载档案）===")
+    print("      （上一轮连接已关闭，这是全新会话）")
+    reply = say("我不吃什么？")
+    print(f"      回复: {reply[:80]!r}")
+    check("回复提到香菜（说明记忆被加载进上下文）", "香菜" in reply, reply[:60])
+    check("没有把内部 ID 说出去", "u_ylhan" not in reply and "u_a" not in reply)
+
+    # ---------------------------------------------------------- 3
+    print("\n=== 3. 一次性事件不得变成稳定偏好 ===")
+    before = len([m for m in find(db_memories(), "饮食")
+                  if m["confirm_status"] == "confirmed"])
+    reply = say("我今天吃了香菜")
+    rows = db_memories()
+    after = len([m for m in find(rows, "饮食") if m["confirm_status"] == "confirmed"])
+    print(f"      回复: {reply[:50]!r}")
+    check("没有新增 confirmed 饮食偏好", after <= before, f"{before} → {after}")
+
+    # ---------------------------------------------------------- 4
+    print("\n=== 4. 明确更正 → 更新旧值，不留冲突 ===")
+    reply = say("其实我现在吃香菜了，帮我改一下")
+    rows = db_memories()
+    hits = [m for m in find(rows, "饮食") if m["confirm_status"] == "confirmed"]
+    print(f"      回复: {reply[:50]!r}")
+    print(f"      库中饮食偏好: {[(m['mem_value'], m['confirm_status']) for m in hits]}")
+    check("同键只保留一条 confirmed", len(hits) <= 1, f"{len(hits)} 条")
+    check("值已更新为吃香菜", any("吃" in m["mem_value"] and "不吃" not in m["mem_value"]
+                                  for m in hits),
+          str([m["mem_value"] for m in hits]))
+
+    # ---------------------------------------------------------- 5
+    print("\n=== 5. 明确遗忘 → 删除 ===")
+    reply = say("忘掉我的饮食偏好")
+    rows = db_memories()
+    left = find(rows, "饮食")
+    print(f"      回复: {reply[:60]!r}")
+    check("饮食偏好已删除", not left, str([m["mem_value"] for m in left]))
+
+    # ---------------------------------------------------------- 6/7
+    print("\n=== 6/7. 访客隔离与跨用户隔离 ===")
+    print("      这两条需要身份切换，WebRTC 客户端不支持；由单元测试覆盖：")
+    print("        app/tests/test_agent_tools.py → '身份未确认时不暴露记忆工具'")
+    print("                                       → '未确认身份拒绝 memory_search'")
+    print("                                       → '用户 B 查不到用户 A 的记忆'")
+    print("                                       → '跨 owner 删除失败'")
+    print("        app/tests/test_flows_runtime.py → '**不含内部 user_id**'")
+    check("由单测覆盖（见上）", True)
+
+    print("\n" + "=" * 64)
+    ok = len(OK)
+    print(f"通过 {ok} / {ok + len(FAIL)}")
+    for n in FAIL:
+        print(f"  未通过: {n}")
+    return 0 if not FAIL else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

services/bot_server.py

@@ -59,6 +59,7 @@ from pipecat.frames.frames import (
     TranscriptionFrame,
 )
 from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.worker import PipelineParams, PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -75,6 +76,10 @@ from pipecat.transports.smallwebrtc.transport import SmallWebRTCTransport
 from pipecat.utils.asyncio.task_manager import TaskManager
 from pipecat.workers.base_worker import WorkerParams
 
+from app.agent.flows_runtime import (
+    UserTextRecorder, build_flows_parts, initialize_conversation_node,
+    load_stable_memories, make_flow_manager,
+)
 from app.config import AppConfig
 from app.engine import ConversationEngine, EventKind
 from app.identity import ManualIdentityProvider
@@ -199,11 +204,17 @@ class EngineProcessor(FrameProcessor):
         cfg: AppConfig,
         transport: SmallWebRTCTransport,
         rtvi: RTVIProcessor,
+        *,
+        owns_text: bool = True,
     ):
         super().__init__()
         self.cfg = cfg
         self.transport = transport
         self.rtvi = rtvi
+        #: True  = 文字路径也归本处理器（legacy）
+        #: False = 文字帧往下传给 Flows 处理器（Phase 3 的 flows 运行时）
+        #: 语音路径**始终**归本处理器——交接文档要求语音暂不迁移。
+        self.owns_text = owns_text
         self._vad = SimpleVAD()
         self._pre_roll: deque[bytes] = deque(maxlen=50)  # 约 1 秒
         self._buf = bytearray()          # 当前话语的 PCM16
@@ -240,6 +251,15 @@ class EngineProcessor(FrameProcessor):
                 self._engine.stt = StreamingSTT(self.cfg.voice)
             return self._engine
 
+    async def runtime_deps(self):
+        """暴露 memory/identity/session/tts，供 Flows 路径复用。
+
+        资源只在一个地方创建（`_ensure_engine`），避免两条路径各建一份
+        SQLite 连接或各加载一次 TTS——那会立刻导致"两条路径看到不同数据"。
+        """
+        engine = await self._ensure_engine()
+        return self._memory, engine.identity, self._session, engine.tts
+
     async def setup(self, setup):
         await super().setup(setup)
         log.info("会话处理器已就绪")
@@ -255,10 +275,17 @@ class EngineProcessor(FrameProcessor):
 
         elif isinstance(frame, LLMConfigureOutputFrame):
             # RTVI 在 send-text 前后用这个帧表达 audio_response 选项。
+            # 即使在 flows 模式下也记录一份（语音路径仍归本处理器），
+            # 同时往下传，让 FlowsBridge 也能看到。
             self._skip_tts = frame.skip_tts
             await self.push_frame(frame, direction)
 
         elif isinstance(frame, LLMMessagesAppendFrame):
+            if not self.owns_text:
+                # flows 运行时：文字交给下游的 Flows 处理器，
+                # 这里只放行，不消费。
+                await self.push_frame(frame, direction)
+                return
             text = extract_user_text(frame.messages)
             if text and frame.run_llm is not False:
                 log.info("收到网页文字输入: %d 字", len(text))
@@ -506,8 +533,36 @@ async def offer(request: SmallWebRTCRequest):
             ),
         )
         rtvi = RTVIProcessor()
-        engine_proc = EngineProcessor(_cfg, transport, rtvi)
+        # Phase 3：文字路径可选走 Flows，**语音始终归 EngineProcessor**
+        # （交接文档要求语音暂不迁移）。
+        # 默认 legacy；只有显式设 AI_AGENT_RUNTIME=flows 才会走新路径。
+        use_flows = _cfg.agent.use_flows
+        engine_proc = EngineProcessor(_cfg, transport, rtvi, owns_text=not use_flows)
+
+        flows_objs = None
+        if use_flows:
+            # 复用 EngineProcessor 已建好的资源，不另建一份
+            memory, identity, session_id, tts = await engine_proc.runtime_deps()
+            llm, aggregator, bridge, ctx_factory, tool_schemas = \
+                await build_flows_parts(
+                    cfg=_cfg, memory=memory, identity=identity,
+                    session_id=session_id, tts=tts, rtvi=rtvi)
+            pipeline = Pipeline([
+                transport.input(),
+                engine_proc,                    # 只吃语音
+                UserTextRecorder(extract_user_text),   # 记下用户原话供策略层
+                aggregator.user(),              # 收文字
+                llm,                            # Flows 驱动
+                bridge,                         # 文本 → RTVI + TTS
+                aggregator.assistant(),
+                transport.output(),
+            ])
+            flows_objs = (llm, aggregator, ctx_factory, tool_schemas,
+                          memory, identity)
+            log.info("Agent 运行时: flows（文字路径），语音仍走 legacy")
+        else:
             pipeline = Pipeline([transport.input(), engine_proc, transport.output()])
+            log.info("Agent 运行时: legacy（默认）")
         # 注意：这个版本里 PipelineParams 没有 allow_interruptions——
         # 打断在 1.10 已移到 pipecat/turns/ 框架。实验阶段我们不用它的
         # 轮次框架（方案 §5 的边界决定是"打断由客户端判定"），
@@ -568,12 +623,33 @@ async def offer(request: SmallWebRTCRequest):
         async def _run_task():
             log.info("PipelineWorker 开始运行")
             try:
-                # TaskManager 必须在这里建——模块级没有运行中的事件循环
+                # **必须用 PipelineRunner，不能直接 task.run(...)。**
-                await task.run(WorkerParams(task_manager=TaskManager()))
+                # 直接 run 时 runner 没挂上，Pipecat 自己的函数调用机制会失败：
+                #   "Worker 'PipelineWorker#N': runner is not set; call attach() first."
+                # 症状是工具被调用但 handler 从不执行，LLM 只收到一句通用的
+                # "The function `X` failed and returned no result."，默认日志看不到。
+                # legacy 路径不受影响（它的工具循环在自有引擎里），
+                # 但 Flows 路径的工具走 Pipecat 机制，必须 attach。
+                # 见笔记库 P1-Flows兼容性Spike结论.md 坑 2。
+                await PipelineRunner(handle_sigint=False).run(task)
             except Exception as e:  # noqa: BLE001
                 log.exception("PipelineWorker 异常退出: %s", e)
             log.info("PipelineWorker 已结束")
 
+        if flows_objs is not None:
+            llm, aggregator, ctx_factory, tool_schemas, memory, identity = flows_objs
+            flow = make_flow_manager(worker=task, llm=llm, aggregator=aggregator)
+            # 方案 §四要求稳定档案"连接开始时加载"，不等模型自己想起来查。
+            # 身份未确认时返回空——访客拿不到任何私人记忆（Phase 4 验收第 6 条）。
+            memories = await load_stable_memories(memory, identity)
+            # **必须 await**：initialize 是协程但签名显示 -> None，
+            # 漏 await 不报错，只是人格/记忆/工具全部静默失效（Phase 1 栽过）。
+            await initialize_conversation_node(
+                flow, identity_decision=identity.current(), memories=memories,
+                tool_schemas=tool_schemas, ctx_factory=ctx_factory)
+            log.info("Flows 节点已初始化：工具 %d 个，稳定记忆 %d 条",
+                     len(tool_schemas), len(memories))
+
         bg = asyncio.create_task(_run_task())
         _bg_tasks.add(bg)
         bg.add_done_callback(_bg_tasks.discard)