feat(backend): AI 对话聊天记录 repo + append/list 端点（AC-2）

在 C2 扩(AC-1) `ai_messages` 表上落 AC-2 后端（计划 §3）：
- repo `SqlAiMessageRepo`：批 append（seq=批内 0 基位置、同 thread_id、只 flush）
  + `list_for_chapter` newest-first（created_at DESC, seq DESC）、chapter_no 过滤含
  项目级 NULL union、可选 kind、limit/offset。
- schemas `ai_messages.py`：批级 append 信封 + Literal 枚举权威（kind/role）+ content
  200k 上限 + meta 序列化兜底。
- 端点 `POST/GET /projects/{id}/ai-messages`（tag ai-messages，已注册）：显式记录写入，
  五个生成端点保持只读不变（守 #3）；项目 404、校验 422、POST 端点提交。
- 注入缝 `get_ai_message_repo`。

TDD：repo 真 pg 单测（seq/定序/union/分页/flush-only）9 passed；端点集成测试
（round-trip/union/kind/404/422×5/commit 边界）12 passed。门禁全绿：ruff/format/mypy
238 files/alembic check 无漂移/pytest 992 passed；新模块 cov router+schemas 100% / repo 97%。

apps/api/tests/test_ai_messages.py

@@ -0,0 +1,325 @@
+"""AC-2 POST/GET /projects/{id}/ai-messages 端点（内存替身，无 DB/无网络）。
+
+docs/design/ai-chat-history-plan.md §3.3 / §6。覆盖：
+- POST 201 + 回显持久化视图（seq/id/created_at）+ 端点 commit 边界；
+- append→list 回环；项目级(chapter_no null) vs 本章 union；
+- 校验 → 422（非法 role、超上限 content、空 messages、超大 meta）；
+- 未知项目 → 404；响应 key 集合不泄漏意外字段。
+
+owner_id 全程 stub（单用户原型）。AI 一律不触达（纯 CRUD）。
+"""
+
+from __future__ import annotations
+
+import os
+import uuid
+from datetime import UTC, datetime, timedelta
+
+import httpx
+import pytest
+from cryptography.fernet import Fernet
+from fakes_projects import FakeProjectRepo, FakeSession
+from ww_core.domain.ai_message_repo import AiMessageView, AiTurnRow
+from ww_core.domain.project_repo import ProjectView
+
+# 期望的响应 bubble 字段集合（防泄漏意外字段）。
+_EXPECTED_KEYS = {
+    "id",
+    "project_id",
+    "chapter_no",
+    "thread_id",
+    "seq",
+    "kind",
+    "tool_key",
+    "role",
+    "content",
+    "meta",
+    "created_at",
+}
+
+
+class _FakeAiMessageRepo:
+    """镜像 `SqlAiMessageRepo` 语义的内存版（批内 seq、newest-first union）。"""
+
+    def __init__(self) -> None:
+        self.rows: list[AiMessageView] = []
+        self._clock = 0
+
+    async def append(
+        self,
+        project_id: uuid.UUID,
+        *,
+        thread_id: uuid.UUID,
+        chapter_no: int | None,
+        kind: str,
+        tool_key: str | None,
+        rows: list[AiTurnRow],
+    ) -> list[AiMessageView]:
+        self._clock += 1
+        created = datetime(2026, 7, 9, tzinfo=UTC) + timedelta(seconds=self._clock)
+        out: list[AiMessageView] = []
+        for i, r in enumerate(rows):
+            view = AiMessageView(
+                id=uuid.uuid4(),
+                project_id=project_id,
+                chapter_no=chapter_no,
+                thread_id=thread_id,
+                seq=i,
+                kind=kind,
+                tool_key=tool_key,
+                role=r.role,
+                content=r.content,
+                meta=dict(r.meta),
+                created_at=created,
+            )
+            self.rows.append(view)
+            out.append(view)
+        return out
+
+    async def list_for_chapter(
+        self,
+        project_id: uuid.UUID,
+        chapter_no: int | None,
+        *,
+        kind: str | None = None,
+        limit: int,
+        offset: int,
+    ) -> list[AiMessageView]:
+        matched = [r for r in self.rows if r.project_id == project_id]
+        if chapter_no is not None:
+            matched = [r for r in matched if r.chapter_no in (chapter_no, None)]
+        else:
+            matched = [r for r in matched if r.chapter_no is None]
+        if kind is not None:
+            matched = [r for r in matched if r.kind == kind]
+        matched.sort(key=lambda r: (r.created_at, r.seq), reverse=True)
+        return matched[offset : offset + limit]
+
+
+def _make_client() -> tuple[httpx.AsyncClient, _FakeAiMessageRepo, FakeProjectRepo, FakeSession]:
+    os.environ.setdefault("CREDENTIAL_ENC_KEY", Fernet.generate_key().decode())
+    from ww_api.main import create_app
+    from ww_api.services.credentials import STUB_OWNER_ID
+    from ww_api.services.project_deps import get_ai_message_repo, get_project_repo
+    from ww_db import get_session
+
+    repo = _FakeAiMessageRepo()
+    project_repo = FakeProjectRepo()
+    session = FakeSession()
+
+    # 一个已存在的项目（owner = stub），供 404 对照——直接种入 fake 的行表（避开事件循环）。
+    project_id = uuid.uuid4()
+    project_repo.rows[project_id] = (STUB_OWNER_ID, ProjectView(id=project_id, title="AC-2"))
+
+    app = create_app()
+    app.dependency_overrides[get_ai_message_repo] = lambda: repo
+    app.dependency_overrides[get_project_repo] = lambda: project_repo
+    app.dependency_overrides[get_session] = lambda: session
+    transport = httpx.ASGITransport(app=app)
+    client = httpx.AsyncClient(transport=transport, base_url="http://test")
+    client.project_id = project_id  # type: ignore[attr-defined]
+    return client, repo, project_repo, session
+
+
+def _refine_batch(chapter_no: int | None = 5) -> dict[str, object]:
+    return {
+        "thread_id": str(uuid.uuid4()),
+        "chapter_no": chapter_no,
+        "kind": "refine",
+        "messages": [
+            {"role": "author", "content": "原选段"},
+            {"role": "ai", "content": "重写段", "meta": {"version_no": 1, "segment": "原选段"}},
+        ],
+    }
+
+
+@pytest.mark.asyncio
+async def test_append_returns_201_persisted_views_and_commits() -> None:
+    client, repo, _pr, session = _make_client()
+    pid = client.project_id  # type: ignore[attr-defined]
+    async with client:
+        resp = await client.post(f"/projects/{pid}/ai-messages", json=_refine_batch(5))
+    assert resp.status_code == 201, resp.text
+    body = resp.json()
+    msgs = body["messages"]
+    assert [m["seq"] for m in msgs] == [0, 1]
+    assert [m["role"] for m in msgs] == ["author", "ai"]
+    assert msgs[1]["meta"] == {"version_no": 1, "segment": "原选段"}
+    assert all(set(m.keys()) == _EXPECTED_KEYS for m in msgs)
+    assert all(m["id"] and m["created_at"] for m in msgs)
+    assert session.commits == 1
+    assert len(repo.rows) == 2
+
+
+@pytest.mark.asyncio
+async def test_append_then_list_roundtrip() -> None:
+    client, _repo, _pr, _session = _make_client()
+    pid = client.project_id  # type: ignore[attr-defined]
+    async with client:
+        await client.post(f"/projects/{pid}/ai-messages", json=_refine_batch(5))
+        resp = await client.get(f"/projects/{pid}/ai-messages", params={"chapter_no": 5})
+    assert resp.status_code == 200
+    msgs = resp.json()["messages"]
+    # newest-first：批内 seq DESC → ai(seq1) 先于 author(seq0)。
+    assert [m["content"] for m in msgs] == ["重写段", "原选段"]
+
+
+@pytest.mark.asyncio
+async def test_list_chapter_union_includes_project_level() -> None:
+    client, _repo, _pr, _session = _make_client()
+    pid = client.project_id  # type: ignore[attr-defined]
+    async with client:
+        await client.post(f"/projects/{pid}/ai-messages", json=_refine_batch(5))
+        # 项目级工具箱生成（chapter_no 省略 → NULL）。
+        await client.post(
+            f"/projects/{pid}/ai-messages",
+            json={
+                "thread_id": str(uuid.uuid4()),
+                "kind": "generator",
+                "tool_key": "expand",
+                "messages": [{"role": "ai", "content": "扩写预览"}],
+            },
+        )
+        chapter_view = await client.get(f"/projects/{pid}/ai-messages", params={"chapter_no": 5})
+        toolbox_view = await client.get(f"/projects/{pid}/ai-messages")
+    chapter_contents = {m["content"] for m in chapter_view.json()["messages"]}
+    toolbox_contents = {m["content"] for m in toolbox_view.json()["messages"]}
+    assert chapter_contents == {"原选段", "重写段", "扩写预览"}  # 本章 ∪ 项目级
+    assert toolbox_contents == {"扩写预览"}  # 仅项目级
+
+
+@pytest.mark.asyncio
+async def test_list_kind_filter() -> None:
+    client, _repo, _pr, _session = _make_client()
+    pid = client.project_id  # type: ignore[attr-defined]
+    async with client:
+        await client.post(f"/projects/{pid}/ai-messages", json=_refine_batch(2))
+        await client.post(
+            f"/projects/{pid}/ai-messages",
+            json={
+                "thread_id": str(uuid.uuid4()),
+                "chapter_no": 2,
+                "kind": "rewrite",
+                "messages": [{"role": "ai", "content": "整章新版"}],
+            },
+        )
+        resp = await client.get(
+            f"/projects/{pid}/ai-messages", params={"chapter_no": 2, "kind": "rewrite"}
+        )
+    msgs = resp.json()["messages"]
+    assert [m["kind"] for m in msgs] == ["rewrite"]
+    assert [m["content"] for m in msgs] == ["整章新版"]
+
+
+@pytest.mark.asyncio
+async def test_append_unknown_project_returns_404() -> None:
+    client, repo, _pr, session = _make_client()
+    async with client:
+        resp = await client.post(f"/projects/{uuid.uuid4()}/ai-messages", json=_refine_batch(5))
+    assert resp.status_code == 404
+    assert resp.json()["error"]["code"] == "NOT_FOUND"
+    assert session.commits == 0
+    assert repo.rows == []
+
+
+@pytest.mark.asyncio
+async def test_list_unknown_project_returns_404() -> None:
+    client, _repo, _pr, _session = _make_client()
+    async with client:
+        resp = await client.get(f"/projects/{uuid.uuid4()}/ai-messages", params={"chapter_no": 1})
+    assert resp.status_code == 404
+    assert resp.json()["error"]["code"] == "NOT_FOUND"
+
+
+@pytest.mark.asyncio
+async def test_append_invalid_role_returns_422() -> None:
+    client, _repo, _pr, session = _make_client()
+    pid = client.project_id  # type: ignore[attr-defined]
+    async with client:
+        resp = await client.post(
+            f"/projects/{pid}/ai-messages",
+            json={
+                "thread_id": str(uuid.uuid4()),
+                "chapter_no": 5,
+                "kind": "refine",
+                "messages": [{"role": "system", "content": "x"}],
+            },
+        )
+    assert resp.status_code == 422
+    assert session.commits == 0
+
+
+@pytest.mark.asyncio
+async def test_append_invalid_kind_returns_422() -> None:
+    client, _repo, _pr, _session = _make_client()
+    pid = client.project_id  # type: ignore[attr-defined]
+    async with client:
+        resp = await client.post(
+            f"/projects/{pid}/ai-messages",
+            json={
+                "thread_id": str(uuid.uuid4()),
+                "kind": "translate",  # 非五类枚举 → 422
+                "messages": [{"role": "ai", "content": "x"}],
+            },
+        )
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_append_empty_messages_returns_422() -> None:
+    client, _repo, _pr, _session = _make_client()
+    pid = client.project_id  # type: ignore[attr-defined]
+    async with client:
+        resp = await client.post(
+            f"/projects/{pid}/ai-messages",
+            json={"thread_id": str(uuid.uuid4()), "kind": "refine", "messages": []},
+        )
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_append_empty_content_returns_422() -> None:
+    client, _repo, _pr, _session = _make_client()
+    pid = client.project_id  # type: ignore[attr-defined]
+    async with client:
+        resp = await client.post(
+            f"/projects/{pid}/ai-messages",
+            json={
+                "thread_id": str(uuid.uuid4()),
+                "kind": "refine",
+                "messages": [{"role": "ai", "content": ""}],
+            },
+        )
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_append_content_over_cap_returns_422() -> None:
+    client, _repo, _pr, _session = _make_client()
+    pid = client.project_id  # type: ignore[attr-defined]
+    async with client:
+        resp = await client.post(
+            f"/projects/{pid}/ai-messages",
+            json={
+                "thread_id": str(uuid.uuid4()),
+                "kind": "rewrite",
+                "messages": [{"role": "ai", "content": "x" * 200_001}],
+            },
+        )
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_append_oversize_meta_returns_422() -> None:
+    client, _repo, _pr, _session = _make_client()
+    pid = client.project_id  # type: ignore[attr-defined]
+    async with client:
+        resp = await client.post(
+            f"/projects/{pid}/ai-messages",
+            json={
+                "thread_id": str(uuid.uuid4()),
+                "kind": "refine",
+                "messages": [{"role": "ai", "content": "x", "meta": {"blob": "y" * 60_000}}],
+            },
+        )
+    assert resp.status_code == 422

