fix(subagent): refuse the disabled backend tools instead of only hiding them

.github/workflows/system-tests.yml

@@ -156,7 +156,8 @@ jobs:
       - name: Verify Run result causality
         run: docker compose exec -T api uv run --no-sync --no-dev pytest test/integration/api/test_agent_run_result_causality.py -q
       - name: Verify Message audit HTTP contract
-        run: docker compose exec -T -e TEST_USERNAME="$E2E_USERNAME" -e TEST_PASSWORD="$E2E_PASSWORD" api uv run --no-sync --no-dev pytest test/integration/api/test_chat_router.py::test_thread_message_audits_return_persisted_facts_without_leaking_into_history -q
+        timeout-minutes: 3
+        run: docker compose exec -T -e TEST_USERNAME="$E2E_USERNAME" -e TEST_PASSWORD="$E2E_PASSWORD" api uv run --no-sync --no-dev pytest test/integration/api/test_chat_router.py::test_thread_message_audits_return_persisted_facts_without_leaking_into_history -q --setup-show -o faulthandler_timeout=60
       - name: Verify deterministic Agent assembled path
         run: docker compose exec -T -e E2E_USERNAME -e E2E_PASSWORD api uv run --no-sync --no-dev pytest test/e2e/test_deterministic_agent_path_e2e.py -q
       - name: Verify identity transaction and replayable secret publication
@@ -212,7 +213,7 @@ jobs:
             uv run --no-sync --no-dev pytest \
               test/integration/api/test_task_router.py::test_enqueue_document_creates_task -q
       - name: Runtime logs on failure
-        if: failure()
+        if: failure() || cancelled()
         run: docker compose logs api worker sandbox-provisioner --tail 300
       - name: Stop runtime topology
         if: always()

backend/package/yuxi/agents/backends/composite.py

@@ -127,9 +127,11 @@ def create_agent_filesystem_middleware(
     tool_token_limit_before_evict: int | None = None,
     *,
     backend: CompositeBackend,
+    disabled_tools: frozenset[str] = frozenset(),
 ) -> FilesystemMiddleware:
+    """构造文件系统中间件，在 ToolNode 注册前排除禁用工具。"""
     return YuxiFilesystemMiddleware(
         backend=backend,
         tool_token_limit_before_evict=tool_token_limit_before_evict,
-        tools=list(_AGENT_FS_TOOLS),
+        tools=[name for name in _AGENT_FS_TOOLS if name not in disabled_tools],
     )

backend/package/yuxi/agents/buildin/subagent/graph.py

@@ -4,6 +4,7 @@
 from langchain.agents import create_agent
 from langchain.agents.middleware import ModelRetryMiddleware, TodoListMiddleware
 from langchain.agents.middleware.types import AgentMiddleware
