social-app/docs/plans/2026-02-25-agent-chat-gap-closure-plan.md

Agent Chat Gap Closure Implementation Plan

For Claude: REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

Goal: 按未完成 Task 顺序补齐 Agent Chat Core 缺口，形成可运行、可测试、可审计的后端链路。

Architecture: 复用已完成的数据层与路由骨架，在 core/agent_chat 补齐编排、成本与多模态能力，并通过 v1/agent_chat/service.py 统一持久化与事件顺序。全流程以 session.id + messages.seq 作为一致性锚点，保证事件输出与落库一致。

Tech Stack: FastAPI, SQLAlchemy, Pydantic, pytest, CrewAI, AG-UI adapter, DashScope SDK, Supabase Storage。

---

Task 1: 补齐 Spike 结论文档

Files:

• Create: docs/plans/2026-02-25-agent-chat-crewai-ag-ui-spike-notes.md

Step 1: 写失败校验（文档存在性）

test -f docs/plans/2026-02-25-agent-chat-crewai-ag-ui-spike-notes.md

Step 2: 运行并确认失败

Run: test -f docs/plans/2026-02-25-agent-chat-crewai-ag-ui-spike-notes.md Expected: non-zero exit code。

Step 3: 写最小文档实现

- CrewAI 版本探测结论
- AG-UI 官方 CrewAI 集成可用性结论
- DashScope FunASR usage 字段策略
- 不可用时的最小兜底映射策略

Step 4: 运行并确认通过

Run: test -f docs/plans/2026-02-25-agent-chat-crewai-ag-ui-spike-notes.md Expected: zero exit code。

Task 5: 补齐编排与成本追踪

Files:

• Create: backend/src/core/agent_chat/events.py
• Create: backend/src/core/agent_chat/cost_tracker.py
• Create: backend/src/core/agent_chat/orchestrator.py
• Test: backend/tests/unit/core/agent_chat/test_cost_tracker.py
• Test: backend/tests/unit/core/agent_chat/test_orchestrator_pipeline.py

Step 1: 写失败测试

def test_normalize_usage_and_cost_aggregation():
    assert False


def test_orchestrator_runs_three_stages_in_order():
    assert False

Step 2: 运行并确认失败

Run: PYTHONPATH=backend/src uv run pytest backend/tests/unit/core/agent_chat/test_cost_tracker.py backend/tests/unit/core/agent_chat/test_orchestrator_pipeline.py -v Expected: FAIL。

Step 3: 写最小实现

class CostTracker:
    def add_usage(self, usage: dict) -> None: ...
    def total(self) -> dict: ...


class AgentChatOrchestrator:
    async def run(self, command):
        # intent -> execution -> organization
        ...

Step 4: 运行并确认通过

Run: PYTHONPATH=backend/src uv run pytest backend/tests/unit/core/agent_chat/test_cost_tracker.py backend/tests/unit/core/agent_chat/test_orchestrator_pipeline.py -v Expected: PASS。

Task 6: 补齐 AG-UI 适配层缺口

Files:

• Create: backend/src/core/agent_chat/agui_adapter.py
• Modify: backend/src/core/agent_chat/event_bridge.py
• Modify: backend/src/v1/agent_chat/service.py
• Test: backend/tests/unit/core/agent_chat/test_agui_adapter.py
• Test: backend/tests/integration/test_agent_chat_event_persistence.py

Step 1: 写失败测试

def test_agui_adapter_maps_internal_events_to_protocol_events():
    assert False

Step 2: 运行并确认失败

Run: PYTHONPATH=backend/src uv run pytest backend/tests/unit/core/agent_chat/test_agui_adapter.py backend/tests/integration/test_agent_chat_event_persistence.py -v Expected: FAIL。

Step 3: 写最小实现

class AguiAdapter:
    def to_command(self, request): ...
    def to_protocol_event(self, event): ...

