feat: add system tools and state updates

backend/db/models.py

@@ -154,6 +154,8 @@ class Assistant(Base):
     enable_interrupt: Mapped[bool] = mapped_column(Boolean, default=True)
     turn_config: Mapped[dict] = mapped_column(JSON, default=dict)
     startup: Mapped[dict] = mapped_column(JSON, default=dict)
+    # Prompt 助手级系统工具；Workflow 的权限保存在各 Agent 节点中。
+    system_tools: Mapped[list] = mapped_column(JSON, default=list)
     vision_enabled: Mapped[bool] = mapped_column(Boolean, default=False)
     vision_model_resource_id: Mapped[str | None] = mapped_column(
         String(40),

backend/db/session.py

@@ -54,12 +54,13 @@ async def sync_default_tools() -> None:
             "id": "tool_end_call_default",
             "name": "结束对话",
             "function_name": "end_call",
-            "type": "end_call",
+            "type": "system",
             "description": "当用户明确要求结束对话，或任务已完成时调用。",
             "definition": {
                 "schema_version": 1,
-                "type": "end_call",
+                "type": "system",
                 "config": {
+                    "kind": "end_conversation",
                     "message_type": "none",
                     "custom_message": "",
                     "capture_reason": True,

backend/migrations/versions/20260804_0010_add_assistant_system_tools.py

@@ -0,0 +1,32 @@
+"""add assistant system tools
+
+Revision ID: 20260804_0010
+Revises: 20260801_0009
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "20260804_0010"
+down_revision: str | Sequence[str] | None = "20260801_0009"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "assistants",
+        sa.Column(
+            "system_tools",
+            sa.JSON(),
+            server_default=sa.text("'[]'"),
+            nullable=False,
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("assistants", "system_tools")

backend/migrations/versions/20260804_0011_tool_end_call_to_system.py

@@ -0,0 +1,44 @@
+"""move end conversation tools into the system category
+
+Revision ID: 20260804_0011
+Revises: 20260804_0010
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+
+
+revision: str = "20260804_0011"
+down_revision: str | Sequence[str] | None = "20260804_0010"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        UPDATE tools
+        SET type = 'system',
+            definition = jsonb_set(
+                jsonb_set(definition, '{type}', '"system"'::jsonb),
+                '{config,kind}',
+                '"end_conversation"'::jsonb,
+                true
+            )
+        WHERE type = 'end_call'
+        """
+    )
+
+
+def downgrade() -> None:
+    op.execute(
+        """
+        UPDATE tools
+        SET type = 'end_call',
+            definition = (definition #- '{config,kind}')
+                || '{"type":"end_call"}'::jsonb
+        WHERE type = 'system'
+          AND definition #>> '{config,kind}' = 'end_conversation'
+        """
+    )

backend/models.py

@@ -114,6 +114,8 @@ class AssistantConfig(BaseModel):
     # every item in ``tools``.
     tools: list[RuntimeTool] = Field(default_factory=list)
     llm_tool_ids: list[str] | None = None
+    # 助手级系统工具仅供 Prompt Pipeline；Workflow 在 Agent 节点中配置。
+    system_tools: list[str] = Field(default_factory=list)
     knowledge_base_id: str | None = None
     knowledge_base_name: str = ""
     knowledge_base_description: str = ""

backend/routes/assistants.py

@@ -36,6 +36,35 @@ def _validate_workflow(body: AssistantUpsert) -> None:
         return
     body.graph = normalize_graph(body.graph or {})
     errors = validate_graph(body.graph)
+    declared_variables = set(body.dynamic_variable_definitions)
+    for node in body.graph.get("nodes") or []:
+        node_id = str(node.get("id") or "")
+        node_type = node.get("type")
+        data = node.get("data") or {}
+        if node_type == "update_state":
+            unknown = sorted(
+                set((data.get("assignments") or {}).keys()) - declared_variables
+            )
+            if unknown:
+                errors.append(
+                    f"Update State 节点 {node_id} 引用了未声明变量:"
+                    + ",".join(unknown)
+                )
+        elif node_type == "agent":
+            authorized = set(data.get("stateVariableNames") or [])
+            unknown = sorted(authorized - declared_variables)
+            if unknown:
+                errors.append(
+                    f"Agent 节点 {node_id} 授权了未声明变量:"
+                    + ",".join(unknown)
+                )
+            if (
+                "update_state" in (data.get("systemTools") or [])
+                and not authorized
+            ):
+                errors.append(
+                    f"Agent 节点 {node_id} 启用更新状态工具时必须授权至少一个变量"
+                )
     if errors:
         raise HTTPException(400, "工作流校验失败:" + ";".join(errors))
     # Graph settings are the source of truth. The flat flag is only a session
@@ -283,6 +312,7 @@ async def _to_out(session: AsyncSession, assistant: Assistant) -> AssistantOut:
         enable_interrupt=assistant.enable_interrupt,
         turn_config=assistant.turn_config or {},
         startup=assistant.startup or {},
+        system_tools=assistant.system_tools or [],
         vision_enabled=assistant.vision_enabled,
         vision_model_resource_id=assistant.vision_model_resource_id,
         model_resource_ids=await _resource_ids(session, assistant.id),
@@ -359,6 +389,7 @@ async def duplicate_assistant(
         enable_interrupt=source.enable_interrupt,
         turn_config=dict(source.turn_config or {}),
         startup=dict(source.startup or {}),
+        system_tools=list(source.system_tools or []),
         vision_enabled=source.vision_enabled,
         vision_model_resource_id=source.vision_model_resource_id,
         knowledge_base_id=source.knowledge_base_id,

backend/schemas.py

@@ -19,7 +19,7 @@ ModelType = Literal["LLM", "ASR", "TTS", "Realtime", "Embedding", "Agent"]
 AssistantType = Literal["prompt", "workflow", "dify", "fastgpt", "opencode"]
 TurnEndStrategy = Literal["silence", "smart_turn"]
 KnowledgeRetrievalMode = Literal["automatic", "on_demand"]
-ToolType = Literal["end_call", "http", "mcp", "client"]
+ToolType = Literal["system", "http", "mcp", "client"]
 ToolStatus = Literal["active", "archived", "draft"]
 McpServerStatus = Literal["active", "archived", "draft"]
 McpTransport = Literal["streamable_http", "sse"]
@@ -30,6 +30,12 @@ ClientToolResponseWaitMode = Literal["timeout", "session"]
 DynamicVariableType = Literal["string", "number", "boolean"]
 PromptEntryMode = Literal["wait_user", "generate"]
 PromptOpeningMode = Literal["interruptible", "playback", "confirmation"]
+SystemToolKind = Literal[
+    "end_conversation",
+    "update_state",
+    "skip_turn",
+    "request_human_handoff",
+]
 
 # 外部应用类型:其 config.apiKey 是该助手私有密钥,读时打码 / 写时哨兵
 EXTERNAL_TYPES = {"dify", "fastgpt", "opencode"}
@@ -153,6 +159,8 @@ class AssistantUpsert(CamelModel):
     startup: StartupConfig = Field(default_factory=StartupConfig)
     vision_enabled: bool = False
     vision_model_resource_id: str | None = None
+    # 内置系统工具开关(仅 prompt 类型可用,类似 ElevenLabs Agent 的系统工具)。
+    system_tools: list[SystemToolKind] = Field(default_factory=list, max_length=4)
 
     model_resource_ids: dict[ModelType, str] = Field(default_factory=dict)
     knowledge_base_id: str | None = None
@@ -191,6 +199,8 @@ class AssistantUpsert(CamelModel):
         for field in ("prompt", "api_url", "api_key", "app_id"):
             if field not in allowed:
                 setattr(self, field, "")
+        if self.type != "prompt":
+            self.system_tools = []
         if "graph" not in allowed:
             self.graph = {}
         if self.type == "workflow":
@@ -210,6 +220,15 @@ class AssistantUpsert(CamelModel):
         # 外部托管大脑只能 cascade,拦住不兼容的 realtime
         if self.runtime_mode == "realtime" and self.type not in REALTIME_CAPABLE_TYPES:
             raise ValueError(f"类型 {self.type} 不支持 realtime 运行模式")
+        if self.type == "prompt":
+            self.system_tools = list(dict.fromkeys(self.system_tools))
+            if self.runtime_mode == "realtime" and self.system_tools:
+                raise ValueError("Prompt Realtime 模式暂不支持系统工具")
+            if (
+                "update_state" in self.system_tools
+                and not self.dynamic_variable_definitions
+            ):
+                raise ValueError("启用更新状态工具前必须声明至少一个动态变量")
         return self
 
 
@@ -250,7 +269,8 @@ class ToolParameter(CamelModel):
     required: bool = True
 
 
-class EndCallToolConfig(CamelModel):
+class SystemToolConfig(CamelModel):
+    kind: Literal["end_conversation"] = "end_conversation"
     message_type: Literal["none", "custom"] = "none"
     custom_message: str = ""
     capture_reason: bool = True
@@ -275,10 +295,10 @@ class HttpToolConfig(CamelModel):
         return value
 
 
-class EndCallToolDefinition(CamelModel):
+class SystemToolDefinition(CamelModel):
     schema_version: int = 1
-    type: Literal["end_call"] = "end_call"
-    config: EndCallToolConfig = Field(default_factory=EndCallToolConfig)
+    type: Literal["system"] = "system"
+    config: SystemToolConfig = Field(default_factory=SystemToolConfig)
 
 
 class HttpToolDefinition(CamelModel):
@@ -326,7 +346,7 @@ class McpToolDefinition(CamelModel):
 
 ToolDefinition = Annotated[
     Union[
-        EndCallToolDefinition,
+        SystemToolDefinition,
         HttpToolDefinition,
         ClientToolDefinition,
         McpToolDefinition,

backend/services/brains/prompt_brain.py

@@ -43,14 +43,14 @@ from services.message_stage import (
     MessageStageRunner,
     MessageStageSpec,
 )
-from services.runtime_variables import DynamicVariableStore
+from services.runtime_variables import DynamicVariableError, DynamicVariableStore
+from services.system_tools import SYSTEM_TOOL_KINDS, state_update_properties
 from services.tool_executor import ToolExecutionError, ToolExecutor
 from services.tool_policy import policy_for_tool
 
 
 PREFLIGHT_TIMEOUT_SECONDS = 30
 
-
 class PromptBrain(BaseBrain):
     spec = BrainSpec(
         type="prompt",
@@ -123,22 +123,37 @@ class PromptBrain(BaseBrain):
             set(cfg.llm_tool_ids) if cfg.llm_tool_ids is not None else None
         )
         schemas: list[FunctionSchema] = []
+        registered_names: set[str] = set()
         for tool in cfg.tools:
             if llm_tool_ids is not None and tool.id not in llm_tool_ids:
                 continue
-            if tool.type == "end_call":
+            if tool.type == "system":
                 schema, handler = self._make_end_call_tool(tool, runtime)
             elif tool.type in {"http", "mcp", "client"}:
                 schema, handler = self._make_remote_tool(tool, runtime)
             else:
                 continue
             schemas.append(schema)
+            registered_names.add(schema.name)
             policy = policy_for_tool(tool)
             runtime.llm.register_function(
                 tool.function_name,
                 handler,
                 cancel_on_interruption=policy.cancel_on_interruption,
             )
+        for kind in cfg.system_tools or []:
+            if kind not in SYSTEM_TOOL_KINDS:
+                logger.warning(f"忽略未知系统工具: {kind}")
+                continue
+            schema, handler = self._make_system_tool(kind, runtime)
+            if schema.name in registered_names:
+                logger.warning(
+                    f"跳过系统工具 {schema.name}: 与已绑定工具函数名冲突"
+                )
+                continue
+            registered_names.add(schema.name)
+            schemas.append(schema)
+            runtime.llm.register_function(schema.name, handler)
         runtime.set_tools(schemas)
 
     async def run_preflight(self) -> None:
@@ -582,3 +597,162 @@ class PromptBrain(BaseBrain):
             required=["reason"] if capture_reason else [],
         )
         return schema, end_call
+
+    # ---------- 内置系统工具(助手配置 system_tools 开关) ----------
+
+    def _make_system_tool(self, kind: str, runtime: BrainRuntime):
+        if kind == "end_conversation":
+            return self._make_end_conversation_tool(runtime)
+        if kind == "update_state":
+            return self._make_update_state_tool(runtime)
+        if kind == "skip_turn":
+            return self._make_skip_turn_tool()
+        if kind == "request_human_handoff":
+            return self._make_handoff_tool(runtime)
+        raise ValueError(f"未知系统工具: {kind}")
+
+    def _make_end_conversation_tool(self, runtime: BrainRuntime):
+        """结束本次对话,等待模型已生成的告别语播完后再挂断。"""
+
+        async def end_conversation(params: FunctionCallParams) -> None:
+            reason = str(
+                params.arguments.get("reason") or "end_conversation"
+            ).strip()
+            self._waiting_for_generated_end_speech = True
+            runtime.call_end.begin(reason)
+            await params.result_callback(
+                {"status": "success", "action": "ending_call"},
+                properties=FunctionCallResultProperties(run_llm=False),
+            )
+
+        schema = FunctionSchema(
+            name="end_conversation",
+            description=(
+                "礼貌地结束本次对话。当用户明确告别、表示任务已完成"
+                "或要求挂断时调用。"
+            ),
+            properties={
+                "reason": {
+                    "type": "string",
+                    "description": "结束对话的简短原因。",
+                }
+            },
+            required=[],
+        )
+        return schema, end_conversation
+
+    def _make_update_state_tool(self, runtime: BrainRuntime):
+        """更新已声明的动态变量(会话状态),并让模型继续当前回答。"""
+
+        writable = state_update_properties(self._cfg.dynamic_variable_definitions)
+
+        async def update_state(params: FunctionCallParams) -> None:
+            state = dict(params.arguments or {})
+            try:
+                changed = self._store.assign_declared_many(state)
+            except DynamicVariableError as exc:
+                await params.result_callback(
+                    {"status": "error", "message": f"状态更新失败: {exc}"}
+                )
+                return
+            if changed:
+                self._refresh_prompt()
+                await runtime.queue_frame(
+                    OutputTransportMessageUrgentFrame(
+                        message={
+                            "type": "session-variables",
+                            "reason": "update_state",
+                            "variables": self._store.public_values(),
+                            "changed": changed,
+                        }
+                    )
+                )
+            await params.result_callback(
+                {
+                    "status": "success",
+                    "changed": changed,
+                    "variables": self._store.public_values(),
+                }
+            )
+
+        schema = FunctionSchema(
+            name="update_state",
+            description=(
+                "静默更新本次对话中已经声明并明确列出的动态变量。"
+                "只提交本轮获得或确认的信息，更新后继续当前回答。"
+            ),
+            properties=writable,
+            required=[],
+        )
+        return schema, update_state
+
+    def _make_skip_turn_tool(self):
+        """跳过当前轮次,不生成任何语音回复。"""
+
+        async def skip_turn(params: FunctionCallParams) -> None:
+            reason = str(params.arguments.get("reason") or "").strip()
+            result = {"status": "success", "action": "skip_turn"}
+            if reason:
+                result["reason"] = reason
+            await params.result_callback(
+                result,
+                properties=FunctionCallResultProperties(run_llm=False),
+            )
+
+        schema = FunctionSchema(
+            name="skip_turn",
+            description=(
+                "跳过当前轮次,不生成任何语音回复。"
+                "仅当用户明确要求稍等、话还没说完，或输入可确认只是噪音时调用。"
+            ),
+            properties={
+                "reason": {
+                    "type": "string",
+                    "description": "跳过本轮的原因(可选)。",
+                }
+            },
+            required=[],
+        )
+        return schema, skip_turn
+
+    def _make_handoff_tool(self, runtime: BrainRuntime):
+        """提交人工接管请求；请求完成前保持当前 AI 会话可用。"""
+
+        async def request_human_handoff(params: FunctionCallParams) -> None:
+            reason = str(
+                params.arguments.get("reason") or "human_handoff"
+            ).strip()
+            await runtime.queue_frame(
+                OutputTransportMessageUrgentFrame(
+                    message={
+                        "type": "handoff-requested",
+                        "source": "prompt-system-tool",
+                        "reason": reason,
+                        "message": "用户请求转接人工服务。",
+                    }
+                )
+            )
+            await params.result_callback(
+                {
+                    "status": "requested",
+                    "action": "human_handoff_requested",
+                    "message": "人工接管请求已提交，请告知用户正在等待人工响应。",
+                }
+            )
+
+        schema = FunctionSchema(
+            name="request_human_handoff",
+            description=(
+                "提交人工接管请求。当用户明确要求人工服务、投诉升级或 AI 无法"
+                "解决时调用。该工具只提交请求，不代表人工已经接通；调用后继续"
+                "回复用户并说明正在等待人工响应。"
+            ),
+            properties={
+                "reason": {
+                    "type": "string",
+                    "description": "转接人工的原因。",
+                }
+            },
+            required=[],
+        )
+        return schema, request_human_handoff

backend/services/brains/workflow_brain.py

@@ -56,7 +56,8 @@ from services.message_stage import (
     MessageStageRunner,
     MessageStageSpec,
 )
-from services.runtime_variables import DynamicVariableStore
+from services.runtime_variables import DynamicVariableError, DynamicVariableStore
+from services.system_tools import SYSTEM_TOOL_KINDS, state_update_properties
 from services.tool_executor import ToolExecutionError, ToolExecutor
 from services.tool_policy import policy_for_tool
 from services.workflow.agent import WorkflowAgentStage
@@ -184,6 +185,7 @@ class WorkflowBrain(BaseBrain):
         self._output: WorkflowOutput | None = None
         self._agent_stage: WorkflowAgentStage | None = None
         self._ended = False
+        self._waiting_for_generated_end_speech = False
         self._next_message_token = 1
         self._pending_message: _MessageContinuation | None = None
 
@@ -229,6 +231,7 @@ class WorkflowBrain(BaseBrain):
             runtime=runtime,
         )
         self._ended = False
+        self._waiting_for_generated_end_speech = False
         self._next_message_token = 1
         self._pending_message = None
         self._manager = ConfiguredFlowManager(
@@ -494,18 +497,30 @@ class WorkflowBrain(BaseBrain):
             return None
         return decision.edge
 
+    async def on_assistant_text_start(self, _turn_id: str) -> None:
+        if self._runtime is not None:
+            self._runtime.call_end.begin_response()
+
     async def on_assistant_text_end(
         self,
         _turn_id: str,
         content: str,
         interrupted: bool,
     ) -> None:
-        if not content or interrupted or self._ended:
-            return
-        self._store.record("agent", content, completed_agent_turn=True)
-        self._state.consume_user_turn()
-        if self._engine.node_type(self._state.current_node_id) == "agent":
-            self._state.status = WorkflowStatus.WAITING_USER
+        if content and not interrupted and not self._ended:
+            self._store.record("agent", content, completed_agent_turn=True)
+            self._state.consume_user_turn()
+            if self._engine.node_type(self._state.current_node_id) == "agent":
+                self._state.status = WorkflowStatus.WAITING_USER
+        if (
+            self._waiting_for_generated_end_speech
+            and self._runtime is not None
+            and self._runtime.call_end.ending
+        ):
+            self._waiting_for_generated_end_speech = False
+            await self._runtime.call_end.finish_after_current_speech(
+                has_text=bool(content.strip()) and not interrupted
+            )
 
     async def _refresh_agent_prompt(self, node_id: str) -> None:
         await self._require_agent_stage().refresh_prompt(node_id)
@@ -528,15 +543,43 @@ class WorkflowBrain(BaseBrain):
     ) -> NodeConfig:
         stage = self._engine.agent_stage_config(node_id)
         functions: list[FlowsFunctionSchema] = []
+        registered_names: set[str] = set()
+
+        def append_function(function: FlowsFunctionSchema | None) -> None:
+            if function is None:
+                return
+            function_name = getattr(function, "name", "")
+            if not function_name:
+                # Runtime-provided global functions are opaque in a few
+                # adapters; FlowManager remains responsible for those names.
+                functions.append(function)
+                return
+            if function_name in registered_names:
+                logger.warning(
+                    f"跳过 Agent {node_id} 的函数 {function_name}: 函数名冲突"
+                )
+                return
+            registered_names.add(function_name)
+            functions.append(function)
+
         for tool_id in stage.tool_ids:
             tool = self._tool_by_id.get(str(tool_id))
             if tool and tool.type in {"http", "mcp", "client"}:
-                functions.append(self._flow_tool(tool, node_id))
-        knowledge_function = self._knowledge_function(node_id)
-        if knowledge_function:
-            functions.append(knowledge_function)
+                append_function(self._flow_tool(tool, node_id))
+        append_function(self._knowledge_function(node_id))
         if stage.vision_enabled and self._require_runtime().vision_function:
-            functions.append(self._require_runtime().vision_function)
+            append_function(self._require_runtime().vision_function)
+        for kind in stage.system_tools:
+            if kind not in SYSTEM_TOOL_KINDS:
+                logger.warning(f"忽略 Agent {node_id} 的未知系统工具: {kind}")
+                continue
+            append_function(
+                self._workflow_system_tool(
+                    kind,
+                    node_id=node_id,
+                    state_variable_names=stage.state_variable_names,
+                )
+            )
         return self._require_agent_stage().node_config(
             node_id,
             functions=functions,
@@ -693,6 +736,159 @@ class WorkflowBrain(BaseBrain):
             ),
         )
 
+    def _workflow_system_tool(
+        self,
+        kind: str,
+        *,
+        node_id: str,
+        state_variable_names: tuple[str, ...],
+    ) -> FlowsFunctionSchema:
+        """Build one platform-owned tool scoped to the active Agent node."""
+        if kind == "update_state":
+            return self._workflow_update_state_tool(
+                node_id,
+                state_variable_names=state_variable_names,
+            )
+        if kind == "skip_turn":
+            return self._workflow_skip_turn_tool()
+        if kind == "request_human_handoff":
+            return self._workflow_handoff_tool(node_id)
+        if kind == "end_conversation":
+            return self._workflow_end_conversation_tool()
+        raise ValueError(f"未知系统工具: {kind}")
+
+    def _workflow_update_state_tool(
+        self,
+        node_id: str,
+        *,
+        state_variable_names: tuple[str, ...],
+    ) -> FlowsFunctionSchema:
+        allowed = frozenset(state_variable_names)
+
+        async def handler(args, _flow_manager):
+            values = dict(args or {})
+            unauthorized = sorted(set(values) - allowed)
+            if unauthorized:
+                return {
+                    "status": "error",
+                    "message": "状态变量未获当前节点授权: " + ",".join(unauthorized),
+                }
+            try:
+                changed = self._store.assign_declared_many(values)
+            except DynamicVariableError as exc:
+                return {"status": "error", "message": f"状态更新失败: {exc}"}
+            if changed:
+                await self._emit_variables(
+                    reason="update_state",
+                    node_id=node_id,
+                    changed=changed,
+                )
+                await self._refresh_agent_prompt(node_id)
+            return {
+                "status": "success",
+                "changed": changed,
+                "variables": self._store.public_values(),
+            }
+
+        return FlowsFunctionSchema(
+            name="update_state",
+            description=(
+                "静默更新当前阶段明确授权的动态变量。"
+                "只提交本轮获得或确认的信息，更新后继续当前回答。"
+            ),
+            properties=state_update_properties(
+                self._cfg.dynamic_variable_definitions if self._cfg else {},
+                allowed_names=state_variable_names,
+            ),
+            required=[],
+            handler=handler,
+        )
+
+    @staticmethod
+    def _workflow_skip_turn_tool() -> FlowsFunctionSchema:
+        async def handler(args, _flow_manager):
+            reason = str((args or {}).get("reason") or "").strip()
+            result = {"status": "success", "action": "skip_turn"}
+            if reason:
+                result["reason"] = reason
+            return result
+
+        setattr(handler, "_suppress_followup_llm", True)
+        return FlowsFunctionSchema(
+            name="skip_turn",
+            description=(
+                "跳过当前轮次，不生成任何语音回复。"
+                "仅当用户明确要求稍等、话还没说完，或输入可确认只是噪音时调用。"
+            ),
+            properties={
+                "reason": {
+                    "type": "string",
+                    "description": "跳过本轮的原因（可选）。",
+                }
+            },
+            required=[],
+            handler=handler,
+        )
+
+    def _workflow_handoff_tool(self, node_id: str) -> FlowsFunctionSchema:
+        async def handler(args, _flow_manager):
+            reason = str((args or {}).get("reason") or "human_handoff").strip()
+            await self._require_runtime().queue_frame(
+                OutputTransportMessageUrgentFrame(
+                    message={
+                        "type": "handoff-requested",
+                        "source": "workflow-system-tool",
+                        "nodeId": node_id,
+                        "reason": reason,
+                        "message": "用户请求转接人工服务。",
+                    }
+                )
+            )
+            self._store.values["system__handoff_status"] = "requested"
+            return {
+                "status": "requested",
+                "action": "human_handoff_requested",
+                "message": "人工接管请求已提交，请告知用户正在等待人工响应。",
+            }
+
+        return FlowsFunctionSchema(
+            name="request_human_handoff",
+            description=(
+                "提交人工接管请求。当用户明确要求人工服务、投诉升级或 AI 无法"
+                "解决时调用。该工具只提交请求，不代表人工已经接通；调用后继续"
+                "回复用户并说明正在等待人工响应。"
+            ),
+            properties={
+                "reason": {"type": "string", "description": "转接人工的原因。"}
+            },
+            required=[],
+            handler=handler,
+        )
+
+    def _workflow_end_conversation_tool(self) -> FlowsFunctionSchema:
+        async def handler(args, _flow_manager):
+            reason = str((args or {}).get("reason") or "end_conversation").strip()
+            self._waiting_for_generated_end_speech = True
+            self._require_runtime().call_end.begin(reason)
+            return {"status": "success", "action": "ending_call"}
+
+        setattr(handler, "_suppress_followup_llm", True)
+        return FlowsFunctionSchema(
+            name="end_conversation",
+            description=(
+                "礼貌地结束本次对话。当用户明确告别、表示任务已完成"
+                "或要求挂断时调用。"
+            ),
+            properties={
+                "reason": {
+                    "type": "string",
+                    "description": "结束对话的简短原因。",
+                }
+            },
+            required=[],
+            handler=handler,
+        )
+
     def _flow_managed_transition_config(
         self,
         node_config: NodeConfig,
@@ -838,6 +1034,8 @@ class WorkflowBrain(BaseBrain):
                 outcome = await self._enter_action(node_id)
                 if not outcome.should_route:
                     return self._passive_node_config(node_id, context_messages)
+            elif node_type == "update_state":
+                await self._enter_update_state(node_id)
             elif node_type == "message":
                 self._prepare_message_continuation(
                     node_id,
@@ -862,6 +1060,22 @@ class WorkflowBrain(BaseBrain):
             node_id = str(edge.get("target") or "")
         raise RuntimeError("工作流连续自动跳转超过安全上限")
 
+    async def _enter_update_state(self, node_id: str) -> None:
+        """Apply one deterministic, atomic dynamic-variable update."""
+        self._state.enter(node_id, WorkflowStatus.RUNNING_ACTION)
+        await self._emit_node_active(node_id)
+        raw_assignments = self._engine.data(node_id).get("assignments") or {}
+        rendered = self._store.render_data(raw_assignments)
+        if not isinstance(rendered, dict):
+            raise DynamicVariableError("Update State 节点赋值必须是对象")
+        changed = self._store.assign_declared_many(rendered)
+        if changed:
+            await self._emit_variables(
+                reason="update_state",
+                node_id=node_id,
+                changed=changed,
+            )
+
     def _prepare_message_continuation(
         self,
         node_id: str,

backend/services/config_resolver.py

@@ -266,6 +266,7 @@ async def resolve_runtime_config(
         enableInterrupt=assistant.enable_interrupt,
         turnConfig=assistant.turn_config or {},
         startup=assistant.startup or {},
+        system_tools=assistant.system_tools or [],
         tools=runtime_tools,
         llm_tool_ids=llm_tool_ids,
         knowledge_base_id=assistant.knowledge_base_id,

backend/services/node_specs.py

@@ -11,15 +11,30 @@ from services.message_policy import (
     MESSAGE_CONFIRMATION,
     MESSAGE_PLAYBACK,
 )
+from services.system_tools import SYSTEM_TOOL_KINDS, normalize_system_tools
 
 
 SPEC_VERSION = "3"
-NODE_TYPES = {"start", "agent", "message", "action", "handoff", "end"}
+NODE_TYPES = {
+    "start",
+    "agent",
+    "message",
+    "action",
+    "update_state",
+    "handoff",
+    "end",
+}
 EDGE_MODES = {"llm", "expression", "always"}
 AGENT_ENTRY_MODES = {"wait_user", "generate"}
 ACTION_RESULT_ASSIGNMENT_MODES = {"inherit", "override", "none"}
 ACTION_USER_INPUT_POLICIES = {"queue", "block"}
-AUTOMATIC_NODE_TYPES = {"start", "message", "action", "handoff"}
+AUTOMATIC_NODE_TYPES = {
+    "start",
+    "message",
+    "action",
+    "update_state",
+    "handoff",
+}
 EXPRESSION_OPERATORS = {
     "eq",
     "neq",
@@ -98,6 +113,24 @@ NODE_SPECS: list[dict[str, Any]] = [
             {"key": "name", "label": "节点名称", "type": "text", "default": "Action"},
         ],
     },
+    {
+        "name": "update_state",
+        "displayName": "Update State",
+        "category": "execution_node",
+        "description": "原子更新已声明的动态变量，然后使用新状态继续路由。",
+        "icon": "Braces",
+        "accent": "mint",
+        "addable": True,
+        "constraints": {"minIncoming": 1, "minOutgoing": 0},
+        "fields": [
+            {
+                "key": "name",
+                "label": "节点名称",
+                "type": "text",
+                "default": "Update State",
+            },
+        ],
+    },
     {
         "name": "handoff",
         "displayName": "Handoff",
@@ -160,6 +193,11 @@ def _normalize_agent_data(data: dict[str, Any]) -> None:
     if data.get("entryMode") not in AGENT_ENTRY_MODES:
         data["entryMode"] = "wait_user"
     data.pop("entrySpeech", None)
+    data["systemTools"] = list(normalize_system_tools(data.get("systemTools")))
+    state_names = data.get("stateVariableNames")
+    data["stateVariableNames"] = list(
+        dict.fromkeys(str(name) for name in state_names or [] if str(name))
+    )
     if "inheritGlobalConfig" not in data:
         has_node_overrides = any(
             (
@@ -194,6 +232,12 @@ def _normalize_action_data(data: dict[str, Any]) -> None:
     data.pop("speech", None)
 
 
+def _normalize_update_state_data(data: dict[str, Any]) -> None:
+    """Keep deterministic state updates as one explicit assignment object."""
+    assignments = data.get("assignments")
+    data["assignments"] = dict(assignments) if isinstance(assignments, dict) else {}
+
+
 def _normalize_message_data(data: dict[str, Any]) -> None:
     """Fill the small built-in Message contract used by runtime and editor."""
     data.setdefault("speech", "")
@@ -246,6 +290,8 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]:
                 _normalize_message_data(data)
             elif node.get("type") == "action":
                 _normalize_action_data(data)
+            elif node.get("type") == "update_state":
+                _normalize_update_state_data(data)
         return source
 
     nodes = source.get("nodes") or []
@@ -262,6 +308,7 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]:
         "agent": "agent",
         "message": "message",
         "action": "action",
+        "update_state": "update_state",
         "handoff": "handoff",
         "end": "end",
     }
@@ -283,6 +330,8 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]:
             _normalize_message_data(data)
         elif new_type == "action":
             _normalize_action_data(data)
+        elif new_type == "update_state":
+            _normalize_update_state_data(data)
         elif new_type == "start":
             prompt = str(data.pop("prompt", "") or "").strip()
             data.pop("greeting", None)
@@ -390,6 +439,16 @@ def validate_graph(graph: dict[str, Any]) -> list[str]:
             entry_mode = data.get("entryMode", "wait_user")
             if entry_mode not in AGENT_ENTRY_MODES:
                 errors.append(f"Agent 节点 {node_id} 的进入模式无效:{entry_mode}")
+            system_tools = data.get("systemTools", [])
+            if not isinstance(system_tools, list) or any(
+                tool not in SYSTEM_TOOL_KINDS for tool in system_tools
+            ):
+                errors.append(f"Agent 节点 {node_id} 的系统工具配置无效")
+            state_names = data.get("stateVariableNames", [])
+            if not isinstance(state_names, list) or any(
+                not isinstance(name, str) for name in state_names
+            ):
+                errors.append(f"Agent 节点 {node_id} 的状态变量授权必须是列表")
         elif node_type == "message":
             data = node.get("data") or {}
             speech = data.get("speech")
@@ -447,6 +506,13 @@ def validate_graph(graph: dict[str, Any]) -> list[str]:
                 errors.append(
                     f"Action 节点 {node_id} 的用户输入策略无效:{input_policy}"
                 )
+        elif node_type == "update_state":
+            data = node.get("data") or {}
+            assignments = data.get("assignments")
+            if not isinstance(assignments, dict) or not assignments:
+                errors.append(
+                    f"Update State 节点 {node_id} 必须配置至少一个变量赋值"
+                )
 
     if counts["start"] != 1:
         errors.append("工作流必须有且仅有一个 Start 节点")

backend/services/system_tools.py

@@ -0,0 +1,54 @@
+"""Shared contracts for platform-owned conversation system tools."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable, Mapping
+from typing import Any
+
+
+SYSTEM_TOOL_KINDS = frozenset(
+    {
+        "end_conversation",
+        "update_state",
+        "skip_turn",
+        "request_human_handoff",
+    }
+)
+
+
+def normalize_system_tools(values: Iterable[Any] | None) -> tuple[str, ...]:
+    """Return known tool names once each while preserving editor order."""
+    return tuple(
+        dict.fromkeys(
+            str(value)
+            for value in values or ()
+            if str(value) in SYSTEM_TOOL_KINDS
+        )
+    )
+
+
+def state_update_properties(
+    definitions: Mapping[str, Mapping[str, Any]] | None,
+    *,
+    allowed_names: Iterable[str] | None = None,
+) -> dict[str, dict[str, Any]]:
+    """Build an explicit LLM schema for writable declared variables."""
+    definitions = definitions or {}
+    names = (
+        list(dict.fromkeys(str(name) for name in allowed_names))
+        if allowed_names is not None
+        else list(definitions)
+    )
+    properties: dict[str, dict[str, Any]] = {}
+    for name in names:
+        definition = definitions.get(name)
+        if not isinstance(definition, Mapping):
+            continue
+        variable_type = str(definition.get("type") or "string")
+        if variable_type not in {"string", "number", "boolean"}:
+            continue
+        properties[name] = {
+            "type": variable_type,
+            "description": f"更新动态变量 {name}。",
+        }
+    return properties

backend/services/workflow_engine.py

@@ -11,6 +11,7 @@ from typing import Any
 
 from services.node_specs import normalize_graph
 from services.runtime_variables import DynamicVariableStore
+from services.system_tools import normalize_system_tools
 
 
 @dataclass(frozen=True)
@@ -24,6 +25,8 @@ class AgentStageConfig:
     vision_enabled: bool
     vision_model_resource_id: str | None
     tool_ids: tuple[str, ...]
+    system_tools: tuple[str, ...]
+    state_variable_names: tuple[str, ...]
     knowledge_base_id: str | None
     knowledge_mode: str
     knowledge_top_n: int
@@ -145,6 +148,16 @@ class WorkflowEngine:
                 str(source.get("visionModelResourceId") or "") or None
             ),
             tool_ids=tuple(str(tool_id) for tool_id in source.get("toolIds") or []),
+            # System tools and their writable state scope always belong to the
+            # Agent node. They are permissions, not inheritable model config.
+            system_tools=normalize_system_tools(data.get("systemTools")),
+            state_variable_names=tuple(
+                dict.fromkeys(
+                    str(name)
+                    for name in data.get("stateVariableNames") or []
+                    if str(name)
+                )
+            ),
             knowledge_base_id=knowledge_base_id or None,
             knowledge_mode=(
                 str(source.get("knowledgeMode") or "automatic")

backend/tests/test_brains.py

@@ -154,6 +154,46 @@ class BrainRegistryTests(unittest.TestCase):
         )
         self.assertIn("user_name", assistant.dynamic_variable_definitions)
 
+    def test_system_tools_are_prompt_pipeline_only(self):
+        assistant = AssistantUpsert(
+            name="prompt",
+            type="prompt",
+            systemTools=["end_conversation", "skip_turn", "end_conversation"],
+        )
+        self.assertEqual(assistant.system_tools, ["end_conversation", "skip_turn"])
+
+        workflow = AssistantUpsert(
+            name="workflow",
+            type="workflow",
+            systemTools=["update_state"],
+            graph={},
+        )
+        self.assertEqual(workflow.system_tools, [])
+
+        with self.assertRaises(ValueError):
+            AssistantUpsert(
+                name="realtime prompt",
+                type="prompt",
+                runtimeMode="realtime",
+                systemTools=["skip_turn"],
+            )
+
+    def test_system_tools_reject_unknown_kind(self):
+        with self.assertRaises(ValueError):
+            AssistantUpsert(
+                name="prompt",
+                type="prompt",
+                systemTools=["update_state", "magic"],
+            )
+
+    def test_prompt_update_state_requires_a_declared_variable(self):
+        with self.assertRaisesRegex(ValueError, "必须声明至少一个动态变量"):
+            AssistantUpsert(
+                name="prompt",
+                type="prompt",
+                systemTools=["update_state"],
+            )
+
     def test_workflow_keeps_dynamic_variables_and_tool_bindings(self):
         assistant = AssistantUpsert(
             name="workflow",
@@ -757,7 +797,7 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(prompts, ["面板状态：true"])
         self.assertEqual(queued_frames, [])
 
-    async def test_end_call_tool_is_owned_by_prompt_brain(self):
+    async def test_end_conversation_system_tool_is_owned_by_prompt_brain(self):
         brain = build_brain(
             AssistantConfig(
                 type="prompt",
@@ -766,9 +806,10 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase):
                         id="end-call",
                         name="结束通话",
                         function_name="end_call",
-                        type="end_call",
+                        type="system",
                         definition={
                             "config": {
+                                "kind": "end_conversation",
                                 "message_type": "none",
                                 "capture_reason": True,
                             }
@@ -792,8 +833,13 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase):
                         id="end-call",
                         name="结束通话",
                         function_name="end_call",
-                        type="end_call",
-                        definition={"config": {"capture_reason": True}},
+                        type="system",
+                        definition={
+                            "config": {
+                                "kind": "end_conversation",
+                                "capture_reason": True,
+                            }
+                        },
                     )
                 ],
             ),
@@ -819,13 +865,18 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase):
         self.assertTrue(call_end.response_started)
         self.assertEqual(params.result["action"], "ending_call")
 
-    async def test_end_call_waits_for_prompt_generated_closing_speech(self):
+    async def test_system_end_conversation_waits_for_generated_closing_speech(self):
         tool = RuntimeTool(
             id="end-call",
             name="结束通话",
             function_name="end_call",
-            type="end_call",
-            definition={"config": {"message_type": "none"}},
+            type="system",
+            definition={
+                "config": {
+                    "kind": "end_conversation",
+                    "message_type": "none",
+                }
+            },
         )
         cfg = AssistantConfig(type="prompt", tools=[tool])
         brain = build_brain(cfg)
@@ -855,6 +906,279 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase):
         self.assertTrue(call_end.armed)
         self.assertTrue(call_end.waited_for_text)
 
