From 65dbd407a98657cabb4a8319521a7d4180b18e86 Mon Sep 17 00:00:00 2001 From: shiyu Date: Fri, 2 Oct 2026 21:58:04 +0800 Subject: [PATCH] fix: enforce agent permissions and persist tool approvals Validate MCP arguments and recursive operation scopes, persist approval batches, and recheck permissions for queued work. Handle remote metadata requests independently of proxy settings and update Ant Design component APIs. --- README.md | 9 + README_zh.md | 9 + domain/agent/api.py | 7 +- domain/agent/approvals.py | 87 +++ domain/agent/execution.py | 254 ++++++++ domain/agent/mcp.py | 92 ++- domain/agent/service.py | 611 +++++++----------- domain/agent/tools/base.py | 11 +- domain/agent/tools/processors.py | 5 +- domain/agent/tools/vfs.py | 47 +- domain/agent/tools/web_fetch.py | 16 +- domain/agent/types.py | 1 + domain/config/service.py | 4 +- domain/notices/service.py | 4 +- domain/permission/execution.py | 93 +++ domain/processors/service.py | 33 + domain/tasks/task_queue.py | 15 + domain/virtual_fs/file_ops.py | 9 + domain/virtual_fs/processing.py | 5 + domain/virtual_fs/transfer.py | 11 + models/database.py | 35 + tests/test_agent_boundaries.py | 488 ++++++++++++++ tests/test_config_version.py | 71 ++ tests/test_notices.py | 43 ++ web/src/api/agent.ts | 7 +- web/src/components/AiAgentWidget.tsx | 27 +- web/src/layout/SideNav.tsx | 2 +- web/src/pages/AdaptersPage.tsx | 2 +- web/src/pages/AuditLogsPage.tsx | 2 +- .../components/ContextMenu.tsx | 2 +- .../components/FileDetailModal.tsx | 2 +- .../components/Modals/UploadModal.tsx | 2 +- web/src/pages/LoginPage.tsx | 4 +- web/src/pages/ProcessorsPage.tsx | 6 +- web/src/pages/RegisterPage.tsx | 2 +- web/src/pages/SetupPage.tsx | 2 +- .../pages/SystemSettingsPage/BackupPage.tsx | 4 +- .../SystemSettingsPage/SystemSettingsPage.tsx | 2 +- .../components/AiSettingsTab.tsx | 14 +- .../components/EmailSettingsTab.tsx | 4 +- .../components/ProtocolMappingsTab.tsx | 4 +- .../components/VectorDbSettingsTab.tsx | 14 +- web/src/pages/TasksPage.tsx | 2 +- .../components/PathRuleEditorDrawer.tsx | 4 +- .../UsersPage/components/RoleEditorDrawer.tsx | 6 +- .../UsersPage/components/UserEditorDrawer.tsx | 2 +- 46 files changed, 1568 insertions(+), 508 deletions(-) create mode 100644 domain/agent/approvals.py create mode 100644 domain/agent/execution.py create mode 100644 domain/permission/execution.py create mode 100644 tests/test_agent_boundaries.py create mode 100644 tests/test_config_version.py create mode 100644 tests/test_notices.py diff --git a/README.md b/README.md index 2aecba0..98cc82b 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,15 @@ Run asynchronous background tasks — file indexing, data backups, scheduled job An integrated AI agent with built-in tools for VFS operations, web fetching, and file processing — bringing intelligent automation directly into your cloud storage. +#### MCP and Agent boundaries + +- The MCP endpoint is `/api/mcp/`. External clients authenticate with a Foxel Bearer Token and own their confirmation UI; the server always enforces the authenticated user's path permissions. +- `time`, directory listing, metadata, text reads, search, processor listing, and `web_fetch` can run directly. `web_fetch` supports GET/POST/PUT/PATCH/DELETE/HEAD/OPTIONS, including external write requests, without an additional Foxel approval step. +- Agent writes, creates, deletes, moves, copies, renames, and `processors_run` calls create database-backed approval batches. Approval requests must include the returned `approval_batch_id`; execution uses the server-stored arguments, rejects expired or cross-user requests, and does not re-execute repeated confirmations. +- Recursive changes and directory processing validate all enumerated children before starting. Any denied or unverifiable path rejects the batch; queued work rechecks permissions and stops after a scope change. +- Tools return structured error codes such as `permission_denied`, `invalid_arguments`, `permission_scope_unverifiable`, `approval_expired`, `request_timeout`, and `execution_failed` without exposing internal exceptions or credentials. +- Agent conversations currently support OpenAI, Anthropic, and Ollama interfaces, with at most eight tool-call rounds per request. Processor task IDs can be tracked in the task queue. + ### 🌐 Protocol Mappings Access your files through familiar protocols: diff --git a/README_zh.md b/README_zh.md index 61f946d..118ed47 100644 --- a/README_zh.md +++ b/README_zh.md @@ -75,6 +75,15 @@ 内置 AI Agent，提供 VFS 操作、网页抓取、文件处理等工具，将智能自动化能力直接融入你的云盘。 +#### MCP 与 Agent 边界 + +- MCP 服务地址为 `/api/mcp/`，外部客户端使用 Foxel Bearer Token；直接调用由客户端负责确认，服务端始终执行当前用户的路径权限。 +- `time`、目录浏览、元信息、文本读取、搜索、处理器列表及 `web_fetch` 可直接调用。`web_fetch` 支持 GET/POST/PUT/PATCH/DELETE/HEAD/OPTIONS，包含外部写请求但不在 Foxel 内额外审批。 +- Agent 对文件写入、创建、删除、移动、复制、重命名和 `processors_run` 创建数据库审批批次。审批请求必须携带返回的 `approval_batch_id`；参数以服务端保存的原始调用为准，过期或跨用户请求会拒绝，重复确认不会再次执行。 +- 目录变更和目录处理在执行前检查全部可见子项，任一项缺少权限或无法完整枚举则整批拒绝。后台任务会重新检查权限，权限变化后停止后续操作。 +- 工具错误使用结构化错误码，例如 `permission_denied`、`invalid_arguments`、`permission_scope_unverifiable`、`approval_expired`、`request_timeout` 和 `execution_failed`，不会回传底层异常或凭据。 +- Agent 当前支持 OpenAI、Anthropic、Ollama 对话接口；单次最多 8 轮工具调用。处理器返回任务 ID 后需要在任务队列查看进度。 + ### 🌐 协议映射 通过熟悉的协议访问你的文件： diff --git a/domain/agent/api.py b/domain/agent/api.py index d4d8c7c..55c40d2 100644 --- a/domain/agent/api.py +++ b/domain/agent/api.py @@ -6,7 +6,7 @@ from fastapi.responses import StreamingResponse from api.response import success from domain.audit import AuditAction, audit from domain.auth import User, get_current_active_user -from .service import AgentService +from .service import AgentService, validate_request from .types import AgentChatRequest @@ -14,7 +14,7 @@ router = APIRouter(prefix="/api/agent", tags=["agent"]) @router.post("/chat") -@audit(action=AuditAction.CREATE, description="Agent 对话", body_fields=["auto_execute", "approved_mcp_call_ids", "rejected_mcp_call_ids"]) +@audit(action=AuditAction.CREATE, description="Agent 对话", body_fields=["auto_execute", "approval_batch_id", "approved_mcp_call_ids", "rejected_mcp_call_ids"]) async def chat( request: Request, payload: AgentChatRequest, @@ -25,12 +25,13 @@ async def chat( @router.post("/chat/stream") -@audit(action=AuditAction.CREATE, description="Agent 对话（SSE）", body_fields=["auto_execute", "approved_mcp_call_ids", "rejected_mcp_call_ids"]) +@audit(action=AuditAction.CREATE, description="Agent 对话（SSE）", body_fields=["auto_execute", "approval_batch_id", "approved_mcp_call_ids", "rejected_mcp_call_ids"]) async def chat_stream( request: Request, payload: AgentChatRequest, current_user: Annotated[User, Depends(get_current_active_user)], ): + await validate_request(payload, current_user) return StreamingResponse( AgentService.chat_stream(payload, current_user), media_type="text/event-stream", diff --git a/domain/agent/approvals.py b/domain/agent/approvals.py new file mode 100644 index 0000000..7f69ee8 --- /dev/null +++ b/domain/agent/approvals.py @@ -0,0 +1,87 @@ +import json +import uuid +from datetime import datetime, timedelta, timezone +from typing import Any + +from fastapi import HTTPException +from tortoise.transactions import in_transaction + +from models.database import AgentApprovalBatch, AgentApprovalCall +from .tools.base import tool_error, tool_result_to_content + + +def now(): + return datetime.now(timezone.utc) + + +def pending_call(call: AgentApprovalCall) -> dict[str, Any]: + return {"id": call.call_id, "name": call.name, "arguments": call.arguments, + "requires_confirmation": call.requires_confirmation, "status": call.status} + + +def result_message(call: AgentApprovalCall) -> dict[str, Any]: + content = call.result or tool_result_to_content(tool_error("operation_in_progress")) + return {"role": "tool", "mcp_call_id": call.call_id, "content": content} + + +async def create_batch(user_id: int, assistant: dict[str, Any], tool_index: dict[str, Any]) -> AgentApprovalBatch: + async with in_transaction() as connection: + batch = await AgentApprovalBatch.create( + id=str(uuid.uuid4()), user_id=user_id, assistant_message=assistant, + expires_at=now() + timedelta(hours=24), using_db=connection, + ) + await AgentApprovalCall.bulk_create([ + AgentApprovalCall(batch_id=batch.id, call_id=call["id"], position=position, + name=call["name"], arguments=call.get("arguments", {}), + requires_confirmation=bool(tool_index.get(call["name"], {}).get("meta", {}).get("requires_confirmation", True))) + for position, call in enumerate(assistant["mcp_calls"]) + ], using_db=connection) + return batch + + +async def load_batch(batch_id: str, user_id: int, approved: set[str], rejected: set[str]): + batch = await AgentApprovalBatch.get_or_none(id=batch_id, user_id=user_id) + if batch is None: + raise HTTPException(400, detail="invalid_approval_batch") + calls = await AgentApprovalCall.filter(batch_id=batch.id).order_by("position") + if approved & rejected or (approved | rejected) - {call.call_id for call in calls}: + raise HTTPException(400, detail="invalid_approval_ids") + if batch.expires_at <= now(): + await AgentApprovalCall.filter(batch_id=batch.id, status="pending").update( + status="expired", result=tool_result_to_content(tool_error("approval_expired")), updated_at=now(), + ) + raise HTTPException(400, detail="approval_expired") + cutoff = now() - timedelta(days=7) + expired_ids = await AgentApprovalBatch.filter(expires_at__lt=now()).values_list("id", flat=True) + if expired_ids: + await AgentApprovalCall.filter(batch_id__in=expired_ids, status="pending").update( + status="expired", result=tool_result_to_content(tool_error("approval_expired")), updated_at=now(), + ) + stale = await AgentApprovalBatch.filter(expires_at__lt=cutoff, executing=False).exclude( + continuation_status="running" + ).values_list("id", flat=True) + for stale_id in stale: + if not await AgentApprovalCall.filter(batch_id=stale_id, status__in=["pending", "running"]).exists() and not await AgentApprovalCall.filter(batch_id=stale_id, updated_at__gte=cutoff).exists(): + await AgentApprovalBatch.filter(id=stale_id).delete() + return batch, calls + + +async def execute_call(call: AgentApprovalCall, execute, *, rejected: bool = False) -> AgentApprovalCall: + claimed = await AgentApprovalCall.filter(id=call.id, status="pending").update(status="running", updated_at=now()) + if not claimed: + return await AgentApprovalCall.get(id=call.id) + if rejected: + content = tool_result_to_content({"canceled": True, "reason": "user_rejected"}) + status = "rejected" + else: + # Cancellation intentionally leaves running: retrying an uncertain write is unsafe. + try: + content = await execute(call.name, call.arguments) + except Exception: + content = tool_result_to_content(tool_error("execution_failed")) + try: + status = "failed" if json.loads(content).get("ok") is False else "succeeded" + except (ValueError, AttributeError): + status = "succeeded" + await AgentApprovalCall.filter(id=call.id, status="running").update(status=status, result=content, updated_at=now()) + return await AgentApprovalCall.get(id=call.id) diff --git a/domain/agent/execution.py b/domain/agent/execution.py new file mode 100644 index 0000000..fa116e1 --- /dev/null +++ b/domain/agent/execution.py @@ -0,0 +1,254 @@ +import codecs +import json +from pathlib import PurePosixPath +from typing import Any +from urllib.parse import urlsplit + +import httpx +from fastapi import HTTPException + +from domain.auth import User +from domain.permission.execution import ( + ExecutionError, execution_context, normalize_path, require_path, require_paths, +) +from domain.permission.types import PathAction +from domain.virtual_fs import VirtualFSService + +from .tools import get_tool +from .tools.base import normalize_tool_result, tool_error + + +def validate_arguments(name: str, arguments: dict[str, Any]) -> dict[str, Any]: + spec = get_tool(name) + if spec is None: + raise ExecutionError("unsupported_capability") + args = {key: value for key, value in arguments.items() if value is not None} + properties = spec.parameters.get("properties", {}) + if set(args) - set(properties) or any(key not in args for key in spec.parameters.get("required", [])): + raise ExecutionError("invalid_arguments") + types = {"string": str, "integer": int, "number": (int, float), "boolean": bool, "object": dict, "array": list} + for key, value in args.items(): + expected = types.get(properties[key].get("type")) + if expected and (not isinstance(value, expected) or (isinstance(value, bool) and expected != bool)): + raise ExecutionError("invalid_arguments") + if key in {"path", "src", "dst", "save_to"}: + args[key] = normalize_path(value) + for key, low, high in (("page", 1, None), ("page_size", 1, 100), ("top_k", 1, 100), + ("max_chars", 1, 100000), ("max_depth", 0, None)): + if key in args and (args[key] < low or (high is not None and args[key] > high)): + raise ExecutionError("invalid_arguments") + for key, values in (("sort_by", {"name", "size", "mtime"}), ("sort_order", {"asc", "desc"}), + ("mode", {"filename", "vector"})): + if key in args and args[key] not in values: + raise ExecutionError("invalid_arguments") + if "encoding" in args: + try: + codecs.lookup(args["encoding"]) + except LookupError: + raise ExecutionError("invalid_arguments") from None + if "q" in args and not args["q"].strip(): + raise ExecutionError("invalid_arguments") + if name == "web_fetch": + url = urlsplit(args["url"]) + method = args.get("method", "GET").upper() + if url.scheme not in {"http", "https"} or not url.hostname or method not in { + "GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS", + }: + raise ExecutionError("invalid_arguments") + args["method"] = method + for key in ("headers", "params"): + if key in args and any(not isinstance(value, str) for value in args[key].values()): + raise ExecutionError("invalid_arguments") + if "suffix" in args and (not args["suffix"].strip() or any(c in args["suffix"] for c in ("/", "\\", "\x00"))): + raise ExecutionError("invalid_arguments") + return args + + +async def snapshot_tree(root: str, *, optional: bool = False) -> dict[str, bool]: + """Enumerate unfiltered paths, failing closed if pagination cannot prove completeness.""" + try: + info = await VirtualFSService.stat(root) + except FileNotFoundError: + if optional: + return {} + raise + except HTTPException as exc: + if optional and exc.status_code == 404: + return {} + raise + if not isinstance(info, dict): + raise ExecutionError("permission_scope_unverifiable") + result = {root: bool(info.get("is_dir"))} + stack = [root] if result[root] else [] + while stack: + directory = stack.pop() + page, cursor, seen_cursors = 1, None, set() + while True: + try: + listing = await VirtualFSService.list_virtual_dir(directory, page, 100, "name", "asc", cursor) + except Exception as exc: + raise ExecutionError("permission_scope_unverifiable") from exc + if not isinstance(listing, dict) or not isinstance(listing.get("items"), list): + raise ExecutionError("permission_scope_unverifiable") + items = listing["items"] + for item in items: + name = item.get("name") + if not isinstance(name, str) or not name or "/" in name or "\\" in name: + raise ExecutionError("permission_scope_unverifiable") + path = normalize_path(directory.rstrip("/") + "/" + name) + if path in result or len(result) >= 10000: + raise ExecutionError("permission_scope_unverifiable") + result[path] = bool(item.get("is_dir")) + if result[path]: + stack.append(path) + if listing.get("pagination_mode") == "cursor": + if not listing.get("has_next"): + break + cursor = listing.get("next_cursor") + if not cursor or cursor in seen_cursors: + raise ExecutionError("permission_scope_unverifiable") + seen_cursors.add(cursor) + else: + total = listing.get("total") + if not isinstance(total, int) or (not items and page * 100 < total): + raise ExecutionError("permission_scope_unverifiable") + if len(items) < min(100, max(0, total - (page - 1) * 100)): + raise ExecutionError("permission_scope_unverifiable") + if page * 100 >= total: + break + page += 1 + return result + + +async def prepare_scope(name: str, args: dict[str, Any], user: User) -> dict[str, Any]: + from domain.processors import ProcessorService + scope: dict[str, Any] = {"user_id": user.id, "permissions": [], "trees": {}} + permissions: set[tuple[str, str]] = set() + + async def output_path(path: str): + permissions.add((path, PathAction.WRITE)) + parent = str(PurePosixPath(path).parent) + while parent != "/": + try: + await VirtualFSService.stat(parent) + break + except FileNotFoundError: + permissions.add((parent, PathAction.WRITE)) + except HTTPException as exc: + if exc.status_code != 404: + raise + permissions.add((parent, PathAction.WRITE)) + parent = str(PurePosixPath(parent).parent) + + async def tree(root: str, actions: list[str], optional: bool = False): + await require_paths([(root, action) for action in actions]) + snapshot = await snapshot_tree(root, optional=optional) + if snapshot: + scope["trees"][root] = snapshot + for path in snapshot or {root: False}: + permissions.update((path, action) for action in actions) + return snapshot + + if name in {"vfs_move", "vfs_rename", "vfs_copy"}: + src, dst = args["src"], args["dst"] + if src == "/" or dst == "/" or src == dst or dst.startswith(src + "/") or src.startswith(dst + "/"): + raise ExecutionError("invalid_arguments") + actions = [PathAction.READ] + ([] if name == "vfs_copy" else [PathAction.DELETE]) + await require_paths([(src, action) for action in actions] + [(dst, PathAction.WRITE)]) + source = await tree(src, actions) + for path in source: + permissions.add((dst + path[len(src):], PathAction.WRITE)) + # Native overwrite can remove the entire destination subtree. + if args.get("overwrite"): + await tree(dst, [PathAction.WRITE, PathAction.DELETE], optional=True) + await output_path(dst) + elif name == "vfs_delete": + if args["path"] == "/": + raise ExecutionError("invalid_arguments") + await tree(args["path"], [PathAction.DELETE, PathAction.READ]) + elif name == "processors_run": + root = args["path"] + await require_path(root, PathAction.READ) + processor = ProcessorService.get_processor(args["processor_type"]) + if processor is None: + raise ExecutionError("unsupported_capability") + config = args.get("config", {}) + for field in getattr(processor, "config_schema", []): + value = config.get(field["key"], field.get("default")) + if field.get("required") and (value is None or value == ""): + raise ExecutionError("invalid_arguments") + if value is not None: + kind = field.get("type") + if kind == "number" and (isinstance(value, bool) or not isinstance(value, (float, int))): + raise ExecutionError("invalid_arguments") + if kind == "string" and not isinstance(value, str): + raise ExecutionError("invalid_arguments") + if kind == "select" and value not in [option["value"] for option in field.get("options", [])]: + raise ExecutionError("invalid_arguments") + source = await tree(root, [PathAction.READ]) + is_dir = source[root] + produces = bool(getattr(processor, "produces_file", False)) + if is_dir and getattr(processor, "supports_directory", False) and produces and not user.is_admin: + raise ExecutionError("permission_scope_unverifiable") + directory_mode = is_dir and ("max_depth" in args or "suffix" in args) + overwrite = args.get("overwrite", directory_mode) + if is_dir and args.get("save_to"): + raise ExecutionError("invalid_arguments") + if produces: + if is_dir: + if not overwrite and not args.get("suffix"): + raise ExecutionError("invalid_arguments") + for path, child_is_dir in source.items(): + if not child_is_dir: + target = path + if not overwrite: + p = PurePosixPath(path) + target = str(p.with_name(p.stem + args["suffix"] + p.suffix)) + permissions.add((target, PathAction.WRITE)) + else: + target = root if overwrite else args.get("save_to") + if not target: + raise ExecutionError("permission_scope_unverifiable") + await output_path(target) + else: + action = PathAction.WRITE if name in {"vfs_write_text", "vfs_mkdir"} else PathAction.READ + if "path" in args: + if action == PathAction.WRITE and args["path"] == "/": + raise ExecutionError("invalid_arguments") + permissions.add((args["path"], action)) + if action == PathAction.WRITE: + await require_path(args["path"], action) + await output_path(args["path"]) + await require_paths(sorted(permissions)) + scope["permissions"] = [list(item) for item in sorted(permissions)] + return scope + + +def exception_result(exc: Exception) -> dict[str, Any]: + if isinstance(exc, ExecutionError): + return tool_error(exc.code) + if isinstance(exc, (ValueError, TypeError, LookupError, OverflowError)): + return tool_error("invalid_arguments") + if isinstance(exc, (FileNotFoundError,)) or isinstance(exc, HTTPException) and exc.status_code == 404: + return tool_error("path_not_found") + if isinstance(exc, HTTPException): + code = {403: "permission_denied", 400: "invalid_arguments", 409: "path_conflict", 501: "unsupported_capability"}.get(exc.status_code, "execution_failed") + return tool_error(code) + if isinstance(exc, httpx.TimeoutException): + return tool_error("request_timeout") + return tool_error("execution_failed") + + +async def execute_tool(name: str, arguments: dict[str, Any], user: User | None, current_path: str | None = None) -> dict[str, Any]: + try: + if user is None or user.disabled: + raise ExecutionError("permission_denied") + args = validate_arguments(name, arguments) + with execution_context({"user_id": user.id, "current_path": current_path}): + scope = await prepare_scope(name, args, user) + with execution_context(scope): + result = await get_tool(name).handler(args) + # Pydantic search results must remain structured JSON rather than repr strings. + return normalize_tool_result(json.loads(json.dumps(result, ensure_ascii=False, default=lambda value: value.model_dump(mode="json")))) + except Exception as exc: + return exception_result(exc) diff --git a/domain/agent/mcp.py b/domain/agent/mcp.py index 7cbd7fc..70ca090 100644 --- a/domain/agent/mcp.py +++ b/domain/agent/mcp.py @@ -1,6 +1,7 @@ import inspect import json from contextlib import asynccontextmanager +from contextvars import ContextVar from datetime import timedelta from typing import Annotated, Any, Literal from urllib.parse import quote, unquote @@ -10,6 +11,7 @@ from mcp.client import Client from mcp.client.streamable_http import streamable_http_client from mcp.server.auth.provider import AccessToken from mcp.server.auth.settings import AuthSettings +from mcp.server.auth.middleware.auth_context import get_access_token from mcp.server.mcpserver import Context, MCPServer from mcp.types import ToolAnnotations from pydantic import Field @@ -17,11 +19,24 @@ from pydantic import Field from domain.auth import AuthService, User from domain.processors import ProcessorService -from .tools import get_tool, mcp_tool_descriptors -from .tools.base import McpToolDescriptor, normalize_tool_result, tool_result_to_content +from .tools import mcp_tool_descriptors +from .tools.base import McpToolDescriptor, tool_result_to_content +from .execution import execute_tool +from domain.permission.execution import ExecutionError, normalize_path INTERNAL_MCP_BASE_URL = "http://127.0.0.1:8000/" CURRENT_PATH_HEADER = "x-foxel-current-path" +_resource_context: ContextVar[Context | None] = ContextVar("foxel_resource_context", default=None) + + +class FoxelMCPServer(MCPServer): + async def read_resource(self, uri, context=None): + # The SDK does not inject Context into fixed-URI resource handlers. + token = _resource_context.set(context) + try: + return await super().read_resource(uri, context) + finally: + _resource_context.reset(token) def _normalize_path(path: str | None) -> str | None: @@ -32,14 +47,17 @@ def _normalize_path(path: str | None) -> str | None: return None if not value.startswith("/"): value = "/" + value - return value.rstrip("/") or "/" + return normalize_path(value) def _header_current_path(ctx: Context | None) -> str | None: request = ctx.request_context.request if ctx and ctx.request_context else None if request is None: return None - return _normalize_path(request.headers.get(CURRENT_PATH_HEADER)) + try: + return _normalize_path(request.headers.get(CURRENT_PATH_HEADER)) + except ExecutionError: + return None def _field_annotation(schema: dict[str, Any], required: bool) -> tuple[Any, Any]: @@ -69,7 +87,15 @@ def _field_annotation(schema: dict[str, Any], required: bool) -> tuple[Any, Any] if not required and default is None: annotation = annotation | None - if description: + if enum_values: + annotation = Annotated[annotation, Field(description=description)] + elif raw_type in {"string", "integer", "number", "boolean"}: + annotation = Annotated[annotation, Field( + description=description, strict=True, + ge=schema.get("minimum") if raw_type in {"integer", "number"} else None, + le=schema.get("maximum") if raw_type in {"integer", "number"} else None, + )] + elif description: annotation = Annotated[annotation, Field(description=description)] return annotation, default @@ -94,15 +120,8 @@ def _build_tool_signature(descriptor: McpToolDescriptor) -> inspect.Signature: def _build_tool_wrapper(descriptor: McpToolDescriptor): - async def wrapper(**kwargs: Any) -> dict[str, Any]: - spec = get_tool(descriptor.name) - if not spec: - return normalize_tool_result({"error": f"unknown_tool: {descriptor.name}"}) - try: - result = await spec.handler(kwargs) - return normalize_tool_result(result) - except Exception as exc: # noqa: BLE001 - return normalize_tool_result({"error": str(exc)}) + async def wrapper(ctx: Context, **kwargs: Any) -> dict[str, Any]: + return await _tool_resource(descriptor.name, kwargs, _header_current_path(ctx)) wrapper.__name__ = descriptor.name wrapper.__doc__ = descriptor.description @@ -110,6 +129,16 @@ def _build_tool_wrapper(descriptor: McpToolDescriptor): return wrapper +async def _authenticated_user() -> User | None: + token = get_access_token() + if token is None: + return None + try: + return await AuthService.get_current_active_user(await AuthService.get_current_user(token.token)) + except Exception: + return None + + class FoxelMcpTokenVerifier: async def verify_token(self, token: str) -> AccessToken | None: try: @@ -119,7 +148,7 @@ class FoxelMcpTokenVerifier: return AccessToken(token=token, client_id=user.username, scopes=[]) -MCP_SERVER = MCPServer( +MCP_SERVER = FoxelMCPServer( name="Foxel MCP", instructions="Foxel 内置 MCP 服务，提供文件系统、网页抓取、时间与处理器相关能力。", token_verifier=FoxelMcpTokenVerifier(), @@ -150,7 +179,7 @@ for descriptor in mcp_tool_descriptors(): mime_type="application/json", ) def current_path_resource() -> dict[str, Any]: - return {"current_path": None} + return {"current_path": _header_current_path(_resource_context.get())} @MCP_SERVER.resource( @@ -162,9 +191,11 @@ def current_path_resource() -> dict[str, Any]: ) def tool_confirmation_policy_resource() -> dict[str, Any]: return { - "read_tools": [tool.name for tool in mcp_tool_descriptors() if not tool.requires_confirmation], + "read_tools": [tool.name for tool in mcp_tool_descriptors() if tool.annotations.get("readOnlyHint")], + "unconfirmed_tools": [tool.name for tool in mcp_tool_descriptors() if not tool.requires_confirmation], "write_tools": [tool.name for tool in mcp_tool_descriptors() if tool.requires_confirmation], "rule": "直接调用 MCP tool 时不额外审批；通过 agent 代表用户执行写操作时需要审批。", + "web_fetch": "所有 HTTP 方法免审批，包含可能修改外部系统的请求；免审批不代表只读。", } @@ -179,15 +210,8 @@ def processors_index_resource() -> dict[str, Any]: return {"processors": ProcessorService.list_processors()} -async def _tool_resource(tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]: - spec = get_tool(tool_name) - if not spec: - return normalize_tool_result({"error": f"unknown_tool: {tool_name}"}) - try: - result = await spec.handler(arguments) - return normalize_tool_result(result) - except Exception as exc: # noqa: BLE001 - return normalize_tool_result({"error": str(exc)}) +async def _tool_resource(tool_name: str, arguments: dict[str, Any], current_path: str | None = None) -> dict[str, Any]: + return await execute_tool(tool_name, arguments, await _authenticated_user(), current_path) @MCP_SERVER.resource( @@ -197,8 +221,8 @@ async def _tool_resource(tool_name: str, arguments: dict[str, Any]) -> dict[str, description="读取指定路径的文件或目录元信息；path 需要 URL 编码。", mime_type="application/json", ) -async def vfs_stat_resource(path: str) -> dict[str, Any]: - return await _tool_resource("vfs_stat", {"path": "/" + unquote(path).lstrip("/")}) +async def vfs_stat_resource(path: str, ctx: Context) -> dict[str, Any]: + return await _tool_resource("vfs_stat", {"path": "/" + unquote(path).lstrip("/")}, _header_current_path(ctx)) @MCP_SERVER.resource( @@ -208,8 +232,8 @@ async def vfs_stat_resource(path: str) -> dict[str, Any]: description="读取文本文件内容；path 需要 URL 编码。", mime_type="application/json", ) -async def vfs_text_resource(path: str) -> dict[str, Any]: - return await _tool_resource("vfs_read_text", {"path": "/" + unquote(path).lstrip("/")}) +async def vfs_text_resource(path: str, ctx: Context) -> dict[str, Any]: + return await _tool_resource("vfs_read_text", {"path": "/" + unquote(path).lstrip("/")}, _header_current_path(ctx)) @MCP_SERVER.resource( @@ -219,8 +243,8 @@ async def vfs_text_resource(path: str) -> dict[str, Any]: description="列出目录内容；path 需要 URL 编码。", mime_type="application/json", ) -async def vfs_dir_resource(path: str) -> dict[str, Any]: - return await _tool_resource("vfs_list_dir", {"path": "/" + unquote(path).lstrip("/")}) +async def vfs_dir_resource(path: str, ctx: Context) -> dict[str, Any]: + return await _tool_resource("vfs_list_dir", {"path": "/" + unquote(path).lstrip("/")}, _header_current_path(ctx)) @MCP_SERVER.resource( @@ -230,8 +254,8 @@ async def vfs_dir_resource(path: str) -> dict[str, Any]: description="搜索文件；query 需要 URL 编码。", mime_type="application/json", ) -async def vfs_search_resource(query: str) -> dict[str, Any]: - return await _tool_resource("vfs_search", {"q": unquote(query)}) +async def vfs_search_resource(query: str, ctx: Context) -> dict[str, Any]: + return await _tool_resource("vfs_search", {"q": unquote(query)}, _header_current_path(ctx)) @MCP_SERVER.prompt(name="browse_path", title="Browse Path", description="生成浏览目录的推荐提示词。") diff --git a/domain/agent/service.py b/domain/agent/service.py index dfd8353..2fb842d 100644 --- a/domain/agent/service.py +++ b/domain/agent/service.py @@ -1,443 +1,264 @@ import asyncio import json import uuid -from typing import Any, Dict, List, Optional, Tuple +from contextlib import aclosing +from typing import Any import httpx from fastapi import HTTPException from domain.ai import AIProviderService, MissingModelError, chat_completion, chat_completion_stream from domain.auth import User +from domain.permission.execution import ExecutionError, normalize_path +from models.database import AgentApprovalBatch, AgentApprovalCall +from .approvals import create_batch, execute_call, load_batch, pending_call, result_message from .mcp import mcp_client_session, mcp_content_to_text -from .tools import tool_result_to_content -from .types import AgentChatRequest, PendingMcpCall +from .tools.base import tool_error, tool_result_to_content +from .types import AgentChatRequest -def _normalize_path(path: Optional[str]) -> Optional[str]: - if not path: - return None - value = str(path).strip().replace("\\", "/") - if not value: - return None - if not value.startswith("/"): - value = "/" + value - return value.rstrip("/") or "/" - - -def _build_system_prompt(current_path: Optional[str]) -> str: - lines = [ - "你是 Foxel 的 AI 助手。", - "你可以通过 MCP 工具对文件/目录进行查询、读写、移动、复制、删除，以及运行处理器（processor）。", - "", - "可用工具：", - "- time：获取服务器当前时间（精确到秒，英文星期），支持 year/month/day/hour/minute/second 偏移。", - "- web_fetch：抓取网页（HTTP 请求），支持 GET/POST/PUT/PATCH/DELETE/HEAD/OPTIONS，返回状态/标题/正文/链接等。", - "- vfs_list_dir：浏览目录（列出 entries + pagination）。", - "- vfs_stat：查看文件/目录信息。", - "- vfs_read_text：读取文本文件内容（不支持二进制）。", - "- vfs_search：搜索文件（vector/filename）。", - "- vfs_write_text：写入文本文件内容（覆盖）。", - "- vfs_mkdir：创建目录。", - "- vfs_delete：删除文件或目录。", - "- vfs_move：移动路径。", - "- vfs_copy：复制路径。", - "- vfs_rename：重命名路径。", - "- processors_list：获取可用处理器列表（含 type/name/config_schema/produces_file/supports_directory）。", - "- processors_run：运行处理器处理文件或目录（会返回 task_id 或 task_ids）。", - "", - "规则：", - "1) 读操作（web_fetch/vfs_list_dir/vfs_stat/vfs_read_text/vfs_search）可直接调用工具。", - "2) 写/改/删操作（vfs_write_text/vfs_mkdir/vfs_delete/vfs_move/vfs_copy/vfs_rename/processors_run）默认需要用户确认；只有在开启自动执行时才应直接执行。", - "3) 用户未给出明确路径时先追问；若提供了“当前文件管理目录”，可以基于它把相对描述补全为绝对路径（以 / 开头）。", - "4) 修改文件内容：先读取（vfs_read_text）→给出改动点→确认后再写入（vfs_write_text）。", - "5) processors_run 返回任务 id 后，说明任务已提交，可在任务队列查看进度。", - "6) 回答语言跟随用户；用户用英文则用英文，用户用中文则用中文。回答尽量简洁。", - ] - if current_path: - lines.append("") - lines.append(f"当前文件管理目录：{current_path}") - return "\n".join(lines) - - -def _ensure_mcp_call_ids(message: Dict[str, Any]) -> Dict[str, Any]: - mcp_calls = message.get("mcp_calls") - if not isinstance(mcp_calls, list): - return message - - changed = False - for idx, call in enumerate(mcp_calls): - if not isinstance(call, dict): - continue - call_id = call.get("id") - if isinstance(call_id, str) and call_id.strip(): - continue - call["id"] = f"call_{idx}" - changed = True - - if changed: - message["mcp_calls"] = mcp_calls - return message - - -def _extract_pending(mcp_call: Dict[str, Any], requires_confirmation: bool) -> PendingMcpCall: - arguments = mcp_call.get("arguments") if isinstance(mcp_call.get("arguments"), dict) else {} - return PendingMcpCall( - id=str(mcp_call.get("id") or ""), - name=str(mcp_call.get("name") or ""), - arguments=arguments, - requires_confirmation=requires_confirmation, +def _build_system_prompt(current_path: str | None) -> str: + text = ( + "你是 Foxel 的 AI 助手。使用提供的 MCP 工具查询、读取、写入、移动、复制、删除文件，" + "抓取网页、获取时间或提交处理器任务。回答语言跟随用户，尽量简洁。\n" + "查询工具可直接执行。web_fetch 的所有 HTTP 方法均免审批，但可能修改外部系统。\n" + "文件写入、创建、删除、移动、复制、重命名和 processors_run 默认需要用户确认；" + "开启自动执行时可直接执行。审批不授予路径权限。\n" + "路径必须是以 / 开头的绝对路径；未提供明确路径时追问，或基于当前文件管理目录补全。\n" + "修改文本前先读取并说明改动，再申请写入。processors_run 返回 task_id 只代表已提交，" + "请用户到任务队列查看进度。当前不提供任务查询、外部 MCP、后台持续 Agent 或长期记忆。" ) + return text + (f"\n当前文件管理目录：{current_path}" if current_path else "") -def _find_last_assistant_mcp_calls(messages: List[Dict[str, Any]]) -> Tuple[int, Dict[str, Any]]: - for idx in range(len(messages) - 1, -1, -1): - msg = messages[idx] - if not isinstance(msg, dict): +def _ensure_mcp_call_ids(message: dict[str, Any]) -> dict[str, Any]: + calls = message.get("mcp_calls") + if not isinstance(calls, list): + return message + seen, valid = set(), [] + for call in calls: + if not isinstance(call, dict) or not isinstance(call.get("name"), str): continue - if msg.get("role") != "assistant": - continue - mcp_calls = msg.get("mcp_calls") - if isinstance(mcp_calls, list) and mcp_calls: - return idx, msg - raise HTTPException(status_code=400, detail="没有可确认的待执行操作") + call = dict(call) + call_id = call.get("id") + if not isinstance(call_id, str) or not call_id.strip() or call_id in seen or len(call_id) > 255: + call_id = "call_" + uuid.uuid4().hex + seen.add(call_id) + call["id"] = call_id + if not isinstance(call.get("arguments"), dict): + call["arguments"] = {} + valid.append(call) + return {**message, "mcp_calls": valid} -def _existing_mcp_result_ids(messages: List[Dict[str, Any]]) -> set[str]: - ids: set[str] = set() - for msg in messages: - if not isinstance(msg, dict): - continue - if msg.get("role") != "tool": - continue - call_id = msg.get("mcp_call_id") - if isinstance(call_id, str) and call_id.strip(): - ids.add(call_id) - return ids - - -def _tool_requires_confirmation(tool_descriptor: Dict[str, Any]) -> bool: - meta = tool_descriptor.get("meta") if isinstance(tool_descriptor.get("meta"), dict) else {} - if "requires_confirmation" in meta: - return bool(meta.get("requires_confirmation")) - annotations = tool_descriptor.get("annotations") if isinstance(tool_descriptor.get("annotations"), dict) else {} - return not bool(annotations.get("readOnlyHint")) +def _tool_requires_confirmation(descriptor: dict[str, Any]) -> bool: + meta = descriptor.get("meta") or {} + return bool(meta.get("requires_confirmation", not (descriptor.get("annotations") or {}).get("readOnlyHint"))) async def _choose_chat_ability() -> str: - tools_model = await AIProviderService.get_default_model("tools") - return "tools" if tools_model else "chat" + ability = "tools" if await AIProviderService.get_default_model("tools") else "chat" + model = await AIProviderService.get_default_model(ability) + if model is None: + raise MissingModelError(f"未配置默认 {ability} 模型，请前往系统设置完成配置。") + if getattr(model, "provider", None) is None: + await model.fetch_related("provider") + if str(model.provider.api_format).lower() not in {"openai", "anthropic", "ollama"}: + raise MissingModelError("Agent 当前支持 OpenAI、Anthropic 和 Ollama 对话接口。") + return ability def _sse(event: str, data: Any) -> bytes: - payload = json.dumps(data, ensure_ascii=False, separators=(",", ":")) - return f"event: {event}\ndata: {payload}\n\n".encode("utf-8") + return ("event: " + event + "\ndata: " + json.dumps(data, ensure_ascii=False, separators=(",", ":")) + "\n\n").encode() -def _format_exc(exc: BaseException) -> str: - text = str(exc) - return text if text else exc.__class__.__name__ - - -async def _list_mcp_tools(session) -> List[Dict[str, Any]]: +async def _list_mcp_tools(session) -> list[dict[str, Any]]: result = await session.list_tools() - tools: List[Dict[str, Any]] = [] - for item in result.tools: - annotations = getattr(item, "annotations", None) - meta = getattr(item, "meta", None) - tools.append( - { - "name": str(getattr(item, "name", "") or ""), - "description": str(getattr(item, "description", "") or ""), - "input_schema": getattr(item, "input_schema", None) or {}, - "annotations": annotations.model_dump(exclude_none=True) if annotations is not None else {}, - "meta": meta if isinstance(meta, dict) else {}, - } - ) - return tools + return [{"name": item.name, "description": item.description or "", "input_schema": item.input_schema or {}, + "annotations": item.annotations.model_dump(exclude_none=True) if item.annotations else {}, + "meta": item.meta or {}} for item in result.tools] -async def _execute_mcp_call(session, name: str, arguments: Dict[str, Any]) -> str: - result = await session.call_tool(name, arguments) - return mcp_content_to_text(result.content, result.structured_content) +async def _execute_mcp_call(session, name: str, arguments: dict[str, Any]) -> str: + try: + result = await session.call_tool(name, arguments) + if getattr(result, "is_error", False): + return tool_result_to_content(tool_error("invalid_arguments")) + return mcp_content_to_text(result.content, result.structured_content) + except Exception: + return tool_result_to_content(tool_error("execution_failed")) + + +async def validate_request(req: AgentChatRequest, user: User): + approved, rejected = set(req.approved_mcp_call_ids), set(req.rejected_mcp_call_ids) + if approved & rejected: + raise HTTPException(400, detail="invalid_approval_ids") + if (approved or rejected) and not req.approval_batch_id: + raise HTTPException(400, detail="approval_batch_required: 请重新生成待审批操作。") + if req.context and req.context.current_path: + try: + normalize_path(req.context.current_path) + except ExecutionError: + raise HTTPException(400, detail="invalid_arguments") from None + if req.approval_batch_id: + return await load_batch(req.approval_batch_id, user.id, approved, rejected) + answered = {m.get("mcp_call_id") for m in req.messages if m.get("role") == "tool"} + if any(c.get("id") not in answered for m in req.messages if m.get("role") == "assistant" + for c in (m.get("mcp_calls") or []) if isinstance(c, dict)): + raise HTTPException(400, detail="approval_batch_required: 请重新生成待审批操作。") + return None class AgentService: @classmethod - async def chat(cls, req: AgentChatRequest, user: Optional[User]) -> Dict[str, Any]: - history: List[Dict[str, Any]] = list(req.messages or []) - current_path = _normalize_path(req.context.current_path if req.context else None) - system_prompt = _build_system_prompt(current_path) - internal_messages: List[Dict[str, Any]] = [{"role": "system", "content": system_prompt}] + history - new_messages: List[Dict[str, Any]] = [] - pending: List[PendingMcpCall] = [] + async def _run(cls, req: AgentChatRequest, user: User, streaming: bool): + validated = await validate_request(req, user) + path = normalize_path(req.context.current_path) if req.context and req.context.current_path else None + history = [dict(m) for m in req.messages if m.get("role") in {"user", "assistant", "tool"}] + messages: list[dict[str, Any]] = [] + batch_id, pending = req.approval_batch_id, [] + replace = bool(validated) + continuation_owned = False + execution_owned = False - approved_ids = {i for i in (req.approved_mcp_call_ids or []) if isinstance(i, str) and i.strip()} - rejected_ids = {i for i in (req.rejected_mcp_call_ids or []) if isinstance(i, str) and i.strip()} + def payload(reason: str): + return {"messages": history + messages if replace else messages, + "replace_messages": replace, "pending_mcp_calls": pending, + "approval_batch_id": batch_id if pending else None, "finish_reason": reason} - async with mcp_client_session(user, current_path) as mcp_session: - tools_schema = await _list_mcp_tools(mcp_session) - tool_index = {tool["name"]: tool for tool in tools_schema if tool.get("name")} - - if approved_ids or rejected_ids: - _, last_call_msg = _find_last_assistant_mcp_calls(internal_messages) - last_call_msg = _ensure_mcp_call_ids(last_call_msg) - mcp_calls = last_call_msg.get("mcp_calls") or [] - call_map: Dict[str, Dict[str, Any]] = { - str(call.get("id")): call - for call in mcp_calls - if isinstance(call, dict) and isinstance(call.get("id"), str) - } - - existing_ids = _existing_mcp_result_ids(internal_messages) - for call_id in approved_ids | rejected_ids: - if call_id in existing_ids: - continue - mcp_call = call_map.get(call_id) - if not mcp_call: - continue - name = str(mcp_call.get("name") or "") - arguments = mcp_call.get("arguments") if isinstance(mcp_call.get("arguments"), dict) else {} - tool_desc = tool_index.get(name) - - if call_id in rejected_ids: - content = tool_result_to_content({"canceled": True, "reason": "user_rejected"}) - elif not tool_desc: - content = tool_result_to_content({"error": f"unknown_tool: {name}"}) - else: - try: - content = await _execute_mcp_call(mcp_session, name, arguments) - except Exception as exc: # noqa: BLE001 - content = tool_result_to_content({"error": str(exc)}) - tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content} - internal_messages.append(tool_msg) - new_messages.append(tool_msg) - - ability = await _choose_chat_ability() - - for _ in range(8): - try: - assistant = await chat_completion( - internal_messages, - ability=ability, - tools=tools_schema, - tool_choice="auto", - timeout=60.0, - ) - except MissingModelError as exc: - raise HTTPException(status_code=400, detail=_format_exc(exc)) from exc - except httpx.HTTPStatusError as exc: - raise HTTPException(status_code=502, detail=f"对话请求失败: {_format_exc(exc)}") from exc - except httpx.RequestError as exc: - raise HTTPException(status_code=502, detail=f"对话请求异常: {_format_exc(exc)}") from exc - - assistant = _ensure_mcp_call_ids(assistant if isinstance(assistant, dict) else {"role": "assistant", "content": ""}) - internal_messages.append(assistant) - new_messages.append(assistant) - - mcp_calls = assistant.get("mcp_calls") - if not isinstance(mcp_calls, list) or not mcp_calls: - break - - pending = [] - for call in mcp_calls: - if not isinstance(call, dict): - continue - call_id = str(call.get("id") or "") - name = str(call.get("name") or "") - arguments = call.get("arguments") if isinstance(call.get("arguments"), dict) else {} - tool_desc = tool_index.get(name) - - if not tool_desc: - content = tool_result_to_content({"error": f"unknown_tool: {name}"}) - tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content} - internal_messages.append(tool_msg) - new_messages.append(tool_msg) - continue - - if _tool_requires_confirmation(tool_desc) and not req.auto_execute: - pending.append(_extract_pending(call, True)) - continue - - try: - content = await _execute_mcp_call(mcp_session, name, arguments) - except Exception as exc: # noqa: BLE001 - content = tool_result_to_content({"error": str(exc)}) - tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content} - internal_messages.append(tool_msg) - new_messages.append(tool_msg) - - if pending: - break - - payload: Dict[str, Any] = {"messages": new_messages} - if pending: - payload["pending_mcp_calls"] = [item.model_dump() for item in pending] - return payload - - @classmethod - async def chat_stream(cls, req: AgentChatRequest, user: Optional[User]): - history: List[Dict[str, Any]] = list(req.messages or []) - current_path = _normalize_path(req.context.current_path if req.context else None) - system_prompt = _build_system_prompt(current_path) - internal_messages: List[Dict[str, Any]] = [{"role": "system", "content": system_prompt}] + history - new_messages: List[Dict[str, Any]] = [] - pending: List[PendingMcpCall] = [] - - approved_ids = {i for i in (req.approved_mcp_call_ids or []) if isinstance(i, str) and i.strip()} - rejected_ids = {i for i in (req.rejected_mcp_call_ids or []) if isinstance(i, str) and i.strip()} + async def finish(reason: str): + data = payload(reason) + if continuation_owned: + await AgentApprovalBatch.filter(id=req.approval_batch_id, continuation_status="running").update( + continuation_status="completed", continuation_result=data, + ) + return data try: - async with mcp_client_session(user, current_path) as mcp_session: - tools_schema = await _list_mcp_tools(mcp_session) - tool_index = {tool["name"]: tool for tool in tools_schema if tool.get("name")} + async with mcp_client_session(user, path) as session: + tools = await _list_mcp_tools(session) + index = {tool["name"]: tool for tool in tools} - if approved_ids or rejected_ids: - _, last_call_msg = _find_last_assistant_mcp_calls(internal_messages) - last_call_msg = _ensure_mcp_call_ids(last_call_msg) - mcp_calls = last_call_msg.get("mcp_calls") or [] - call_map: Dict[str, Dict[str, Any]] = { - str(call.get("id")): call - for call in mcp_calls - if isinstance(call, dict) and isinstance(call.get("id"), str) - } + async def execute(name, arguments): + return await _execute_mcp_call(session, name, arguments) - existing_ids = _existing_mcp_result_ids(internal_messages) - for call_id in approved_ids | rejected_ids: - if call_id in existing_ids: - continue - mcp_call = call_map.get(call_id) - if not mcp_call: - continue - - name = str(mcp_call.get("name") or "") - arguments = mcp_call.get("arguments") if isinstance(mcp_call.get("arguments"), dict) else {} - tool_desc = tool_index.get(name) - - if call_id in rejected_ids: - content = tool_result_to_content({"canceled": True, "reason": "user_rejected"}) - tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content} - internal_messages.append(tool_msg) - new_messages.append(tool_msg) - yield _sse("mcp_call_end", {"mcp_call_id": call_id, "name": name, "message": tool_msg}) - continue - - if not tool_desc: - content = tool_result_to_content({"error": f"unknown_tool: {name}"}) - tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content} - internal_messages.append(tool_msg) - new_messages.append(tool_msg) - yield _sse("mcp_call_end", {"mcp_call_id": call_id, "name": name, "message": tool_msg}) - continue - - yield _sse("mcp_call_start", {"mcp_call_id": call_id, "name": name}) - try: - content = await _execute_mcp_call(mcp_session, name, arguments) - except Exception as exc: # noqa: BLE001 - content = tool_result_to_content({"error": str(exc)}) - tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content} - internal_messages.append(tool_msg) - new_messages.append(tool_msg) - yield _sse("mcp_call_end", {"mcp_call_id": call_id, "name": name, "message": tool_msg}) + if validated: + batch, calls = validated + if batch.continuation_result is not None: + yield "done", batch.continuation_result + return + approved, rejected = set(req.approved_mcp_call_ids), set(req.rejected_mcp_call_ids) + # Replace the last tool-call turn with its authoritative batch. + last = next((i for i in range(len(history) - 1, -1, -1) + if history[i].get("role") == "assistant" and history[i].get("mcp_calls")), None) + history = history + [batch.assistant_message] if last is None else history[:last] + [batch.assistant_message] + execution_owned = bool(await AgentApprovalBatch.filter(id=batch.id, executing=False).update(executing=True)) + calls = await AgentApprovalCall.filter(batch_id=batch.id).order_by("position") + if not execution_owned or any(call.status == "running" for call in calls): + messages.extend(result_message(call) for call in calls if call.result is not None) + pending = [pending_call(call) for call in calls if call.status in {"pending", "running"}] + yield "pending", {"pending_mcp_calls": pending, "approval_batch_id": batch.id} + yield "done", {**payload("operation_in_progress"), "approval_batch_id": batch.id} + return + for call in calls: + if call.call_id in approved | rejected: + if call.status == "pending" and call.call_id in approved: + yield "mcp_call_start", {"mcp_call_id": call.call_id, "name": call.name} + call = await execute_call(call, execute, rejected=call.call_id in rejected) + yield "mcp_call_end", {"mcp_call_id": call.call_id, "name": call.name, "message": result_message(call)} + if call.status == "running": + break + calls = await AgentApprovalCall.filter(batch_id=batch.id).order_by("position") + messages.extend(result_message(call) for call in calls if call.result is not None) + pending = [pending_call(call) for call in calls if call.status in {"pending", "running"}] + if pending: + yield "pending", {"pending_mcp_calls": pending, "approval_batch_id": batch_id} + yield "done", await finish("operation_in_progress" if any(c["status"] == "running" for c in pending) else "pending_approval") + return + continuation_owned = bool(await AgentApprovalBatch.filter(id=batch.id, continuation_status="pending").update(continuation_status="running")) + if not continuation_owned: + batch = await AgentApprovalBatch.get(id=batch.id) + data = batch.continuation_result or {**payload("operation_in_progress"), "approval_batch_id": batch.id} + yield "done", data + return ability = await _choose_chat_ability() - + internal = [{"role": "system", "content": _build_system_prompt(path)}] + history + messages for _ in range(8): - assistant_event_id = str(uuid.uuid4()) - yield _sse("assistant_start", {"id": assistant_event_id}) - - assistant_message: Dict[str, Any] | None = None - try: - async for event in chat_completion_stream( - internal_messages, - ability=ability, - tools=tools_schema, - tool_choice="auto", - timeout=60.0, - ): - event_type = event.get("type") - if event_type == "delta": - delta = event.get("delta") - if isinstance(delta, str) and delta: - yield _sse("assistant_delta", {"id": assistant_event_id, "delta": delta}) - elif event_type == "message": - msg = event.get("message") - if isinstance(msg, dict): - assistant_message = msg - except MissingModelError as exc: - raise HTTPException(status_code=400, detail=_format_exc(exc)) from exc - except httpx.HTTPStatusError as exc: - raise HTTPException(status_code=502, detail=f"对话请求失败: {_format_exc(exc)}") from exc - except httpx.RequestError as exc: - raise HTTPException(status_code=502, detail=f"对话请求异常: {_format_exc(exc)}") from exc - - if not assistant_message: - assistant_message = {"role": "assistant", "content": ""} - - assistant_message = _ensure_mcp_call_ids(assistant_message) - internal_messages.append(assistant_message) - new_messages.append(assistant_message) - yield _sse("assistant_end", {"id": assistant_event_id, "message": assistant_message}) - - mcp_calls = assistant_message.get("mcp_calls") - if not isinstance(mcp_calls, list) or not mcp_calls: - break - - pending = [] - for call in mcp_calls: - if not isinstance(call, dict): + event_id = str(uuid.uuid4()) + yield "assistant_start", {"id": event_id} + assistant = None + if streaming: + async for event in chat_completion_stream(internal, ability=ability, tools=tools, tool_choice="auto", timeout=60.0): + if event.get("type") == "delta" and event.get("delta"): + yield "assistant_delta", {"id": event_id, "delta": event["delta"]} + elif event.get("type") == "message": + assistant = event.get("message") + else: + assistant = await chat_completion(internal, ability=ability, tools=tools, tool_choice="auto", timeout=60.0) + assistant = _ensure_mcp_call_ids(assistant or {"role": "assistant", "content": ""}) + internal.append(assistant) + messages.append(assistant) + yield "assistant_end", {"id": event_id, "message": assistant} + calls = assistant.get("mcp_calls") or [] + if not calls: + yield "done", await finish("completed") + return + needs_approval = not req.auto_execute and any( + _tool_requires_confirmation(index.get(call["name"], {})) for call in calls + ) + batch = await create_batch(user.id, assistant, index) if needs_approval else None + records = await AgentApprovalCall.filter(batch_id=batch.id).order_by("position") if batch else [] + for position, call in enumerate(calls): + if needs_approval and _tool_requires_confirmation(index.get(call["name"], {})): continue - call_id = str(call.get("id") or "") - name = str(call.get("name") or "") - arguments = call.get("arguments") if isinstance(call.get("arguments"), dict) else {} - tool_desc = tool_index.get(name) + yield "mcp_call_start", {"mcp_call_id": call["id"], "name": call["name"]} + if batch: + record = await execute_call(records[position], execute) + message = result_message(record) + else: + message = {"role": "tool", "mcp_call_id": call["id"], "content": await execute(call["name"], call["arguments"])} + messages.append(message) + internal.append(message) + yield "mcp_call_end", {"mcp_call_id": call["id"], "name": call["name"], "message": message} + if batch: + batch_id = batch.id + records = await AgentApprovalCall.filter(batch_id=batch.id).order_by("position") + pending = [pending_call(call) for call in records if call.status in {"pending", "running"}] + yield "pending", {"pending_mcp_calls": pending, "approval_batch_id": batch_id} + yield "done", await finish("pending_approval") + return + messages.append({"role": "assistant", "content": "已达到本次 8 轮调用上限，已执行结果保留。请继续对话。"}) + yield "done", await finish("iteration_limit") + except asyncio.CancelledError: + raise + except Exception as exc: + if isinstance(exc, MissingModelError): + content, reason = str(exc), "model_unavailable" + elif isinstance(exc, httpx.TimeoutException): + content, reason = "模型请求超时。", "request_timeout" + else: + content, reason = "请求执行失败，请稍后重试。", "execution_failed" + messages.append({"role": "assistant", "content": content}) + yield "done", await finish(reason) + finally: + if execution_owned: + await AgentApprovalBatch.filter(id=req.approval_batch_id).update(executing=False) - if not tool_desc: - content = tool_result_to_content({"error": f"unknown_tool: {name}"}) - tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content} - internal_messages.append(tool_msg) - new_messages.append(tool_msg) - yield _sse("mcp_call_end", {"mcp_call_id": call_id, "name": name, "message": tool_msg}) - continue - - if _tool_requires_confirmation(tool_desc) and not req.auto_execute: - pending.append(_extract_pending(call, True)) - continue - - yield _sse("mcp_call_start", {"mcp_call_id": call_id, "name": name}) - try: - content = await _execute_mcp_call(mcp_session, name, arguments) - except Exception as exc: # noqa: BLE001 - content = tool_result_to_content({"error": str(exc)}) - tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content} - internal_messages.append(tool_msg) - new_messages.append(tool_msg) - yield _sse("mcp_call_end", {"mcp_call_id": call_id, "name": name, "message": tool_msg}) - - if pending: - yield _sse("pending", {"pending_mcp_calls": [item.model_dump() for item in pending]}) - break - - payload: Dict[str, Any] = {"messages": new_messages} - if pending: - payload["pending_mcp_calls"] = [item.model_dump() for item in pending] - yield _sse("done", payload) + @classmethod + async def chat(cls, req: AgentChatRequest, user: User) -> dict[str, Any]: + async with aclosing(cls._run(req, user, streaming=False)) as events: + async for event, data in events: + if event == "done": + return data + return {"messages": []} + @classmethod + async def chat_stream(cls, req: AgentChatRequest, user: User): + try: + async with aclosing(cls._run(req, user, streaming=True)) as events: + async for event, data in events: + yield _sse(event, data) except asyncio.CancelledError: return - except HTTPException as exc: - detail = exc.detail - content = detail if isinstance(detail, str) else str(detail) - if not content.strip(): - content = f"请求失败({exc.status_code})" - new_messages.append({"role": "assistant", "content": content}) - payload: Dict[str, Any] = {"messages": new_messages} - if pending: - payload["pending_mcp_calls"] = [item.model_dump() for item in pending] - yield _sse("done", payload) - except Exception as exc: # noqa: BLE001 - new_messages.append({"role": "assistant", "content": f"服务端异常: {_format_exc(exc)}"}) - payload: Dict[str, Any] = {"messages": new_messages} - if pending: - payload["pending_mcp_calls"] = [item.model_dump() for item in pending] - yield _sse("done", payload) diff --git a/domain/agent/tools/base.py b/domain/agent/tools/base.py index 9ab550d..3f799d1 100644 --- a/domain/agent/tools/base.py +++ b/domain/agent/tools/base.py @@ -121,6 +121,10 @@ def _build_error_payload(code: str, message: str, detail: Any = None) -> Dict[st return payload +def tool_error(code: str, message: str | None = None) -> Dict[str, Any]: + return _build_error_payload(code, message or code) + + def _normalize_tool_result(result: Any) -> Dict[str, Any]: if isinstance(result, dict) and "ok" in result: payload = dict(result) @@ -144,7 +148,8 @@ def _normalize_tool_result(result: Any) -> Dict[str, Any]: if isinstance(result, dict) and "error" in result: error = result.get("error") message = _stringify_value(error.get("message") if isinstance(error, dict) else error) - return _build_error_payload("error", message, detail=error) + code = str(error.get("code") or "execution_failed") if isinstance(error, dict) else "execution_failed" + return _build_error_payload(code, message) view = _build_view(result) summary = _build_summary(view) @@ -166,8 +171,8 @@ def tool_result_to_content(result: Any) -> str: def tool_spec_to_mcp_descriptor(spec: ToolSpec) -> McpToolDescriptor: read_only = not spec.requires_confirmation annotations: Dict[str, Any] = { - "readOnlyHint": read_only, - "destructiveHint": bool(spec.requires_confirmation), + "readOnlyHint": read_only and spec.name != "web_fetch", + "destructiveHint": bool(spec.requires_confirmation) or spec.name == "web_fetch", } if spec.name == "web_fetch": annotations["openWorldHint"] = True diff --git a/domain/agent/tools/processors.py b/domain/agent/tools/processors.py index 601747f..e8ddc3d 100644 --- a/domain/agent/tools/processors.py +++ b/domain/agent/tools/processors.py @@ -4,6 +4,8 @@ from domain.processors import ProcessDirectoryRequest, ProcessRequest, Processor from domain.virtual_fs import VirtualFSService from .base import ToolSpec +from domain.permission.execution import require_path +from domain.permission.types import PathAction async def _processors_list(_: Dict[str, Any]) -> Dict[str, Any]: @@ -12,6 +14,7 @@ async def _processors_list(_: Dict[str, Any]) -> Dict[str, Any]: async def _processors_run(args: Dict[str, Any]) -> Dict[str, Any]: path = str(args.get("path") or "") + path = await require_path(path, PathAction.READ) processor_type = str(args.get("processor_type") or "") config = args.get("config") if not isinstance(config, dict): @@ -85,7 +88,7 @@ TOOLS: Dict[str, ToolSpec] = { "config": {"type": "object", "description": "处理器配置，按 processors_list 返回的 config_schema 填写"}, "overwrite": {"type": "boolean", "description": "是否覆盖原文件/目录内文件"}, "save_to": {"type": "string", "description": "保存到指定路径（仅文件模式，且 overwrite=false 时使用）"}, - "max_depth": {"type": "integer", "description": "目录遍历深度（仅目录模式）"}, + "max_depth": {"type": "integer", "minimum": 0, "description": "目录遍历深度（仅目录模式）"}, "suffix": {"type": "string", "description": "目录批处理时的输出后缀（仅 produces_file 且 overwrite=false）"}, }, "required": ["path", "processor_type"], diff --git a/domain/agent/tools/vfs.py b/domain/agent/tools/vfs.py index 0cbe851..6a33a27 100644 --- a/domain/agent/tools/vfs.py +++ b/domain/agent/tools/vfs.py @@ -3,7 +3,9 @@ from typing import Any, Dict, Optional from domain.virtual_fs import VirtualFSService from domain.virtual_fs.search import VirtualFSSearchService -from .base import ToolSpec +from .base import ToolSpec, tool_error +from domain.permission.execution import current_user_id, require_path, require_paths +from domain.permission.types import PathAction def _normalize_vfs_path(value: Any) -> str: @@ -25,20 +27,25 @@ def _require_vfs_path(value: Any, field: str) -> str: async def _vfs_list_dir(args: Dict[str, Any]) -> Dict[str, Any]: path = _normalize_vfs_path(args.get("path") or "/") or "/" + path = await require_path(path, PathAction.READ) page = int(args.get("page") or 1) page_size = int(args.get("page_size") or 50) sort_by = str(args.get("sort_by") or "name") sort_order = str(args.get("sort_order") or "asc") - return await VirtualFSService.list_directory(path, page, page_size, sort_by, sort_order) + return await VirtualFSService.list_directory_with_permission( + path, current_user_id(), page, page_size, sort_by, sort_order, args.get("cursor") + ) async def _vfs_stat(args: Dict[str, Any]) -> Any: path = _require_vfs_path(args.get("path"), "path") + path = await require_path(path, PathAction.READ) return await VirtualFSService.stat(path) async def _vfs_read_text(args: Dict[str, Any]) -> Dict[str, Any]: path = _require_vfs_path(args.get("path"), "path") + path = await require_path(path, PathAction.READ) encoding = str(args.get("encoding") or "utf-8") max_chars = int(args.get("max_chars") or 8000) @@ -47,7 +54,7 @@ async def _vfs_read_text(args: Dict[str, Any]) -> Dict[str, Any]: try: text = bytes(data).decode(encoding) except UnicodeDecodeError: - return {"error": "binary_or_invalid_text", "path": path} + return tool_error("unsupported_capability", "binary_or_invalid_text") elif isinstance(data, str): text = data else: @@ -70,6 +77,7 @@ async def _vfs_write_text(args: Dict[str, Any]) -> Dict[str, Any]: path = _require_vfs_path(args.get("path"), "path") if path == "/": raise ValueError("invalid_path") + path = await require_path(path, PathAction.WRITE) encoding = str(args.get("encoding") or "utf-8") content = str(args.get("content") or "") data = content.encode(encoding) @@ -79,11 +87,13 @@ async def _vfs_write_text(args: Dict[str, Any]) -> Dict[str, Any]: async def _vfs_mkdir(args: Dict[str, Any]) -> Dict[str, Any]: path = _require_vfs_path(args.get("path"), "path") + path = await require_path(path, PathAction.WRITE) return await VirtualFSService.mkdir(path) async def _vfs_delete(args: Dict[str, Any]) -> Dict[str, Any]: path = _require_vfs_path(args.get("path"), "path") + path = await require_path(path, PathAction.DELETE) return await VirtualFSService.delete(path) @@ -92,6 +102,7 @@ async def _vfs_move(args: Dict[str, Any]) -> Dict[str, Any]: dst = _require_vfs_path(args.get("dst"), "dst") if src == "/" or dst == "/": raise ValueError("invalid_path") + await require_paths([(src, PathAction.READ), (src, PathAction.DELETE), (dst, PathAction.WRITE)]) overwrite = bool(args.get("overwrite") or False) return await VirtualFSService.move(src, dst, overwrite) @@ -101,6 +112,7 @@ async def _vfs_copy(args: Dict[str, Any]) -> Dict[str, Any]: dst = _require_vfs_path(args.get("dst"), "dst") if src == "/" or dst == "/": raise ValueError("invalid_path") + await require_paths([(src, PathAction.READ), (dst, PathAction.WRITE)]) overwrite = bool(args.get("overwrite") or False) return await VirtualFSService.copy(src, dst, overwrite) @@ -110,6 +122,7 @@ async def _vfs_rename(args: Dict[str, Any]) -> Dict[str, Any]: dst = _require_vfs_path(args.get("dst"), "dst") if src == "/" or dst == "/": raise ValueError("invalid_path") + await require_paths([(src, PathAction.READ), (src, PathAction.DELETE), (dst, PathAction.WRITE)]) overwrite = bool(args.get("overwrite") or False) return await VirtualFSService.rename(src, dst, overwrite) @@ -122,7 +135,14 @@ async def _vfs_search(args: Dict[str, Any]) -> Dict[str, Any]: top_k = int(args.get("top_k") or 10) page = int(args.get("page") or 1) page_size = int(args.get("page_size") or 10) - return await VirtualFSSearchService.search(q, top_k, mode, page, page_size) + result = await VirtualFSSearchService.search(q, top_k, mode, page, page_size) + from domain.permission.service import PermissionService + allowed = await PermissionService.filter_paths_by_permission( + current_user_id(), [str(item.path) for item in result.get("items", [])], PathAction.READ + ) + allowed_set = set(allowed) + result["items"] = [item for item in result.get("items", []) if str(item.path) in allowed_set] + return result TOOLS: Dict[str, ToolSpec] = { @@ -133,10 +153,11 @@ TOOLS: Dict[str, ToolSpec] = { "type": "object", "properties": { "path": {"type": "string", "description": "目录路径（绝对路径，如 /foo/bar）"}, - "page": {"type": "integer", "description": "页码（从 1 开始）"}, - "page_size": {"type": "integer", "description": "每页条数"}, - "sort_by": {"type": "string", "description": "排序字段：name/size/mtime"}, - "sort_order": {"type": "string", "description": "排序顺序：asc/desc"}, + "page": {"type": "integer", "minimum": 1, "description": "页码（从 1 开始）"}, + "page_size": {"type": "integer", "minimum": 1, "maximum": 100, "description": "每页条数"}, + "sort_by": {"type": "string", "enum": ["name", "size", "mtime"], "description": "排序字段：name/size/mtime"}, + "sort_order": {"type": "string", "enum": ["asc", "desc"], "description": "排序顺序：asc/desc"}, + "cursor": {"type": "string", "description": "游标分页的下一页游标"}, }, "required": ["path"], "additionalProperties": False, @@ -166,7 +187,7 @@ TOOLS: Dict[str, ToolSpec] = { "properties": { "path": {"type": "string", "description": "文件路径（绝对路径，如 /foo/bar.md）"}, "encoding": {"type": "string", "description": "文本编码（默认 utf-8）"}, - "max_chars": {"type": "integer", "description": "最多返回的字符数（默认 8000）"}, + "max_chars": {"type": "integer", "minimum": 1, "maximum": 100000, "description": "最多返回的字符数（默认 8000）"}, }, "required": ["path"], "additionalProperties": False, @@ -273,10 +294,10 @@ TOOLS: Dict[str, ToolSpec] = { "type": "object", "properties": { "q": {"type": "string", "description": "搜索关键词"}, - "mode": {"type": "string", "description": "搜索模式：vector/filename（默认 vector）"}, - "top_k": {"type": "integer", "description": "返回数量（vector 模式使用，默认 10）"}, - "page": {"type": "integer", "description": "页码（filename 模式使用，默认 1）"}, - "page_size": {"type": "integer", "description": "分页大小（filename 模式使用，默认 10）"}, + "mode": {"type": "string", "enum": ["vector", "filename"], "description": "搜索模式：vector/filename（默认 vector）"}, + "top_k": {"type": "integer", "minimum": 1, "maximum": 100, "description": "返回数量（vector 模式使用，默认 10）"}, + "page": {"type": "integer", "minimum": 1, "description": "页码（filename 模式使用，默认 1）"}, + "page_size": {"type": "integer", "minimum": 1, "maximum": 100, "description": "分页大小（filename 模式使用，默认 10）"}, }, "required": ["q"], "additionalProperties": False, diff --git a/domain/agent/tools/web_fetch.py b/domain/agent/tools/web_fetch.py index 6f8374a..f233d77 100644 --- a/domain/agent/tools/web_fetch.py +++ b/domain/agent/tools/web_fetch.py @@ -5,6 +5,7 @@ from urllib.parse import urljoin import httpx from .base import ToolSpec +from domain.permission.execution import ExecutionError class _HtmlTextExtractor(HTMLParser): @@ -99,10 +100,19 @@ async def _web_fetch(args: Dict[str, Any]) -> Dict[str, Any]: request_kwargs["content"] = str(body) async with httpx.AsyncClient(timeout=20.0, follow_redirects=True) as client: - resp = await client.request(method, url, **request_kwargs) + async with client.stream(method, url, **request_kwargs) as resp: + chunks = [] + size = 0 + async for chunk in resp.aiter_bytes(): + size += len(chunk) + if size > 2 * 1024 * 1024: + raise ExecutionError("response_too_large") + chunks.append(chunk) + await resp.aclose() + body_bytes = b"".join(chunks) content_type = resp.headers.get("content-type") or "" - text = resp.text or "" + text = body_bytes.decode(resp.encoding or "utf-8", errors="replace") is_html = "html" in content_type.lower() if not is_html: probe = text.lstrip()[:200].lower() @@ -167,7 +177,7 @@ TOOLS: Dict[str, ToolSpec] = { "type": "object", "properties": { "url": {"type": "string", "description": "目标 URL"}, - "method": {"type": "string", "description": "请求方法（默认 GET）"}, + "method": {"type": "string", "description": "请求方法（默认 GET，所有方法免审批）"}, "headers": {"type": "object", "description": "请求头", "additionalProperties": {"type": "string"}}, "params": {"type": "object", "description": "查询参数", "additionalProperties": {"type": "string"}}, "json": {"type": "object", "description": "JSON 请求体"}, diff --git a/domain/agent/types.py b/domain/agent/types.py index 0e695c7..7c4114c 100644 --- a/domain/agent/types.py +++ b/domain/agent/types.py @@ -10,6 +10,7 @@ class AgentChatContext(BaseModel): class AgentChatRequest(BaseModel): messages: List[Dict[str, Any]] = Field(default_factory=list) auto_execute: bool = False + approval_batch_id: Optional[str] = None approved_mcp_call_ids: List[str] = Field(default_factory=list) rejected_mcp_call_ids: List[str] = Field(default_factory=list) context: Optional[AgentChatContext] = None diff --git a/domain/config/service.py b/domain/config/service.py index 97fbc2d..3925ee1 100644 --- a/domain/config/service.py +++ b/domain/config/service.py @@ -92,7 +92,7 @@ class ConfigService: if current_time - cache["timestamp"] < 3600 and cache["data"]: return cache["data"] try: - async with httpx.AsyncClient(timeout=10.0) as client: + async with httpx.AsyncClient(timeout=10.0, trust_env=False) as client: resp = await client.get( "https://api.github.com/repos/DrizzleTime/Foxel/releases/latest", follow_redirects=True, @@ -106,7 +106,7 @@ class ConfigService: cache["timestamp"] = current_time cache["data"] = version_info return version_info - except httpx.RequestError: + except httpx.HTTPError: if cache["data"]: return cache["data"] return LatestVersionInfo() diff --git a/domain/notices/service.py b/domain/notices/service.py index f27525b..e4297d5 100644 --- a/domain/notices/service.py +++ b/domain/notices/service.py @@ -113,7 +113,9 @@ class NoticeService: async def _fetch_remote_notices(cls) -> list[dict[str, Any]]: results: list[dict[str, Any]] = [] page = 1 - async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client: + # The notice endpoint is public and should not inherit malformed local + # proxy variables (for example, ``ALL_PROXY=socks://...``). + async with httpx.AsyncClient(timeout=15.0, follow_redirects=True, trust_env=False) as client: while True: resp = await client.get( REMOTE_NOTICES_URL, diff --git a/domain/permission/execution.py b/domain/permission/execution.py new file mode 100644 index 0000000..0c218c0 --- /dev/null +++ b/domain/permission/execution.py @@ -0,0 +1,93 @@ +"""Request-local authority propagated only to MCP-originated background tasks.""" + +from contextlib import contextmanager +from contextvars import ContextVar +from copy import deepcopy +from typing import Any + +from .service import PermissionService +from models.database import UserAccount + + +class ExecutionError(Exception): + def __init__(self, code: str): + self.code = code + super().__init__(code) + + +_scope: ContextVar[dict[str, Any] | None] = ContextVar("foxel_execution_scope", default=None) + + +def normalize_path(value: str) -> str: + if not isinstance(value, str) or not value.strip() or "\x00" in value: + raise ExecutionError("invalid_arguments") + parts = value.strip().replace("\\", "/").split("/") + if any(part in {".", ".."} for part in parts): + raise ExecutionError("invalid_arguments") + return "/" + "/".join(part for part in parts if part) + + +@contextmanager +def execution_context(scope: dict[str, Any] | None): + token = _scope.set(scope) + try: + yield + finally: + _scope.reset(token) + + +def execution_scope() -> dict[str, Any] | None: + return deepcopy(_scope.get()) + + +def current_user_id() -> int: + scope = _scope.get() + if scope is None: + raise ExecutionError("permission_denied") + return scope["user_id"] + + +async def require_path(path: str, action: str) -> str: + path = normalize_path(path) + await require_paths([(path, action)]) + return path + + +async def require_paths(paths: list[tuple[str, str]]) -> None: + user_id = current_user_id() + if not await UserAccount.filter(id=user_id, disabled=False).exists(): + raise ExecutionError("permission_denied") + PermissionService.clear_cache(user_id) + for path, action in paths: + if not await PermissionService.check_path_permission(user_id, normalize_path(path), action): + raise ExecutionError("permission_denied") + + +async def guard_path(path: str, action: str) -> None: + scope = _scope.get() + if scope is None: + return + path = normalize_path(path) + permissions = scope.get("permissions") + if permissions is not None and [path, action] not in permissions: + raise ExecutionError("permission_scope_changed") + await validate_scope() + await require_path(path, action) + + +async def validate_scope() -> None: + scope = _scope.get() + if scope is None or "permissions" not in scope: + return + await require_paths([tuple(item) for item in scope["permissions"]]) + from domain.agent.execution import snapshot_tree + for root, expected in scope.get("trees", {}).items(): + actual = await snapshot_tree(root) + if actual != expected: + raise ExecutionError("permission_scope_changed") + + +def forget_tree(root: str) -> None: + scope = _scope.get() + if scope is not None: + scope.get("trees", {}).pop(normalize_path(root), None) diff --git a/domain/processors/service.py b/domain/processors/service.py index a3061b7..0bb99cd 100644 --- a/domain/processors/service.py +++ b/domain/processors/service.py @@ -17,6 +17,8 @@ from .types import ( ProcessRequest, UpdateSourceRequest, ) +from domain.permission.execution import guard_path, validate_scope, execution_scope +from domain.permission.types import PathAction class ProcessorService: @@ -91,6 +93,7 @@ class ProcessorService: else: overwrite = False suffix = None + payload = { "path": req.path, "processor_type": req.processor_type, @@ -104,6 +107,7 @@ class ProcessorService: @classmethod async def scan_directory(cls, req: ProcessDirectoryRequest): + await validate_scope() if req.max_depth is not None and req.max_depth < 0: raise HTTPException(400, detail="max_depth must be >= 0") @@ -130,6 +134,34 @@ class ProcessorService: overwrite = False suffix = None + scope = execution_scope() + if scope is not None: + snapshot = scope["trees"].get(req.path) + if snapshot is None: + raise HTTPException(400, detail="permission_scope_unverifiable") + supported = {str(ext).lower().lstrip(".") for ext in schema.get("supported_exts", [])} + scheduled = 0 + for source, is_dir in snapshot.items(): + if is_dir: + continue + relative = Path(source).relative_to(req.path) + if req.max_depth is not None and len(relative.parts) - 1 > req.max_depth: + continue + extension = Path(source).suffix.lower().lstrip(".") + if supported and extension not in supported: + continue + await guard_path(source, PathAction.READ) + output = None + if produces_file and not overwrite: + p = Path(source) + output = str(p.with_name(p.stem + suffix + p.suffix)) + await task_queue_service.add_task("process_file", { + "path": source, "processor_type": req.processor_type, "config": req.config, + "save_to": output, "overwrite": overwrite, + }) + scheduled += 1 + return {"scheduled": scheduled} + supported_exts = schema.get("supported_exts") or [] allowed_exts = { ext.lower().lstrip('.') @@ -196,6 +228,7 @@ class ProcessorService: if not matches_extension(child_rel): continue absolute_path = build_absolute_path(adapter_model.path, child_rel) + await guard_path(absolute_path, PathAction.READ) save_to = None if produces_file and not overwrite and suffix: save_to = apply_suffix(absolute_path, suffix) diff --git a/domain/tasks/task_queue.py b/domain/tasks/task_queue.py index 9d36886..ae5d280 100644 --- a/domain/tasks/task_queue.py +++ b/domain/tasks/task_queue.py @@ -3,6 +3,7 @@ from typing import Dict, Any from pydantic import BaseModel, Field import uuid from enum import Enum +from domain.permission.execution import execution_context, execution_scope, validate_scope class TaskStatus(str, Enum): @@ -43,6 +44,14 @@ class TaskQueueService: self._worker_seq: int = 0 async def add_task(self, name: str, task_info: Dict[str, Any]) -> Task: + scope = execution_scope() + if scope is not None: + if name == "process_file" and any( + tree.get(task_info.get("path")) is False for tree in scope.get("trees", {}).values() + ): + path = task_info["path"] + scope["trees"] = {path: {path: False}} + task_info = {**task_info, "_execution_scope": scope} task = Task(name=name, task_info=task_info) self._tasks[task.id] = task await self._queue.put(task) @@ -70,9 +79,15 @@ class TaskQueueService: task.meta = (task.meta or {}) | meta async def _execute_task(self, task: Task): + scope = task.task_info.get("_execution_scope") + with execution_context(scope): + await self._execute_scoped_task(task) + + async def _execute_scoped_task(self, task: Task): task.status = TaskStatus.RUNNING try: + await validate_scope() # Local import to avoid circular dependency during module load. from domain.virtual_fs import VirtualFSService diff --git a/domain/virtual_fs/file_ops.py b/domain/virtual_fs/file_ops.py index 9292125..d288229 100644 --- a/domain/virtual_fs/file_ops.py +++ b/domain/virtual_fs/file_ops.py @@ -5,6 +5,8 @@ from fastapi import HTTPException from fastapi.responses import Response from domain.tasks import TaskService +from domain.permission.execution import guard_path, validate_scope, forget_tree +from domain.permission.types import PathAction from .thumbnail import is_raw_filename, raw_bytes_to_jpeg from .listing import VirtualFSListingMixin @@ -36,6 +38,7 @@ class VirtualFSFileOpsMixin(VirtualFSListingMixin): @classmethod async def read_file(cls, path: str) -> Union[bytes, Any]: + await guard_path(path, PathAction.READ) adapter_instance, _, root, rel = await cls.resolve_adapter_and_rel(path) if rel.endswith("/") or rel == "": raise HTTPException(400, detail="Path is a directory") @@ -44,6 +47,7 @@ class VirtualFSFileOpsMixin(VirtualFSListingMixin): @classmethod async def write_file(cls, path: str, data: bytes): + await guard_path(path, PathAction.WRITE) adapter_instance, adapter_model, root, rel = await cls.resolve_adapter_and_rel(path) if rel.endswith("/"): raise HTTPException(400, detail="Invalid file path") @@ -55,6 +59,7 @@ class VirtualFSFileOpsMixin(VirtualFSListingMixin): @classmethod async def write_file_stream(cls, path: str, data_iter: AsyncIterator[bytes], overwrite: bool = True): + await guard_path(path, PathAction.WRITE) adapter_instance, adapter_model, root, rel = await cls.resolve_adapter_and_rel(path) if rel.endswith("/"): raise HTTPException(400, detail="Invalid file path") @@ -91,6 +96,7 @@ class VirtualFSFileOpsMixin(VirtualFSListingMixin): @classmethod async def make_dir(cls, path: str): + await guard_path(path, PathAction.WRITE) adapter_instance, _, root, rel = await cls.resolve_adapter_and_rel(path) if not rel: return @@ -99,11 +105,14 @@ class VirtualFSFileOpsMixin(VirtualFSListingMixin): @classmethod async def delete_path(cls, path: str): + await validate_scope() + await guard_path(path, PathAction.DELETE) adapter_instance, _, root, rel = await cls.resolve_adapter_and_rel(path) if not rel: raise HTTPException(400, detail="Cannot delete root") delete_func = await cls._ensure_method(adapter_instance, "delete") await delete_func(root, rel) + forget_tree(path) await TaskService.trigger_tasks("file_deleted", path) @classmethod diff --git a/domain/virtual_fs/processing.py b/domain/virtual_fs/processing.py index 42fa41a..a6afe8b 100644 --- a/domain/virtual_fs/processing.py +++ b/domain/virtual_fs/processing.py @@ -4,6 +4,8 @@ from fastapi import HTTPException from fastapi.responses import Response from .transfer import VirtualFSTransferMixin +from domain.permission.execution import guard_path +from domain.permission.types import PathAction class VirtualFSProcessingMixin(VirtualFSTransferMixin): @@ -18,6 +20,8 @@ class VirtualFSProcessingMixin(VirtualFSTransferMixin): ) -> Any: from domain.processors import get_processor + await guard_path(path, PathAction.READ) + processor = get_processor(processor_type) if not processor: raise HTTPException(400, detail=f"Processor {processor_type} not found") @@ -81,6 +85,7 @@ class VirtualFSProcessingMixin(VirtualFSTransferMixin): if not matches_extension(child_rel): continue absolute_path = cls._build_absolute_path(adapter_model.path, child_rel) + await guard_path(absolute_path, PathAction.READ) data = b"" if requires_input_bytes: data = await cls.read_file(absolute_path) diff --git a/domain/virtual_fs/transfer.py b/domain/virtual_fs/transfer.py index 5d7cc85..368280d 100644 --- a/domain/virtual_fs/transfer.py +++ b/domain/virtual_fs/transfer.py @@ -6,6 +6,8 @@ import aiofiles from fastapi import HTTPException from .file_ops import VirtualFSFileOpsMixin +from domain.permission.execution import guard_path, validate_scope +from domain.permission.types import PathAction class VirtualFSTransferMixin(VirtualFSFileOpsMixin): @@ -13,6 +15,7 @@ class VirtualFSTransferMixin(VirtualFSFileOpsMixin): async def move_path( cls, src: str, dst: str, overwrite: bool = False, return_debug: bool = True, allow_cross: bool = False ): + await validate_scope() adapter_s, adapter_model_s, root_s, rel_s = await cls.resolve_adapter_and_rel(src) adapter_d, adapter_model_d, root_d, rel_d = await cls.resolve_adapter_and_rel(dst) debug_info = { @@ -95,6 +98,7 @@ class VirtualFSTransferMixin(VirtualFSFileOpsMixin): @classmethod async def rename_path(cls, src: str, dst: str, overwrite: bool = False, return_debug: bool = True): + await validate_scope() adapter_s, adapter_model_s, root_s, rel_s = await cls.resolve_adapter_and_rel(src) adapter_d, adapter_model_d, root_d, rel_d = await cls.resolve_adapter_and_rel(dst) debug_info = { @@ -167,6 +171,7 @@ class VirtualFSTransferMixin(VirtualFSFileOpsMixin): async def copy_path( cls, src: str, dst: str, overwrite: bool = False, return_debug: bool = True, allow_cross: bool = False ): + await validate_scope() adapter_s, adapter_model_s, root_s, rel_s = await cls.resolve_adapter_and_rel(src) adapter_d, adapter_model_d, root_d, rel_d = await cls.resolve_adapter_and_rel(dst) debug_info = { @@ -461,9 +466,15 @@ class VirtualFSTransferMixin(VirtualFSFileOpsMixin): async def ensure_dir(rel_path: str): if not rel_path or rel_path in ensured_dirs: return + absolute = cls._build_absolute_path(adapter_model_d.path, rel_path) + exists = getattr(adapter_d, "exists", None) + if callable(exists) and await exists(root_d, rel_path): + ensured_dirs.add(rel_path) + return parent = cls._parent_rel(rel_path) if parent: await ensure_dir(parent) + await guard_path(absolute, PathAction.WRITE) try: await mkdir_func(root_d, rel_path) except FileExistsError: diff --git a/models/database.py b/models/database.py index 73c0507..25b970f 100644 --- a/models/database.py +++ b/models/database.py @@ -180,6 +180,41 @@ class AIDefaultModel(Model): table = "ai_default_models" +class AgentApprovalBatch(Model): + id = fields.CharField(max_length=36, pk=True) + user: fields.ForeignKeyRelation[UserAccount] = fields.ForeignKeyField( + "models.UserAccount", related_name="agent_approvals", on_delete=fields.CASCADE + ) + assistant_message = fields.JSONField() + continuation_status = fields.CharField(max_length=20, default="pending") + continuation_result = fields.JSONField(null=True) + executing = fields.BooleanField(default=False) + expires_at = fields.DatetimeField() + created_at = fields.DatetimeField(auto_now_add=True) + + class Meta: + table = "agent_approval_batches" + + +class AgentApprovalCall(Model): + id = fields.IntField(pk=True) + batch: fields.ForeignKeyRelation[AgentApprovalBatch] = fields.ForeignKeyField( + "models.AgentApprovalBatch", related_name="calls", on_delete=fields.CASCADE + ) + call_id = fields.CharField(max_length=255) + position = fields.IntField() + name = fields.CharField(max_length=100) + arguments = fields.JSONField() + requires_confirmation = fields.BooleanField(default=True) + status = fields.CharField(max_length=20, default="pending") + result = fields.TextField(null=True) + updated_at = fields.DatetimeField(auto_now=True) + + class Meta: + table = "agent_approval_calls" + unique_together = (("batch", "call_id"),) + + class AutomationTask(Model): id = fields.IntField(pk=True) name = fields.CharField(max_length=100) diff --git a/tests/test_agent_boundaries.py b/tests/test_agent_boundaries.py new file mode 100644 index 0000000..71e28d1 --- /dev/null +++ b/tests/test_agent_boundaries.py @@ -0,0 +1,488 @@ +import asyncio +import json +import unittest +import tempfile +from pathlib import Path +from contextlib import asynccontextmanager +from datetime import timedelta +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import httpx +from fastapi import FastAPI, HTTPException +from tortoise import Tortoise + +from api.response import cursor_page, page +from domain.agent.api import router +from domain.agent.approvals import create_batch, execute_call, load_batch, now +from domain.agent.execution import execute_tool, snapshot_tree, validate_arguments +from domain.agent.mcp import MCP_HTTP_APP, mcp_client_session +from domain.agent.service import AgentService, _choose_chat_ability +from domain.agent.tools import mcp_tool_descriptors +from domain.agent.tools.base import tool_error, tool_result_to_content +from domain.agent.types import AgentChatRequest +from domain.ai import MissingModelError +from domain.auth import User, get_current_active_user +from domain.permission.execution import ExecutionError, execution_context, guard_path, validate_scope +from domain.permission.service import PermissionService +from domain.tasks.task_queue import TaskQueueService, TaskStatus +from domain.virtual_fs import VirtualFSService +from models.database import AgentApprovalBatch, AgentApprovalCall, PathRule, Role, StorageAdapter, UserAccount, UserRole + + +class DatabaseTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + await Tortoise.init(db_url="sqlite://:memory:", modules={"models": ["models.database"]}, _enable_global_fallback=True) + await Tortoise.generate_schemas() + self.account = await UserAccount.create(username="member", hashed_password="unused") + self.user = User(id=self.account.id, username="member") + self.admin_account = await UserAccount.create(username="admin", hashed_password="unused", is_admin=True) + self.admin = User(id=self.admin_account.id, username="admin", is_admin=True) + self.role = await Role.create(name="member") + await UserRole.create(user_id=self.user.id, role_id=self.role.id) + await PathRule.create(role_id=self.role.id, path_pattern="/allowed/**", can_read=True, can_write=True, can_delete=True) + await PathRule.create(role_id=self.role.id, path_pattern="/allowed/secret/**", can_read=False, can_write=False, can_delete=False, priority=10) + PermissionService.clear_cache() + + async def asyncTearDown(self): + PermissionService.clear_cache() + await Tortoise.close_connections() + + def assert_error(self, result, code): + self.assertFalse(result["ok"], result) + self.assertEqual(result["error"]["code"], code, result) + + +class BoundaryTests(DatabaseTests): + async def test_real_processor_scan_uses_snapshot_and_child_scopes(self): + from domain.adapters import runtime_registry + from domain.adapters.providers.local import LocalAdapter + from domain.processors import TYPE_MAP, ProcessorService + from domain.tasks import task_queue_service + processor = SimpleNamespace(produces_file=True, supported_exts=["jpg"], supports_directory=False, requires_input_bytes=True, + process=AsyncMock(return_value=b"processed")) + with tempfile.TemporaryDirectory() as temporary: + mount = await StorageAdapter.create(name="local", type="local", path="/allowed", config={"root": temporary}) + adapter = LocalAdapter(mount) + root = Path(temporary) + (root / "photos").mkdir() + (root / "photos/a.jpg").write_bytes(b"a") + (root / "photos/b.jpg").write_bytes(b"b") + queue = TaskQueueService() + with patch.object(runtime_registry, "get", return_value=adapter), patch.dict(TYPE_MAP, {"image_watermark": lambda: processor}), patch.object(task_queue_service, "add_task", new=queue.add_task): + result = await execute_tool("processors_run", {"path": "/allowed/photos", "processor_type": "image_watermark", "config": {"text": "x"}, "max_depth": 0, "overwrite": False, "suffix": "_out"}, self.user) + self.assertTrue(result["ok"], result) + scan = queue.get_task(result["data"]["task_id"]) + await queue._execute_task(scan) + self.assertEqual(scan.status, TaskStatus.SUCCESS, scan.error) + children = [task for task in queue.get_all_tasks() if task.name == "process_file"] + self.assertEqual(len(children), 2) + for child in children: + self.assertEqual(list(child.task_info["_execution_scope"]["trees"]), [child.task_info["path"]]) + await queue._execute_task(child) + self.assertEqual(child.status, TaskStatus.SUCCESS, child.error) + self.assertEqual((root / "photos/a_out.jpg").read_bytes(), b"processed") + self.assertEqual((root / "photos/b_out.jpg").read_bytes(), b"processed") + + async def test_real_local_recursive_delete_and_copy(self): + from domain.adapters import runtime_registry + from domain.adapters.providers.local import LocalAdapter + with tempfile.TemporaryDirectory() as temporary: + mount = await StorageAdapter.create(name="local", type="local", path="/allowed", config={"root": temporary}) + adapter = LocalAdapter(mount) + with patch.object(runtime_registry, "get", return_value=adapter): + root = Path(temporary) + (root / "source/sub").mkdir(parents=True) + (root / "source/sub/a.txt").write_text("hello") + result = await execute_tool("vfs_copy", {"src": "/allowed/source", "dst": "/allowed/copied"}, self.user) + self.assertTrue(result["ok"], result) + self.assertEqual((root / "copied/sub/a.txt").read_text(), "hello") + result = await execute_tool("vfs_delete", {"path": "/allowed/copied"}, self.user) + self.assertTrue(result["ok"], result) + self.assertFalse((root / "copied").exists()) + (root / "source/secret").mkdir() + (root / "source/secret/a.txt").write_text("private") + await PathRule.create(role_id=self.role.id, path_pattern="/allowed/source/secret/**", can_read=False, can_write=False, can_delete=False, priority=20) + denied = await execute_tool("vfs_delete", {"path": "/allowed/source"}, self.user) + self.assert_error(denied, "permission_denied") + self.assertTrue((root / "source/sub/a.txt").exists()) + + async def test_real_cross_mount_task_preserves_scope(self): + from domain.adapters import runtime_registry + from domain.adapters.providers.local import LocalAdapter + from domain.tasks import task_queue_service + with tempfile.TemporaryDirectory() as left, tempfile.TemporaryDirectory() as right, tempfile.TemporaryDirectory() as staging: + source_mount = await StorageAdapter.create(name="source", type="local", path="/allowed/source", config={"root": left}) + target_mount = await StorageAdapter.create(name="target", type="local", path="/allowed/target", config={"root": right}) + source_adapter, target_adapter = LocalAdapter(source_mount), LocalAdapter(target_mount) + (Path(left) / "folder").mkdir() + (Path(left) / "folder/a.txt").write_text("cross mount") + (Path(right) / "out").mkdir() + (Path(right) / "out/old.txt").write_text("old") + queue = TaskQueueService() + instances = {source_mount.id: source_adapter, target_mount.id: target_adapter} + with patch.object(runtime_registry, "get", side_effect=instances.get), patch.object(task_queue_service, "add_task", new=queue.add_task), patch.object(VirtualFSService, "CROSS_TRANSFER_TEMP_ROOT", Path(staging)): + result = await execute_tool("vfs_move", {"src": "/allowed/source/folder", "dst": "/allowed/target/out", "overwrite": True}, self.user) + self.assertTrue(result["ok"], result) + task = queue.get_task(result["data"]["task_id"]) + self.assertEqual(task.task_info["_execution_scope"]["user_id"], self.user.id) + await queue._execute_task(task) + self.assertEqual(task.status, TaskStatus.SUCCESS, task.error) + self.assertEqual((Path(right) / "out/a.txt").read_text(), "cross mount") + self.assertFalse((Path(right) / "out/old.txt").exists()) + self.assertFalse((Path(left) / "folder").exists()) + + + async def test_all_file_tools_deny_before_storage_access(self): + cases = [ + ("vfs_list_dir", {"path": "/private"}), + ("vfs_stat", {"path": "/private"}), + ("vfs_read_text", {"path": "/private"}), + ("vfs_write_text", {"path": "/private", "content": "x"}), + ("vfs_mkdir", {"path": "/private"}), + ("vfs_delete", {"path": "/private"}), + ("vfs_copy", {"src": "/private", "dst": "/allowed/out"}), + ("vfs_move", {"src": "/private", "dst": "/allowed/out"}), + ("vfs_rename", {"src": "/private", "dst": "/allowed/out"}), + ("processors_run", {"path": "/private", "processor_type": "image_watermark"}), + ] + with patch.object(VirtualFSService, "stat", AsyncMock()) as stat: + for name, args in cases: + with self.subTest(tool=name): + self.assert_error(await execute_tool(name, args, self.user), "permission_denied") + stat.assert_not_awaited() + + async def test_admin_can_write_and_paths_are_canonical(self): + with patch.object(VirtualFSService, "write_file", AsyncMock()) as written: + result = await execute_tool("vfs_write_text", {"path": "//allowed\\docs//a.txt/", "content": "hello"}, self.admin) + self.assertTrue(result["ok"], result) + written.assert_awaited_once_with("/allowed/docs/a.txt", b"hello") + + async def test_no_identity_and_disabled_identity_denied(self): + self.assert_error(await execute_tool("time", {}, None), "permission_denied") + self.assert_error(await execute_tool("time", {}, self.user.model_copy(update={"disabled": True})), "permission_denied") + + async def test_invalid_inputs_do_not_execute(self): + cases = [ + ("vfs_stat", {"path": "/allowed/../private"}), + ("vfs_stat", {"path": "/allowed/./a"}), + ("vfs_stat", {"path": "/allowed/\x00a"}), + ("vfs_read_text", {"path": "/allowed/a", "max_chars": 0}), + ("vfs_read_text", {"path": "/allowed/a", "max_chars": 100001}), + ("vfs_read_text", {"path": "/allowed/a", "encoding": "bad-encoding"}), + ("vfs_list_dir", {"path": "/allowed", "page_size": 101}), + ("vfs_list_dir", {"path": "/allowed", "page": True}), + ("vfs_list_dir", {"path": "/allowed", "sort_by": "invalid"}), + ("vfs_search", {"q": "x", "mode": "unknown"}), + ("vfs_search", {"q": "x", "top_k": 101}), + ("processors_run", {"path": "/allowed", "processor_type": "x", "max_depth": -1}), + ("web_fetch", {"url": "file:///etc/passwd"}), + ("web_fetch", {"url": "https://example.com", "method": "TRACE"}), + ("vfs_write_text", {"path": "/allowed/a", "content": "x", "user_id": self.admin.id}), + ] + for name, args in cases: + with self.subTest(tool=name, args=args): + self.assert_error(await execute_tool(name, args, self.user), "invalid_arguments") + + async def test_directory_listing_filters_children_and_preserves_cursor(self): + listing = cursor_page([{"name": "ok", "is_dir": False}, {"name": "secret", "is_dir": True}], 50, next_cursor="next") + with patch.object(VirtualFSService, "list_virtual_dir", AsyncMock(return_value=listing)) as listed: + result = await execute_tool("vfs_list_dir", {"path": "/allowed", "cursor": "start"}, self.user) + self.assertEqual([item["name"] for item in result["data"]["entries"]], ["ok"]) + self.assertEqual(result["data"]["pagination"]["next_cursor"], "next") + self.assertEqual(listed.await_args.args[-1], "start") + + async def test_search_filters_private_snippets(self): + from domain.virtual_fs.types import SearchResultItem + from domain.virtual_fs.search import VirtualFSSearchService + items = [SearchResultItem(id="1", path="/allowed/a", score=1, snippet="public"), + SearchResultItem(id="2", path="/private/a", score=1, snippet="private-secret")] + with patch.object(VirtualFSSearchService, "search", AsyncMock(return_value={"items": items})): + result = await execute_tool("vfs_search", {"q": "x"}, self.user) + self.assertEqual(result["data"]["items"][0]["path"], "/allowed/a") + self.assertNotIn("private-secret", json.dumps(result)) + + async def test_recursive_denial_has_zero_mutations(self): + tree = {"/allowed": True, "/allowed/ok": False, "/allowed/secret": True, "/allowed/secret/a": False} + with patch("domain.agent.execution.snapshot_tree", AsyncMock(return_value=tree)), patch.object(VirtualFSService, "delete", AsyncMock()) as deleted: + result = await execute_tool("vfs_delete", {"path": "/allowed"}, self.user) + self.assert_error(result, "permission_denied") + deleted.assert_not_awaited() + + async def test_transfer_checks_source_read_delete_and_target_children(self): + source = {"/allowed/from": True, "/allowed/from/a": False} + with patch("domain.agent.execution.snapshot_tree", AsyncMock(return_value=source)), patch.object(VirtualFSService, "stat", AsyncMock(return_value={"is_dir": True})), patch.object(VirtualFSService, "copy", AsyncMock()) as copied: + result = await execute_tool("vfs_copy", {"src": "/allowed/from", "dst": "/allowed/secret/out"}, self.user) + self.assert_error(result, "permission_denied") + copied.assert_not_awaited() + + async def test_snapshot_supports_cursor_and_rejects_repeated_cursor(self): + with patch.object(VirtualFSService, "stat", AsyncMock(return_value={"is_dir": True})), patch.object(VirtualFSService, "list_virtual_dir", AsyncMock(side_effect=[cursor_page([{"name": "a"}], 100, next_cursor="x"), cursor_page([{"name": "b"}], 100)])): + tree = await snapshot_tree("/allowed") + self.assertEqual(set(tree), {"/allowed", "/allowed/a", "/allowed/b"}) + with patch.object(VirtualFSService, "stat", AsyncMock(return_value={"is_dir": True})), patch.object(VirtualFSService, "list_virtual_dir", AsyncMock(return_value=cursor_page([], 100, next_cursor="x"))): + with self.assertRaisesRegex(ExecutionError, "permission_scope_unverifiable"): + await snapshot_tree("/allowed") + + async def test_snapshot_unknown_total_fails_closed(self): + with patch.object(VirtualFSService, "stat", AsyncMock(return_value={"is_dir": True})), patch.object(VirtualFSService, "list_virtual_dir", AsyncMock(return_value={"items": [], "total": None})): + with self.assertRaisesRegex(ExecutionError, "permission_scope_unverifiable"): + await snapshot_tree("/allowed") + + async def test_new_child_and_permission_revocation_stop_task(self): + scope = {"user_id": self.user.id, "permissions": [["/allowed", "read"]], "trees": {"/allowed": {"/allowed": True}}} + with execution_context(scope), patch("domain.agent.execution.snapshot_tree", AsyncMock(return_value={"/allowed": True, "/allowed/new": False})): + with self.assertRaisesRegex(ExecutionError, "permission_scope_changed"): + await validate_scope() + await UserAccount.filter(id=self.user.id).update(disabled=True) + with execution_context({"user_id": self.user.id, "permissions": [["/allowed", "write"]]}): + with self.assertRaisesRegex(ExecutionError, "permission_denied"): + await guard_path("/allowed", "write") + + async def test_background_scope_and_child_identity_are_preserved(self): + queue = TaskQueueService() + scope = {"user_id": self.user.id, "permissions": [["/allowed/a", "read"]], "trees": {"/allowed/a": {"/allowed/a": False}}} + with execution_context(scope): + task = await queue.add_task("process_file", {"path": "/allowed/a", "processor_type": "vector_index", "config": {}}) + self.assertEqual(task.task_info["_execution_scope"]["user_id"], self.user.id) + scope["permissions"].clear() + self.assertTrue(task.task_info["_execution_scope"]["permissions"]) + with patch("domain.agent.execution.snapshot_tree", AsyncMock(return_value={"/allowed/a": False})), patch.object(VirtualFSService, "process_file", AsyncMock()) as processed: + await UserAccount.filter(id=self.user.id).update(disabled=True) + await queue._execute_task(task) + self.assertEqual(task.status, TaskStatus.FAILED) + processed.assert_not_awaited() + + async def test_processor_requires_output_write_and_valid_config(self): + with patch("domain.agent.execution.snapshot_tree", AsyncMock(return_value={"/allowed/a.jpg": False})), patch.object(VirtualFSService, "path_is_directory", AsyncMock(return_value=False)), patch("domain.agent.tools.processors.ProcessorService.process_file", AsyncMock()) as processed: + self.assert_error(await execute_tool("processors_run", {"path": "/allowed/a.jpg", "processor_type": "image_watermark", "config": {"text": "x"}, "save_to": "/private/out.jpg"}, self.user), "permission_denied") + self.assert_error(await execute_tool("processors_run", {"path": "/allowed/a.jpg", "processor_type": "image_watermark", "config": {}}, self.user), "invalid_arguments") + self.assert_error(await execute_tool("processors_run", {"path": "/allowed/a.jpg", "processor_type": "image_watermark", "config": {"text": "x"}}, self.user), "permission_scope_unverifiable") + processed.assert_not_awaited() + + async def test_web_methods_remain_unconfirmed_with_accurate_annotations(self): + descriptor = next(d for d in mcp_tool_descriptors() if d.name == "web_fetch") + self.assertFalse(descriptor.requires_confirmation) + self.assertFalse(descriptor.annotations["readOnlyHint"]) + self.assertTrue(descriptor.annotations["destructiveHint"]) + self.assertTrue(descriptor.annotations["openWorldHint"]) + transport = httpx.MockTransport(lambda req: httpx.Response(409, text="conflict", request=req)) + original_client = httpx.AsyncClient + with patch("domain.agent.tools.web_fetch.httpx.AsyncClient", lambda **kw: original_client(transport=transport, **kw)): + for method in ["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"]: + result = await execute_tool("web_fetch", {"url": "https://example.com", "method": method}, self.user) + self.assertTrue(result["ok"], result) + self.assertEqual(result["data"]["status_code"], 409) + + async def test_web_response_limit_and_timeout(self): + original_client = httpx.AsyncClient + for handler, code in [(lambda req: httpx.Response(200, content=b"x" * (2 * 1024 * 1024 + 1)), "response_too_large"), + (lambda req: (_ for _ in ()).throw(httpx.ReadTimeout("secret")), "request_timeout")]: + with patch("domain.agent.tools.web_fetch.httpx.AsyncClient", lambda **kw: original_client(transport=httpx.MockTransport(handler), **kw)): + self.assert_error(await execute_tool("web_fetch", {"url": "https://example.com"}, self.user), code) + + async def test_internal_exception_is_redacted(self): + with patch.object(VirtualFSService, "read_file", AsyncMock(side_effect=RuntimeError("secret password /etc/private"))): + result = await execute_tool("vfs_read_text", {"path": "/allowed/a"}, self.user) + self.assert_error(result, "execution_failed") + self.assertNotIn("password", json.dumps(result)) + + async def test_real_mcp_resources_auth_context_and_tools(self): + with patch("domain.agent.mcp.AuthService.create_access_token", AsyncMock(return_value="test-token")), patch("domain.agent.mcp.AuthService.get_current_user", AsyncMock(return_value=self.user)): + async with MCP_HTTP_APP.router.lifespan_context(MCP_HTTP_APP): + async with mcp_client_session(self.user, "//allowed//docs") as session: + tools = await session.list_tools() + self.assertEqual(len(tools.tools), 14) + read_tool = next(tool for tool in tools.tools if tool.name == "vfs_read_text") + bounds = read_tool.input_schema["properties"]["max_chars"] + self.assertIn("100000", json.dumps(bounds)) + invalid = await session.call_tool("vfs_read_text", {"path": "/allowed/a", "max_chars": True}) + self.assertTrue(invalid.is_error) + current = await session.read_resource("foxel://context/current-path") + self.assertEqual(json.loads(current.contents[0].text)["current_path"], "/allowed/docs") + denied = await session.call_tool("vfs_read_text", {"path": "/private"}) + self.assert_error(json.loads(denied.content[0].text), "permission_denied") + denied_resource = await session.read_resource("foxel://vfs/text/private") + self.assert_error(json.loads(denied_resource.contents[0].text), "permission_denied") + + +class ApprovalTests(DatabaseTests): + async def asyncSetUp(self): + await super().asyncSetUp() + self.assistant = {"role": "assistant", "content": "", "mcp_calls": [ + {"id": "one", "name": "vfs_write_text", "arguments": {"path": "/allowed/one", "content": "original"}}, + {"id": "two", "name": "vfs_mkdir", "arguments": {"path": "/allowed/two"}}, + ]} + self.index = {d.name: {"meta": d.meta} for d in mcp_tool_descriptors()} + self.executed = [] + + async def call_tool(name, arguments): + self.executed.append((name, arguments)) + return SimpleNamespace(content=[], structured_content={"ok": True, "data": arguments}) + + @asynccontextmanager + async def session(*args): + yield SimpleNamespace(call_tool=call_tool) + + self.patches = [ + patch("domain.agent.service.mcp_client_session", session), + patch("domain.agent.service._list_mcp_tools", AsyncMock(return_value=[{"name": k, **v} for k, v in self.index.items()])), + patch("domain.agent.service._choose_chat_ability", AsyncMock(return_value="tools")), + ] + for p in self.patches: + p.start() + self.model = AsyncMock(return_value={"role": "assistant", "content": "done"}) + self.model_patch = patch("domain.agent.service.chat_completion", self.model) + self.model_patch.start() + + async def asyncTearDown(self): + self.model_patch.stop() + for p in reversed(self.patches): + p.stop() + await super().asyncTearDown() + + async def batch(self): + return await create_batch(self.user.id, self.assistant, self.index) + + def request(self, batch, approved=(), rejected=(), messages=None): + return AgentChatRequest(messages=messages or [self.assistant], approval_batch_id=batch.id, + approved_mcp_call_ids=list(approved), rejected_mcp_call_ids=list(rejected)) + + async def test_partial_approval_uses_saved_arguments_and_blocks_model(self): + batch = await self.batch() + tampered = json.loads(json.dumps(self.assistant)) + tampered["mcp_calls"][0]["arguments"]["content"] = "tampered" + result = await AgentService.chat(self.request(batch, approved=["one"], messages=[tampered]), self.user) + self.assertEqual(self.executed[0][1]["content"], "original") + self.assertEqual([p["id"] for p in result["pending_mcp_calls"]], ["two"]) + self.assertEqual(result["approval_batch_id"], batch.id) + self.assertTrue(result["replace_messages"]) + self.model.assert_not_awaited() + final = await AgentService.chat(self.request(batch, rejected=["two"], messages=result["messages"]), self.user) + self.assertEqual(final["pending_mcp_calls"], []) + wire = self.model.await_args.args[0] + self.assertEqual([m["mcp_call_id"] for m in wire if m["role"] == "tool"], ["one", "two"]) + self.assertEqual(len(self.executed), 1) + + async def test_all_approvals_execute_in_original_order(self): + batch = await self.batch() + result = await AgentService.chat(self.request(batch, approved=["two", "one"]), self.user) + self.assertEqual([name for name, _ in self.executed], ["vfs_write_text", "vfs_mkdir"]) + self.assertEqual(result["finish_reason"], "completed") + repeated = await AgentService.chat(self.request(batch, approved=["two", "one"]), self.user) + self.assertEqual(repeated, result) + self.assertEqual(len(self.executed), 2) + self.model.assert_awaited_once() + + async def test_concurrent_batch_requests_do_not_reorder_operations(self): + batch = await self.batch() + started, release = asyncio.Event(), asyncio.Event() + + async def execute(name, args): + self.executed.append((name, args)) + started.set() + await release.wait() + return tool_result_to_content({"written": True}) + + async def execute_mcp(session, name, args): + return await execute(name, args) + + with patch("domain.agent.service._execute_mcp_call", new=execute_mcp): + first = asyncio.create_task(AgentService.chat(self.request(batch, approved=["one", "two"]), self.user)) + await asyncio.wait_for(started.wait(), timeout=3) + second = await AgentService.chat(self.request(batch, approved=["two"]), self.user) + self.assertEqual(second["finish_reason"], "operation_in_progress") + self.assertEqual(len(self.executed), 1) + release.set() + await first + self.assertEqual([name for name, _ in self.executed], ["vfs_write_text", "vfs_mkdir"]) + + async def test_unknown_cross_user_conflicting_and_legacy_ids_rejected(self): + batch = await self.batch() + for req, user in [(self.request(batch, approved=["unknown"]), self.user), + (self.request(batch, approved=["one"], rejected=["one"]), self.user), + (self.request(batch, approved=["one"]), self.admin), + (AgentChatRequest(messages=[self.assistant], approved_mcp_call_ids=["one"]), self.user)]: + with self.subTest(req=req): + with self.assertRaises(HTTPException) as error: + await AgentService.chat(req, user) + self.assertEqual(error.exception.status_code, 400) + self.assertFalse(self.executed) + + async def test_expired_calls_require_regeneration(self): + batch = await self.batch() + await AgentApprovalBatch.filter(id=batch.id).update(expires_at=now() - timedelta(seconds=1)) + with self.assertRaisesRegex(HTTPException, "approval_expired"): + await AgentService.chat(self.request(batch, approved=["one"]), self.user) + self.assertEqual(await AgentApprovalCall.filter(batch_id=batch.id, status="expired").count(), 2) + self.assertFalse(self.executed) + + async def test_concurrent_claim_and_cancelled_write_never_retry(self): + batch = await self.batch() + record = await AgentApprovalCall.get(batch_id=batch.id, call_id="one") + started, release = asyncio.Event(), asyncio.Event() + execute = AsyncMock() + + async def blocking(*args): + started.set() + await release.wait() + return tool_result_to_content({"written": True}) + + execute.side_effect = blocking + first = asyncio.create_task(execute_call(record, execute)) + await started.wait() + second = await execute_call(record, execute) + self.assertEqual(second.status, "running") + first.cancel() + with self.assertRaises(asyncio.CancelledError): + await first + third = await execute_call(record, execute) + self.assertEqual(third.status, "running") + execute.assert_awaited_once() + + async def test_running_operation_keeps_pending_and_never_calls_model(self): + batch = await self.batch() + await AgentApprovalCall.filter(batch_id=batch.id, call_id="one").update(status="running") + result = await AgentService.chat(self.request(batch, approved=["one", "two"]), self.user) + self.assertEqual(result["finish_reason"], "operation_in_progress") + self.assertEqual(len(result["pending_mcp_calls"]), 2) + self.model.assert_not_awaited() + self.assertFalse(self.executed) + + async def test_http_and_sse_partial_approval_parity_and_400(self): + app = FastAPI() + app.include_router(router) + app.dependency_overrides[get_current_active_user] = lambda: self.user + batch = await self.batch() + async with httpx.AsyncClient(transport=httpx.ASGITransport(app), base_url="http://test") as client: + invalid = await client.post("/api/agent/chat/stream", json={"approved_mcp_call_ids": ["one"]}) + self.assertEqual(invalid.status_code, 400) + response = await client.post("/api/agent/chat", json=self.request(batch, approved=["one"]).model_dump()) + self.assertEqual(response.status_code, 200, response.text) + normal = response.json()["data"] + response = await client.post("/api/agent/chat/stream", json=self.request(batch, approved=["one"]).model_dump()) + self.assertEqual(response.status_code, 200, response.text) + events = [part for part in response.text.split("\n\n") if part.startswith("event: done")] + streamed = json.loads(events[-1].split("data: ", 1)[1]) + self.assertEqual(streamed, normal) + self.assertEqual(len(self.executed), 1) + + async def test_iteration_limit_is_visible_and_keeps_results(self): + self.model.return_value = {"role": "assistant", "content": "", "mcp_calls": [{"id": "time", "name": "time", "arguments": {}}]} + result = await AgentService.chat(AgentChatRequest(messages=[{"role": "user", "content": "loop"}]), self.user) + self.assertEqual(result["finish_reason"], "iteration_limit") + self.assertEqual(self.model.await_count, 8) + self.assertEqual(len([m for m in result["messages"] if m["role"] == "tool"]), 8) + + async def test_missing_model_and_unsupported_provider_are_explicit(self): + self.patches[-1].stop() + with patch("domain.agent.service.AIProviderService.get_default_model", AsyncMock(return_value=None)): + result = await AgentService.chat(AgentChatRequest(), self.user) + self.assertEqual(result["finish_reason"], "model_unavailable") + with patch("domain.agent.service.AIProviderService.get_default_model", AsyncMock(return_value=SimpleNamespace(provider=SimpleNamespace(api_format="gemini")))): + with self.assertRaises(MissingModelError): + await _choose_chat_ability() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_config_version.py b/tests/test_config_version.py new file mode 100644 index 0000000..2aeb392 --- /dev/null +++ b/tests/test_config_version.py @@ -0,0 +1,71 @@ +import os +import unittest +from unittest.mock import patch + +import httpx + +from domain.config.service import ConfigService +from domain.config.types import LatestVersionInfo + + +class LatestVersionTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.cache = {"timestamp": 0.0, "data": None} + cache_patch = patch.object(ConfigService, "_latest_version_cache", self.cache) + cache_patch.start() + self.addCleanup(cache_patch.stop) + + async def test_fetch_with_unsupported_environment_proxy(self): + requests = [] + + async def respond(transport, request): + requests.append(request) + return httpx.Response( + 200, + json={"tag_name": "v2.3.0", "body": "Release notes"}, + request=request, + ) + + proxy_env = { + "ALL_PROXY": "socks://127.0.0.1:7897", + "all_proxy": "socks://127.0.0.1:7897", + "HTTP_PROXY": "http://127.0.0.1:7897", + "HTTPS_PROXY": "http://127.0.0.1:7897", + } + # Exercise real client initialization, including environment proxy parsing. + with patch.dict(os.environ, proxy_env, clear=True): + with patch.object(httpx.AsyncHTTPTransport, "handle_async_request", respond): + result = await ConfigService.get_latest_version() + cached_result = await ConfigService.get_latest_version() + + self.assertEqual(result.latest_version, "v2.3.0") + self.assertEqual(result.body, "Release notes") + self.assertIs(cached_result, result) + self.assertEqual(len(requests), 1) + self.assertEqual( + str(requests[0].url), + "https://api.github.com/repos/DrizzleTime/Foxel/releases/latest", + ) + + async def test_failed_fetch_returns_cached_or_empty_info(self): + stale_info = LatestVersionInfo(latest_version="v2.2.3", body="Cached notes") + for failure in ("connection", 403, 500): + for cached_info in (None, stale_info): + with self.subTest(failure=failure, cached=bool(cached_info)): + self.cache.update(timestamp=0.0, data=cached_info) + + async def respond(transport, request): + if failure == "connection": + raise httpx.ConnectError("Connection failed", request=request) + return httpx.Response(failure, request=request) + + with patch.object(httpx.AsyncHTTPTransport, "handle_async_request", respond): + result = await ConfigService.get_latest_version() + + self.assertEqual(result, cached_info or LatestVersionInfo()) + self.assertEqual(self.cache["timestamp"], 0.0) + self.assertIs(self.cache["data"], cached_info) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_notices.py b/tests/test_notices.py new file mode 100644 index 0000000..cd8c66e --- /dev/null +++ b/tests/test_notices.py @@ -0,0 +1,43 @@ +import os +import unittest +from unittest.mock import patch + +import httpx + +from domain.notices.service import NoticeService, REMOTE_NOTICES_URL + + +class NoticeFetchTests(unittest.IsolatedAsyncioTestCase): + async def test_fetch_with_unsupported_environment_proxy(self): + requests = [] + + async def respond(transport, request): + requests.append(request) + page = int(request.url.params["page"]) + return httpx.Response( + 200, + json={"items": [{"id": page}], "total": 2, "pageSize": 1}, + request=request, + ) + + proxy_env = { + "ALL_PROXY": "socks://127.0.0.1:7897", + "all_proxy": "socks://127.0.0.1:7897", + "HTTP_PROXY": "http://127.0.0.1:7897", + "HTTPS_PROXY": "http://127.0.0.1:7897", + } + # Keep real client initialization so environment proxy parsing is exercised. + with patch.dict(os.environ, proxy_env, clear=True): + with patch.object(httpx.AsyncHTTPTransport, "handle_async_request", respond): + with patch("domain.notices.service.VERSION", "v1.2.3"): + items = await NoticeService._fetch_remote_notices() + + self.assertEqual(items, [{"id": 1}, {"id": 2}]) + self.assertEqual(len(requests), 2) + for request in requests: + self.assertEqual(str(request.url).split("?")[0], REMOTE_NOTICES_URL) + self.assertEqual(request.url.params["version"], "1.2.3") + + +if __name__ == "__main__": + unittest.main() diff --git a/web/src/api/agent.ts b/web/src/api/agent.ts index f1d683e..465ed22 100644 --- a/web/src/api/agent.ts +++ b/web/src/api/agent.ts @@ -9,6 +9,7 @@ export interface AgentChatContext { export interface AgentChatRequest { messages: AgentChatMessage[]; auto_execute?: boolean; + approval_batch_id?: string | null; approved_mcp_call_ids?: string[]; rejected_mcp_call_ids?: string[]; context?: AgentChatContext; @@ -25,11 +26,15 @@ export interface PendingMcpCall { name: string; arguments: Record; requires_confirmation: boolean; + status?: 'pending' | 'running'; } export interface AgentChatResponse { messages: AgentChatMessage[]; pending_mcp_calls?: PendingMcpCall[]; + approval_batch_id?: string | null; + replace_messages?: boolean; + finish_reason?: string; } export type AgentSseEvent = @@ -38,7 +43,7 @@ export type AgentSseEvent = | { event: 'assistant_end'; data: { id: string; message: AgentChatMessage } } | { event: 'mcp_call_start'; data: { mcp_call_id: string; name: string } } | { event: 'mcp_call_end'; data: { mcp_call_id: string; name: string; message: AgentChatMessage } } - | { event: 'pending'; data: { pending_mcp_calls: PendingMcpCall[] } } + | { event: 'pending'; data: { pending_mcp_calls: PendingMcpCall[]; approval_batch_id?: string | null } } | { event: 'done'; data: AgentChatResponse }; export const agentApi = { diff --git a/web/src/components/AiAgentWidget.tsx b/web/src/components/AiAgentWidget.tsx index c4d7cfe..b9bc931 100644 --- a/web/src/components/AiAgentWidget.tsx +++ b/web/src/components/AiAgentWidget.tsx @@ -109,6 +109,7 @@ const AiAgentWidget = memo(function AiAgentWidget({ currentPath, open, onOpenCha const [loading, setLoading] = useState(false); const [messages, setMessages] = useState([]); const [pending, setPending] = useState([]); + const [approvalBatchId, setApprovalBatchId] = useState(null); const [expandedTools, setExpandedTools] = useState>({}); const [expandedRaw, setExpandedRaw] = useState>({}); const [runningTools, setRunningTools] = useState>({}); @@ -193,6 +194,7 @@ const AiAgentWidget = memo(function AiAgentWidget({ currentPath, open, onOpenCha { messages: payload.messages, auto_execute: autoExecute, + approval_batch_id: approvalBatchId, context: effectivePath ? { current_path: effectivePath } : undefined, approved_mcp_call_ids: payload.approved_mcp_call_ids, rejected_mcp_call_ids: payload.rejected_mcp_call_ids, @@ -267,14 +269,16 @@ const AiAgentWidget = memo(function AiAgentWidget({ currentPath, open, onOpenCha case 'pending': { const items = Array.isArray((evt.data as any)?.pending_mcp_calls) ? (evt.data as any).pending_mcp_calls : []; setPending(items); + setApprovalBatchId(evt.data.approval_batch_id || null); return; } case 'done': { const base = baseMessagesRef.current || []; const newMessages = Array.isArray((evt.data as any)?.messages) ? (evt.data as any).messages : []; const nextPending = Array.isArray((evt.data as any)?.pending_mcp_calls) ? (evt.data as any).pending_mcp_calls : []; - setMessages([...base, ...newMessages]); + setMessages(evt.data.replace_messages ? newMessages : [...base, ...newMessages]); setPending(nextPending); + setApprovalBatchId(evt.data.approval_batch_id || null); setRunningTools({}); assistantIndexRef.current = {}; return; @@ -297,7 +301,7 @@ const AiAgentWidget = memo(function AiAgentWidget({ currentPath, open, onOpenCha } } } - }, [autoExecute, effectivePath, t]); + }, [autoExecute, approvalBatchId, effectivePath, t]); const handleSend = useCallback(async () => { const text = input.trim(); @@ -318,6 +322,7 @@ const AiAgentWidget = memo(function AiAgentWidget({ currentPath, open, onOpenCha streamControllerRef.current?.abort(); setMessages([]); setPending([]); + setApprovalBatchId(null); setExpandedTools({}); setExpandedRaw({}); setRunningTools({}); @@ -332,13 +337,13 @@ const AiAgentWidget = memo(function AiAgentWidget({ currentPath, open, onOpenCha }, [messages, runStream]); const approveAll = useCallback(async () => { - const ids = pending.map((p) => p.id).filter(Boolean); + const ids = pending.filter((p) => p.status !== 'running').map((p) => p.id).filter(Boolean); if (ids.length === 0) return; await runStream({ messages, approved_mcp_call_ids: ids }); }, [messages, pending, runStream]); const rejectAll = useCallback(async () => { - const ids = pending.map((p) => p.id).filter(Boolean); + const ids = pending.filter((p) => p.status !== 'running').map((p) => p.id).filter(Boolean); if (ids.length === 0) return; await runStream({ messages, rejected_mcp_call_ids: ids }); }, [messages, pending, runStream]); @@ -444,7 +449,7 @@ const AiAgentWidget = memo(function AiAgentWidget({ currentPath, open, onOpenCha if (metaEntries.length === 0 && !title) return null; return ( <> - + {title ? ( {title} ) : null} @@ -614,7 +619,7 @@ const AiAgentWidget = memo(function AiAgentWidget({ currentPath, open, onOpenCha {t('AI Agent')} {t('Auto execute')} - + 0} /> - @@ -781,7 +786,7 @@ const AiAgentWidget = memo(function AiAgentWidget({ currentPath, open, onOpenCha const args = p.arguments || {}; const key = `pending:${p.id}`; const expanded = !!expandedTools[key]; - const running = Object.prototype.hasOwnProperty.call(runningTools, p.id); + const running = p.status === 'running' || Object.prototype.hasOwnProperty.call(runningTools, p.id); const summary = renderToolArgsSummary(args); return (
@@ -801,14 +806,14 @@ const AiAgentWidget = memo(function AiAgentWidget({ currentPath, open, onOpenCha type="primary" onClick={() => void approveOne(p.id)} loading={loading && running} - disabled={loading && !running} + disabled={loading || p.status === 'running'} > {t('Execute')} diff --git a/web/src/layout/SideNav.tsx b/web/src/layout/SideNav.tsx index dcf5ef2..aab672f 100644 --- a/web/src/layout/SideNav.tsx +++ b/web/src/layout/SideNav.tsx @@ -275,7 +275,7 @@ const SideNav = memo(function SideNav({ open={open} onClose={onClose} title={null} - width={280} + size={280} styles={{ body: { padding: 0 } }} > {renderNavBody(false, false)} diff --git a/web/src/pages/AdaptersPage.tsx b/web/src/pages/AdaptersPage.tsx index b67ef91..7cbd325 100644 --- a/web/src/pages/AdaptersPage.tsx +++ b/web/src/pages/AdaptersPage.tsx @@ -282,7 +282,7 @@ const AdaptersPage = memo(function AdaptersPage() { /> { setOpen(false); setEditing(null); }} destroyOnHidden diff --git a/web/src/pages/AuditLogsPage.tsx b/web/src/pages/AuditLogsPage.tsx index ba8207d..74807f5 100644 --- a/web/src/pages/AuditLogsPage.tsx +++ b/web/src/pages/AuditLogsPage.tsx @@ -264,7 +264,7 @@ const AuditLogsPage = memo(function AuditLogsPage() { styles={{ body: { maxHeight: '70vh', overflowY: 'auto' } }} > {selectedLog && ( - + = (props) => { placement="bottom" onClose={onClose} title={entry ? t('Actions') : t('Quick Actions')} - height="auto" + size="auto" styles={{ body: { padding: 8 } }} >
e.stopPropagation()}> diff --git a/web/src/pages/FileExplorerPage/components/FileDetailModal.tsx b/web/src/pages/FileExplorerPage/components/FileDetailModal.tsx index 3abce68..ed4a817 100644 --- a/web/src/pages/FileExplorerPage/components/FileDetailModal.tsx +++ b/web/src/pages/FileExplorerPage/components/FileDetailModal.tsx @@ -99,7 +99,7 @@ export const FileDetailModal: React.FC = ({ entry, loading, data, onClose borderBottom: idx === total - 1 ? 'none' : `1px solid ${token.colorSplit}`, }} > - + {entry?.chunk_id && ( {t('Chunk ID')}: {entry.chunk_id} diff --git a/web/src/pages/FileExplorerPage/components/Modals/UploadModal.tsx b/web/src/pages/FileExplorerPage/components/Modals/UploadModal.tsx index 7988a0d..eab9671 100644 --- a/web/src/pages/FileExplorerPage/components/Modals/UploadModal.tsx +++ b/web/src/pages/FileExplorerPage/components/Modals/UploadModal.tsx @@ -125,7 +125,7 @@ const UploadModal: React.FC = ({ , ]} > - +
diff --git a/web/src/pages/LoginPage.tsx b/web/src/pages/LoginPage.tsx index af66262..0002635 100644 --- a/web/src/pages/LoginPage.tsx +++ b/web/src/pages/LoginPage.tsx @@ -83,7 +83,7 @@ export default function LoginPage() { }} >
- +
Foxel Logo @@ -161,7 +161,7 @@ export default function LoginPage() { Foxel 旨在提供一个安全、高效且智能的文件管理解决方案，帮助您轻松组织、访问和共享您的数字资产。
- + diff --git a/web/src/pages/ProcessorsPage.tsx b/web/src/pages/ProcessorsPage.tsx index c5470a1..e092d6d 100644 --- a/web/src/pages/ProcessorsPage.tsx +++ b/web/src/pages/ProcessorsPage.tsx @@ -354,7 +354,7 @@ const ProcessorsPage = memo(function ProcessorsPage() { {item.type} - +
{t('Supported Extensions')}: {item.supported_exts?.length ? ( @@ -446,7 +446,7 @@ const ProcessorsPage = memo(function ProcessorsPage() { {isDirectory && supportsDirectory && ( - + )} {isDirectory && !supportsDirectory && ( - + - +
{t('Create Account')} {t('Sign up to your Foxel account')} diff --git a/web/src/pages/SetupPage.tsx b/web/src/pages/SetupPage.tsx index 9c8e9b3..01ea789 100644 --- a/web/src/pages/SetupPage.tsx +++ b/web/src/pages/SetupPage.tsx @@ -394,7 +394,7 @@ const SetupPage = () => {
- + {currentStep > 0 && (