apps/api/ww_api/main.py

@@ -20,6 +20,7 @@ from ww_api.middleware import (
     request_id_middleware,
 )
 from ww_api.routers import (
+    ai_messages,
     chain,
     foreshadow,
     generation,
@@ -123,6 +124,7 @@ def create_app() -> FastAPI:
     app.include_router(rules.router)
     app.include_router(style.router)
     app.include_router(templates.router)
+    app.include_router(ai_messages.router)
     app.include_router(generation.router)
     app.include_router(generation.skills_router)
     app.include_router(toolbox.router)

apps/api/ww_api/routers/ai_messages.py

@@ -0,0 +1,114 @@
+"""AI 对话（聊天记录）端点（AC-2 / ai-chat-history-plan §3.3）。
+
+作者↔AI 五类往复交换（refine/rewrite/clarify/continue/generator）落库，供每章「AI 对话」
+抽屉重放。**本端点是显式的记录写入**（前端在一轮交换结束后 POST 一批）——不是生成端点；
+五个生成端点保持只读、逐字节不变（守不变量 #3）。`ai_messages` 是侧记录，assemble()/agent
+永不读它（守 #1/#6）。
+
+- POST /projects/{id}/ai-messages   一批 append（201；项目不存在 → 404）。
+- GET  /projects/{id}/ai-messages    列出（newest-first；`chapter_no` 过滤含项目级 NULL；分页）。
+
+提交边界：`AiMessageRepo.append` 只 flush，POST 端点写后 `commit()`（仿模板/规则写侧）。
+GET 只读不 commit。
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Query, Request
+from sqlalchemy.ext.asyncio import AsyncSession
+from ww_core.domain import AiMessageRepo, AiTurnRow, ProjectRepo
+from ww_core.domain.ai_message_repo import AiMessageView as AiMessageRepoView
+from ww_db import get_session
+from ww_shared import AppError, ErrorCode
+
+from ww_api.logging_config import get_logger
+from ww_api.pagination import PageDep
+from ww_api.schemas.ai_messages import (
+    AiKind,
+    AiMessageAppendRequest,
+    AiMessageListResponse,
+    AiMessageView,
+)
+from ww_api.services.credentials import STUB_OWNER_ID
+from ww_api.services.project_deps import get_ai_message_repo, get_project_repo
+
+log = get_logger("ww.api.ai_messages")
+
+router = APIRouter(prefix="/projects", tags=["ai-messages"])
+
+AiMessageRepoDep = Annotated[AiMessageRepo, Depends(get_ai_message_repo)]
+ProjectRepoDep = Annotated[ProjectRepo, Depends(get_project_repo)]
+SessionDep = Annotated[AsyncSession, Depends(get_session)]
+
+
+def _to_response_view(v: AiMessageRepoView) -> AiMessageView:
+    """repo 视图 → API 视图（同字段镜像；避免路由碰 repo 视图类型泄漏）。"""
+    return AiMessageView(
+        id=v.id,
+        project_id=v.project_id,
+        chapter_no=v.chapter_no,
+        thread_id=v.thread_id,
+        seq=v.seq,
+        kind=v.kind,
+        tool_key=v.tool_key,
+        role=v.role,
+        content=v.content,
+        meta=v.meta,
+        created_at=v.created_at,
+    )
+
+
+@router.post("/{project_id}/ai-messages", status_code=201)
+async def append_ai_messages(
+    project_id: uuid.UUID,
+    body: AiMessageAppendRequest,
+    request: Request,
+    project_repo: ProjectRepoDep,
+    repo: AiMessageRepoDep,
+    session: SessionDep,
+) -> AiMessageListResponse:
+    """一批往复 bubble append（201）。项目不存在 → 404；非法 kind/role/空/超上限 → 422。"""
+    request_id = getattr(request.state, "request_id", None)
+    project = await project_repo.get(STUB_OWNER_ID, project_id)
+    if project is None:
+        raise AppError(ErrorCode.NOT_FOUND, f"project not found: {project_id}")
+    views = await repo.append(
+        project_id,
+        thread_id=body.thread_id,
+        chapter_no=body.chapter_no,
+        kind=body.kind,
+        tool_key=body.tool_key,
+        rows=[AiTurnRow(role=m.role, content=m.content, meta=m.meta) for m in body.messages],
+    )
+    await session.commit()
+    log.info(
+        "ai_messages_appended",
+        project_id=str(project_id),
+        request_id=request_id,
+        thread_id=str(body.thread_id),
+        kind=body.kind,
+        count=len(views),
+    )
+    return AiMessageListResponse(messages=[_to_response_view(v) for v in views])
+
+
+@router.get("/{project_id}/ai-messages")
+async def list_ai_messages(
+    project_id: uuid.UUID,
+    project_repo: ProjectRepoDep,
+    repo: AiMessageRepoDep,
+    page: PageDep,
+    chapter_no: Annotated[int | None, Query(ge=1)] = None,
+    kind: AiKind | None = None,
+) -> AiMessageListResponse:
+    """列出 AI 对话（newest-first）。`chapter_no` → 本章 ∪ 项目级(NULL)；缺省 → 仅项目级。"""
+    project = await project_repo.get(STUB_OWNER_ID, project_id)
+    if project is None:
+        raise AppError(ErrorCode.NOT_FOUND, f"project not found: {project_id}")
+    views = await repo.list_for_chapter(
+        project_id, chapter_no, kind=kind, limit=page.limit, offset=page.offset
+    )
+    return AiMessageListResponse(messages=[_to_response_view(v) for v in views])

apps/api/ww_api/schemas/ai_messages.py

@@ -0,0 +1,74 @@
+"""AI 对话（聊天记录）端点的请求/响应 schema（AC-2 / ai-chat-history-plan §3.2）。
+
+snake_case；前端经 OpenAPI 生成 TS 类型消费（改字段 → 前端 `pnpm gen:api`）。
+`Literal` 是 kind/role 的**枚举权威**（DB 列自由 Text，边界在此收窄）：非法 kind/role → 422。
+批级字段（thread_id/chapter_no/kind/tool_key）提到请求顶层，`messages` 是本批各条 bubble。
+长文进 `content`（有手稿级上限），`meta` 有序列化上界兜底防绕过 content 上限。
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from datetime import datetime
+from typing import Annotated, Any, Literal
+
+from pydantic import BaseModel, Field, StringConstraints, field_validator
+
+# kind/role 的枚举权威（DB 列自由 Text；加新 kind 改这里 + 零迁移）。
+AiKind = Literal["refine", "rewrite", "clarify", "continue", "generator"]
+AiRole = Literal["author", "ai"]
+
+# content 上界 = 手稿上限（rewrite 整章文本级）；命名常量非魔法数（CR-H9）。
+AI_MESSAGE_CONTENT_MAX = 200_000
+# meta 序列化上界兜底：防长文绕过 content 上限塞进 meta。
+_META_MAX_SERIALIZED = 50_000
+# 一次 append 的最大条数（一次交换的多轮 bubble）。
+_MAX_TURNS_PER_CALL = 50
+
+
+class AiMessageInput(BaseModel):
+    """一条待落库 bubble（角色 + 人读文本 + 结构化 meta）。"""
+
+    role: AiRole
+    content: Annotated[str, StringConstraints(min_length=1, max_length=AI_MESSAGE_CONTENT_MAX)]
+    meta: dict[str, Any] = Field(default_factory=dict)
+
+    @field_validator("meta")
+    @classmethod
+    def _bound_meta(cls, v: dict[str, Any]) -> dict[str, Any]:
+        if len(json.dumps(v, ensure_ascii=False)) > _META_MAX_SERIALIZED:
+            raise ValueError("meta too large; put long text in content")
+        return v
+
+
+class AiMessageAppendRequest(BaseModel):
+    """POST /projects/{id}/ai-messages：一次交换的一批 bubble（批级字段提顶层）。"""
+
+    thread_id: uuid.UUID
+    chapter_no: Annotated[int, Field(ge=1)] | None = None  # NULL = 项目级(工具箱)
+    kind: AiKind
+    tool_key: str | None = None
+    messages: Annotated[list[AiMessageInput], Field(min_length=1, max_length=_MAX_TURNS_PER_CALL)]
+
+
+class AiMessageView(BaseModel):
+    """聊天留痕视图（append/list 回显；snake_case，镜像 repo view）。"""
+
+    id: uuid.UUID
+    project_id: uuid.UUID
+    chapter_no: int | None
+    thread_id: uuid.UUID
+    seq: int
+    kind: str
+    tool_key: str | None
+    role: str
+    content: str
+    meta: dict[str, Any] = Field(default_factory=dict)
+    created_at: datetime
+
+
+class AiMessageListResponse(BaseModel):
+    """append/list 信封：`messages` 复数键（repo 约定，非 `items`）。"""
+
+    messages: list[AiMessageView]

apps/api/ww_api/services/project_deps.py

@@ -18,6 +18,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from ww_config import get_settings
 from ww_core.domain import ForeshadowLedgerRepo, SqlForeshadowLedgerRepo
+from ww_core.domain.ai_message_repo import AiMessageRepo, SqlAiMessageRepo
 from ww_core.domain.chapter_repo import ChapterRepo, SqlChapterRepo
 from ww_core.domain.character_repo import CharacterWriteRepo, SqlCharacterWriteRepo
 from ww_core.domain.digest_repo import DigestAppendRepo, SqlDigestAppendRepo
@@ -160,6 +161,16 @@ def get_rule_write_repo(
     return SqlRuleWriteRepo(session)
 
 
+def get_ai_message_repo(
+    session: Annotated[AsyncSession, Depends(get_session)],
+) -> AiMessageRepo:
+    """AI 对话留痕 repo（POST/GET /projects/{id}/ai-messages；append 只 flush，端点提交）。
+
+    测试经 `app.dependency_overrides[get_ai_message_repo]` 注入内存 fake。
+    """
+    return SqlAiMessageRepo(session)
+
+
 async def get_template_repo(
     session: Annotated[AsyncSession, Depends(get_session)],
 ) -> TemplateRepo:

packages/core/tests/test_ai_message_repo.py

@@ -0,0 +1,293 @@
+"""AC-2 `SqlAiMessageRepo` 单测（真 pg——seq/定序是 SQL 语义，fake 无法验证）。
+
+docs/design/ai-chat-history-plan.md §3.1 / §6。验证 append-only 侧表的两操作：
+- `append`：一批多条 bubble → 批内 `seq` = 0..n-1、同 `thread_id`、meta round-trip、
+  项目级 `chapter_no=None` 持久化；只 flush（提交交调用方）。
+- `list_for_chapter`：`(created_at DESC, seq DESC)` newest-first + 分页；`chapter_no`
+  非 None → 本章 ∪ 项目级(NULL)；None → 仅项目级；可选 `kind` 过滤。
+
+关键 gotcha（本测据此设计）：Postgres `now()` 是事务起始恒定时刻——同一事务里两批共享
+`created_at`。故要拿到「跨批 created_at 不同」必须**每批各自 commit**（模拟每次 POST 独立
+事务），批内再靠 `seq` 定序。无 pg → skip。
+"""
+
+from __future__ import annotations
+
+import uuid
+from collections.abc import AsyncIterator
+from datetime import UTC, datetime
+
+import pytest
+from pydantic import ValidationError
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession
+from ww_core.domain.ai_message_repo import (
+    AiMessageRepo,
+    AiMessageView,
+    AiTurnRow,
+    SqlAiMessageRepo,
+)
+from ww_db import get_sessionmaker
+from ww_db.models import AiMessage, Project, User
+
+
+@pytest.fixture
+async def seeded() -> AsyncIterator[tuple[AsyncSession, uuid.UUID]]:
+    """真 pg session + 一个 seed 好的 project（owner user）。无 pg → skip；结束清理。"""
+    get_sessionmaker.cache_clear()
+    maker = get_sessionmaker()
+    try:
+        async with maker() as probe:
+            await probe.execute(select(1))
+    except Exception:
+        pytest.skip("postgres not reachable")
+
+    owner_id = uuid.uuid4()
+    project_id = uuid.uuid4()
+    async with maker() as setup:
+        setup.add(User(id=owner_id, email=f"ac2-{owner_id}@local", display_name="ac2"))
+        await setup.flush()  # user 先落库，满足 project.owner_id FK。
+        setup.add(Project(id=project_id, owner_id=owner_id, title="AC-2 测试作品"))
+        await setup.commit()
+
+    session = maker()
+    try:
+        yield session, project_id
+    finally:
+        await session.close()
+        async with maker() as cleanup:
+            await cleanup.execute(delete(AiMessage).where(AiMessage.project_id == project_id))
+            await cleanup.execute(delete(Project).where(Project.id == project_id))
+            await cleanup.execute(delete(User).where(User.id == owner_id))
+            await cleanup.commit()
+        await maker.kw["bind"].dispose()
+        get_sessionmaker.cache_clear()
+
+
+async def test_append_assigns_seq_0_based_and_shares_thread_id(
+    seeded: tuple[AsyncSession, uuid.UUID],
+) -> None:
+    # Arrange
+    session, project_id = seeded
+    repo: AiMessageRepo = SqlAiMessageRepo(session)
+    thread_id = uuid.uuid4()
+
+    # Act：一次交换的 3 条 bubble 作为一批 append。
+    views = await repo.append(
+        project_id,
+        thread_id=thread_id,
+        chapter_no=5,
+        kind="refine",
+        tool_key=None,
+        rows=[
+            AiTurnRow(role="author", content="原选段"),
+            AiTurnRow(role="ai", content="重写段", meta={"version_no": 1, "segment": "原选段"}),
+            AiTurnRow(role="author", content="再改一版"),
+        ],
+    )
+
+    # Assert：批内 seq = 0..n-1；同 thread_id；字段/meta round-trip；有 id/created_at。
+    assert [v.seq for v in views] == [0, 1, 2]
+    assert {v.thread_id for v in views} == {thread_id}
+    assert [v.role for v in views] == ["author", "ai", "author"]
+    assert [v.content for v in views] == ["原选段", "重写段", "再改一版"]
+    assert views[1].meta == {"version_no": 1, "segment": "原选段"}
+    assert all(v.chapter_no == 5 and v.kind == "refine" for v in views)
+    assert all(isinstance(v.id, uuid.UUID) and v.created_at is not None for v in views)
+
+
+async def test_append_project_level_chapter_no_none_persists(
+    seeded: tuple[AsyncSession, uuid.UUID],
+) -> None:
+    # Arrange
+    session, project_id = seeded
+    repo = SqlAiMessageRepo(session)
+
+    # Act：工具箱项目级生成（chapter_no=None）。
+    views = await repo.append(
+        project_id,
+        thread_id=uuid.uuid4(),
+        chapter_no=None,
+        kind="generator",
+        tool_key="expand",
+        rows=[AiTurnRow(role="ai", content="扩写预览", meta={"tool_key": "expand"})],
+    )
+
+    # Assert
+    assert views[0].chapter_no is None
+    assert views[0].tool_key == "expand"
+
+
+async def test_list_newest_first_created_at_desc_then_seq_desc(
+    seeded: tuple[AsyncSession, uuid.UUID],
+) -> None:
+    # Arrange：两批各自 commit（模拟两次 POST 独立事务）→ created_at 不同。
+    session, project_id = seeded
+    repo = SqlAiMessageRepo(session)
+    await repo.append(
+        project_id,
+        thread_id=uuid.uuid4(),
+        chapter_no=1,
+        kind="rewrite",
+        tool_key=None,
+        rows=[AiTurnRow(role="author", content="A0"), AiTurnRow(role="ai", content="A1")],
+    )
+    await session.commit()
+    await repo.append(
+        project_id,
+        thread_id=uuid.uuid4(),
+        chapter_no=1,
+        kind="rewrite",
+        tool_key=None,
+        rows=[AiTurnRow(role="author", content="B0"), AiTurnRow(role="ai", content="B1")],
+    )
+    await session.commit()
+
+    # Act
+    rows = await repo.list_for_chapter(project_id, 1, limit=50, offset=0)
+
+    # Assert：新批（B）整体先于旧批（A）；批内 seq DESC。
+    assert [r.content for r in rows] == ["B1", "B0", "A1", "A0"]
+
+
+async def test_list_chapter_union_includes_project_level(
+    seeded: tuple[AsyncSession, uuid.UUID],
+) -> None:
+    # Arrange：第 5 章 + 项目级(NULL) + 第 6 章 各一批。
+    session, project_id = seeded
+    repo = SqlAiMessageRepo(session)
+    await _append(repo, session, project_id, chapter_no=5, kind="refine", content="ch5")
+    await _append(repo, session, project_id, chapter_no=None, kind="generator", content="proj")
+    await _append(repo, session, project_id, chapter_no=6, kind="refine", content="ch6")
+
+    # Act：看第 5 章抽屉 = 本章 ∪ 项目级，绝不含第 6 章。
+    rows = await repo.list_for_chapter(project_id, 5, limit=50, offset=0)
+
+    # Assert
+    contents = {r.content for r in rows}
+    assert contents == {"ch5", "proj"}
+    assert "ch6" not in contents
+
+
+async def test_list_project_only_when_chapter_no_none(
+    seeded: tuple[AsyncSession, uuid.UUID],
+) -> None:
+    # Arrange
+    session, project_id = seeded
+    repo = SqlAiMessageRepo(session)
+    await _append(repo, session, project_id, chapter_no=5, kind="refine", content="ch5")
+    await _append(repo, session, project_id, chapter_no=None, kind="generator", content="proj")
+
+    # Act：工具箱页（chapter_no=None）只看项目级。
+    rows = await repo.list_for_chapter(project_id, None, limit=50, offset=0)
+
+    # Assert
+    assert [r.content for r in rows] == ["proj"]
+
+
+async def test_list_kind_filter(seeded: tuple[AsyncSession, uuid.UUID]) -> None:
+    # Arrange
+    session, project_id = seeded
+    repo = SqlAiMessageRepo(session)
+    await _append(repo, session, project_id, chapter_no=2, kind="refine", content="r")
+    await _append(repo, session, project_id, chapter_no=2, kind="rewrite", content="w")
+
+    # Act
+    rows = await repo.list_for_chapter(project_id, 2, kind="refine", limit=50, offset=0)
+
+    # Assert
+    assert [r.kind for r in rows] == ["refine"]
+    assert [r.content for r in rows] == ["r"]
+
+
+async def test_list_pagination_limit_offset(
+    seeded: tuple[AsyncSession, uuid.UUID],
+) -> None:
+    # Arrange：三批各一条（newest-first = c,b,a）。
+    session, project_id = seeded
+    repo = SqlAiMessageRepo(session)
+    await _append(repo, session, project_id, chapter_no=3, kind="continue", content="a")
+    await _append(repo, session, project_id, chapter_no=3, kind="continue", content="b")
+    await _append(repo, session, project_id, chapter_no=3, kind="continue", content="c")
+
+    # Act + Assert：limit=1 取最新；offset=1 取第二新。
+    first = await repo.list_for_chapter(project_id, 3, limit=1, offset=0)
+    second = await repo.list_for_chapter(project_id, 3, limit=1, offset=1)
+    assert [r.content for r in first] == ["c"]
+    assert [r.content for r in second] == ["b"]
+
+
+async def test_append_only_flushes_not_commits(
+    seeded: tuple[AsyncSession, uuid.UUID],
+) -> None:
+    # Arrange：append 只 flush；另一独立 session 在本 session commit 前看不到行。
+    session, project_id = seeded
+    repo = SqlAiMessageRepo(session)
+    thread_id = uuid.uuid4()
+    await repo.append(
+        project_id,
+        thread_id=thread_id,
+        chapter_no=9,
+        kind="refine",
+        tool_key=None,
+        rows=[AiTurnRow(role="ai", content="uncommitted")],
+    )
+
+    # Act + Assert：未提交 → 别的连接查不到。
+    maker = get_sessionmaker()
+    async with maker() as other:
+        before = (
+            (await other.execute(select(AiMessage).where(AiMessage.thread_id == thread_id)))
+            .scalars()
+            .all()
+        )
+    assert before == []
+
+    # 提交后可见。
+    await session.commit()
+    async with maker() as other:
+        after = (
+            (await other.execute(select(AiMessage).where(AiMessage.thread_id == thread_id)))
+            .scalars()
+            .all()
+        )
+    assert len(after) == 1
+
+
+def test_view_is_frozen() -> None:
+    view = AiMessageView(
+        id=uuid.uuid4(),
+        project_id=uuid.uuid4(),
+        chapter_no=None,
+        thread_id=uuid.uuid4(),
+        seq=0,
+        kind="refine",
+        tool_key=None,
+        role="ai",
+        content="x",
+        meta={},
+        created_at=datetime.now(UTC),
+    )
+    with pytest.raises(ValidationError):
+        view.content = "y"
+
+
+async def _append(
+    repo: SqlAiMessageRepo,
+    session: AsyncSession,
+    project_id: uuid.UUID,
+    *,
+    chapter_no: int | None,
+    kind: str,
+    content: str,
+) -> None:
+    """一条 ai bubble 一批 + commit（各批独立事务 → created_at 递增）。"""
+    await repo.append(
+        project_id,
+        thread_id=uuid.uuid4(),
+        chapter_no=chapter_no,
+        kind=kind,
+        tool_key=None,
+        rows=[AiTurnRow(role="ai", content=content)],
+    )
+    await session.commit()

packages/core/ww_core/domain/__init__.py

@@ -2,6 +2,12 @@
 
 from __future__ import annotations
 
+from ww_core.domain.ai_message_repo import (
+    AiMessageRepo,
+    AiMessageView,
+    AiTurnRow,
+    SqlAiMessageRepo,
+)
 from ww_core.domain.chapter_repo import (
     ChapterDraftView,
     ChapterRepo,
@@ -80,6 +86,10 @@ from ww_core.domain.world_entity_repo import (
 )
 
 __all__ = [
+    "AiMessageRepo",
+    "AiMessageView",
+    "AiTurnRow",
+    "SqlAiMessageRepo",
     "ChapterDraftView",
     "ChapterRepo",
     "ChapterView",

packages/core/ww_core/domain/ai_message_repo.py

@@ -0,0 +1,160 @@
+"""作者↔AI 往复聊天记录 Repository（append-only 侧表；AC-2 / ai-chat-history-plan §3.1）。
+
+`ai_messages` 是「说了什么」的真源，**非手稿真源**：assemble()/agent 永不读它（守 #1/#3/#6）。
+两操作：
+- `append`：一次交换的多条 bubble 作为**一批**落库，批内 `seq` = enumerate 0 基位置、同
+  `thread_id`；只 `flush()`（提交交端点事务）。Postgres `now()` 事务起始恒定 → 同批共享
+  `created_at`，靠 `seq` 批内定序；跨批 `created_at` 不同自然定序。
+- `list_for_chapter`：按 project(+可选 chapter_no) 过滤，`(created_at DESC, seq DESC)`
+  newest-first + `limit/offset` 分页。`chapter_no` 非 None → 本章 ∪ 项目级(NULL)；None →
+  仅项目级。可选 `kind` 过滤。
+
+视图 frozen、snake_case、与 ORM 解耦（路由不碰 SQLAlchemy 行）。JSONB `meta` 写入建**新
+dict**（不原地 mutate，见 memory/gotchas）。
+"""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+from typing import Any, Protocol
+
+from pydantic import BaseModel, Field
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from ww_db.models import AiMessage
+
+
+class AiTurnRow(BaseModel):
+    """一批中的一条消息（批级字段 thread_id/chapter_no/kind/tool_key 作 `append` 形参）。"""
+
+    role: str
+    content: str
+    meta: dict[str, Any] = Field(default_factory=dict)
+
+
+class AiMessageView(BaseModel):
+    """聊天留痕只读快照（snake_case，frozen）——镜像端点出参字段。"""
+
+    model_config = {"frozen": True}
+
+    id: uuid.UUID
+    project_id: uuid.UUID
+    chapter_no: int | None
+    thread_id: uuid.UUID
+    seq: int
+    kind: str
+    tool_key: str | None
+    role: str
+    content: str
+    meta: dict[str, Any]
+    created_at: datetime
+
+
+class AiMessageRepo(Protocol):
+    """聊天留痕读写接口（append 批 + newest-first 列表；append 只 flush，端点提交）。"""
+
+    async def append(
+        self,
+        project_id: uuid.UUID,
+        *,
+        thread_id: uuid.UUID,
+        chapter_no: int | None,
+        kind: str,
+        tool_key: str | None,
+        rows: list[AiTurnRow],
+    ) -> list[AiMessageView]:
+        """一批消息 append（seq = 批内 0 基位置；同 thread_id）。只 flush 不 commit。"""
+        ...
+
+    async def list_for_chapter(
+        self,
+        project_id: uuid.UUID,
+        chapter_no: int | None,
+        *,
+        kind: str | None = None,
+        limit: int,
+        offset: int,
+    ) -> list[AiMessageView]:
+        """按 project(+chapter_no) 过滤，newest-first（created_at DESC, seq DESC）分页。"""
+        ...
+
+
+def _to_view(row: AiMessage) -> AiMessageView:
+    return AiMessageView(
+        id=row.id,
+        project_id=row.project_id,
+        chapter_no=row.chapter_no,
+        thread_id=row.thread_id,
+        seq=row.seq,
+        kind=row.kind,
+        tool_key=row.tool_key,
+        role=row.role,
+        content=row.content,
+        meta=dict(row.meta or {}),
+        created_at=row.created_at,
+    )
+
+
+class SqlAiMessageRepo:
+    """SQLAlchemy 实现：批 append + newest-first 列表（只 flush 不 commit）。"""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self._s = session
+
+    async def append(
+        self,
+        project_id: uuid.UUID,
+        *,
+        thread_id: uuid.UUID,
+        chapter_no: int | None,
+        kind: str,
+        tool_key: str | None,
+        rows: list[AiTurnRow],
+    ) -> list[AiMessageView]:
+        orm_rows = [
+            AiMessage(
+                project_id=project_id,
+                chapter_no=chapter_no,
+                thread_id=thread_id,
+                seq=i,  # 批内 0 基位置：同批共享 created_at，靠 seq 稳定定序。
+                kind=kind,
+                tool_key=tool_key,
+                role=row.role,
+                content=row.content,
+                meta=dict(row.meta),  # 新 dict，避免原地 mutate（JSONB gotcha）。
+            )
+            for i, row in enumerate(rows)
+        ]
+        self._s.add_all(orm_rows)
+        await self._s.flush()
+        for row_ in orm_rows:
+            await self._s.refresh(row_)  # 回填 id / created_at。
+        return [_to_view(r) for r in orm_rows]
+
+    async def list_for_chapter(
+        self,
+        project_id: uuid.UUID,
+        chapter_no: int | None,
+        *,
+        kind: str | None = None,
+        limit: int,
+        offset: int,
+    ) -> list[AiMessageView]:
+        stmt = select(AiMessage).where(AiMessage.project_id == project_id)
+        if chapter_no is not None:
+            # 本章 ∪ 项目级(NULL)——正是抽屉 union。
+            stmt = stmt.where(
+                (AiMessage.chapter_no == chapter_no) | (AiMessage.chapter_no.is_(None))
+            )
+        else:
+            stmt = stmt.where(AiMessage.chapter_no.is_(None))  # 工具箱页仅项目级。
+        if kind is not None:
+            stmt = stmt.where(AiMessage.kind == kind)
+        stmt = (
+            stmt.order_by(AiMessage.created_at.desc(), AiMessage.seq.desc())
+            .limit(limit)
+            .offset(offset)
+        )
+        rows = (await self._s.execute(stmt)).scalars().all()
+        return [_to_view(r) for r in rows]