+    async def test_system_tools_register_with_fixed_names(self):
+        cfg = AssistantConfig(
+            type="prompt",
+            dynamic_variable_definitions={
+                "user_name": {
+                    "type": "string",
+                    "required": False,
+                    "default": None,
+                }
+            },
+            system_tools=[
+                "end_conversation",
+                "update_state",
+                "skip_turn",
+                "request_human_handoff",
+            ],
+        )
+        brain = build_brain(cfg)
+        llm = FakeLLM()
+        visible_tools = []
+
+        await brain.setup(
+            cfg,
+            BrainRuntime(
+                context=LLMContext(messages=[]),
+                llm=llm,
+                queue_frame=lambda _frame: None,
+                set_system_prompt=lambda _prompt: None,
+                set_tools=lambda tools: visible_tools.extend(tools or []),
+                call_end=FakeCallEnd(),
+            ),
+        )
+
+        self.assertEqual(
+            [tool.name for tool in visible_tools],
+            [
+                "end_conversation",
+                "update_state",
+                "skip_turn",
+                "request_human_handoff",
+            ],
+        )
+        update_schema = next(
+            tool for tool in visible_tools if tool.name == "update_state"
+        )
+        self.assertEqual(
+            update_schema.properties,
+            {
+                "user_name": {
+                    "type": "string",
+                    "description": "更新动态变量 user_name。",
+                }
+            },
+        )
+        for name in (
+            "end_conversation",
+            "update_state",
+            "skip_turn",
+            "request_human_handoff",
+        ):
+            self.assertIn(name, llm.functions)
+
+    async def test_end_conversation_ends_call_after_generated_speech(self):
+        cfg = AssistantConfig(
+            type="prompt",
+            system_tools=["end_conversation"],
+        )
+        brain = build_brain(cfg)
+        llm = FakeLLM()
+        call_end = FakeCallEnd()
+
+        await brain.setup(
+            cfg,
+            BrainRuntime(
+                context=LLMContext(messages=[]),
+                llm=llm,
+                queue_frame=lambda _frame: None,
+                set_system_prompt=lambda _prompt: None,
+                set_tools=lambda _tools: None,
+                call_end=call_end,
+            ),
+        )
+
+        await brain.on_assistant_text_start("turn-1")
+        params = FakeFunctionParams({"reason": "用户已完成咨询"})
+        await llm.functions["end_conversation"](params)
+        self.assertEqual(call_end.reason, "用户已完成咨询")
+        self.assertTrue(call_end.ending)
+        self.assertFalse(call_end.finished)
+        self.assertEqual(params.result["action"], "ending_call")
+        self.assertFalse(params.properties.run_llm)
+
+        await brain.on_assistant_text_end(
+            "turn-1",
+            "好的，再见。",
+            False,
+        )
+        self.assertFalse(call_end.finished)
+        self.assertTrue(call_end.armed)
+
+    async def test_end_conversation_finishes_when_no_speech(self):
+        cfg = AssistantConfig(
+            type="prompt",
+            system_tools=["end_conversation"],
+        )
+        brain = build_brain(cfg)
+        llm = FakeLLM()
+        call_end = FakeCallEnd()
+
+        await brain.setup(
+            cfg,
+            BrainRuntime(
+                context=LLMContext(messages=[]),
+                llm=llm,
+                queue_frame=lambda _frame: None,
+                set_system_prompt=lambda _prompt: None,
+                set_tools=lambda _tools: None,
+                call_end=call_end,
+            ),
+        )
+
+        await brain.on_assistant_text_start("turn-1")
+        await llm.functions["end_conversation"](FakeFunctionParams({}))
+        await brain.on_assistant_text_end("turn-1", "", False)
+        self.assertTrue(call_end.finished)
+
+    async def test_update_state_updates_declared_variables(self):
+        cfg = AssistantConfig(
+            type="prompt",
+            prompt="当前用户名: {{user_name}}",
+            dynamic_variable_definitions={
+                "user_name": {"type": "string", "required": False, "default": None}
+            },
+            system_tools=["update_state"],
+        )
+        brain = build_brain(cfg)
+        llm = FakeLLM()
+        prompts = []
+        queued_frames = []
+
+        async def collect(frame):
+            queued_frames.append(frame)
+
+        await brain.setup(
+            cfg,
+            BrainRuntime(
+                context=LLMContext(messages=[]),
+                llm=llm,
+                queue_frame=collect,
+                set_system_prompt=prompts.append,
+                set_tools=lambda _tools: None,
+                call_end=FakeCallEnd(),
+            ),
+        )
+
+        params = FakeFunctionParams({"user_name": "王小明"})
+        await llm.functions["update_state"](params)
+
+        self.assertEqual(params.result["status"], "success")
+        self.assertEqual(params.result["changed"], ["user_name"])
+        self.assertIsNone(params.properties)
+        self.assertEqual(
+            [frame.message for frame in queued_frames],
+            [
+                {
+                    "type": "session-variables",
+                    "reason": "update_state",
+                    "variables": {"user_name": "王小明"},
+                    "changed": ["user_name"],
+                }
+            ],
+        )
+        self.assertTrue(prompts[-1].endswith("当前用户名: 王小明"))
+
+    async def test_update_state_rejects_undeclared_variable(self):
+        cfg = AssistantConfig(
+            type="prompt",
+            system_tools=["update_state"],
+        )
+        brain = build_brain(cfg)
+        llm = FakeLLM()
+
+        await brain.setup(
+            cfg,
+            BrainRuntime(
+                context=LLMContext(messages=[]),
+                llm=llm,
+                queue_frame=lambda _frame: None,
+                set_system_prompt=lambda _prompt: None,
+                set_tools=lambda _tools: None,
+                call_end=FakeCallEnd(),
+            ),
+        )
+
+        params = FakeFunctionParams({"unknown_key": "x"})
+        await llm.functions["update_state"](params)
+        self.assertEqual(params.result["status"], "error")
+        self.assertIn("未声明", params.result["message"])
+
+    async def test_skip_turn_suppresses_response(self):
+        cfg = AssistantConfig(
+            type="prompt",
+            system_tools=["skip_turn"],
+        )
+        brain = build_brain(cfg)
+        llm = FakeLLM()
+
+        await brain.setup(
+            cfg,
+            BrainRuntime(
+                context=LLMContext(messages=[]),
+                llm=llm,
+                queue_frame=lambda _frame: None,
+                set_system_prompt=lambda _prompt: None,
+                set_tools=lambda _tools: None,
+                call_end=FakeCallEnd(),
+            ),
+        )
+
+        params = FakeFunctionParams({"reason": "无需回复"})
+        await llm.functions["skip_turn"](params)
+        self.assertEqual(params.result["action"], "skip_turn")
+        self.assertEqual(params.result["reason"], "无需回复")
+        self.assertFalse(params.properties.run_llm)
+
+    async def test_request_human_handoff_keeps_call_available(self):
+        cfg = AssistantConfig(
+            type="prompt",
+            system_tools=["request_human_handoff"],
+        )
+        brain = build_brain(cfg)
+        llm = FakeLLM()
+        call_end = FakeCallEnd()
+        queued_frames = []
+
+        async def collect(frame):
+            queued_frames.append(frame)
+
+        await brain.setup(
+            cfg,
+            BrainRuntime(
+                context=LLMContext(messages=[]),
+                llm=llm,
+                queue_frame=collect,
+                set_system_prompt=lambda _prompt: None,
+                set_tools=lambda _tools: None,
+                call_end=call_end,
+            ),
+        )
+
+        await brain.on_assistant_text_start("turn-1")
+        params = FakeFunctionParams({"reason": "用户要求人工客服"})
+        await llm.functions["request_human_handoff"](params)
+
+        self.assertEqual(
+            [frame.message for frame in queued_frames],
+            [
+                {
+                    "type": "handoff-requested",
+                    "source": "prompt-system-tool",
+                    "reason": "用户要求人工客服",
+                    "message": "用户请求转接人工服务。",
+                }
+            ],
+        )
+        self.assertEqual(call_end.reason, "")
+        self.assertFalse(call_end.ending)
+        self.assertEqual(params.result["action"], "human_handoff_requested")
+        self.assertIsNone(params.properties)
+
+        await brain.on_assistant_text_end("turn-1", "", False)
+        self.assertFalse(call_end.finished)
+
     async def test_http_tool_renders_secrets_and_updates_prompt_variable(self):
         requests = []
 