Step 4: 运行并确认通过

Run: PYTHONPATH=backend/src uv run pytest backend/tests/unit/core/agent_chat/test_agui_adapter.py backend/tests/integration/test_agent_chat_event_persistence.py -v Expected: PASS。

Task 7: 补齐多模态输入编排

Files:

• Create: backend/src/core/agent_chat/multimodal.py
• Modify: backend/src/core/agent_chat/storage_adapter.py
• Modify: backend/src/core/agent_chat/tools/asr_fun_asr.py
• Test: backend/tests/unit/core/agent_chat/test_multimodal.py

Step 1: 写失败测试

def test_multimodal_validates_and_builds_attachment_context():
    assert False

Step 2: 运行并确认失败

Run: PYTHONPATH=backend/src uv run pytest backend/tests/unit/core/agent_chat/test_multimodal.py -v Expected: FAIL。

Step 3: 写最小实现

class MultimodalProcessor:
    async def build_context(self, attachments): ...

Step 4: 运行并确认通过

Run: PYTHONPATH=backend/src uv run pytest backend/tests/unit/core/agent_chat/test_multimodal.py backend/tests/unit/core/agent_chat/test_storage_adapter.py backend/tests/unit/core/agent_chat/test_asr_fun_asr_tool.py -v Expected: PASS。

Task 8: 补齐会话审计与 recent 规则

Files:

• Modify: backend/src/v1/agent_chat/service.py
• Modify: backend/src/v1/agent_chat/router.py
• Test: backend/tests/unit/core/agent_chat/test_session_title_strategy.py
• Test: backend/tests/integration/test_agent_chat_session_recent_selection.py
• Test: backend/tests/integration/test_agent_chat_session_persistence.py

Step 1: 写失败测试

def test_title_generated_from_first_user_message():
    assert False


def test_recent_session_selected_by_last_activity_at_desc():
    assert False

Step 2: 运行并确认失败

Run: PYTHONPATH=backend/src uv run pytest backend/tests/unit/core/agent_chat/test_session_title_strategy.py backend/tests/integration/test_agent_chat_session_recent_selection.py backend/tests/integration/test_agent_chat_session_persistence.py -v Expected: FAIL。

Step 3: 写最小实现

def build_session_title(first_message: str) -> str: ...

Step 4: 运行并确认通过

Run: PYTHONPATH=backend/src uv run pytest backend/tests/unit/core/agent_chat/test_session_title_strategy.py backend/tests/integration/test_agent_chat_session_recent_selection.py backend/tests/integration/test_agent_chat_session_persistence.py -v Expected: PASS。

Task 9: 补齐 E2E 与运行文档闭环

Files:

• Create: backend/tests/e2e/test_agent_chat_flow.py
• Create: backend/tests/e2e/test_agent_chat_recent_session_home.py
• Modify: docs/runtime/runtime-runbook.md

Step 1: 写失败 E2E 用例

def test_agent_chat_text_image_audio_document_flow():
    assert False

Step 2: 运行并确认失败

Run: PYTHONPATH=backend/src uv run pytest backend/tests/e2e/test_agent_chat_flow.py -v Expected: FAIL。

Step 3: 写最小实现与文档补充

- bootstrap gate 执行顺序
- agent_chat 验证命令

Step 4: 全量验证

Run: make runtime-bootstrap-gate Expected: bootstrap 通过。

Run: PYTHONPATH=backend/src uv run pytest backend/tests/unit/core/agent_chat -v Expected: PASS。

Run: PYTHONPATH=backend/src uv run pytest backend/tests/integration -k agent_chat -v Expected: PASS。

Run: PYTHONPATH=backend/src uv run pytest backend/tests/e2e/test_agent_chat_flow.py backend/tests/e2e/test_agent_chat_recent_session_home.py -v Expected: PASS。

Run: PYTHONPATH=backend/src uv run pip check Expected: no broken requirements。