feat(chat): hand out fresh attachment links and clean up on delete

Links issued at upload time expire, so the client now asks for a new one when
it renders an attachment. Which object gets signed is decided from the
conversation's own messages and never from anything the caller sends — an
endpoint that signed a caller-supplied object name would hand out the whole
bucket to anyone with an account. Ownership of the conversation is the entire
authorization story; attachments recorded before this feature have no object
name and read as gone rather than falling back to a guess.

Deleting a conversation now drops its files. The delete is soft with no way
back, so nothing is left to read them. Cleanup is best-effort: a storage
outage must not leave the user with a conversation that refuses to disappear.

Refs: features/v2.6.0/043-chat-file-permanent-storage (T007-T009)

features/v2.6.0/043-chat-file-permanent-storage/tasks.md

@@ -12,7 +12,7 @@
 | spec.md | ✅ 已评审 | 用户已确认（存量不可恢复 / 会话删除即清理 / 按需换发链接 / 三场景一并处理 / 对象名唯一化纳入范围） |
 | design.md | ✅ 已评审 | 用户已确认；接手时第一入口 |
 | tasks.md | ✅ 已拆解 | 方案调整后重排波次 |
-| 实现 | 🟡 进行中 | 6 / 14 完成（Wave 1-2 已完成）|
+| 实现 | 🟡 进行中 | 9 / 14 完成（Wave 1-3 已完成）|
 
 ---
 
@@ -70,7 +70,7 @@
 
 ### Wave 3 — 换发链接与清理
 
-- [ ] **T007**: 换发链接的鉴权逻辑 + 单元测试
+- [x] **T007**: 换发链接的鉴权逻辑 + 单元测试
   **文件**: `src/backend/bisheng/chat_session/domain/chat.py`（或同模块 service）
   `src/backend/test/chat_session/test_attachment_link.py`（新建）
   **逻辑**: `resolve_attachment_url(chat_id, file_id, login_user)` —— ①载入会话 ②校验请求者为会话所属用户 ③在该会话消息的 files 中查 `file_id` 取**对象名** ④签发短时效链接。**对象名只从服务端数据取，绝不使用入参**（design §3 决策 3、§5 坑 6）
@@ -78,13 +78,13 @@
   **覆盖 AC**: AC-04, AC-08
   **依赖**: 无
 
-- [ ] **T008**: 换发链接端点
+- [x] **T008**: 换发链接端点
   **文件**: `src/backend/bisheng/chat_session/api/endpoints/chat.py`
   **逻辑**: 新增端点，入参 `chat_id` + `file_id`，委托 T007；不新增错误码，复用既有未授权 / 未找到响应
   **覆盖 AC**: AC-04, AC-08
   **依赖**: T006
 
-- [ ] **T009**: 会话删除时清理附件
+- [x] **T009**: 会话删除时清理附件
   **文件**: `src/backend/bisheng/chat_session/domain/chat.py`（`delete_session`）
   **逻辑**: 软删会话后，从该会话的消息 files 中取出 `object_name` 并逐个删除（上传时拿不到会话 ID，无法按前缀清扫——design §5 坑 8）。**清理失败只记日志，不得让删除会话失败**（spec §3）
   **覆盖 AC**: AC-03
@@ -92,7 +92,7 @@
 
 ### Wave 4 — 前端（client）
 
-- [ ] **T009**: 共用「消息图片」组件 + 换发链接接入
+- [x] **T009**: 共用「消息图片」组件 + 换发链接接入
   **文件**: `src/frontend/client/src/components/Chat/Messages/Content/MessageImage.tsx`（新建，基于既有 `Image.tsx` / `DialogImage.tsx` 提取）
   `src/frontend/client/src/api/chatApi.ts`（新增换发链接请求方法）
   **逻辑**: 渲染时调换发接口取链接 → 缩略图 → 点击全屏（右上角关闭）；**换发失败或图片加载失败** → 渲染占位「图片已失效，无法查看」（design §3 决策 4）；老消息无对象名字段时直接走失效分支（向后兼容）

src/backend/bisheng/chat_session/api/endpoints/chat.py

@@ -1,19 +1,13 @@
-import asyncio
-from typing import Optional, Union
+from typing import Union
 
 from fastapi import APIRouter, Body, Query, Request
 from fastapi.params import Depends
