# Agent Multimodal Smoke Implementation Plan > **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. **Goal:** 完成 agent 三条主链路(runs/events/history)真实冒烟,并支持 RunAgentInput 图片信息在发送链路落 Supabase Storage、在 messages.metadata 持久化、在 history 返回中可渲染。 **Architecture:** 在 `v1/agent` 服务层新增“用户消息持久化 + 图片附件上传”步骤:`enqueue_run` 时解析用户消息 content block,图片上传到 `config.storage.bucket`,将路径写入 `messages.metadata`。运行时继续通过 AgentScope pipeline 输出 AG-UI 事件,SSE 从 Redis stream 订阅,历史查询从 `messages` 回放并附带附件信息。 **Tech Stack:** FastAPI, SQLAlchemy AsyncSession, Supabase Storage Admin Client, Redis SSE stream, AG-UI, pytest/httpx。 --- ### Task 1: 用户消息图片附件上传与落库 **Files:** - Create: `backend/src/v1/agent/attachment_storage.py` - Modify: `backend/src/v1/agent/service.py` - Modify: `backend/src/v1/agent/repository.py` - Test: `backend/tests/unit/v1/agent/test_service.py` **Step 1: 写失败测试(RED)** ```python @pytest.mark.asyncio async def test_enqueue_run_persists_user_message_with_uploaded_image_metadata() -> None: ... ``` **Step 2: 运行单测验证失败** Run: `uv run pytest tests/unit/v1/agent/test_service.py::test_enqueue_run_persists_user_message_with_uploaded_image_metadata -q` Expected: FAIL(缺少附件上传/metadata 持久化行为) **Step 3: 最小实现(GREEN)** ```python class AgentAttachmentStorage: async def upload_bytes(...): ... class AgentService: async def enqueue_run(...): # 解析 user content blocks # 上传图片到 storage # repository 持久化 user message(metadata 包含 bucket/path) ... ``` **Step 4: 运行单测验证通过** Run: `uv run pytest tests/unit/v1/agent/test_service.py::test_enqueue_run_persists_user_message_with_uploaded_image_metadata -q` Expected: PASS ### Task 2: history 渲染附件路径 **Files:** - Modify: `backend/src/v1/agent/repository.py` - Test: `backend/tests/unit/v1/agent/test_repository.py` **Step 1: 写失败测试(RED)** ```python @pytest.mark.asyncio async def test_history_includes_user_message_attachments_from_metadata() -> None: ... ``` **Step 2: 运行测试验证失败** Run: `uv run pytest tests/unit/v1/agent/test_repository.py::test_history_includes_user_message_attachments_from_metadata -q` Expected: FAIL(history 尚未渲染 attachments) **Step 3: 最小实现(GREEN)** ```python if role == "user" and isinstance(metadata.get("attachments"), list): payload["attachments"] = metadata["attachments"] ``` **Step 4: 运行测试验证通过** Run: `uv run pytest tests/unit/v1/agent/test_repository.py::test_history_includes_user_message_attachments_from_metadata -q` Expected: PASS ### Task 3: 真实冒烟 runs + SSE + history(含图片输入) **Files:** - Modify: `backend/tests/integration/v1/agent/test_sse_flow_live.py` **Step 1: 写失败测试(RED)** ```python @pytest.mark.asyncio @pytest.mark.live async def test_agent_runs_events_history_live_with_image_input() -> None: ... ``` **Step 2: 运行 live 测试验证失败(实现前或环境不完整)** Run: `AGENT_LIVE_INTEGRATION=1 AGENT_LIVE_EMAIL=... AGENT_LIVE_PASSWORD=... uv run pytest tests/integration/v1/agent/test_sse_flow_live.py::test_agent_runs_events_history_live_with_image_input -q -s` Expected: FAIL(缺 metadata/path 或 history 不含附件) **Step 3: 最小实现(GREEN)** ```python # live 测试流程: # 1) 登录拿 token # 2) POST /runs 发送 text + image(data) # 3) SSE 订阅直到 RUN_FINISHED/RUN_ERROR # 4) GET /runs/{thread_id}/history # 5) SQL 校验 sessions/messages 字段与 metadata.attachments ``` **Step 4: 运行 live 测试验证通过** Run: `AGENT_LIVE_INTEGRATION=1 AGENT_LIVE_EMAIL=... AGENT_LIVE_PASSWORD=... uv run pytest tests/integration/v1/agent/test_sse_flow_live.py::test_agent_runs_events_history_live_with_image_input -q -s` Expected: PASS ### Task 4: 全量收口验证与安全门禁 **Files:** - Modify (if needed): `backend/src/v1/agent/*`, `backend/tests/*` **Step 1: 回归测试** Run: `uv run pytest tests/unit/v1/agent tests/unit/core/agentscope tests/integration/v1/agent -q` Expected: PASS **Step 2: 静态检查** Run: `uv run ruff check src/v1/agent src/core/agentscope tests/unit/v1/agent tests/integration/v1/agent` Expected: PASS Run: `uv run basedpyright src/v1/agent src/core/agentscope tests/unit/v1/agent tests/integration/v1/agent` Expected: 0 errors **Step 3: 评审门禁** Run agents: `security-reviewer`, `refactor-cleaner`, `code-reviewer` Expected: 无未解决 CRITICAL/HIGH