Refactor file handling and permissions for platform attachments

- Updated the storage path for multimedia files received via the WeChat adapter to be organized by session identifiers.
- Enhanced the local permissions model to restrict non-admin users from accessing shared temporary files.
- Implemented tests to ensure that attachment permissions are enforced correctly based on user roles and session ownership.
- Adjusted the handling of temporary files in various tests to ensure isolation and proper cleanup.
- Improved the integration of platform files utility functions across the codebase for consistency in file path management.

astrbot/core/agent/runners/tool_loop_agent_runner.py

@@ -1,5 +1,6 @@
 import asyncio
 import copy
+import os
 import sys
 import time
 import traceback
@@ -28,6 +29,7 @@ from astrbot import logger
 from astrbot.core.agent.message import ImageURLPart, TextPart, ThinkPart
 from astrbot.core.agent.tool import FunctionTool, ToolSet
 from astrbot.core.agent.tool_image_cache import tool_image_cache
+from astrbot.core.computer.local_file_security import open_file_in_allowed_roots
 from astrbot.core.exceptions import EmptyModelOutputError
 from astrbot.core.message.components import Json
 from astrbot.core.message.message_event_result import (
@@ -380,7 +382,7 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
         if self.tool_result_overflow_dir is None:
             raise ValueError("tool_result_overflow_dir is not configured")
 
-        overflow_dir = Path(self.tool_result_overflow_dir).resolve(strict=False)
+        overflow_dir = Path(self.tool_result_overflow_dir).absolute()
         safe_tool_call_id = (
             "".join(
                 ch if ch.isalnum() or ch in {"-", "_", "."} else "_"
@@ -393,7 +395,20 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
 
         def _run() -> str:
             overflow_dir.mkdir(parents=True, exist_ok=True)
-            overflow_path.write_text(content, encoding="utf-8")
+            if sys.platform == "win32":
+                with overflow_path.open("x", encoding="utf-8") as output:
+                    output.write(content)
+            else:
+                # The workspace is writable by tools; do not follow a replaced
+                # root or file when the host spills a tool result into it.
+                descriptor = open_file_in_allowed_roots(
+                    str(overflow_path),
+                    (overflow_dir,),
+                    access="write",
+                    create_parents=True,
+                )
+                with os.fdopen(descriptor, "w", encoding="utf-8") as output:
+                    output.write(content)
             return str(overflow_path)
 
         return await asyncio.to_thread(_run)

astrbot/core/astr_main_agent.py

@@ -114,6 +114,10 @@ from astrbot.core.utils.media_utils import (
     is_recoverable_image_error,
     normalize_model_image_max_size,
 )
+from astrbot.core.utils.platform_files import (
+    retain_platform_file,
+    update_platform_image_path,
+)
 from astrbot.core.utils.quoted_message.settings import (
     SETTINGS as DEFAULT_QUOTED_MESSAGE_SETTINGS,
 )
@@ -813,9 +817,24 @@ async def _append_video_attachment(
     video: Video,
     *,
     quoted: bool = False,
+    event: AstrMessageEvent | None = None,
 ) -> None:
+    """Attach a video, retaining platform input when an event is available.
+
+    Args:
+        req: Request receiving the attachment notice.
+        video: Incoming video component.
+        quoted: Whether the video belongs to a quoted message.
+        event: Owner used to localize late or quoted platform attachments.
+    """
     try:
         video_path = await video.convert_to_file_path()
+        if event is not None:
+            video_path = await retain_platform_file(
+                video_path,
+                event.unified_msg_origin,
+            )
+            video.file = video.path = video.url = video_path
     except Exception as exc:  # noqa: BLE001
         if quoted:
             logger.debug(
@@ -1456,6 +1475,22 @@ async def collect_initial_request(
                 if isinstance(comp, Image):
                     try:
                         image_path = await comp.convert_to_file_path()
+                        source_ref = comp.url or comp.file or ""
+                        if not is_file_uri(source_ref):
+                            try:
+                                source_is_local = Path(source_ref).is_file()
+                            except OSError as exc:
+                                if not is_recoverable_image_error(exc):
+                                    raise
+                                source_is_local = False
+                            if not source_is_local and Path(image_path).is_file():
+                                event.track_temporary_local_file(image_path)
+                        image_path = await retain_platform_file(
+                            image_path, event.unified_msg_origin
+                        )
+                        update_platform_image_path(
+                            comp, image_path, event.message_obj.message
+                        )
                     except Exception as exc:
                         if not is_recoverable_image_error(exc):
                             raise
@@ -1468,24 +1503,23 @@ async def collect_initial_request(
                         continue
                     req.image_urls.append(image_path)
                     attachment_paths.append(image_path)
-                    # Adopt sources created after PreProcess, before another
-                    # attachment or conversation lookup can fail or be cancelled.
-                    source_ref = comp.url or comp.file or ""
-                    if not is_file_uri(source_ref):
-                        try:
-                            source_is_local = Path(source_ref).is_file()
-                        except OSError as exc:
-                            if not is_recoverable_image_error(exc):
-                                raise
-                            source_is_local = False
-                        if not source_is_local and Path(image_path).is_file():
-                            event.track_temporary_local_file(image_path)
                 elif isinstance(comp, Record):
                     audio_path = await comp.convert_to_file_path()
+                    audio_path = await retain_platform_file(
+                        audio_path,
+                        event.unified_msg_origin,
+                    )
+                    comp.file = comp.path = comp.url = audio_path
                     req.audio_urls.append(audio_path)
                     _append_audio_attachment(req, audio_path)
                 elif isinstance(comp, File):
                     file_path = await comp.get_file()
+                    if file_path:
+                        file_path = await retain_platform_file(
+                            file_path,
+                            event.unified_msg_origin,
+                        )
+                        comp.file_ = file_path
                     file_name = comp.name or os.path.basename(file_path)
                     req.extra_user_content_parts.append(
                         TextPart(
@@ -1493,7 +1527,7 @@ async def collect_initial_request(
                         )
                     )
                 elif isinstance(comp, Video):
-                    await _append_video_attachment(req, comp)
+                    await _append_video_attachment(req, comp, event=event)
             # quoted message attachments
             reply_comps = [
                 comp for comp in event.message_obj.message if isinstance(comp, Reply)
@@ -1510,6 +1544,25 @@ async def collect_initial_request(
                             has_embedded_image = True
                             try:
                                 image_path = await reply_comp.convert_to_file_path()
+                                source_ref = reply_comp.url or reply_comp.file or ""
+                                if not is_file_uri(source_ref):
+                                    try:
+                                        source_is_local = Path(source_ref).is_file()
+                                    except OSError as exc:
+                                        if not is_recoverable_image_error(exc):
+                                            raise
+                                        source_is_local = False
+                                    if (
+                                        not source_is_local
+                                        and Path(image_path).is_file()
+                                    ):
+                                        event.track_temporary_local_file(image_path)
+                                image_path = await retain_platform_file(
+                                    image_path, event.unified_msg_origin
+                                )
+                                update_platform_image_path(
+                                    reply_comp, image_path, event.message_obj.message
+                                )
                             except Exception as exc:
                                 if not is_recoverable_image_error(exc):
                                     raise
@@ -1523,23 +1576,26 @@ async def collect_initial_request(
                                 continue
                             req.image_urls.append(image_path)
                             attachment_paths.append(image_path)
-                            source_ref = reply_comp.url or reply_comp.file or ""
-                            if not is_file_uri(source_ref):
-                                try:
-                                    source_is_local = Path(source_ref).is_file()
-                                except OSError as exc:
-                                    if not is_recoverable_image_error(exc):
-                                        raise
-                                    source_is_local = False
-                                if not source_is_local and Path(image_path).is_file():
-                                    event.track_temporary_local_file(image_path)
                             quoted_image_refs.add(image_path)
                         elif isinstance(reply_comp, Record):
                             audio_path = await reply_comp.convert_to_file_path()
+                            audio_path = await retain_platform_file(
+                                audio_path,
+                                event.unified_msg_origin,
+                            )
+                            reply_comp.file = reply_comp.path = reply_comp.url = (
+                                audio_path
+                            )
                             req.audio_urls.append(audio_path)
                             _append_quoted_audio_attachment(req, audio_path)
                         elif isinstance(reply_comp, File):
                             file_path = await reply_comp.get_file()
+                            if file_path:
+                                file_path = await retain_platform_file(
+                                    file_path,
+                                    event.unified_msg_origin,
+                                )
+                                reply_comp.file_ = file_path
                             file_name = reply_comp.name or os.path.basename(file_path)
                             req.extra_user_content_parts.append(
                                 TextPart(
@@ -1550,7 +1606,9 @@ async def collect_initial_request(
                                 )
                             )
                         elif isinstance(reply_comp, Video):
-                            await _append_video_attachment(req, reply_comp, quoted=True)
+                            await _append_video_attachment(
+                                req, reply_comp, quoted=True, event=event
+                            )
 
                 # Fallback quoted image extraction for reply-id-only payloads, or when
                 # embedded reply chain only contains placeholders (e.g. [Forward Message], [Image]).
@@ -1587,9 +1645,19 @@ async def collect_initial_request(
                         for image_ref in fallback_images:
                             if image_ref in req.image_urls:
                                 continue
-                            req.image_urls.append(image_ref)
+                            image = Image(file=image_ref)
+                            image_path = await image.convert_to_file_path()
+                            retained_path = await retain_platform_file(
+                                image_path,
+                                event.unified_msg_origin,
+                            )
+                            image.file = image.path = image.url = retained_path
+                            if comp.chain is None:
+                                comp.chain = []
+                            comp.chain.append(image)
+                            req.image_urls.append(retained_path)
                             fallback_quoted_image_count += 1
-                            quoted_image_refs.add(image_ref)
+                            quoted_image_refs.add(retained_path)
                     except Exception as exc:  # noqa: BLE001
                         logger.warning(
                             "Failed to resolve fallback quoted images for umo=%s, reply_id=%s: %s",
@@ -1613,8 +1681,7 @@ async def collect_initial_request(
         image = next((part for part in quote.chain if isinstance(part, Image)), None)
         if image:
             quote_image_ref = image.url or image.file
-    # Keep adopted source paths usable after cleanup, but retain provisional
-    # ownership until collection succeeds so errors and cancellation can clean up.
+    # Retained images outlive event cleanup; staging files and previews remain owned.
     for image_path in attachment_paths:
         event.untrack_temporary_local_file(image_path)
     return req, quote_image_ref
@@ -1899,7 +1966,15 @@ async def build_main_agent(
         fallback_providers=fallback_providers,
         request_max_retries=config.request_max_retries,
         tool_result_overflow_dir=(
-            get_astrbot_system_tmp_path()
+            (
+                str(
+                    await _get_workspace_path_for_umo(
+                        event.unified_msg_origin, plugin_context
+                    )
+                )
+                if config.computer_use_runtime == "local"
+                else get_astrbot_system_tmp_path()
+            )
             if req.func_tool and req.func_tool.get_tool("astrbot_file_read_tool")
             else None
         ),

astrbot/core/pipeline/preprocess_stage/stage.py

@@ -6,7 +6,16 @@ from collections.abc import AsyncGenerator
 from pathlib import Path
 
 from astrbot.core import logger
-from astrbot.core.message.components import Image, Plain, Record, Reply
+from astrbot.core.message.components import (
+    File,
+    Image,
+    Node,
+    Nodes,
+    Plain,
+    Record,
+    Reply,
+    Video,
+)
 from astrbot.core.platform.astr_message_event import AstrMessageEvent
 from astrbot.core.utils.astrbot_path import get_astrbot_temp_path
 from astrbot.core.utils.media_utils import (
@@ -16,6 +25,10 @@ from astrbot.core.utils.media_utils import (
     file_uri_to_path,
     is_file_uri,
 )
+from astrbot.core.utils.platform_files import (
+    retain_platform_file,
+    update_platform_image_path,
+)
 
 from ..context import PipelineContext
 from ..stage import Stage, register_stage
@@ -67,7 +80,12 @@ class PreProcessStage(Stage):
         event: AstrMessageEvent,
         component: Image,
     ) -> None:
-        """Resolve and validate an image while preserving cleanup ownership."""
+        """Resolve an image in place while preserving cleanup ownership.
+
+        Args:
+            event: Incoming event with its finalized session identity.
+            component: Image component to localize in place.
+        """
         media_ref = component.url or component.file
         image_path: str | None = None
         materialized = False
@@ -90,11 +108,9 @@ class PreProcessStage(Stage):
                 event.untrack_temporary_local_file(image_path)
             raise
 
-        component.file = image_path
-        component.path = image_path
-        # Image.convert_to_file_path() prefers url, so keep it aligned.
-        component.url = image_path
-        event.untrack_temporary_local_file(image_path)
+        retained_path = await retain_platform_file(image_path, event.unified_msg_origin)
+        update_platform_image_path(component, retained_path, event.get_messages())
+        event.untrack_temporary_local_file(retained_path)
 
     async def process(
         self,
@@ -162,60 +178,72 @@ class PreProcessStage(Stage):
                             logger.debug(f"Path mapping: {url} -> {component.url}")
                     message_chain[idx] = component
 
-        # Localize source images and normalize audio for downstream processing.
-        message_chain = event.get_messages()
-        for idx, component in enumerate(message_chain):
-            if isinstance(component, Record):
+        # WakingCheckStage has finalized the UMO before attachments are retained.
+        # Traverse quoted and forwarded chains as well as the incoming message.
+        pending = list(reversed(event.get_messages()))
+        while pending:
+            component = pending.pop()
+            if isinstance(component, Reply) and component.chain:
+                pending.extend(reversed(component.chain))
+            elif isinstance(component, Node):
+                pending.extend(reversed(component.content))
+            elif isinstance(component, Nodes):
+                pending.extend(reversed(component.nodes))
+            elif isinstance(component, Image):
+                try:
+                    await self._normalize_image_component(event, component)
+                except Exception as exc:
+                    logger.warning(
+                        "Image processing failed for %s: %s",
+                        describe_media_ref(component.url or component.file),
+                        exc,
+                    )
+            elif isinstance(component, Record):
                 try:
                     original_path = await component.convert_to_file_path()
                     self._track_temp_media(event, original_path)
                     record_path = await ensure_wav(original_path)
                     self._track_temp_media(event, record_path)
-                    component.file = record_path
-                    component.path = record_path
-                    message_chain[idx] = component
-                except Exception as e:
-                    logger.warning(f"Voice processing failed: {e}")
-            elif isinstance(component, Image):
-                try:
-                    await self._normalize_image_component(event, component)
-                    message_chain[idx] = component
-                except Exception as e:
-                    media_ref = component.url or component.file
-                    logger.warning(
-                        "Image processing failed for %s: %s",
-                        describe_media_ref(media_ref),
-                        e,
+                    retained_path = await retain_platform_file(
+                        record_path,
+                        event.unified_msg_origin,
                     )
-
-        # Also normalize media components inside Reply chains.
-        for component in event.get_messages():
-            if isinstance(component, Reply) and component.chain:
-                for idx, reply_comp in enumerate(component.chain):
-                    if isinstance(reply_comp, Record):
-                        try:
-                            original_path = await reply_comp.convert_to_file_path()
-                            self._track_temp_media(event, original_path)
-                            record_path = await ensure_wav(original_path)
-                            self._track_temp_media(event, record_path)
-                            reply_comp.file = record_path
-                            reply_comp.path = record_path
-                            component.chain[idx] = reply_comp
-                        except Exception as e:
-                            logger.warning(
-                                f"Voice processing in reply chain failed: {e}"
-                            )
-                    elif isinstance(reply_comp, Image):
-                        try:
-                            await self._normalize_image_component(event, reply_comp)
-                            component.chain[idx] = reply_comp
-                        except Exception as e:
-                            media_ref = reply_comp.url or reply_comp.file
-                            logger.warning(
-                                "Image processing in reply chain failed for %s: %s",
-                                describe_media_ref(media_ref),
-                                e,
-                            )
+                    component.file = retained_path
+                    component.path = retained_path
+                    component.url = retained_path
+                    event.untrack_temporary_local_file(retained_path)
+                except Exception as exc:
+                    logger.warning("Voice processing failed: %s", exc)
+            elif isinstance(component, File | Video):
+                try:
+                    source_ref = (
+                        component.file_
+                        if isinstance(component, File)
+                        else component.file or component.url
+                    )
+                    was_local = self._is_existing_local_image_ref(source_ref)
+                    source_path = (
+                        await component.get_file()
+                        if isinstance(component, File)
+                        else await component.convert_to_file_path()
+                    )
+                    if not source_path:
+                        continue
+                    if not was_local:
+                        self._track_temp_media(event, source_path)
+                    retained_path = await retain_platform_file(
+                        source_path,
+                        event.unified_msg_origin,
+                    )
+                    if isinstance(component, File):
+                        component.file_ = retained_path
+                    else:
+                        component.file = retained_path
+                        component.path = retained_path
+                        component.url = retained_path
+                    event.untrack_temporary_local_file(retained_path)
+                except Exception as exc:
+                    logger.warning("Attachment localization failed: %s", exc)
 
         # STT
         if self.stt_settings.get("enable", False):

astrbot/core/star/context.py

@@ -261,6 +261,10 @@ class Context:
             AstrAgentContext,
         )
         from astrbot.core.astr_agent_tool_exec import FunctionToolExecutor
+        from astrbot.core.tools.computer_tools.util import (
+            is_local_runtime,
+            workspace_root_for_context,
+        )
 
         prov = await self.provider_manager.get_provider_by_id(chat_provider_id)
         if not prov or not isinstance(prov, Provider):
@@ -291,6 +295,10 @@ class Context:
             )
         agent_runner = ToolLoopAgentRunner()
         tool_executor = FunctionToolExecutor()
+        run_context = AgentContextWrapper(
+            context=agent_context,
+            tool_call_timeout=tool_call_timeout,
+        )
 
         streaming = kwargs.get("stream", False)
 
@@ -300,9 +308,12 @@ class Context:
             if k not in ["stream", "agent_hooks", "agent_context"]
         }
         if request.func_tool and request.func_tool.get_tool("astrbot_file_read_tool"):
-            other_kwargs.setdefault(
-                "tool_result_overflow_dir", get_astrbot_system_tmp_path()
-            )
+            if "tool_result_overflow_dir" not in other_kwargs:
+                other_kwargs["tool_result_overflow_dir"] = (
+                    str(await workspace_root_for_context(run_context))
+                    if is_local_runtime(run_context)
+                    else get_astrbot_system_tmp_path()
+                )
             other_kwargs.setdefault(
                 "read_tool", request.func_tool.get_tool("astrbot_file_read_tool")
             )
@@ -310,10 +321,7 @@ class Context:
         await agent_runner.reset(
             provider=prov,
             request=request,
-            run_context=AgentContextWrapper(
-                context=agent_context,
-                tool_call_timeout=tool_call_timeout,
-            ),
+            run_context=run_context,
             tool_executor=tool_executor,
             agent_hooks=agent_hooks,
             streaming=streaming,

astrbot/core/tools/computer_tools/fs.py

@@ -13,9 +13,10 @@ Local behavior follows each role's `filesystem_scope` permission:
   depends on host OS permissions.
 - `workspace`: read/grep are restricted to globally installed Skills,
   plugin-provided Skills, built-in plugin Skills, the current session or project
-  workspace, and AstrBot temporary directories. Write/edit are restricted to the
-  current workspace and temporary directories. Administrators may also update
-  globally installed Skills; plugin-provided and built-in Skills remain read-only.
+  workspace, and the current session's platform attachments. Write/edit are
+  restricted to the current workspace. Administrators may also read and write
+  all platform attachments and globally installed Skills; plugin-provided and
+  built-in Skills remain read-only. Shared temporary roots are not exposed.
 - Upload and download tools are not exposed in Local mode.
 
 Remote Sandbox behavior still follows `computer_use_require_admin`:
@@ -51,9 +52,8 @@ from astrbot.core.utils.astrbot_path import (
     get_astrbot_builtin_plugin_path,
     get_astrbot_plugin_path,
     get_astrbot_skills_path,
-    get_astrbot_system_tmp_path,
-    get_astrbot_temp_path,
 )
+from astrbot.core.utils.platform_files import platform_files_root
 
 from ..registry import builtin_tool
 from . import util as computer_util
@@ -86,6 +86,7 @@ def _restricted_env_path_labels(
     *,
     include_installed_skills: bool,
     include_plugin_skills: bool,
+    is_admin: bool = False,
     current_workspace_root: Path | None = None,
 ) -> list[str]:
     """Return labels for directories allowed by a workspace-scoped Local policy."""
@@ -99,13 +100,9 @@ def _restricted_env_path_labels(
                 "astrbot/builtin_stars/*/skills",
             ]
         )
-    labels.extend(
-        [
-            str(current_workspace_root or _workspace_root(umo)),
-            get_astrbot_system_tmp_path(),
-            get_astrbot_temp_path(),
-        ]
-    )
+    labels.append(str(current_workspace_root or _workspace_root(umo)))
+    if include_plugin_skills or is_admin:
+        labels.append(str(platform_files_root(None if is_admin else umo)))
     return labels
 
 
@@ -139,14 +136,24 @@ def _plugin_skill_roots() -> tuple[Path, ...]:
 def _read_allowed_roots(
     umo: str,
     current_workspace_root: Path | None = None,
+    *,
+    is_admin: bool = False,
 ) -> tuple[Path, ...]:
-    """Return roots readable by a workspace-scoped Local policy."""
+    """Return roots readable by a workspace-scoped Local policy.
+
+    Args:
+        umo: Final unified message origin for attachment isolation.
+        current_workspace_root: Resolved session or project workspace, if available.
+        is_admin: Whether all sessions' attachments may be read.
+
+    Returns:
+        Trusted roots for reading and searching local files.
+    """
     return (
         Path(get_astrbot_skills_path()).resolve(strict=False),
         *_plugin_skill_roots(),
         current_workspace_root or _workspace_root(umo),
-        Path(get_astrbot_system_tmp_path()).resolve(strict=False),
-        Path(get_astrbot_temp_path()).resolve(strict=False),
+        platform_files_root(None if is_admin else umo),
     )
 
 
@@ -154,18 +161,22 @@ def _write_allowed_roots(
     umo: str,
     current_workspace_root: Path | None = None,
     *,
-    include_installed_skills: bool = False,
+    is_admin: bool = False,
 ) -> tuple[Path, ...]:
-    """Return writable roots for a workspace-scoped Local policy."""
+    """Return writable roots for a workspace-scoped Local policy.
+
+    Args:
+        umo: Unified message origin used for the fallback workspace.
+        current_workspace_root: Resolved session or project workspace, if available.
+        is_admin: Whether installed skills and all platform attachments are writable.
+
+    Returns:
+        Trusted roots for creating and editing local files.
+    """
     return (
-        *(
-            (Path(get_astrbot_skills_path()).resolve(strict=False),)
-            if include_installed_skills
-            else ()
-        ),
+        *((Path(get_astrbot_skills_path()).resolve(strict=False),) if is_admin else ()),
         current_workspace_root or _workspace_root(umo),
-        Path(get_astrbot_system_tmp_path()).resolve(strict=False),
-        Path(get_astrbot_temp_path()).resolve(strict=False),
+        *((platform_files_root(),) if is_admin else ()),
     )
 
 
@@ -267,7 +278,7 @@ def _normalize_rw_path(
     local_env: bool,
     umo: str,
     write: bool = False,
-    allow_installed_skill_write: bool = False,
+    is_admin: bool = False,
     current_workspace_root: Path | None = None,
 ) -> str:
     normalized_path = _resolve_tool_path(
@@ -283,10 +294,10 @@ def _normalize_rw_path(
             _write_allowed_roots(
                 umo,
                 current_workspace_root,
-                include_installed_skills=allow_installed_skill_write,
+                is_admin=is_admin,
             )
             if write
-            else _read_allowed_roots(umo, current_workspace_root)
+            else _read_allowed_roots(umo, current_workspace_root, is_admin=is_admin)
         )
     if restricted and not _is_path_within_allowed_roots(
         normalized_path,
@@ -297,8 +308,9 @@ def _normalize_rw_path(
         allowed = ", ".join(
             _restricted_env_path_labels(
                 umo,
-                include_installed_skills=not write or allow_installed_skill_write,
+                include_installed_skills=not write or is_admin,
                 include_plugin_skills=not write,
+                is_admin=is_admin,
                 current_workspace_root=current_workspace_root,
             )
         )
@@ -383,6 +395,7 @@ class FileReadTool(FunctionTool):
                     restricted=restricted,
                     local_env=local_env,
                     umo=context.context.event.unified_msg_origin,
+                    is_admin=context.context.event.role == "admin",
                     current_workspace_root=current_workspace_root,
                 )
                 if local_env
@@ -407,6 +420,7 @@ class FileReadTool(FunctionTool):
                     _read_allowed_roots(
                         context.context.event.unified_msg_origin,
                         current_workspace_root,
+                        is_admin=context.context.event.role == "admin",
                     ),
                     access="read",
                 )
@@ -487,7 +501,7 @@ class FileWriteTool(FunctionTool):
                     local_env=local_env,
                     umo=context.context.event.unified_msg_origin,
                     write=True,
-                    allow_installed_skill_write=(context.context.event.role == "admin"),
+                    is_admin=(context.context.event.role == "admin"),
                     current_workspace_root=current_workspace_root,
                 )
                 if local_env
@@ -503,14 +517,16 @@ class FileWriteTool(FunctionTool):
             if restricted:
                 if current_workspace_root is not None:
                     current_workspace_root.mkdir(parents=True, exist_ok=True)
+                if context.context.event.role == "admin" and Path(
+                    normalized_path
+                ).is_relative_to(platform_files_root()):
+                    platform_files_root().mkdir(parents=True, exist_ok=True)
                 file_descriptor = open_file_in_allowed_roots(
                     normalized_path,
                     _write_allowed_roots(
                         context.context.event.unified_msg_origin,
                         current_workspace_root,
-                        include_installed_skills=(
-                            context.context.event.role == "admin"
-                        ),
+                        is_admin=context.context.event.role == "admin",
                     ),
                     access="write",
                     create_parents=True,
@@ -603,7 +619,7 @@ class FileEditTool(FunctionTool):
                     local_env=local_env,
                     umo=umo,
                     write=True,
-                    allow_installed_skill_write=(context.context.event.role == "admin"),
+                    is_admin=(context.context.event.role == "admin"),
                     current_workspace_root=current_workspace_root,
                 )
                 if local_env
@@ -630,9 +646,7 @@ class FileEditTool(FunctionTool):
                     _write_allowed_roots(
                         umo,
                         current_workspace_root,
-                        include_installed_skills=(
-                            context.context.event.role == "admin"
-                        ),
+                        is_admin=context.context.event.role == "admin",
                     ),
                     access="edit",
                 )
@@ -791,6 +805,7 @@ class GrepTool(FunctionTool):
         local_env: bool,
         umo: str,
         current_workspace_root: Path | None = None,
+        is_admin: bool = False,
     ) -> list[str]:
         normalized = (
             [
@@ -808,7 +823,9 @@ class GrepTool(FunctionTool):
             if restricted:
                 return [
                     str(root)
-                    for root in _read_allowed_roots(umo, current_workspace_root)
+                    for root in _read_allowed_roots(
+                        umo, current_workspace_root, is_admin=is_admin
+                    )
                     if root.exists()
                 ]
             if local_env:
@@ -822,7 +839,9 @@ class GrepTool(FunctionTool):
                 if not _is_path_within_allowed_roots(
                     path,
                     umo=umo,
-                    allowed_roots=_read_allowed_roots(umo, current_workspace_root),
+                    allowed_roots=_read_allowed_roots(
+                        umo, current_workspace_root, is_admin=is_admin
+                    ),
                     current_workspace_root=current_workspace_root,
                 )
             ]
@@ -832,6 +851,7 @@ class GrepTool(FunctionTool):
                         umo,
                         include_installed_skills=True,
                         include_plugin_skills=True,
+                        is_admin=is_admin,
                         current_workspace_root=current_workspace_root,
                     )
                 )
@@ -873,6 +893,7 @@ class GrepTool(FunctionTool):
                     restricted=restricted,
                     local_env=local_env,
                     umo=context.context.event.unified_msg_origin,
+                    is_admin=context.context.event.role == "admin",
                     current_workspace_root=current_workspace_root,
                 )
                 if local_env
@@ -898,6 +919,7 @@ class GrepTool(FunctionTool):
                         for root in _read_allowed_roots(
                             context.context.event.unified_msg_origin,
                             current_workspace_root,
+                            is_admin=context.context.event.role == "admin",
                         )
                         if path_object == root or path_object.is_relative_to(root)
                     ]
@@ -1056,9 +1078,9 @@ class FileDownloadTool(FunctionTool):
         try:
             name = _remote_basename(remote_path) or os.path.basename(remote_path)
 
-            local_path = os.path.join(
-                get_astrbot_temp_path(), f"sandbox_{uuid.uuid4().hex[:4]}_{name}"
-            )
+            workspace = await workspace_root_for_context(context)
+            workspace.mkdir(parents=True, exist_ok=True)
+            local_path = str(workspace / f"sandbox_{uuid.uuid4().hex}_{name}")
 
             # Download file from sandbox
             await sb.download_file(remote_path, local_path)

astrbot/core/tools/computer_tools/python.py

@@ -164,11 +164,15 @@ class LocalPythonTool(FunctionTool):
             if sandboxed and local_policy.filesystem_scope == "workspace":
                 umo = context.context.event.unified_msg_origin
                 sandbox_roots = {
-                    "readable_roots": _read_allowed_roots(umo, current_workspace_root),
+                    "readable_roots": _read_allowed_roots(
+                        umo,
+                        current_workspace_root,
+                        is_admin=context.context.event.role == "admin",
+                    ),
                     "writable_roots": _write_allowed_roots(
                         umo,
                         current_workspace_root,
-                        include_installed_skills=context.context.event.role == "admin",
+                        is_admin=context.context.event.role == "admin",
                     ),
                 }
             result = await sb.python.exec(

astrbot/core/tools/computer_tools/shell.py

@@ -142,13 +142,14 @@ class ExecuteShellTool(FunctionTool):
                     umo = context.context.event.unified_msg_origin
                     sandbox_roots = {
                         "readable_roots": _read_allowed_roots(
-                            umo, current_workspace_root
+                            umo,
+                            current_workspace_root,
+                            is_admin=context.context.event.role == "admin",
                         ),
                         "writable_roots": _write_allowed_roots(
                             umo,
                             current_workspace_root,
-                            include_installed_skills=context.context.event.role
-                            == "admin",
+                            is_admin=context.context.event.role == "admin",
                         ),
                     }
                 started_at = monotonic()

astrbot/core/tools/message_tools.py

@@ -27,25 +27,19 @@ from astrbot.core.tools.computer_tools.util import (
     workspace_root_for_context,
 )
 from astrbot.core.tools.registry import builtin_tool
-from astrbot.core.utils.astrbot_path import (
-    get_astrbot_system_tmp_path,
-    get_astrbot_temp_path,
-)
+from astrbot.core.utils.platform_files import platform_files_root
 
 
 def _file_send_allowed_roots(
     umo: str | None,
     current_workspace_root: Path | None = None,
+    *,
+    is_admin: bool = False,
 ) -> tuple[Path, ...]:
     roots = []
     if umo:
         roots.append(current_workspace_root or workspace_root(umo))
-    roots.extend(
-        [
-            Path(get_astrbot_temp_path()).resolve(strict=False),
-            Path(get_astrbot_system_tmp_path()).resolve(strict=False),
-        ]
-    )
+        roots.append(platform_files_root(None if is_admin else umo))
     return tuple(roots)
 
 
@@ -65,7 +59,9 @@ def _can_send_local_file(
     current_workspace_root: Path | None = None,
 ) -> bool:
     umo = context.context.event.unified_msg_origin
-    allowed_roots = _file_send_allowed_roots(umo, current_workspace_root)
+    allowed_roots = _file_send_allowed_roots(
+        umo, current_workspace_root, is_admin=context.context.event.role == "admin"
+    )
     if _is_path_within(local_path, allowed_roots):
         return True
     return is_local_runtime(context) and not _is_restricted_local_env(context)
@@ -141,11 +137,7 @@ class SendMessageToUserTool(FunctionTool[AstrAgentContext]):
         if not path:
             raise FileNotFoundError(f"{component_type} path is empty")
 
-        current_workspace_root = (
-            await workspace_root_for_context(context)
-            if is_local_runtime(context)
-            else None
-        )
+        current_workspace_root = await workspace_root_for_context(context)
 
         # Relative host paths are resolved only inside the user's workspace.
         if not os.path.isabs(path):
@@ -171,6 +163,7 @@ class SendMessageToUserTool(FunctionTool[AstrAgentContext]):
                         for root in _file_send_allowed_roots(
                             context.context.event.unified_msg_origin,
                             current_workspace_root,
+                            is_admin=context.context.event.role == "admin",
                         )
                     )
                     raise PermissionError(
@@ -194,8 +187,9 @@ class SendMessageToUserTool(FunctionTool[AstrAgentContext]):
             result = await sb.shell.exec(f"test -f {quoted_path} && echo '_&exists_'")
             if "_&exists_" in json.dumps(result):
                 name = _remote_basename(path) or os.path.basename(path)
-                local_path = os.path.join(
-                    get_astrbot_temp_path(), f"sandbox_{uuid.uuid4().hex[:4]}_{name}"
+                current_workspace_root.mkdir(parents=True, exist_ok=True)
+                local_path = str(
+                    current_workspace_root / f"sandbox_{uuid.uuid4().hex}_{name}"
                 )
                 await sb.download_file(path, local_path)
                 logger.info(f"Downloaded file from sandbox: {path} -> {local_path}")

astrbot/core/utils/platform_files.py

@@ -0,0 +1,104 @@
+"""Session-scoped storage for incoming platform attachments."""
+
+import asyncio
+import shutil
+import uuid
+from pathlib import Path
+
+from astrbot.core.message.components import (
+    BaseMessageComponent,
+    Image,
+    Node,
+    Nodes,
+    Reply,
+)
+from astrbot.core.utils.astrbot_path import get_astrbot_temp_path
+from astrbot.core.workspace import normalize_umo_for_workspace
+
+
+def platform_files_root(umo: str | None = None) -> Path:
+    """Return the attachment root, optionally restricted to one session.
+
+    Args:
+        umo: Final unified message origin, or None for all platform attachments.
+
+    Returns:
+        Absolute attachment directory using the workspace naming convention.
+
+    Raises:
+        ValueError: If the normalized session name targets a parent directory.
+    """
+    root = Path(get_astrbot_temp_path()) / "platform_files"
+    if umo is not None:
+        name = normalize_umo_for_workspace(umo)
+        if name in {".", ".."}:
+            raise ValueError("Invalid platform attachment session name.")
+        root /= name
+    return root.resolve(strict=False)
+
+
+async def retain_platform_file(path: str, umo: str) -> str:
+    """Copy an incoming attachment into its final session directory.
+
+    Args:
+        path: Existing local attachment, possibly owned by an adapter or cache.
+        umo: Final unified message origin after session isolation is applied.
+
+    Returns:
+        Absolute retained path. Already localized attachments keep their path.
+
+    Raises:
+        OSError: If the source cannot be read or the destination cannot be written.
+    """
+    source = Path(path).resolve(strict=True)
+    root = platform_files_root(umo)
+    if source.is_relative_to(root):
+        return str(source)
+    root.mkdir(parents=True, exist_ok=True)
+    destination = root / f"{uuid.uuid4().hex}{source.suffix[:20]}"
+    copy_task = asyncio.create_task(
+        asyncio.to_thread(shutil.copyfile, source, destination)
+    )
+    try:
+        # Do not move shared adapter caches or files owned by external platforms.
+        await asyncio.shield(copy_task)
+    except asyncio.CancelledError:
+        # A worker thread cannot be cancelled; finish it before deleting its output.
+        try:
+            await copy_task
+        finally:
+            destination.unlink(missing_ok=True)
+        raise
+    except OSError:
+        destination.unlink(missing_ok=True)
+        raise
+    return str(destination)
+
+
+def update_platform_image_path(
+    image: Image, path: str, message_chain: list[BaseMessageComponent]
+) -> None:
+    """Update an image and repeated references directly in the message segments.
+
+    Args:
+        image: Image whose original reference has just been localized.
+        path: Retained local image path.
+        message_chain: Current message, including quoted or forwarded segments.
+    """
+    source_ref = image.url or image.file
+    pending = list(message_chain)
+    while pending:
+        component = pending.pop()
+        if isinstance(component, Reply) and component.chain:
+            pending.extend(component.chain)
+        elif isinstance(component, Node):
+            pending.extend(component.content)
+        elif isinstance(component, Nodes):
+            pending.extend(component.nodes)
+        elif (
+            isinstance(component, Image)
+            and (component.url or component.file) == source_ref
+        ):
+            component.file = component.path = component.url = path
+    # A newly resolved quote may not belong to the message chain yet.
+    image.file = image.path = image.url = path

dashboard/src/assets/mdi-subset/materialdesignicons-subset.css

@@ -1,4 +1,4 @@
-/* Auto-generated MDI subset – 272 icons */
+/* Auto-generated MDI subset – 269 icons */
 /* Do not edit manually. Run: pnpm run subset-icons */
 
 @font-face {
@@ -804,10 +804,6 @@
   content: "\F1353";
 }
 
-.mdi-phone-in-talk::before {
-  content: "\F03F6";
-}
-
 .mdi-pin::before {
   content: "\F0403";
 }
@@ -976,14 +972,6 @@
   content: "\F04D2";
 }
 
-.mdi-stop::before {
-  content: "\F04DB";
-}
-
-.mdi-stop-circle::before {
-  content: "\F0666";
-}
-
 .mdi-subdirectory-arrow-right::before {
   content: "\F060D";
 }

dashboard/src/i18n/locales/en-US/features/config-metadata.json

@@ -312,7 +312,7 @@
         },
         "scopeHints": {
           "none": "No file access allowed",
-          "workspace": "Restricted workspace, temporary directories, and skill files.",
+          "workspace": "Current workspace and skills; members can read their session attachments, while admins can read and write all session attachments.",
           "host": "All files in the environment where AstrBot runs."
         },
         "scopeDetails": {

dashboard/src/i18n/locales/zh-CN/features/config-metadata.json

@@ -328,7 +328,7 @@
         },
         "scopeHints": {
           "none": "不允许访问任何文件",
-          "workspace": "受限的工作区、临时目录和技能文件。",
+          "workspace": "当前工作区和技能文件；成员只读当前会话附件，管理员可读写所有会话附件。",
           "host": "AstrBot 所在运行环境的所有文件。"
         },
         "scopeDetails": {

docs/en/dev/plugin-platform-adapter.md

@@ -247,6 +247,8 @@ If the adapter downloads platform media into AstrBot's temporary directory by it
 message_event.track_temporary_local_file(temp_media_path)
 ```
 
+After the final UMO is established, common preprocessing copies attachments from message components and quoted/forwarded chains into `data/temp/platform_files/<normalized_umo>/` and updates their local paths. Registered staging files are still cleaned up when the event ends; retained copies remain until temporary-directory size cleanup. Do not derive the final attachment directory from the original group ID during download: per-user group isolation may change the session identity afterward.
+
 Finally, in `main.py`, simply import the `fake_platform_adapter` module during initialization. The decorator will handle registration automatically.
 
 ```py

docs/en/platform/weixin_oc.md

@@ -61,9 +61,9 @@ You can also watch `Data & Logs` -> `Logs` in the WebUI to confirm that the adap
 
 ## Media File Storage
 
-Received images, videos, files, and voice messages are downloaded and decrypted into AstrBot's local temporary directory:
+Received images, videos, files, and voice messages are downloaded and decrypted by the adapter, then retained by common preprocessing under the final session identity:
 
-`data/temp`
+`data/temp/platform_files/<normalized_umo>/`
 
 These files are temporary cached files and can be further used by plugins, agents, or the file service.
 

docs/en/use/computer.md

@@ -66,29 +66,24 @@ The local Shell tool includes basic blocking for dangerous commands such as `rm
 
 ### Permission Model
 
-In the local permission table, open the **File access scope** menu to read each option's description. **Workspace** restricts access to the session workspace, temporary directories, and skill files. **Entire environment** (formerly **Host files**) allows access to files permitted by the account running AstrBot. In Docker, this means files inside the container and mounted files, not all files on the Docker host. **Disabled** prevents file access through local Computer Use tools. These descriptions now appear under the corresponding menu options instead of below the table. Hover over or tap the question mark after “All files in the environment where AstrBot runs.” for details about account permissions and Docker file access.
+In the local permission table, open the **File access scope** menu to read each option's description. **Workspace** restricts access to the current workspace, platform attachments, and skill files. **Entire environment** (formerly **Host files**) allows access to files permitted by the account running AstrBot. In Docker, this means files inside the container and mounted files, not all files on the Docker host. **Disabled** prevents file access through local Computer Use tools. These descriptions now appear under the corresponding menu options instead of below the table. Hover over or tap the question mark after “All files in the environment where AstrBot runs.” for details about account permissions and Docker file access.
 
-Computer Use has a separate option:
+Local mode configures `filesystem_scope`, code execution, and network access separately for each role. `none` disables local file tools; `host` permits files accessible to the runtime account. `workspace` applies these directory rules:
 
-- `Require AstrBot admin permission`
+| Directory | Members | Administrators |
+| --- | --- | --- |
+| Current session or project workspace | Read/write | Read/write |
+| `data/skills` | Read-only | Read/write |
+| `data/plugins/*/skills`, `astrbot/builtin_stars/*/skills` | Read-only | Read-only |
+| `data/temp/platform_files/<normalized_umo>/` for the current session | Read-only | Read/write |
+| Other sessions under `platform_files` | No access | Read/write |
+| Other files under shared `data/temp` or the system `.astrbot` temp directory | No blanket access | No blanket access |
 
-This option is enabled by default.
+The common preprocessing stage retains incoming files, images, audio, video, and quoted/forwarded attachments after the session identity is finalized. Directory names reuse workspace UMO normalization, including the final isolated UMO for per-user group sessions. Adapters and shared downloaders may create internal staging files before localization. Members must copy incoming attachments into their workspace to edit them. Retained attachments remain subject to temporary-directory size cleanup and are not permanent storage.
 
-When enabled:
+Read/Grep, write/edit, restricted Shell/Python, and file sending use the corresponding session boundaries. Generated files, sandbox downloads, and local tool overflow output use the current workspace without granting access to shared temporary roots. Existing loose temporary files are not automatically migrated because they lack reliable session ownership; available attachments are localized when their messages pass through preprocessing again.
 
-- Admin users can use Shell, Python, file read, file write, file edit, and Grep search in `local` mode.
-- Non-admin users cannot use Shell or Python.
-- Non-admin users can only use file read, write, edit, and search inside restricted directories. Plugin-provided Skills are read/search-only and cannot be written or edited.
-
-Allowed directories for non-admin users in `local` mode include:
-
-- `data/skills`
-- `data/plugins/*/skills` (read-only, for plugin-provided Skills)
-- Current session's `data/workspaces/{normalized_umo}`
-- AstrBot temporary directories
-- `.astrbot` under the system temporary directory
-
-If `Require AstrBot admin permission` is disabled, regular users behave much closer to admins for Computer Use tools. Do not disable it unless you understand the risk.
+New macOS/Linux configurations default both roles to `workspace`, with code execution disabled for members. Windows defaults to disabled member access and administrator `host` access. Migrated configurations may retain administrator host access. `computer_use_require_admin` still gates third-party sandbox operations; it does not replace local per-role directory policies.
 
 Admin IDs can be configured in:
 

docs/zh/dev/plugin-platform-adapter.md

@@ -240,6 +240,8 @@ elif isinstance(i, Video):
 message_event.track_temporary_local_file(temp_media_path)
 ```
 
+公共预处理阶段会在最终 UMO 确定后，将消息组件及引用／转发中的附件复制到 `data/temp/platform_files/<normalized_umo>/`，并更新组件中的本地路径。登记的中转文件仍会在事件结束时清理，归档副本保留到临时目录容量清理。不要按下载时的原始群 ID 自行确定最终附件目录，因为群聊独立会话可能随后改写会话标识。
+
 最后，main.py 只需这样，在初始化的时候导入 fake_platform_adapter 模块。装饰器会自动注册。
 
 ```py

docs/zh/platform/weixin_oc.md

@@ -63,9 +63,9 @@ AstrBot 支持通过 `个人微信` 适配器接入微信个人号。该适配
 
 ## 多媒体文件保存位置
 
-接收到的图片、视频、文件、语音会被 AstrBot 下载并解密后保存到本地临时目录：
+接收到的图片、视频、文件、语音会先由适配器下载并解密，再由公共预处理阶段按最终会话标识归档：
 
-`data/temp`
+`data/temp/platform_files/<normalized_umo>/`
 
 这些文件属于 AstrBot 的临时缓存文件，后续插件、Agent 或文件服务可以继续读取和处理。
 

docs/zh/use/computer.md

@@ -64,29 +64,24 @@ data/workspaces/{normalized_umo}/notes/todo.txt
 
 ### 权限模型
 
-在本地权限表中，展开“文件访问范围”下拉菜单即可查看各选项的说明。“工作区”允许受限访问会话工作区、临时目录和技能文件；“整个环境”（原“主机文件”）允许访问 AstrBot 运行账户有权限访问的文件。Docker 部署时，“整个环境”指容器内及挂载的文件，并非宿主机全部文件。“关闭”不允许通过本地电脑能力工具访问任何文件。范围说明已从表格下方移到下拉菜单的对应选项中；将鼠标悬停在“AstrBot 所在运行环境的所有文件。”后的问号上，或点按问号，可查看账户权限和 Docker 访问范围的详细说明。
+在本地权限表中，展开“文件访问范围”下拉菜单即可查看各选项的说明。“工作区”允许受限访问当前工作区、平台附件和技能文件；“整个环境”（原“主机文件”）允许访问 AstrBot 运行账户有权限访问的文件。Docker 部署时，“整个环境”指容器内及挂载的文件，并非宿主机全部文件。“关闭”不允许通过本地电脑能力工具访问任何文件。范围说明已从表格下方移到下拉菜单的对应选项中；将鼠标悬停在“AstrBot 所在运行环境的所有文件。”后的问号上，或点按问号，可查看账户权限和 Docker 访问范围的详细说明。
 
-电脑能力还有一个独立开关：
+本地模式按身份分别配置 `filesystem_scope`、代码执行和联网权限。`none` 禁用本地文件工具，`host` 允许访问运行账户有权限访问的文件。`workspace` 使用以下目录规则：
 
-- `需要 AstrBot 管理员权限`
+| 目录 | 普通成员 | 管理员 |
+| --- | --- | --- |
+| 当前会话或项目工作区 | 读写 | 读写 |
+| `data/skills` | 只读 | 读写 |
+| `data/plugins/*/skills`、`astrbot/builtin_stars/*/skills` | 只读 | 只读 |
+| `data/temp/platform_files/<normalized_umo>/`（当前会话） | 只读 | 读写 |
+| 其他会话的 `platform_files` 子目录 | 不可访问 | 读写 |
+| 共享 `data/temp` 和系统临时目录 `.astrbot` 中的其他文件 | 不整体授权 | 不整体授权 |
 
-默认情况下这个开关是开启的。
+平台附件在会话标识确定后的公共预处理阶段归档，覆盖文件、图片、语音、视频和引用／转发中的附件。目录名复用工作区的 UMO 规范化规则；群聊独立会话使用隔离后的最终 UMO。适配器或公共下载器可能先写入内部中转文件，再归档到对应会话目录。原始附件对成员只读，需要编辑时请复制到当前工作区。归档目录仍受临时目录容量清理约束，不是永久存储。
 
-开启后：
+read/Grep、write/edit、受限 Shell/Python 和发送文件工具使用对应的会话边界。生成文件、沙箱下载文件和本地工具长输出放在当前工作区，不需要开放整个临时目录。既有散落临时文件不会自动迁移，因为它们没有可靠的会话归属；旧消息重新经过预处理时会归档其可用附件。
 
-- 管理员可以使用 `local` 模式下的 Shell、Python、文件读取、文件写入、文件编辑和 Grep 搜索。
-- 非管理员不能使用 Shell 和 Python。
-- 非管理员只能在受限目录内使用文件读取、写入、编辑和搜索。插件内置 Skills 只允许读取和搜索，不允许写入或编辑。
-
-非管理员在 `local` 模式下允许访问的目录包括：
-
-- `data/skills`
-- `data/plugins/*/skills`（只读，用于插件内置 Skills）
-- 当前会话的 `data/workspaces/{normalized_umo}`
-- AstrBot 的临时目录
-- 系统临时目录中的 `.astrbot`
-
-关闭“需要 AstrBot 管理员权限”后，普通用户在电脑能力工具上的行为会接近管理员。除非你非常清楚风险，否则不建议关闭。
+macOS/Linux 新配置默认让成员和管理员使用 `workspace`，成员默认不能执行代码；Windows 默认关闭成员访问，管理员使用 `host`。旧配置迁移可能保留管理员的整机访问权限。`computer_use_require_admin` 继续用于第三方沙箱操作，不能替代本地按身份配置的目录权限。
 
 管理员 ID 可在：
 

tests/test_computer_fs_tools.py

@@ -19,6 +19,7 @@ from astrbot.core.computer import file_read_utils, local_file_security
 from astrbot.core.computer.booters.local import LocalBooter
 from astrbot.core.tools.computer_tools import fs as fs_tools
 from astrbot.core.tools.computer_tools import util as computer_util
+from astrbot.core.utils import platform_files
 
 
 def _make_context(
@@ -78,8 +79,8 @@ async def test_sandbox_file_download_handles_windows_remote_filename(
     temp_root.mkdir(parents=True, exist_ok=True)
 
     monkeypatch.setattr(
-        fs_tools,
-        "get_astrbot_temp_path",
+        computer_util,
+        "get_astrbot_workspaces_path",
         lambda: str(temp_root),
     )
 
@@ -117,8 +118,8 @@ async def test_sandbox_file_download_strips_trailing_remote_slash(
     temp_root.mkdir(parents=True, exist_ok=True)
 
     monkeypatch.setattr(
-        fs_tools,
-        "get_astrbot_temp_path",
+        computer_util,
+        "get_astrbot_workspaces_path",
         lambda: str(temp_root),
     )
 
@@ -180,7 +181,7 @@ def _setup_local_fs_tools(
         lambda: str(builtin_plugins_root),
     )
     monkeypatch.setattr(
-        fs_tools,
+        platform_files,
         "get_astrbot_temp_path",
         lambda: str(temp_root),
     )
@@ -207,6 +208,81 @@ def _make_large_text() -> str:
     return "".join(f"line-{index:05d}-{'x' * 48}\n" for index in range(6000))
 
 
+@pytest.mark.skipif(os.name == "nt", reason="Restricted file descriptors require POSIX")
+@pytest.mark.asyncio
+@pytest.mark.parametrize("role", ["member", "admin"])
+@pytest.mark.parametrize("location", ["own", "other", "shared", "workspace"])
+@pytest.mark.parametrize("operation", ["read", "write", "edit"])
+async def test_platform_attachment_file_permissions(
+    monkeypatch, tmp_path, role, location, operation
+):
+    """Enforce session isolation and member read-only access at tool execution."""
+    workspace = _setup_local_fs_tools(monkeypatch, tmp_path)
+    context = _make_context(
+        role=role, local_permissions={role: {"filesystem_scope": "workspace"}}
+    )
+    own = platform_files.platform_files_root(context.context.event.unified_msg_origin)
+    other = platform_files.platform_files_root("qq:friend:another-user")
+    root = {
+        "own": own,
+        "other": other,
+        "shared": tmp_path / "temp",
+        "workspace": workspace,
+    }[location]
+    root.mkdir(parents=True, exist_ok=True)
+    target = root / "attachment.txt"
+    target.write_text("original attachment", encoding="utf-8")
+    allowed = location == "workspace" or (
+        location in {"own", "other"}
+        and (role == "admin" or (location == "own" and operation == "read"))
+    )
+    if operation == "read":
+        result = await fs_tools.FileReadTool().call(context, path=str(target))
+    elif operation == "write":
+        result = await fs_tools.FileWriteTool().call(
+            context, path=str(target), content="updated attachment"
+        )
+    else:
+        result = await fs_tools.FileEditTool().call(
+            context, path=str(target), old="original", new="updated"
+        )
+    assert ("restricted" not in str(result)) is allowed, result
+    assert "Error" not in str(result) if allowed else "Error" in str(result)
+    assert target.read_text() == (
+        "updated attachment"
+        if allowed and operation != "read"
+        else "original attachment"
+    )
+
+
+@pytest.mark.parametrize("role", ["member", "admin"])
+def test_platform_attachment_grep_roots(monkeypatch, tmp_path, role):
+    """Default grep includes only the caller's authorized attachment tree."""
+    workspace = _setup_local_fs_tools(monkeypatch, tmp_path)
+    umo = "qq:friend:user-1"
+    own = platform_files.platform_files_root(umo)
+    other = platform_files.platform_files_root("qq:friend:another-user")
+    own.mkdir(parents=True)
+    other.mkdir(parents=True)
+    kwargs = dict(
+        restricted=True,
+        local_env=True,
+        umo=umo,
+        current_workspace_root=workspace,
+        is_admin=role == "admin",
+    )
+    roots = fs_tools.GrepTool()._normalize_search_paths(None, **kwargs)
+    assert str(own if role == "member" else own.parent) in roots
+    assert str(tmp_path / "temp") not in roots
+    if role == "member":
+        with pytest.raises(PermissionError):
+            fs_tools.GrepTool()._normalize_search_paths(str(other), **kwargs)
+    else:
+        assert fs_tools.GrepTool()._normalize_search_paths(str(other), **kwargs) == [
+            str(other)
+        ]
+
+
 def _make_hardlink_or_skip(source, link) -> None:
     try:
         os.link(source, link)
@@ -760,7 +836,9 @@ async def test_restricted_local_grep_requests_read_only_os_sandbox(
     (workspace / "target.txt").write_text("needle\n", encoding="utf-8")
     missing_root = tmp_path / "missing-temp"
     monkeypatch.setattr(
-        fs_tools, "_read_allowed_roots", lambda *args: (workspace, missing_root)
+        fs_tools,
+        "_read_allowed_roots",
+        lambda *args, **kwargs: (workspace, missing_root),
     )
     calls = []
 

tests/test_local_sandbox_access.py

@@ -23,6 +23,7 @@ from astrbot.core.computer.process_sandbox import (
     unix,
 )
 from astrbot.core.tools.computer_tools import fs, python, shell
+from astrbot.core.utils import platform_files
 from astrbot.dashboard.services import stat_service
 
 requires_local_sandbox = pytest.mark.skipif(
@@ -413,8 +414,7 @@ async def test_local_execution_obeys_file_tool_roots(
     monkeypatch.setattr(
         fs, "get_astrbot_builtin_plugin_path", lambda: str(builtin.parent.parent)
     )
-    monkeypatch.setattr(fs, "get_astrbot_temp_path", lambda: str(temporary))
-    monkeypatch.setattr(fs, "get_astrbot_system_tmp_path", lambda: str(system_temp))
+    monkeypatch.setattr(platform_files, "get_astrbot_temp_path", lambda: str(temporary))
     monkeypatch.setattr(local, "get_astrbot_system_tmp_path", lambda: str(system_temp))
     monkeypatch.setattr(
         shell, "workspace_root_for_context", AsyncMock(return_value=workspace)
@@ -436,7 +436,7 @@ async def test_local_execution_obeys_file_tool_roots(
                     "allow_network": False,
                 }
             },
-        }
+        },
     }
     event = SimpleNamespace(
         role=role,
@@ -449,10 +449,25 @@ async def test_local_execution_obeys_file_tool_roots(
         ),
         tool_call_timeout=20,
     )
-    readable = fs._read_allowed_roots(event.unified_msg_origin, workspace)
-    writable = fs._write_allowed_roots(
-        event.unified_msg_origin, workspace, include_installed_skills=role == "admin"
+    own_attachments = platform_files.platform_files_root(event.unified_msg_origin)
+    other_attachments = platform_files.platform_files_root("test:friend:other")
+    for root in (own_attachments, other_attachments, own_attachments.parent):
+        root.mkdir(parents=True, exist_ok=True)
+        (root / "sample.txt").write_text("approved content", encoding="utf-8")
+    readable = fs._read_allowed_roots(
+        event.unified_msg_origin, workspace, is_admin=role == "admin"
     )
+    writable = fs._write_allowed_roots(
+        event.unified_msg_origin, workspace, is_admin=role == "admin"
+    )
+    blocked = [
+        secret,
+        workspace / "outside-link",
+        temporary / "sample.txt",
+        system_temp / "sample.txt",
+    ]
+    if role == "member":
+        blocked.append(other_attachments / "sample.txt")
     code = f"""
 import runpy
 from pathlib import Path
@@ -469,8 +484,8 @@ for root in {list(map(str, set(readable) - set(writable)))!r}:
     except OSError:
         pass
     else:
-        raise AssertionError("Skill directory is writable: " + root)
-for path in {[str(secret), str(workspace / "outside-link")]!r}:
+        raise AssertionError("Read-only directory is writable: " + root)
+for path in {list(map(str, blocked))!r}:
     try:
         Path(path).read_text()
     except OSError:

tests/test_preprocess_stage.py

@@ -1,23 +1,35 @@
 import base64
 from io import BytesIO
+from pathlib import Path
 from types import SimpleNamespace
 
 import pytest
 
-from astrbot.core.message.components import Image, Plain, Reply
+from astrbot.core.message.components import (
+    File,
+    Image,
+    Node,
+    Nodes,
+    Plain,
+    Record,
+    Reply,
+    Video,
+)
 from astrbot.core.pipeline.preprocess_stage import stage as preprocess_stage
 from astrbot.core.pipeline.preprocess_stage.stage import PreProcessStage
 from astrbot.core.platform.astr_message_event import AstrMessageEvent
-from astrbot.core.utils import media_utils
+from astrbot.core.utils import media_utils, platform_files
 
 
 class FakeEvent:
     def __init__(self, message):
         self.message_obj = SimpleNamespace(message=message, message_str="")
         self.message_str = ""
+        self.unified_msg_origin = "test:GroupMessage:group-user"
         self.is_at_or_wake_command = False
         self.temporary_local_files: list[str] = []
         self._temporary_local_files = self.temporary_local_files
+        self._extras = {}
 
     def get_platform_name(self):
         return "test"
@@ -27,6 +39,21 @@ class FakeEvent:
 
     track_temporary_local_file = AstrMessageEvent.track_temporary_local_file
     untrack_temporary_local_file = AstrMessageEvent.untrack_temporary_local_file
+    get_extra = AstrMessageEvent.get_extra
+    set_extra = AstrMessageEvent.set_extra
+
+
+@pytest.fixture(autouse=True)
+def isolate_attachment_storage(tmp_path, monkeypatch):
+    """Keep retained attachments and staging files inside each test directory."""
+    monkeypatch.setattr(
+        preprocess_stage, "get_astrbot_temp_path", lambda: str(tmp_path)
+    )
+    monkeypatch.setattr(
+        platform_files,
+        "get_astrbot_temp_path",
+        lambda: preprocess_stage.get_astrbot_temp_path(),
+    )
 
 
 @pytest.mark.asyncio
@@ -138,15 +165,20 @@ async def test_preprocess_image_cleanup_preserves_usable_file(
 
     await stage.process(event)
 
-    assert event.temporary_local_files == []
-    assert image.file == image.path == image.url == str(source_path)
+    assert image.file == image.path == image.url
+    retained = Path(image.file)
+    assert retained.parent == platform_files.platform_files_root(
+        event.unified_msg_origin
+    )
+    assert retained.read_bytes() == original
     assert source_path.read_bytes() == original
 
     # Exercise event cleanup to verify that the usable image survives.
     AstrMessageEvent.cleanup_temporary_local_files(
         SimpleNamespace(_temporary_local_files=event.temporary_local_files)
     )
-    assert source_path.read_bytes() == original
+    assert retained.read_bytes() == original
+    assert source_path.exists() is (not pretracked)
     assert Path(await image.convert_to_file_path()).exists()
 
 
@@ -200,7 +232,11 @@ async def test_preprocess_path_mapping_accepts_file_uri(tmp_path):
 
     image = event.get_messages()[0]
     assert isinstance(image, Image)
-    assert image.file == image.path == image.url == str(target_image)
+    assert image.file == image.path == image.url
+    assert Path(image.file).parent == platform_files.platform_files_root(
+        event.unified_msg_origin
+    )
+    assert Path(image.file).read_bytes() == target_image.read_bytes()
 
 
 @pytest.mark.asyncio
@@ -225,4 +261,151 @@ async def test_preprocess_path_mapping_accepts_windows_source_to_posix_target(
 
     image = event.get_messages()[0]
     assert isinstance(image, Image)
-    assert image.file == image.path == image.url == str(target_image)
+    assert image.file == image.path == image.url
+    assert Path(image.file).parent == platform_files.platform_files_root(
+        event.unified_msg_origin
+    )
+    assert Path(image.file).read_bytes() == target_image.read_bytes()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", ["file", "image", "record", "video"])
+@pytest.mark.parametrize("nested", [False, True])
+async def test_incoming_attachments_use_final_session_storage(
+    tmp_path, monkeypatch, kind, nested
+):
+    """Retain all attachment types, including forwarded quotes, through cleanup."""
+    from PIL import Image as PILImage
+
+    from astrbot.core.message import components
+
+    source = tmp_path / "input.bin"
+    source.write_bytes(b"platform attachment")
+    if kind == "file":
+
+        async def download(url, path):
+            Path(path).write_bytes(source.read_bytes())
+
+        monkeypatch.setattr(components, "download_file", download)
+        monkeypatch.setattr(components, "get_astrbot_temp_path", lambda: str(tmp_path))
+        component = File(name="report.txt", url="https://example.test/file")
+    elif kind == "image":
+        source = tmp_path / "input.png"
+        PILImage.new("RGB", (2, 2)).save(source)
+        component = Image.fromFileSystem(str(source))
+    elif kind == "record":
+        import wave
+
+        source = tmp_path / "input.wav"
+        with wave.open(str(source), "wb") as audio:
+            audio.setnchannels(1)
+            audio.setsampwidth(2)
+            audio.setframerate(8000)
+            audio.writeframes(b"\0\0" * 80)
+        component = Record.fromFileSystem(str(source))
+    else:
+        component = Video.fromFileSystem(str(source))
+    chain = [component]
+    if nested:
+        chain = [Reply(id="quoted", chain=[Nodes(nodes=[Node(content=chain)])])]
+    event = FakeEvent(chain)
+    event.unified_msg_origin = "platform-instance:GroupMessage:group-isolated-user"
+    event.track_temporary_local_file(str(source))
+    stage = PreProcessStage()
+    stage.config = {}
+    stage.platform_settings = {}
+    stage.stt_settings = {"enable": False}
+
+    await stage.process(event)
+
+    retained = Path(component.file_ if isinstance(component, File) else component.path)
+    assert retained.parent == platform_files.platform_files_root(
+        event.unified_msg_origin
+    )
+    assert retained.read_bytes() == source.read_bytes()
+    if isinstance(component, File):
+        assert component.name == "report.txt"
+        assert await component.get_file() == str(retained)
+    else:
+        assert component.file == component.path == component.url == str(retained)
+    AstrMessageEvent.cleanup_temporary_local_files(event)
+    assert retained.is_file()
+    assert not source.exists()
+
+    # Reprocessing a retained message must not duplicate or delete the attachment.
+    await stage.process(event)
+    assert len(list(retained.parent.iterdir())) == 1
+    AstrMessageEvent.cleanup_temporary_local_files(event)
+    assert retained.is_file()
+
+
+@pytest.mark.asyncio
+async def test_retained_files_are_distinct_across_messages_and_sessions(tmp_path):
+    """Equal filenames cannot overwrite another message's or session's content."""
+    source = tmp_path / "report.txt"
+    source.write_text("first", encoding="utf-8")
+    first = await platform_files.retain_platform_file(str(source), "test:friend:one")
+    source.write_text("second", encoding="utf-8")
+    second = await platform_files.retain_platform_file(str(source), "test:friend:one")
+    other = await platform_files.retain_platform_file(str(source), "test:friend:two")
+    assert len({first, second, other}) == 3
+    assert Path(first).read_text() == "first"
+    assert Path(second).read_text() == Path(other).read_text() == "second"
+    assert Path(first).parent != Path(other).parent
+
+
+@pytest.mark.asyncio
+async def test_cancelled_attachment_copy_cleans_up_after_worker_finishes(
+    tmp_path, monkeypatch
+):
+    """Cancellation cannot leave a worker writing an orphaned attachment afterward."""
+    import asyncio
+    import threading
+
+    source = tmp_path / "input.txt"
+    source.write_text("content", encoding="utf-8")
+    started = asyncio.Event()
+    release = threading.Event()
+    loop = asyncio.get_running_loop()
+    original_copy = platform_files.shutil.copyfile
+
+    def delayed_copy(src, dst):
+        loop.call_soon_threadsafe(started.set)
+        assert release.wait(timeout=5)
+        return original_copy(src, dst)
+
+    monkeypatch.setattr(platform_files.shutil, "copyfile", delayed_copy)
+    task = asyncio.create_task(
+        platform_files.retain_platform_file(str(source), "test:friend:one")
+    )
+    await asyncio.wait_for(started.wait(), timeout=5)
+    task.cancel()
+    release.set()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert list(platform_files.platform_files_root("test:friend:one").iterdir()) == []
+    assert source.read_text() == "content"
+
+
+@pytest.mark.asyncio
+async def test_duplicate_encoded_images_share_retained_file(tmp_path, monkeypatch):
+    """An image repeated in a quote shares its retained source and is not decoded twice."""
+    from PIL import Image as PILImage
+
+    buffer = BytesIO()
+    PILImage.new("RGB", (2, 2)).save(buffer, format="PNG")
+    ref = "base64://" + base64.b64encode(buffer.getvalue()).decode()
+    original = Image(file=ref)
+    quoted = Image(file=ref)
+    event = FakeEvent([original, Reply(id="quote", chain=[quoted])])
+    monkeypatch.setattr(media_utils, "get_astrbot_temp_path", lambda: str(tmp_path))
+    stage = PreProcessStage()
+    stage.config = {}
+    stage.platform_settings = {}
+    stage.stt_settings = {"enable": False}
+    await stage.process(event)
+    assert original.path == quoted.path
+    root = platform_files.platform_files_root(event.unified_msg_origin)
+    assert list(root.iterdir()) == [Path(original.path)]
+    AstrMessageEvent.cleanup_temporary_local_files(event)
+    assert Path(original.path).read_bytes() == buffer.getvalue()

tests/test_process_stage_images.py

@@ -20,6 +20,7 @@ from astrbot.core.agent.message import (
     TextPart,
     dump_messages_with_checkpoints,
 )
+from astrbot.core.agent.tool import ToolSet
 from astrbot.core.config.default import DEFAULT_CONFIG
 from astrbot.core.message.components import Image, Plain, Reply
 from astrbot.core.pipeline.preprocess_stage import stage as preprocess
@@ -34,7 +35,7 @@ from astrbot.core.platform.platform_metadata import PlatformMetadata
 from astrbot.core.provider.entities import LLMResponse, ProviderRequest
 from astrbot.core.provider.provider import Provider
 from astrbot.core.star.star_handler import EventType
-from astrbot.core.utils import image_input
+from astrbot.core.utils import image_input, platform_files
 from astrbot.core.utils import media_utils as media
 
 
@@ -85,7 +86,7 @@ def source_image(tmp_path, fmt="GIF"):
 @pytest.fixture
 def harness(tmp_path, monkeypatch):
     work = tmp_path / "work"
-    for module in (media, image_input, preprocess):
+    for module in (media, image_input, preprocess, platform_files):
         monkeypatch.setattr(module, "get_astrbot_temp_path", lambda: str(work))
     config = copy.deepcopy(DEFAULT_CONFIG)
     config["provider_settings"].update(
@@ -104,6 +105,7 @@ def harness(tmp_path, monkeypatch):
     )
     context.persona_manager.personas_v3 = []
     context.subagent_orchestrator = None
+    context.get_llm_tool_manager.return_value.get_full_tool_set.return_value = ToolSet()
     context.get_llm_tool_manager.return_value.get_builtin_tool.side_effect = (
         lambda cls, **kwargs: cls(**kwargs)
     )
@@ -219,14 +221,16 @@ async def test_legacy_toggle_does_not_disable_preparation(
     assert Path(event.get_messages()[0].file).read_bytes() == original
     assert str(source) not in event._temporary_local_files
     assert any(
-        str(source) in p.text
+        event.get_messages()[0].file in p.text
         for p in req.extra_user_content_parts
         if isinstance(p, TextPart)
     )
     assert harness.provider.text_chat.await_count == 1
     assert "image_settings" not in harness.provider.text_chat.await_args.kwargs
     visual_path = Path(req.image_urls[0])
-    assert (visual_path == source) == (fmt in {"JPEG", "PNG"})
+    assert (str(visual_path) == event.get_messages()[0].file) == (
+        fmt in {"JPEG", "PNG"}
+    )
     event.cleanup_temporary_local_files()
     assert source.read_bytes() == original
     assert visual_path.exists() == (fmt in {"JPEG", "PNG"})
@@ -549,7 +553,7 @@ async def test_provider_and_history_keep_bounded_previews_and_original_paths(
     texts = [
         part.text for part in req.extra_user_content_parts if isinstance(part, TextPart)
     ]
-    assert any(str(source) in text for text in texts)
+    assert any(part.file in text for text in texts)
     assert all(str(preview) not in text for text in texts)
     provider_messages = harness.provider.text_chat.await_args.kwargs["contexts"]
     history = (
@@ -644,7 +648,7 @@ async def test_caption_animation_notice_uses_actual_image_order(harness, tmp_pat
     caption.text_chat.assert_awaited_once()
     args = caption.text_chat.await_args.kwargs
     assert len(args["image_urls"]) == 2
-    assert args["image_urls"][0] == str(still)
+    assert args["image_urls"][0] == event.get_messages()[0].file
     assert "positions 2 (1-based)" in args["prompt"]
     assert "Describe them as animations" in args["prompt"]
     assert "do not mention the conversion or frame layout" in args["prompt"]
@@ -729,7 +733,7 @@ async def test_oversized_images_explain_omission_and_keep_original_path(
         if part["type"] == "text" and "skipped: exceeds" in part["text"]
     ]
     assert len(notices) == 1
-    assert "64 MiB" in notices[0] and str(source) in notices[0]
+    assert "64 MiB" in notices[0] and attachments[0].file in notices[0]
     advice = [
         part["text"]
         for part in payload
@@ -988,10 +992,10 @@ async def test_third_party_route_retains_original_images(
         "prepare_model_image",
         AsyncMock(side_effect=AssertionError("third party must not prepare")),
     )
-    await process_event(
-        harness, make_event([Image(file=str(source))]), preprocess_first=True
-    )
-    assert seen == [str(source)] and not harness.captured
+    event = make_event([Image(file=str(source))])
+    await process_event(harness, event, preprocess_first=True)
+    assert seen == [event.get_messages()[0].file] and not harness.captured
+    assert Path(seen[0]).read_bytes() == source.read_bytes()
 
 
 @pytest.mark.asyncio
@@ -1094,8 +1098,9 @@ async def test_attachment_sources_survive_event_cleanup(
         assert max(visual_image.size) == 12
 
     owned = [Path(path) for path in event._temporary_local_files]
-    assert len(owned) == 2
+    assert len(owned) >= 2
     assert unrelated in owned
+    assert all(path.is_relative_to(harness.work) for path in owned)
     event.cleanup_temporary_local_files()
     assert attachment_path.read_bytes() == original
     assert all(not path.exists() for path in owned)
@@ -1203,6 +1208,8 @@ async def test_materialized_sources_remain_owned_when_collection_fails(
     )
     if fail_at == "attachment":
         failing_image = MagicMock(spec=Image)
+        failing_image.url = ""
+        failing_image.file = "unavailable-image"
         failing_image.convert_to_file_path = AsyncMock(
             side_effect=failure("unavailable")
         )
@@ -1260,10 +1267,12 @@ async def test_final_image_labels_follow_mixed_visuals_after_hook(
             req.extra_user_content_parts.append(
                 TextPart(text="[Image captions supplied by a plugin]")
             )
-            req.image_urls = ([] if remove_image else [str(animation)]) + [
-                str(still),
-                str(oversized),
-                str(still),
+            req.image_urls = (
+                [] if remove_image else [event.get_messages()[2].chain[0].file]
+            ) + [
+                event.get_messages()[1].file,
+                event.get_messages()[0].file,
+                event.get_messages()[1].file,
             ]
         return False
 
@@ -1272,7 +1281,14 @@ async def test_final_image_labels_follow_mixed_visuals_after_hook(
     assert convert.await_count == 4
     req = harness.captured[0].req
     payload = harness.provider.text_chat.await_args.kwargs["contexts"][-1]["content"]
-    expected = [inserted, still] if remove_image else [inserted, animation, still]
+    archived_oversized = event.get_messages()[0].file
+    archived_still = event.get_messages()[1].file
+    archived_animation = event.get_messages()[2].chain[0].file
+    expected = (
+        [str(inserted), archived_still]
+        if remove_image
+        else [str(inserted), archived_animation, archived_still]
+    )
     images = [
         index for index, part in enumerate(payload) if part["type"] == "image_url"
     ]
@@ -1292,20 +1308,30 @@ async def test_final_image_labels_follow_mixed_visuals_after_hook(
             and part["text"].startswith(f"[Image {number} in quoted message:")
         )
         assert f"original path {original}" in label["text"]
-        assert ("in quoted message" in label["text"]) == (original == animation)
-        assert ("3x3 frame montage" in label["text"]) == (original == animation)
+        assert ("in quoted message" in label["text"]) == (
+            original == archived_animation
+        )
+        assert ("3x3 frame montage" in label["text"]) == (
+            original == archived_animation
+        )
         data = base64.b64decode(payload[position]["image_url"]["url"].split(",", 1)[1])
         assert data == Path(visual_paths[number - 1]).read_bytes()
     texts = [part["text"] for part in payload if part["type"] == "text"]
     assert "[Image captions supplied by a plugin]" in texts
-    for original in (oversized, still, animation, inserted):
-        assert sum(str(original) in text for text in texts) == 1
+    for original in (
+        archived_oversized,
+        archived_still,
+        archived_animation,
+        str(inserted),
+    ):
+        assert sum(original in text for text in texts) == 1
     assert any(
-        str(oversized) in text and "skipped: exceeds 64 MiB" in text for text in texts
+        archived_oversized in text and "skipped: exceeds 64 MiB" in text
+        for text in texts
     )
     if remove_image:
         assert any(
-            str(animation) in text and "not included in this request" in text
+            archived_animation in text and "not included in this request" in text
             for text in texts
         )
     history = (
@@ -1354,7 +1380,8 @@ async def test_captioned_quote_has_status_without_visual_index(
     labels = [
         part["text"]
         for part in payload
-        if part["type"] == "text" and str(source) in part["text"]
+        if part["type"] == "text"
+        and event.get_messages()[0].chain[0].file in part["text"]
     ]
     assert len(labels) == 1
     assert "in quoted message" in labels[0]
@@ -1413,8 +1440,14 @@ async def test_duplicate_current_and_quoted_image_share_one_visual_label(
         for part in content
         if part.get("text", "").startswith("[Image 1 in quoted message:")
     )
-    assert label == f"[Image 1 in quoted message: original path {source}]"
-    assert sum(str(source) in part.get("text", "") for part in content) == 1
+    assert (
+        label
+        == f"[Image 1 in quoted message: original path {event.get_messages()[0].file}]"
+    )
+    assert (
+        sum(event.get_messages()[0].file in part.get("text", "") for part in content)
+        == 1
+    )
 
 
 @pytest.mark.asyncio
@@ -1450,7 +1483,7 @@ async def test_hook_only_new_refs_receive_full_image_policy(
             with PILImage.open(req.image_urls[0]) as image:
                 assert image.size == (90, 45)
             # Reintroducing an already processed original must also reuse its preview.
-            req.image_urls.append(str(initial))
+            req.image_urls.append(event.get_messages()[0].file)
             if origin == "image_urls":
                 req.image_urls.append(str(added))
             elif origin == "image_part":
@@ -1467,7 +1500,7 @@ async def test_hook_only_new_refs_receive_full_image_policy(
     event = make_event([Image(file=str(initial))])
     await process_event(harness, event, preprocess_first=True)
     assert [call.args[0] for call in convert.await_args_list] == [
-        str(initial),
+        event.get_messages()[0].file,
         str(added),
     ]
     assert [call.kwargs["max_size"] for call in convert.await_args_list] == [90, 90]
@@ -1600,3 +1633,85 @@ async def test_only_actually_captioned_images_receive_caption_status(
         assert len(labels) == 1
         assert ("description included as text" in labels[0]) == captioned
         assert ("not included in this request" in labels[0]) != captioned
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("quoted", [False, True])
+@pytest.mark.parametrize("kind", ["image", "audio", "video", "file"])
+async def test_direct_collection_reuses_paths_from_message_segments(
+    harness, tmp_path, monkeypatch, kind, quoted
+):
+    """Late attachments write back their paths and survive repeated collection."""
+    from astrbot.core.message.components import File, Record, Video
+
+    if kind == "image":
+        source = source_image(tmp_path, "PNG")
+        component = Image(file=str(source))
+    else:
+        source = (
+            tmp_path
+            / {"audio": "voice.wav", "video": "clip.mp4", "file": "note.txt"}[kind]
+        )
+        if kind == "audio":
+            import wave
+
+            with wave.open(str(source), "wb") as wav:
+                wav.setnchannels(1)
+                wav.setsampwidth(2)
+                wav.setframerate(16000)
+                wav.writeframes(b"\x00\x00" * 160)
+        else:
+            source.write_bytes(b"platform attachment")
+        component = {
+            "audio": lambda: Record(file=str(source)),
+            "video": lambda: Video(file=str(source)),
+            "file": lambda: File(name=source.name, file=str(source)),
+        }[kind]()
+    event = make_event(
+        [Reply(id="quoted", chain=[component])] if quoted else [component]
+    )
+    monkeypatch.setattr(
+        main, "extract_quoted_message_images", AsyncMock(return_value=[])
+    )
+    config = main.MainAgentBuildConfig(tool_call_timeout=60)
+    await main.collect_initial_request(event, harness.context, config)
+    retained_path = component.file_ if isinstance(component, File) else component.path
+    retained = Path(retained_path)
+    assert retained.parent == platform_files.platform_files_root(
+        event.unified_msg_origin
+    )
+    assert retained.read_bytes() == source.read_bytes()
+    if not isinstance(component, File):
+        assert component.file == component.url == retained_path
+
+    event.set_extra("provider_request", None)
+    await main.collect_initial_request(event, harness.context, config)
+    current_path = component.file_ if isinstance(component, File) else component.path
+    assert current_path == retained_path
+    assert list(retained.parent.iterdir()) == [retained]
+    event.cleanup_temporary_local_files()
+    assert retained.is_file()
+
+
+@pytest.mark.asyncio
+async def test_resolved_quote_images_are_saved_in_reply_chain(
+    harness, tmp_path, monkeypatch
+):
+    """An ID-only reply keeps resolved images in its segments for later collection."""
+    source = source_image(tmp_path, "PNG")
+    extract = AsyncMock(return_value=[str(source)])
+    monkeypatch.setattr(main, "extract_quoted_message_images", extract)
+    reply = Reply(id="quoted")
+    event = make_event([reply])
+    config = main.MainAgentBuildConfig(tool_call_timeout=60)
+    request, _ = await main.collect_initial_request(event, harness.context, config)
+    assert len(reply.chain) == 1
+    image = reply.chain[0]
+    assert isinstance(image, Image)
+    assert image.file == image.url == image.path == request.image_urls[0]
+    assert Path(image.path).read_bytes() == source.read_bytes()
+
+    event.set_extra("provider_request", None)
+    repeated, _ = await main.collect_initial_request(event, harness.context, config)
+    assert repeated.image_urls == request.image_urls
+    extract.assert_awaited_once()

tests/test_tool_loop_agent_runner.py

@@ -2398,3 +2398,22 @@ async def test_small_step_budgets_and_reset(
     )
     assert runner._step_budget_used == 0
     assert runner._step_budget_notified == set()
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(sys.platform == "win32", reason="Restricted opening requires POSIX")
+async def test_tool_overflow_rejects_replaced_workspace_root(tmp_path):
+    """Do not spill host output through a workspace root replaced by a symlink."""
+    workspace = tmp_path / "workspace"
+    outside = tmp_path / "outside"
+    workspace.mkdir()
+    outside.mkdir()
+    runner = ToolLoopAgentRunner()
+    runner.tool_result_overflow_dir = str(workspace)
+    workspace.rmdir()
+    workspace.symlink_to(outside, target_is_directory=True)
+    with pytest.raises(PermissionError):
+        await runner._write_tool_result_overflow_file(
+            tool_call_id="output", content="private tool output"
+        )
+    assert list(outside.iterdir()) == []

tests/unit/test_astr_main_agent.py

@@ -30,6 +30,15 @@ from astrbot.core.provider.manager import ProviderManager
 from astrbot.core.skills.skill_manager import SkillInfo
 from astrbot.core.star.context import Context
 from astrbot.core.star.star import StarMetadata
+from astrbot.core.utils import platform_files
+
+
+@pytest.fixture(autouse=True)
+def isolate_platform_attachments(tmp_path, monkeypatch):
+    """Keep direct-builder attachment storage inside the test directory."""
+    monkeypatch.setattr(
+        platform_files, "get_astrbot_temp_path", lambda: str(tmp_path / "temp")
+    )
 
 
 @pytest.fixture
@@ -96,7 +105,9 @@ def mock_event():
     event.platform_meta = platform_meta
     event.session_id = "session123"
     event.unified_msg_origin = "test_platform:private:session123"
-    event.get_extra.return_value = None
+    extras = {}
+    event.get_extra.side_effect = lambda key, default=None: extras.get(key, default)
+    event.set_extra.side_effect = lambda key, value: extras.__setitem__(key, value)
     event.get_platform_name.return_value = "test_platform"
     event.get_platform_id.return_value = "test_platform"
     event.get_group_id.return_value = None
@@ -1941,7 +1952,9 @@ class TestBuildMainAgent:
         assert result is not None
         request = result.provider_request
         label = "Image 1 in quoted message" if quoted else "Image 1"
-        assert f"[{label}: original path {source_path}]" in [
+        retained_path = image.file
+        assert Path(retained_path).read_bytes() == original
+        assert f"[{label}: original path {retained_path}]" in [
             part.text for part in request.extra_user_content_parts
         ]
         assert len(request.image_urls) == 1
@@ -1950,9 +1963,7 @@ class TestBuildMainAgent:
         with PILImage.open(visual_path) as visual_image:
             assert visual_image.size == (4, 4)
         mock_event.track_temporary_local_file.assert_called_once_with(str(visual_path))
-        mock_event.untrack_temporary_local_file.assert_called_once_with(
-            str(source_path)
-        )
+        mock_event.untrack_temporary_local_file.assert_called_once_with(retained_path)
         AstrMessageEvent.cleanup_temporary_local_files(
             SimpleNamespace(
                 _temporary_local_files=[
@@ -2011,7 +2022,7 @@ class TestBuildMainAgent:
             )
 
         assert result is not None
-        assert result.provider_request.image_urls == [valid_image_path]
+        assert result.provider_request.image_urls == [mock_image.file]
         assert not any(
             "Image Caption" in part.text or "<image_caption>" in part.text
             for part in result.provider_request.extra_user_content_parts
@@ -2271,12 +2282,13 @@ class TestBuildMainAgent:
 
     @pytest.mark.asyncio
     async def test_build_main_agent_with_video_attachment(
-        self, mock_event, mock_context, mock_provider
+        self, mock_event, mock_context, mock_provider, tmp_path
     ):
         """Test building main agent with video attachments."""
         module = ama
-        video_path = str(Path("/path/to/video.mp4"))
-        mock_video = Video(file="file:///path/to/video.mp4")
+        source_path = tmp_path / "video.mp4"
+        source_path.write_bytes(b"video attachment")
+        mock_video = Video.fromFileSystem(str(source_path))
         mock_event.message_obj.message = [mock_video]
 
         mock_context.get_provider_by_id.return_value = None
@@ -2301,18 +2313,21 @@ class TestBuildMainAgent:
             )
 
         assert result is not None
+        video_path = mock_video.file
+        assert Path(video_path).read_bytes() == source_path.read_bytes()
         assert [
             part.text for part in result.provider_request.extra_user_content_parts
-        ] == [f"[Video Attachment: name video.mp4, path {video_path}]"]
+        ] == [f"[Video Attachment: name {Path(video_path).name}, path {video_path}]"]
 
     @pytest.mark.asyncio
     async def test_build_main_agent_with_quoted_video_attachment(
-        self, mock_event, mock_context, mock_provider
+        self, mock_event, mock_context, mock_provider, tmp_path
     ):
         """Test building main agent with quoted video attachments."""
         module = ama
-        video_path = str(Path("/path/to/quoted-video.mp4"))
-        mock_video = Video(file="file:///path/to/quoted-video.mp4")
+        source_path = tmp_path / "quoted-video.mp4"
+        source_path.write_bytes(b"video attachment")
+        mock_video = Video.fromFileSystem(str(source_path))
         mock_reply = Reply(
             id="reply-1",
             chain=[mock_video],
@@ -2343,9 +2358,11 @@ class TestBuildMainAgent:
             )
 
         assert result is not None
+        video_path = mock_video.file
+        assert Path(video_path).read_bytes() == source_path.read_bytes()
         assert (
             "[Video Attachment in quoted message: "
-            f"name quoted-video.mp4, path {video_path}]"
+            f"name {Path(video_path).name}, path {video_path}]"
         ) in [part.text for part in result.provider_request.extra_user_content_parts]
 
     @pytest.mark.asyncio

tests/unit/test_message_tools.py

@@ -410,8 +410,8 @@ async def test_non_admin_can_send_workspace_file(tmp_path, monkeypatch, scope):
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("scope", ["none", "workspace"])
-async def test_non_admin_can_send_temp_file(tmp_path, monkeypatch, scope):
-    """Non-admin users can send generated files under AstrBot temp."""
+async def test_non_admin_cannot_send_shared_temp_file(tmp_path, monkeypatch, scope):
+    """Shared temporary files are not accessible through the send tool."""
     tool = SendMessageToUserTool()
     ctx = _make_context(
         role="member", local_permissions={"member": {"filesystem_scope": scope}}
@@ -421,7 +421,7 @@ async def test_non_admin_can_send_temp_file(tmp_path, monkeypatch, scope):
     output_path = temp_root / "output.txt"
     output_path.write_text("output", encoding="utf-8")
     monkeypatch.setattr(
-        "astrbot.core.tools.message_tools.get_astrbot_temp_path",
+        "astrbot.core.utils.platform_files.get_astrbot_temp_path",
         lambda: str(temp_root),
     )
 
@@ -430,8 +430,40 @@ async def test_non_admin_can_send_temp_file(tmp_path, monkeypatch, scope):
         messages=[{"type": "file", "path": str(output_path)}],
     )
 
-    assert "Message sent to session" in result
-    ctx.context.context.send_message.assert_called_once()
+    assert "restricted" in result
+    ctx.context.context.send_message.assert_not_called()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("role", ["member", "admin"])
+@pytest.mark.parametrize("own_session", [False, True])
+async def test_send_attachment_obeys_session_read_permissions(
+    tmp_path, monkeypatch, role, own_session
+):
+    """Sending must not bypass the member's attachment read boundary."""
+    from astrbot.core.utils import platform_files
+
+    ctx = _make_context(
+        role=role, local_permissions={role: {"filesystem_scope": "workspace"}}
+    )
+    monkeypatch.setattr(platform_files, "get_astrbot_temp_path", lambda: str(tmp_path))
+    umo = (
+        ctx.context.event.unified_msg_origin
+        if own_session
+        else "other:GroupMessage:group"
+    )
+    path = platform_files.platform_files_root(umo) / "file.txt"
+    path.parent.mkdir(parents=True)
+    path.write_text("attachment", encoding="utf-8")
+    result = await SendMessageToUserTool().call(
+        ctx, messages=[{"type": "file", "path": str(path)}]
+    )
+    if role == "admin" or own_session:
+        assert "Message sent to session" in result
+        ctx.context.context.send_message.assert_called_once()
+    else:
+        assert "restricted" in result
+        ctx.context.context.send_message.assert_not_called()
 
 
 @pytest.mark.asyncio
@@ -444,7 +476,7 @@ async def test_send_message_downloads_windows_sandbox_file_with_original_name(
     temp_root = tmp_path / "temp"
     temp_root.mkdir()
     monkeypatch.setattr(
-        "astrbot.core.tools.message_tools.get_astrbot_temp_path",
+        "astrbot.core.tools.computer_tools.util.get_astrbot_workspaces_path",
         lambda: str(temp_root),
     )
 
@@ -492,7 +524,7 @@ async def test_send_message_downloads_trailing_slash_sandbox_file_with_basename(
     temp_root = tmp_path / "temp"
     temp_root.mkdir()
     monkeypatch.setattr(
-        "astrbot.core.tools.message_tools.get_astrbot_temp_path",
+        "astrbot.core.tools.computer_tools.util.get_astrbot_workspaces_path",
         lambda: str(temp_root),
     )