social-app/docs/plans/2026-03-11-agent-multimodal-smoke-implementation.md

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）

@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）

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）

@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）

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）

@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）

# 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