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 && (