+from langchain_core.messages import ToolMessage
 
 from yuxi.agents import BaseAgent, BaseState
 from yuxi.agents.backends import (
@@ -61,6 +62,29 @@ def wrap_model_call(self, request, handler):
     async def awrap_model_call(self, request, handler):
         return await handler(request.override(tools=_filter_disabled_tools(request.tools or [], self.disabled_tools)))
 
+    # 工具列表隐藏不构成执行边界；显式传入的禁用工具调用也必须拒绝。
+    def wrap_tool_call(self, request, handler):
+        denial = self._denied_tool_message(request)
+        return denial if denial is not None else handler(request)
+
+    async def awrap_tool_call(self, request, handler):
+        denial = self._denied_tool_message(request)
+        return denial if denial is not None else await handler(request)
+
+    def _denied_tool_message(self, request) -> ToolMessage | None:
+        """为禁用调用生成与原 tool call 绑定的拒绝结果。"""
+        name = _tool_name(request.tool_call)
+        if name not in self.disabled_tools:
+            return None
+        return ToolMessage(
+            content=(
+                f"工具 {name} 在当前审批模式下对子智能体不可用；请把结果交回主智能体，由主线程按审批流程执行该操作。"
+            ),
+            tool_call_id=request.tool_call.get("id") or "",
+            name=name,
+            status="error",
+        )
+
 
 async def _build_middlewares(context, backend, tool_approval_mode: str):
     # tool_approval_mode is normalized once by the caller (get_graph / SubAgentBackend.get_graph).
@@ -69,6 +93,7 @@ async def _build_middlewares(context, backend, tool_approval_mode: str):
         create_agent_filesystem_middleware(
             getattr(context, "tool_token_limit", DEFAULT_TOOL_RESULT_EVICTION_K_TOKENS) * 1024,
             backend=backend,
+            disabled_tools=_disabled_tools_for(tool_approval_mode),
         ),
         SkillsMiddleware(),
         create_summary_middleware_from_context(context, backend=backend),

backend/package/yuxi/services/agent_config_service.py

@@ -3,7 +3,6 @@
 from typing import Any
 
 from sqlalchemy.ext.asyncio import AsyncSession
-
 from yuxi.agents.context import BaseContext, filter_config_by_role, resolve_agent_resource_options
 from yuxi.repositories.agent_repository import AGENT_RESOURCE_CONFIG_FIELDS
 from yuxi.storage.postgres.models_business import User

backend/package/yuxi/services/chat_service.py

@@ -13,10 +13,10 @@
 """
 
 import asyncio
-from contextlib import aclosing
 import json
 import uuid
 from collections.abc import AsyncIterator, Awaitable, Callable
+from contextlib import aclosing
 from typing import Any, Literal
 
 from langchain.messages import AIMessage, AIMessageChunk, HumanMessage

backend/package/yuxi/services/conversation_service.py

@@ -2,10 +2,10 @@
 from typing import Any
 
 from fastapi import HTTPException
-from yuxi.models.utils import parse_assistant_message_body
 from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
+from yuxi.models.utils import parse_assistant_message_body
 from yuxi.repositories.agent_repository import AgentRepository
 from yuxi.repositories.agent_run_repository import AgentRunRepository
 from yuxi.repositories.conversation_repository import INVOCATION_CONVERSATION_SOURCES, ConversationRepository

backend/test/e2e/test_deterministic_agent_path_e2e.py

@@ -15,7 +15,7 @@
 from e2e_helpers import cancel_run, consume_events, delete_agent, postgres_dsn, wait_for_run
 from yuxi.agents.backends.sandbox import ProvisionerSandboxBackend, get_sandbox_provider
 from yuxi.config import get_skill_projection_dir
-from yuxi.workspace.paths import workspace_uid_dirname
+from yuxi.workspace.paths import user_workspace_dir, workspace_uid_dirname
 
 from test.live_api_cleanup import make_test_conversation_metadata, make_test_conversation_title
 
@@ -216,14 +216,17 @@ async def _create_agent(
     uid: str,
     *,
     system_prompt_suffix: str = "",
+    is_subagent: bool = False,
+    subagents: list[str] | None = None,
 ) -> str:
     slug = f"ci-deterministic-{uuid.uuid4().hex[:8]}"
     response = await client.post(
         "/api/agent",
         json={
             "name": f"Deterministic E2E {slug[-8:]}",
             "slug": slug,
-            "backend_id": "ChatbotAgent",
+            "backend_id": "SubAgentBackend" if is_subagent else "ChatbotAgent",
+            "is_subagent": is_subagent,
             "description": "无外部密钥的 assembled-path 测试智能体",
             "config_json": {
                 "context": {
@@ -234,7 +237,7 @@ async def _create_agent(
                     "mcps": [],
                     "skills": ["image-gen"],
                     "preload_skills": ["image-gen"],
-                    "subagents": [],
+                    "subagents": subagents or [],
                 }
             },
             "share_config": {
@@ -254,6 +257,124 @@ async def _create_agent(
     return slug
 
 
+@pytest.mark.parametrize("mode", ["default", "always_trust"])
+async def test_subagent_worker_enforces_inherited_write_policy(e2e_client, e2e_headers, mode):
+    """真实父子 Run 继承审批模式，回读工具审计与共享 Workdir 文件。"""
+    me = await e2e_client.get("/api/auth/me", headers=e2e_headers)
+    assert me.status_code == 200, me.text
+    uid = str(me.json()["uid"])
+    await _create_provider(e2e_client, e2e_headers)
+    agents = []
+    thread_id = child_thread_id = run_id = workdir_path = probe_path = None
+    try:
+        child_slug = await _create_agent(
+            e2e_client,
+            e2e_headers,
+            uid,
+            is_subagent=True,
+            system_prompt_suffix="DETERMINISTIC_SUBAGENT_CHILD",
+        )
+        agents.append(child_slug)
+        parent_slug = await _create_agent(
+            e2e_client,
+            e2e_headers,
+            uid,
+            subagents=[child_slug],
+            system_prompt_suffix=f"DETERMINISTIC_SUBAGENT_PARENT:{child_slug}",
+        )
+        agents.append(parent_slug)
+        response = await e2e_client.post(
+            "/api/chat/thread",
+            json={
+                "agent_id": parent_slug,
+                "title": make_test_conversation_title("subagent-policy"),
+                "metadata": make_test_conversation_metadata("subagent-policy", e2e=True),
+            },
+            headers=e2e_headers,
+        )
+        assert response.status_code == 200, response.text
+        thread_id = str(response.json()["id"])
+        workdir_path = str(response.json()["workdir_path"])
+        file_name = f"subagent-policy-{uuid.uuid4().hex}.txt"
+        path = f"/home/gem/user-data/{workdir_path}/{file_name}"
+        probe_path = user_workspace_dir(uid) / workdir_path / file_name
+        response = await e2e_client.post(
+            "/api/agent/runs",
+            json={
+                "agent_slug": parent_slug,
+                "thread_id": thread_id,
+                "query": f"{EXPECTED_OUTPUT} SUBAGENT_MODE:{mode} SUBAGENT_PATH:{path}",
+                "tool_approval_mode": mode,
+                "meta": {"request_id": f"subagent-policy-{uuid.uuid4()}"},
+            },
+            headers=e2e_headers,
+        )
+        assert response.status_code == 200, response.text
+        run_id = str(response.json()["run_id"])
+        parent = await wait_for_run(e2e_client, e2e_headers, run_id)
+        assert parent["status"] == "completed", parent
+
+        conn = await asyncpg.connect(postgres_dsn())
+        try:
+            children = await conn.fetch(
+                """
+                SELECT run.id, run.status, run.runtime_scope_id, run.input_payload,
+                       conversation.thread_id
+                FROM agent_runs run JOIN conversations conversation ON conversation.id = run.conversation_id
+                WHERE run.created_by_run_id = $1 AND run.run_type = 'subagent'
+                """,
+                run_id,
+            )
+            assert len(children) == 1, children
+            child = children[0]
+            child_thread_id = str(child["thread_id"])
+            assert child["status"] == "completed", dict(child)
+            assert child["runtime_scope_id"] == thread_id
+            payload = json.loads(child["input_payload"])
+            assert payload["tool_approval_mode"] == mode
+            audit = await conn.fetchrow(
+                """
+                SELECT execution_status, content FROM messages
+                WHERE run_id = $1 AND message_type = 'tool_audit' AND operation_id = 'call-subagent-write'
+                """,
+                child["id"],
+            )
+        finally:
+            await conn.close()
+
+        state = await e2e_client.get(
+            f"/api/chat/thread/{child_thread_id}/state", params={"include_messages": "true"}, headers=e2e_headers
+        )
+        assert state.status_code == 200, state.text
+        assert state.json()["subagent_run"]["run_id"] == child["id"]
+        results = [
+            message for message in state.json()["messages"] if message.get("tool_call_id") == "call-subagent-write"
+        ]
+        assert len(results) == 1, state.json()["messages"]
+        assert results[0]["status"] == ("error" if mode == "default" else "success")
+        assert probe_path.parent.is_dir(), probe_path
+        if mode == "default":
+            assert "不可用" in results[0]["content"]
+            assert not probe_path.exists(), "被拒绝的子智能体调用不能写入共享 Workdir"
+        else:
+            assert audit and audit["execution_status"] == "completed", audit
+            assert probe_path.read_text(encoding="utf-8") == "subagent write verified"
+    finally:
+        if run_id:
+            await cancel_run(e2e_client, e2e_headers, run_id)
+        if probe_path:
+            probe_path.unlink(missing_ok=True)
+        if thread_id:
+            get_sandbox_provider().release(thread_id, uid=uid, workdir_path=workdir_path)
+        for cleanup_thread_id in (child_thread_id, thread_id):
+            if cleanup_thread_id:
+                response = await e2e_client.delete(f"/api/chat/thread/{cleanup_thread_id}", headers=e2e_headers)
+                assert response.status_code in {200, 404}, response.text
+        for slug in reversed(agents):
+            await delete_agent(e2e_client, e2e_headers, slug)
+        await _delete_provider(e2e_client, e2e_headers)
+
+
 async def _assert_persisted_causality(run_id: str, request_id: str) -> None:
     conn = await asyncpg.connect(postgres_dsn())
     try:

backend/test/support/openai_replay_server.py

@@ -48,11 +48,24 @@ def _validate_request(authorization: str | None, request: dict) -> str | None:
         for item in tools or []
         if isinstance(item, dict) and isinstance(item.get("function"), dict)
     }
-    if EXPECTED_PRELOADED_TOOL not in tool_names:
+    subagent_child = "DETERMINISTIC_SUBAGENT_CHILD" in serialized_messages
+    subagent_parent = "DETERMINISTIC_SUBAGENT_PARENT:" in serialized_messages
+    if subagent_child:
+        trusted = "SUBAGENT_MODE:always_trust" in serialized_messages
+        if ("write_file" in tool_names) != trusted or "task" in tool_names:
+            return "subagent_tool_policy_mismatch"
+    elif EXPECTED_PRELOADED_TOOL not in tool_names:
         return "preloaded_tool_missing"
     if LARGE_TOOL_RESULT_MARKER in serialized_messages and "execute" not in tool_names:
         return "execute_tool_missing"
     tool_messages = [message for message in messages if isinstance(message, dict) and message.get("role") == "tool"]
+    if subagent_child or subagent_parent:
+        expected_call = "call-subagent-write" if subagent_child else "call-subagent-task"
+        if subagent_parent and "task" not in tool_names:
+            return "subagent_task_missing"
+        if tool_messages and not any(message.get("tool_call_id") == expected_call for message in tool_messages):
+            return "subagent_tool_result_missing"
+        return None
     if tool_messages and not any(
         (
             message.get("tool_call_id") == EXPECTED_TOOL_CALL_ID
@@ -102,7 +115,16 @@ def _stream_payloads(model: str, messages: list[dict]) -> list[dict]:
     large_result = LARGE_TOOL_RESULT_MARKER in serialized_messages
     tool_call_id = LARGE_TOOL_CALL_ID if large_result else EXPECTED_TOOL_CALL_ID
     tool_name = "execute" if large_result else EXPECTED_PRELOADED_TOOL
-    if large_result:
+    if "DETERMINISTIC_SUBAGENT_CHILD" in serialized_messages:
+        tool_call_id, tool_name = "call-subagent-write", "write_file"
+        path = re.search(r'SUBAGENT_PATH:(/[^\s"\\]+)', serialized_messages).group(1)
+        tool_arguments = json.dumps({"file_path": path, "content": "subagent write verified"})
+    elif "DETERMINISTIC_SUBAGENT_PARENT:" in serialized_messages:
+        tool_call_id, tool_name = "call-subagent-task", "task"
+        slug = re.search(r"DETERMINISTIC_SUBAGENT_PARENT:([\w-]+)", serialized_messages).group(1)
+        description = next(message["content"] for message in reversed(messages) if message.get("role") == "user")
+        tool_arguments = json.dumps({"subagent_slug": slug, "description": description})
+    elif large_result:
         tool_arguments = json.dumps({"command": "yes X | head -c 13000"})
     elif TOOL_ERROR_MARKER in serialized_messages:
         tool_arguments = "{}"