-from loguru import logger
 
-from bisheng.api.services.workflow import WorkFlowService
-from bisheng.api.v1.schema.base_schema import PageList
-from bisheng.api.v1.schemas import AddChatMessages, ChatList, resp_200
+from bisheng.api.v1.schemas import AddChatMessages, resp_200
 from bisheng.chat_session.domain.chat import ChatSessionService
 from bisheng.chat_session.domain.services.chat_message_service import ChatMessageService
 from bisheng.common.dependencies.user_deps import UserPayload
 from bisheng.common.errcode.http_error import UnAuthorizedError
-from bisheng.database.models.flow import FlowStatus, FlowType
-from bisheng.database.models.session import MessageSessionDao
 from bisheng.share_link.api.dependencies import header_share_token_parser
 from bisheng.share_link.domain.models.share_link import ShareLink
 from bisheng.utils import get_request_ip
@@ -21,16 +15,18 @@ from bisheng.utils import get_request_ip
 router = APIRouter()
 
 
-@router.get('/chat/app/list')
-def get_app_chat_list(*,
-                      keyword: Optional[str] = None,
-                      mark_user: Optional[str] = None,
-                      mark_status: Optional[int] = None,
-                      task_id: Optional[int] = Query(default=None, description='Callout TaskID'),
-                      flow_type: Optional[int] = None,
-                      page_num: Optional[int] = 1,
-                      page_size: Optional[int] = 20,
-                      login_user: UserPayload = Depends(UserPayload.get_login_user)):
+@router.get("/chat/app/list")
+def get_app_chat_list(
+    *,
+    keyword: str | None = None,
+    mark_user: str | None = None,
+    mark_status: int | None = None,
+    task_id: int | None = Query(default=None, description="Callout TaskID"),
+    flow_type: int | None = None,
+    page_num: int | None = 1,
+    page_size: int | None = 20,
+    login_user: UserPayload = Depends(UserPayload.get_login_user),
+):
     """Get session list filtered by annotation task."""
     result = ChatSessionService.get_app_chat_list(
         login_user=login_user,
@@ -45,14 +41,16 @@ def get_app_chat_list(*,
     return resp_200(result)
 
 
-@router.get('/chat/history')
-async def get_chat_message(*,
-                           chat_id: str,
-                           flow_id: str,
-                           id: Optional[str] = None,
-                           page_size: Optional[int] = 20,
-                           login_user: UserPayload = Depends(UserPayload.get_login_user),
-                           share_link: Union['ShareLink', None] = Depends(header_share_token_parser)):
+@router.get("/chat/history")
+async def get_chat_message(
+    *,
+    chat_id: str,
+    flow_id: str,
+    id: str | None = None,
+    page_size: int | None = 20,
+    login_user: UserPayload = Depends(UserPayload.get_login_user),
+    share_link: Union["ShareLink", None] = Depends(header_share_token_parser),
+):
     history = await ChatSessionService.get_chat_history(chat_id, flow_id, id, page_size)
 
     if history and login_user.user_id != history[0].user_id:
@@ -61,61 +59,77 @@ async def get_chat_message(*,
     return resp_200(history)
 
 
-@router.get('/chat/info')
-async def get_chat_info(chat_id: str = Query(..., description='Session Uniqueid，chat_id')):
+@router.get("/chat/info")
+async def get_chat_info(chat_id: str = Query(..., description="Session Uniqueid，chat_id")):
     """Get session details by chat_id."""
     res = await ChatSessionService.get_session_info(chat_id)
     return resp_200(res)
 
 
-@router.post('/chat/conversation/rename')
-async def rename(conversationId: str = Body(..., description='Session sid', embed=True),
-                 name: str = Body(..., description='Session name', embed=True),
-                 login_user: UserPayload = Depends(UserPayload.get_login_user)):
+@router.post("/chat/conversation/rename")
+async def rename(
+    conversationId: str = Body(..., description="Session sid", embed=True),
+    name: str = Body(..., description="Session name", embed=True),
+    login_user: UserPayload = Depends(UserPayload.get_login_user),
+):
     await ChatSessionService.rename_session(conversationId, name)
     return resp_200()
 
 
-@router.delete('/chat/{chat_id}', status_code=200)
-async def del_chat_id(*,
-                      request: Request,
-                      chat_id: str,
-                      login_user: UserPayload = Depends(UserPayload.get_login_user)):
+@router.delete("/chat/{chat_id}", status_code=200)
+async def del_chat_id(*, request: Request, chat_id: str, login_user: UserPayload = Depends(UserPayload.get_login_user)):
     await ChatSessionService.delete_session(chat_id, login_user, get_request_ip(request))
     return resp_200()
 
 
-@router.post('/chat/message', status_code=200)
-def add_chat_messages(*,
-                      request: Request,
-                      data: AddChatMessages,
-                      login_user: UserPayload = Depends(UserPayload.get_login_user)):
+@router.get("/chat/{chat_id}/files/{file_id}/url", status_code=200)
+async def get_chat_attachment_url(
+    *, request: Request, chat_id: str, file_id: str, login_user: UserPayload = Depends(UserPayload.get_login_user)
+):
+    """Fresh link for an attachment of this conversation.
+
+    The link issued at upload time expires; the client asks for a new one when
+    it renders. Which object gets signed is decided from the conversation's own
+    messages, never from anything the caller sends.
+    """
+    url = await ChatSessionService.resolve_attachment_url(chat_id, file_id, login_user)
+    return resp_200(data={"url": url})
+
+
+@router.post("/chat/message", status_code=200)
+def add_chat_messages(
+    *, request: Request, data: AddChatMessages, login_user: UserPayload = Depends(UserPayload.get_login_user)
+):
     """Add a full Q&A record. Security check write usage."""
     message_dbs = ChatMessageService.add_qa_messages(data, login_user, get_request_ip(request))
     return resp_200(data=message_dbs)
 
 
-@router.put('/chat/message/{message_id}', status_code=200)
-def update_chat_message(*,
-                        message_id: int,
-                        message: str = Body(embed=True),
-                        category: str = Body(default=None, embed=True),
-                        login_user: UserPayload = Depends(UserPayload.get_login_user)):
+@router.put("/chat/message/{message_id}", status_code=200)
+def update_chat_message(
+    *,
+    message_id: int,
+    message: str = Body(embed=True),
+    category: str = Body(default=None, embed=True),
+    login_user: UserPayload = Depends(UserPayload.get_login_user),
+):
     """Update the content of a message. Security check usage."""
     ChatMessageService.update_message(message_id, message, category, login_user)
     return resp_200()
 
 
-@router.delete('/chat/message/{message_id}', status_code=200)
+@router.delete("/chat/message/{message_id}", status_code=200)
 def del_message_id(*, message_id: str, login_user: UserPayload = Depends(UserPayload.get_login_user)):
     ChatMessageService.delete_message(login_user.user_id, message_id)
     return resp_200()
 
 
-@router.get('/chat/list')
-def get_session_list(page: Optional[int] = Query(default=1, ge=1, le=1000),
-                     limit: Optional[int] = Query(default=10, ge=1, le=100),
-                     login_user: UserPayload = Depends(UserPayload.get_login_user)):
+@router.get("/chat/list")
+def get_session_list(
+    page: int | None = Query(default=1, ge=1, le=1000),
+    limit: int | None = Query(default=10, ge=1, le=100),
+    login_user: UserPayload = Depends(UserPayload.get_login_user),
+):
     """Get session list sorted by update_time descending. Only shows daily chat and linsight sessions."""
     chat_sessions = ChatSessionService.get_user_session_list(login_user.user_id, page, limit)
     return resp_200(chat_sessions)

src/backend/bisheng/chat_session/domain/chat.py

@@ -1,5 +1,4 @@
-from typing import List, Optional
-
+import json
 from loguru import logger
 
 from bisheng.api.services.audit_log import AuditLogService
@@ -8,10 +7,12 @@ from bisheng.api.v1.schema.base_schema import PageList
 from bisheng.api.v1.schema.chat_schema import AppChatList
 from bisheng.api.v1.schema.workflow import WorkflowEventType
 from bisheng.api.v1.schemas import ChatList
+from bisheng.chat_session.domain.services.chat_message_service import _resolve_leaf_tenant_id
+from bisheng.chat_session.utils import get_session_app_type
 from bisheng.common.constants.enums.telemetry import BaseTelemetryTypeEnum
 from bisheng.common.dependencies.user_deps import UserPayload
-from bisheng.common.errcode.http_error import UnAuthorizedError
-from bisheng.common.schemas.telemetry.event_data_schema import NewMessageSessionEventData, DeleteMessageSessionEventData
+from bisheng.common.errcode.http_error import NotFoundError, UnAuthorizedError
+from bisheng.common.schemas.telemetry.event_data_schema import DeleteMessageSessionEventData, NewMessageSessionEventData
 from bisheng.common.services import telemetry_service
 from bisheng.common.services.base import BaseService
 from bisheng.core.logger import trace_id_var
@@ -19,21 +20,20 @@ from bisheng.database.models.assistant import AssistantDao
 from bisheng.database.models.flow import FlowDao, FlowType
 from bisheng.database.models.mark_record import MarkRecordDao, MarkRecordStatus
 from bisheng.database.models.mark_task import MarkTaskDao
+from bisheng.core.storage.minio.minio_manager import get_minio_storage
 from bisheng.database.models.message import ChatMessageDao
 from bisheng.database.models.session import MessageSession, MessageSessionDao, SensitiveStatus
 from bisheng.database.models.user_group import UserGroupDao
-from bisheng.chat_session.domain.services.chat_message_service import _resolve_leaf_tenant_id
-from bisheng.chat_session.utils import get_session_app_type
 from bisheng.user.domain.models.user import UserDao
 
 
 class ChatSessionService:
     """Chat session lifecycle services."""
 
-
     @staticmethod
-    async def get_chat_history(chat_id: str, flow_id: str, message_id: Optional[str] = None,
-                               page_size: Optional[int] = 20) -> List['ChatMessageHistoryResponse']:
+    async def get_chat_history(
+        chat_id: str, flow_id: str, message_id: str | None = None, page_size: int | None = 20
+    ) -> list["ChatMessageHistoryResponse"]:
         """Retrieve chat history for a user."""
         from bisheng.api.v1.schema.chat_schema import ChatMessageHistoryResponse
 
@@ -43,19 +43,16 @@ class ChatSessionService:
         if not session_info or session_info.flow_id != flow_id:
             return []
 
-        history = await ChatMessageDao.afilter_message_by_chat_id(chat_id=chat_id, flow_id=flow_id,
-                                                                  message_id=message_id, page_size=page_size)
+        history = await ChatMessageDao.afilter_message_by_chat_id(
+            chat_id=chat_id, flow_id=flow_id, message_id=message_id, page_size=page_size
+        )
         if history:
             user_info = await UserDao.aget_user(user_id=session_info.user_id)
-            history = ChatMessageHistoryResponse.from_chat_message_objs(
-                history,
-                user_info,
-                session_info
-            )
+            history = ChatMessageHistoryResponse.from_chat_message_objs(history, user_info, session_info)
         return history
 
     @staticmethod
-    async def get_session_info(chat_id: str) -> Optional[MessageSession]:
+    async def get_session_info(chat_id: str) -> MessageSession | None:
         """Get session details with logo URL resolved."""
         res = await MessageSessionDao.async_get_one(chat_id)
         if res:
@@ -67,6 +64,81 @@ class ChatSessionService:
         """Rename a chat session."""
         await MessageSessionDao.update_session_name(conversation_id, name)
 
+    @staticmethod
+    async def _own_conversation_or_raise(chat_id: str, login_user: UserPayload) -> MessageSession:
+        """Load a conversation, refusing anyone who doesn't own it."""
+        session_chat = await MessageSessionDao.async_get_one(chat_id)
+        if not session_chat or session_chat.is_delete:
+            raise NotFoundError.http_exception()
+        if session_chat.user_id != login_user.user_id:
+            raise UnAuthorizedError.http_exception()
+        return session_chat
+
+    @staticmethod
+    def _attachments_of(messages: list) -> list[dict]:
+        """Every attachment recorded across a conversation's messages."""
+        attachments = []
+        for message in messages:
+            if not message.files:
+                continue
+            try:
+                files = json.loads(message.files)
+            except (TypeError, ValueError):
+                logger.warning("unparsable files payload on message of chat {}", getattr(message, "chat_id", "?"))
+                continue
+            attachments.extend(f for f in files or [] if isinstance(f, dict))
+        return attachments
+
+    @staticmethod
+    async def _remove_attachments(chat_id: str) -> None:
+        """Drop the files a deleted conversation was holding.
+
+        Deletion is a soft delete with no way back, so the files have no reader
+        left. Best-effort on purpose: failing to reach storage must not leave
+        the user staring at a conversation that refuses to disappear.
+        """
+        try:
+            messages = await ChatMessageDao.aget_messages_by_chat_id(chat_id=chat_id, limit=1000)
+            object_names = {
+                f["object_name"] for f in ChatSessionService._attachments_of(messages) if f.get("object_name")
+            }
+            if not object_names:
+                return
+            minio_client = await get_minio_storage()
+            for object_name in object_names:
+                try:
+                    await minio_client.remove_object(object_name=object_name)
+                except Exception:
+                    logger.exception("failed to remove attachment {} of chat {}", object_name, chat_id)
+        except Exception:
+            logger.exception("failed to clean up attachments of chat {}", chat_id)
+
+    @staticmethod
+    async def resolve_attachment_url(chat_id: str, file_id: str, login_user: UserPayload) -> str:
+        """Issue a fresh link for one attachment of one conversation.
+
+        The link handed out at upload time expires, so the client comes back
+        here when it renders. The object name is read from what we stored on the
+        conversation's messages and never taken from the caller -- signing a
+        caller-supplied name would open the whole bucket to anyone with an
+        account.
+        """
+        await ChatSessionService._own_conversation_or_raise(chat_id, login_user)
+
+        messages = await ChatMessageDao.aget_messages_by_chat_id(chat_id=chat_id, limit=1000)
+        for attachment in ChatSessionService._attachments_of(messages):
+            if str(attachment.get("file_id")) != str(file_id):
+                continue
+            object_name = attachment.get("object_name")
+            if not object_name:
+                # Written before attachments were made permanent: the file is
+                # gone with the temp bucket. Say so rather than guess.
+                break
+            minio_client = await get_minio_storage()
+            return await minio_client.get_share_link(object_name)
+
+        raise NotFoundError.http_exception()
+
     @staticmethod
     async def delete_session(chat_id: str, login_user: UserPayload, request_ip: str) -> None:
         """Delete a session with audit logging and telemetry."""
@@ -87,6 +159,7 @@ class ChatSessionService:
                 await AuditLogService.delete_chat_workflow(login_user, request_ip, flow_info)
 
         await MessageSessionDao.delete_session(chat_id)
+        await ChatSessionService._remove_attachments(chat_id)
 
         await telemetry_service.log_event(
             user_id=login_user.user_id,
@@ -99,11 +172,11 @@ class ChatSessionService:
     def get_app_chat_list(
         *,
         login_user: UserPayload,
-        keyword: Optional[str] = None,
-        mark_user: Optional[str] = None,
-        mark_status: Optional[int] = None,
-        task_id: Optional[int] = None,
-        flow_type: Optional[int] = None,
+        keyword: str | None = None,
+        mark_user: str | None = None,
+        mark_status: int | None = None,
+        task_id: int | None = None,
+        flow_type: int | None = None,
         page_num: int = 1,
         page_size: int = 20,
     ) -> PageList:
@@ -117,20 +190,20 @@ class ChatSessionService:
         if task_id:
             if not login_user.is_admin():
                 task = MarkTaskDao.get_task_byid(task_id)
-                if str(login_user.user_id) not in task.process_users.split(','):
+                if str(login_user.user_id) not in task.process_users.split(","):
                     raise UnAuthorizedError()
                 if user_groups:
                     task = MarkTaskDao.get_task_byid(task_id)
-                    group_flow_ids = task.app_id.split(',')
+                    group_flow_ids = task.app_id.split(",")
                     if not group_flow_ids:
                         return PageList(list=[], total=0)
                 else:
                     task = MarkTaskDao.get_task_byid(task_id)
-                    if str(login_user.user_id) not in task.process_users.split(','):
+                    if str(login_user.user_id) not in task.process_users.split(","):
                         raise UnAuthorizedError()
-                    group_flow_ids = MarkTaskDao.get_task_byid(task_id).app_id.split(',')
+                    group_flow_ids = MarkTaskDao.get_task_byid(task_id).app_id.split(",")
             else:
-                group_flow_ids = MarkTaskDao.get_task_byid(task_id).app_id.split(',')
+                group_flow_ids = MarkTaskDao.get_task_byid(task_id).app_id.split(",")
 
         if keyword:
             flows = FlowDao.get_flow_list_by_name(name=keyword)
@@ -182,16 +255,16 @@ class ChatSessionService:
                 if mark_status != tmp.mark_status:
                     continue
             if mark_user:
-                users = [int(u) for u in mark_user.split(',')]
+                users = [int(u) for u in mark_user.split(",")]
                 if tmp.mark_id not in users:
                     continue
             result.append(tmp)
 
-        result = result[(page_num - 1) * page_size: page_num * page_size]
+        result = result[(page_num - 1) * page_size : page_num * page_size]
         return PageList(list=result, total=total)
 
     @staticmethod
-    def get_user_session_list(user_id: int, page: int = 1, limit: int = 10) -> List[ChatList]:
+    def get_user_session_list(user_id: int, page: int = 1, limit: int = 10) -> list[ChatList]:
         """List daily chat and linsight sessions for a user, sorted by update_time descending."""
         allowed_flow_types = [FlowType.WORKSTATION.value, FlowType.LINSIGHT.value]
 
@@ -221,7 +294,7 @@ class ChatSessionService:
                 flow_name=one.flow_name,
                 flow_type=one.flow_type,
                 name=one.name,
-                logo=BaseService.get_logo_share_link(one.flow_logo) if one.flow_logo else '',
+                logo=BaseService.get_logo_share_link(one.flow_logo) if one.flow_logo else "",
                 latest_message=latest_messages.get(one.chat_id, None),
                 create_time=one.create_time,
                 update_time=one.update_time,
@@ -235,7 +308,7 @@ class ChatSessionService:
         flow_id: str,
         login_user: UserPayload,
         request_ip: str,
-    ) -> Optional[MessageSession]:
+    ) -> MessageSession | None:
         """Get existing session or create a new one with audit log and telemetry.
 
         Used when adding messages to ensure a session exists.
@@ -254,29 +327,33 @@ class ChatSessionService:
 
         flow_info = FlowDao.get_flow_by_id(flow_id)
         if flow_info:
-            session_info = MessageSessionDao.insert_one(MessageSession(
-                chat_id=chat_id,
-                flow_id=flow_id,
-                flow_type=flow_info.flow_type,
-                flow_name=flow_info.name,
-                user_id=login_user.user_id,
-                sensitive_status=SensitiveStatus.VIOLATIONS.value,
-                tenant_id=leaf_tenant_id,
-            ))
+            session_info = MessageSessionDao.insert_one(
+                MessageSession(
+                    chat_id=chat_id,
+                    flow_id=flow_id,
+                    flow_type=flow_info.flow_type,
+                    flow_name=flow_info.name,
+                    user_id=login_user.user_id,
+                    sensitive_status=SensitiveStatus.VIOLATIONS.value,
+                    tenant_id=leaf_tenant_id,
+                )
+            )
             if flow_info.flow_type == FlowType.WORKFLOW.value:
                 AuditLogService.create_chat_workflow(login_user, request_ip, flow_id, flow_info)
         else:
             assistant_info = AssistantDao.get_one_assistant(flow_id)
             if assistant_info:
-                session_info = MessageSessionDao.insert_one(MessageSession(
-                    chat_id=chat_id,
-                    flow_id=flow_id,
-                    flow_type=FlowType.ASSISTANT.value,
-                    flow_name=assistant_info.name,
-                    user_id=login_user.user_id,
-                    sensitive_status=SensitiveStatus.VIOLATIONS.value,
-                    tenant_id=leaf_tenant_id,
-                ))
+                session_info = MessageSessionDao.insert_one(
+                    MessageSession(
+                        chat_id=chat_id,
+                        flow_id=flow_id,
+                        flow_type=FlowType.ASSISTANT.value,
+                        flow_name=assistant_info.name,
+                        user_id=login_user.user_id,
+                        sensitive_status=SensitiveStatus.VIOLATIONS.value,
+                        tenant_id=leaf_tenant_id,
+                    )
+                )
                 AuditLogService.create_chat_assistant(login_user, request_ip, flow_id)
 
         if session_info:

src/backend/test/chat_session/test_attachment_link.py

@@ -0,0 +1,110 @@
+"""F043: handing out a fresh link for a conversation attachment.
+
+Links issued at upload time expire, so the client asks for a new one when it
+renders. The object name must only ever come from what the server has stored
+for that conversation -- an endpoint that signed whatever object name the
+caller passed would hand out the entire bucket.
+
+See features/v2.6.0/043-chat-file-permanent-storage/design.md §3 decision 3.
+"""
+
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from bisheng.chat_session.domain.chat import ChatSessionService
+
+OWNER_ID = 7
+STRANGER_ID = 99
+
+
+def _user(user_id):
+    user = MagicMock()
+    user.user_id = user_id
+    return user
+
+
+def _message(files):
+    return SimpleNamespace(files=json.dumps(files))
+
+
+@pytest.fixture
+def storage():
+    client = MagicMock()
+    client.get_share_link = AsyncMock(return_value="/bisheng/chat/7/abc.png?sig=fresh")
+    with patch("bisheng.chat_session.domain.chat.get_minio_storage", AsyncMock(return_value=client)):
+        yield client
+
+
+@pytest.fixture
+def conversation():
+    """A conversation owned by OWNER_ID holding one attachment."""
+    session = SimpleNamespace(chat_id="c1", user_id=OWNER_ID, is_delete=False)
+    files = [{"file_id": "f1", "filename": "a.png", "object_name": "chat/7/abc.png"}]
+    with (
+        patch(
+            "bisheng.chat_session.domain.chat.MessageSessionDao.async_get_one",
+            AsyncMock(return_value=session),
+        ),
+        patch(
+            "bisheng.chat_session.domain.chat.ChatMessageDao.aget_messages_by_chat_id",
+            AsyncMock(return_value=[_message(files)]),
+        ),
+    ):
+        yield session
+
+
+class TestResolveAttachmentUrl:
+    async def test_owner_gets_a_fresh_link(self, storage, conversation):
+        url = await ChatSessionService.resolve_attachment_url("c1", "f1", _user(OWNER_ID))
+
+        assert url == "/bisheng/chat/7/abc.png?sig=fresh"
+        # Signed for the object the server recorded, nothing else.
+        assert storage.get_share_link.await_args.args[0] == "chat/7/abc.png"
+
+    async def test_someone_else_is_refused(self, storage, conversation):
+        # AC-04 — conversation ownership is the whole authorization story.
+        with pytest.raises(Exception):
+            await ChatSessionService.resolve_attachment_url("c1", "f1", _user(STRANGER_ID))
+
+        storage.get_share_link.assert_not_awaited()
+
+    async def test_unknown_file_id_is_refused(self, storage, conversation):
+        with pytest.raises(Exception):
+            await ChatSessionService.resolve_attachment_url("c1", "nope", _user(OWNER_ID))
+
+        storage.get_share_link.assert_not_awaited()
+
+    async def test_attachment_without_object_name_is_refused(self, storage):
+        # Messages written before this feature carry no object name; they must
+        # read as "gone", not fall back to guessing at some other object.
+        session = SimpleNamespace(chat_id="c1", user_id=OWNER_ID, is_delete=False)
+        legacy = [{"file_id": "f1", "filename": "a.png", "filepath": "/bisheng-tmp/a.png"}]
+        with (
+            patch(
+                "bisheng.chat_session.domain.chat.MessageSessionDao.async_get_one",
+                AsyncMock(return_value=session),
+            ),
+            patch(
+                "bisheng.chat_session.domain.chat.ChatMessageDao.aget_messages_by_chat_id",
+                AsyncMock(return_value=[_message(legacy)]),
+            ),
+            pytest.raises(Exception),
+        ):
+            await ChatSessionService.resolve_attachment_url("c1", "f1", _user(OWNER_ID))
+
+        storage.get_share_link.assert_not_awaited()
+
+    async def test_missing_conversation_is_refused(self, storage):
+        with (
+            patch(
+                "bisheng.chat_session.domain.chat.MessageSessionDao.async_get_one",
+                AsyncMock(return_value=None),
+            ),
+            pytest.raises(Exception),
+        ):
+            await ChatSessionService.resolve_attachment_url("nope", "f1", _user(OWNER_ID))
+
+        storage.get_share_link.assert_not_awaited()