@@ -962,6 +1286,195 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase):
 
 
 class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
+    async def test_agent_system_tools_are_scoped_and_update_declared_state(self):
+        cfg = prepare_dynamic_config(
+            AssistantConfig(
+                type="workflow",
+                graph={
+                    "specVersion": 3,
+                    "settings": {},
+                    "nodes": [
+                        {"id": "start", "type": "start", "data": {}},
+                        {
+                            "id": "agent",
+                            "type": "agent",
+                            "data": {
+                                "systemTools": [
+                                    "update_state",
+                                    "skip_turn",
+                                    "request_human_handoff",
+                                ],
+                                "stateVariableNames": ["customer_name"],
+                            },
+                        },
+                    ],
+                    "edges": [],
+                },
+                dynamic_variable_definitions={
+                    "customer_name": {
+                        "type": "string",
+                        "required": False,
+                        "default": None,
+                    },
+                    "internal_note": {
+                        "type": "string",
+                        "required": False,
+                        "default": None,
+                    },
+                },
+            ),
+            {},
+            assistant_id="asst_workflow_system_tools",
+        )
+        brain = WorkflowBrain(cfg)
+        queued = []
+
+        async def queue_frame(frame):
+            queued.append(frame)
+
+        brain._runtime = BrainRuntime(
+            context=LLMContext(messages=[]),
+            llm=FakeLLM(),
+            queue_frame=queue_frame,
+            set_system_prompt=lambda _prompt: None,
+            set_tools=lambda _tools: None,
+            call_end=FakeCallEnd(),
+        )
+        brain._agent_stage = SimpleNamespace(
+            node_config=lambda node_id, *, functions, leading_messages: {
+                "name": node_id,
+                "functions": functions,
+                "leading_messages": leading_messages,
+            }
+        )
+        brain._emit_variables = AsyncMock()
+        brain._refresh_agent_prompt = AsyncMock()
+
+        node_config = brain._agent_config("agent")
+        functions = {tool.name: tool for tool in node_config["functions"]}
+        self.assertEqual(
+            list(functions),
+            ["update_state", "skip_turn", "request_human_handoff"],
+        )
+        self.assertEqual(
+            list(functions["update_state"].properties),
+            ["customer_name"],
+        )
+
+        result = await functions["update_state"].handler(
+            {"customer_name": "李白"},
+            None,
+        )
+        self.assertEqual(result["changed"], ["customer_name"])
+        self.assertEqual(brain._store.public_values(), {"customer_name": "李白"})
+        brain._emit_variables.assert_awaited_once_with(
+            reason="update_state",
+            node_id="agent",
+            changed=["customer_name"],
+        )
+        brain._refresh_agent_prompt.assert_awaited_once_with("agent")
+
+        unauthorized = await functions["update_state"].handler(
+            {"internal_note": "不可写"},
+            None,
+        )
+        self.assertEqual(unauthorized["status"], "error")
+        self.assertIn("未获当前节点授权", unauthorized["message"])
+
+        handoff = await functions["request_human_handoff"].handler(
+            {"reason": "用户要求人工"},
+            None,
+        )
+        self.assertEqual(handoff["status"], "requested")
+        self.assertFalse(brain._runtime.call_end.ending)
+        event = next(
+            frame.message
+            for frame in queued
+            if isinstance(frame, OutputTransportMessageUrgentFrame)
+        )
+        self.assertEqual(event["type"], "handoff-requested")
+        self.assertEqual(event["source"], "workflow-system-tool")
+
+    async def test_update_state_node_applies_assignments_atomically(self):
+        cfg = prepare_dynamic_config(
+            AssistantConfig(
+                type="workflow",
+                graph={
+                    "specVersion": 3,
+                    "settings": {},
+                    "nodes": [
+                        {"id": "start", "type": "start", "data": {}},
+                        {
+                            "id": "set_state",
+                            "type": "update_state",
+                            "data": {
+                                "assignments": {
+                                    "confirmed": True,
+                                    "display_name": "{{source_name}}",
+                                }
+                            },
+                        },
+                    ],
+                    "edges": [],
+                },
+                dynamic_variable_definitions={
+                    "source_name": {"type": "string", "default": "李白"},
+                    "display_name": {
+                        "type": "string",
+                        "required": False,
+                        "default": None,
+                    },
+                    "confirmed": {"type": "boolean", "default": False},
+                },
+            ),
+            {},
+            assistant_id="asst_workflow_update_state_node",
+        )
+        brain = WorkflowBrain(cfg)
+        brain._emit_node_active = AsyncMock()
+        brain._emit_variables = AsyncMock()
+
+        await brain._enter_update_state("set_state")
+
+        self.assertEqual(brain._store.public_values()["display_name"], "李白")
+        self.assertTrue(brain._store.public_values()["confirmed"])
+        brain._emit_variables.assert_awaited_once_with(
+            reason="update_state",
+            node_id="set_state",
+            changed=["confirmed", "display_name"],
+        )
+
+    async def test_workflow_end_conversation_waits_for_generated_speech(self):
+        brain = WorkflowBrain(
+            {
+                "specVersion": 3,
+                "settings": {},
+                "nodes": [{"id": "start", "type": "start", "data": {}}],
+                "edges": [],
+            }
+        )
+        call_end = FakeCallEnd()
+        brain._runtime = BrainRuntime(
+            context=LLMContext(messages=[]),
+            llm=FakeLLM(),
+            queue_frame=noop_queue_frame,
+            set_system_prompt=lambda _prompt: None,
+            set_tools=lambda _tools: None,
+            call_end=call_end,
+        )
+        tool = brain._workflow_end_conversation_tool()
+
+        await brain.on_assistant_text_start("turn-1")
+        result = await tool.handler({"reason": "用户告别"}, None)
+        self.assertEqual(result["action"], "ending_call")
+        self.assertTrue(call_end.ending)
+        self.assertEqual(call_end.reason, "用户告别")
+        self.assertTrue(getattr(tool.handler, "_suppress_followup_llm"))
+
+        await brain.on_assistant_text_end("turn-1", "再见。", False)
+        self.assertTrue(call_end.armed)
+        self.assertTrue(call_end.waited_for_text)
+
     async def test_flow_manager_dispatches_native_vision_without_auxiliary_handler(self):
         manager = object.__new__(ConfiguredFlowManager)
         fallback_transition = AsyncMock()

backend/tests/test_workflow_v3.py

@@ -360,6 +360,13 @@ class WorkflowGraphTests(unittest.TestCase):
 
     def test_agent_effective_config_inherits_then_switches_to_override(self):
         graph = valid_graph()
+        agent = next(node for node in graph["nodes"] if node["id"] == "agent")
+        agent["data"].update(
+            {
+                "systemTools": ["update_state", "skip_turn"],
+                "stateVariableNames": ["customer", "order_status"],
+            }
+        )
         graph["settings"].update(
             {
                 "defaultLlmResourceId": "llm_global",
@@ -389,6 +396,11 @@ class WorkflowGraphTests(unittest.TestCase):
         )
         self.assertTrue(engine.uses_vision())
         self.assertEqual(inherited.tool_ids, ("tool_global",))
+        self.assertEqual(inherited.system_tools, ("update_state", "skip_turn"))
+        self.assertEqual(
+            inherited.state_variable_names,
+            ("customer", "order_status"),
+        )
         self.assertEqual(inherited.knowledge_mode, "on_demand")
         self.assertFalse(inherited.enable_interrupt)
         self.assertEqual(
@@ -417,6 +429,7 @@ class WorkflowGraphTests(unittest.TestCase):
         self.assertIsNone(custom.vision_model_resource_id)
         self.assertFalse(engine.uses_vision())
         self.assertEqual(custom.tool_ids, ("tool_agent",))
+        self.assertEqual(custom.system_tools, ("update_state", "skip_turn"))
         self.assertEqual(custom.knowledge_mode, "disabled")
         self.assertTrue(custom.enable_interrupt)
         self.assertEqual(
@@ -424,6 +437,84 @@ class WorkflowGraphTests(unittest.TestCase):
             "smart_turn",
         )
 
+    def test_update_state_node_and_agent_state_scope_are_validated(self):
+        graph = valid_graph()
+        graph["nodes"].insert(
+            1,
+            {
+                "id": "set_state",
+                "type": "update_state",
+                "data": {
+                    "name": "记录状态",
+                    "assignments": {"confirmed": True},
+                },
+            },
+        )
+        graph["edges"][0]["target"] = "set_state"
+        graph["edges"].insert(
+            1,
+            {
+                "id": "after_state",
+                "source": "set_state",
+                "target": "agent",
+                "data": {"mode": "always", "priority": 0},
+            },
+        )
+        agent = next(node for node in graph["nodes"] if node["id"] == "agent")
+        agent["data"].update(
+            {
+                "systemTools": ["update_state"],
+                "stateVariableNames": ["customer"],
+            }
+        )
+        body = AssistantUpsert(
+            name="状态工作流",
+            type="workflow",
+            dynamicVariableDefinitions={
+                "confirmed": {
+                    "type": "boolean",
+                    "required": False,
+                    "default": False,
+                },
+                "customer": {
+                    "type": "string",
+                    "required": False,
+                    "default": None,
+                },
+            },
+            graph=graph,
+        )
+
+        _validate_workflow(body)
+        normalized_state = next(
+            node for node in body.graph["nodes"] if node["type"] == "update_state"
+        )
+        self.assertEqual(normalized_state["data"]["assignments"], {"confirmed": True})
+
+        normalized_state["data"]["assignments"] = {"missing": "x"}
+        with self.assertRaisesRegex(HTTPException, "未声明变量:missing"):
+            _validate_workflow(body)
+
+    def test_agent_update_state_tool_requires_an_authorized_variable(self):
+        graph = valid_graph()
+        agent = next(node for node in graph["nodes"] if node["id"] == "agent")
+        agent["data"]["systemTools"] = ["update_state"]
+        body = AssistantUpsert(
+            name="缺少授权",
+            type="workflow",
+            dynamicVariableDefinitions={
+                "customer": {
+                    "type": "string",
+                    "required": False,
+                    "default": None,
+                }
+            },
+            graph=graph,
+        )
+
+        with self.assertRaisesRegex(HTTPException, "必须授权至少一个变量"):
+            _validate_workflow(body)
+
     def test_vision_resource_creates_isolated_runtime_config(self):
         base = AssistantConfig(type="workflow", model="text-only")
         resource = RuntimeModelResource(