diff --git a/tests/test_chain_continue_volume_e2e.py b/tests/test_chain_continue_volume_e2e.py
new file mode 100644
index 0000000..be27a7f
--- /dev/null
+++ b/tests/test_chain_continue_volume_e2e.py
@@ -0,0 +1,302 @@
+"""F2 续写式链 continue_volume 端到端（真 pg + mock 网关零 token）。
+
+证明续写式链闭环（不变量 #1/#5）：`continue_volume` 每章写作以**上一章 accepted 正文末尾**
+作前文引子（复用 `build_continuation_context`）。两章无冲突全自动：
+- 第二章写作请求的上下文须含第一章 accepted 正文（断前文注入）；
+- DB 真源：两章 accepted + 两 digest 行。
+回归守卫：`draft_volume`（非续写）第二章写作请求**不应**含第一章正文（区分两模式）。
+
+镜像 `tests/test_chain_workflow_e2e.py`：真 `Gateway` + 假适配器（据 `req.output_schema` 分支
+返 parsed/text，绝不联网）+ 真 `SqlAlchemyLedgerSink`；MemorySaver 检查点。无 pg → skip。
+
+关键差异：写章假适配器**逐章返不同正文**（含章号标记），并**记录每次写章请求的 input**——
+故可断言第二章写章请求的 input 含第一章 accepted 正文标记（前文注入），而 draft_volume 不含。
+"""
+
+from __future__ import annotations
+
+import uuid
+from collections.abc import AsyncIterator, Awaitable, Callable
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
+from typing import Annotated, Any
+
+import httpx
+import pytest
+from asgi_lifespan import LifespanManager
+from fastapi import Depends
+from langgraph.checkpoint.memory import MemorySaver
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+from ww_agents import (
+    ContinuityReview,
+    ForeshadowReview,
+    PaceReview,
+    StyleDriftReview,
+)
+from ww_api.services.digest_extraction import ChapterDigestFacts
+from ww_db import get_session, get_sessionmaker
+from ww_db.models import (
+    Chapter,
+    ChapterDigest,
+    ChapterReview,
+    Job,
+    Project,
+    UsageLedger,
+)
+from ww_llm_gateway import Gateway, SqlAlchemyLedgerSink, resolve_route
+from ww_llm_gateway.adapters.base import (
+    Capabilities,
+    ProviderResult,
+    ProviderUsage,
+    StreamChunk,
+)
+from ww_llm_gateway.types import LlmRequest
+
+_PROVIDER = "deepseek"
+_USAGE = ProviderUsage(input_tokens=13, output_tokens=7)
+
+# 每章正文含唯一章号标记，供「前文注入」断言（第二章请求须含第一章标记）。
+_CHAPTER_MARK = "【第{n}章正文END】"
+
+_FORESHADOW_REVIEW = ForeshadowReview(planted=[], resolved=[])
+_PACE_REVIEW = PaceReview(water=[], hook=True, beat_map=[1, 2, 3])
+_STYLE_DRIFT = StyleDriftReview(score=88, segments=[])
+
+
+class _RecordingChainAdapter:
+    """实现 `ProviderAdapter`：写章逐章返带章号标记的正文 + 记录每次写章请求 input。
+
+    - `output_schema is None`（写章 `gateway.run`）：自增计数返第 N 章正文，记录 req.input。
+    - 四审 schema → 无冲突 parsed（让图全自动跑完）。
+    - 其余（ChapterDigestFacts）→ 终稿 digest 提炼。
+    """
+
+    provider = _PROVIDER
+
+    def __init__(self) -> None:
+        self._write_count = 0
+        # (request_input, returned_text) 每次写章一条，按写章顺序。
+        self.write_calls: list[tuple[str, str]] = []
+
+    def capabilities(self) -> Capabilities:
+        return Capabilities(structured_output=True)
+
+    async def complete(self, req: LlmRequest, model: str) -> ProviderResult:
+        schema = req.output_schema
+        if schema is None:
+            self._write_count += 1
+            text = f"灵气如潮{_CHAPTER_MARK.format(n=self._write_count)}"
+            self.write_calls.append((str(req.input), text))
+            return ProviderResult(text=text, parsed=None, usage=_USAGE)
+        if schema is ContinuityReview:
+            review = ContinuityReview(conflicts=[])
+            return ProviderResult(text=review.model_dump_json(), parsed=review, usage=_USAGE)
+        if schema is ForeshadowReview:
+            return ProviderResult(
+                text=_FORESHADOW_REVIEW.model_dump_json(), parsed=_FORESHADOW_REVIEW, usage=_USAGE
+            )
+        if schema is PaceReview:
+            return ProviderResult(
+                text=_PACE_REVIEW.model_dump_json(), parsed=_PACE_REVIEW, usage=_USAGE
+            )
+        if schema is StyleDriftReview:
+            return ProviderResult(
+                text=_STYLE_DRIFT.model_dump_json(), parsed=_STYLE_DRIFT, usage=_USAGE
+            )
+        facts = ChapterDigestFacts(summary="终稿摘要", events=["少年出山"], locations=["山门"])
+        return ProviderResult(text=facts.model_dump_json(), parsed=facts, usage=_USAGE)
+
+    async def stream(self, req: LlmRequest, model: str) -> AsyncIterator[StreamChunk]:
+        yield StreamChunk(usage=_USAGE)
+
+
+@pytest.fixture
+async def e2e_sm() -> AsyncIterator[async_sessionmaker[AsyncSession]]:
+    """真 DB session 工厂；无 pg 时跳过。"""
+    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")
+    yield maker
+    await maker.kw["bind"].dispose()
+    get_sessionmaker.cache_clear()
+
+
+def _gateway_for(adapter: _RecordingChainAdapter, session: AsyncSession) -> Gateway:
+    return Gateway(
+        adapters={_PROVIDER: adapter},
+        ledger=SqlAlchemyLedgerSink(session),
+        resolver=resolve_route,
+    )
+
+
+def _chain_gateway_override(
+    adapter: _RecordingChainAdapter,
+) -> Callable[[AsyncSession], Gateway]:
+    def _override(session: Annotated[AsyncSession, Depends(get_session)]) -> Gateway:
+        return _gateway_for(adapter, session)
+
+    return _override
+
+
+def _builder_override(
+    adapter: _RecordingChainAdapter,
+) -> Callable[[], Callable[[AsyncSession], Awaitable[Gateway]]]:
+    def _get_builder() -> Callable[[AsyncSession], Awaitable[Gateway]]:
+        async def _build(session: AsyncSession) -> Gateway:
+            return _gateway_for(adapter, session)
+
+        return _build
+
+    return _get_builder
+
+
+def _memsaver_override(
+    saver: MemorySaver,
+) -> Callable[[], Callable[[], AbstractAsyncContextManager[MemorySaver]]]:
+    @asynccontextmanager
+    async def _ctx() -> AsyncIterator[MemorySaver]:
+        yield saver
+
+    return lambda: lambda: _ctx()
+
+
+async def _cleanup(e2e_sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> None:
+    async with e2e_sm() as cleanup:
+        await cleanup.execute(delete(UsageLedger).where(UsageLedger.project_id == project_uuid))
+        await cleanup.execute(delete(Job).where(Job.project_id == project_uuid))
+        await cleanup.execute(delete(ChapterDigest).where(ChapterDigest.project_id == project_uuid))
+        await cleanup.execute(delete(ChapterReview).where(ChapterReview.project_id == project_uuid))
+        await cleanup.execute(delete(Chapter).where(Chapter.project_id == project_uuid))
+        await cleanup.execute(delete(Project).where(Project.id == project_uuid))
+        await cleanup.commit()
+
+
+def _build_app(adapter: _RecordingChainAdapter, saver: MemorySaver, e2e_sm: Any) -> Any:
+    from ww_api.main import create_app
+    from ww_api.services.chain_deps import get_checkpointer_factory
+    from ww_api.services.project_deps import (
+        get_chain_gateway,
+        get_chain_gateway_builder,
+        get_digest_gateway_builder,
+        get_session_factory,
+    )
+
+    app = create_app()
+    app.dependency_overrides[get_chain_gateway] = _chain_gateway_override(adapter)
+    app.dependency_overrides[get_chain_gateway_builder] = _builder_override(adapter)
+    app.dependency_overrides[get_session_factory] = lambda: e2e_sm
+    app.dependency_overrides[get_checkpointer_factory] = _memsaver_override(saver)
+    app.dependency_overrides[get_digest_gateway_builder] = _builder_override(adapter)
+    return app
+
+
+async def _run_two_chapter_chain(
+    adapter: _RecordingChainAdapter,
+    e2e_sm: async_sessionmaker[AsyncSession],
+    chain_key: str,
+) -> tuple[uuid.UUID, dict[str, Any]]:
+    """跑两章链（无冲突全自动）→ 返 (project_uuid, job result)。调用方负责 cleanup。"""
+    saver = MemorySaver()
+    app = _build_app(adapter, saver, e2e_sm)
+    transport = httpx.ASGITransport(app=app)
+    async with LifespanManager(app):
+        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+            create_resp = await client.post("/projects", json={"title": f"F2 {chain_key} 作品"})
+            assert create_resp.status_code == 201
+            project_id = create_resp.json()["id"]
+            project_uuid = uuid.UUID(project_id)
+
+            run_resp = await client.post(
+                f"/projects/{project_id}/chains/{chain_key}/run",
+                json={"start_chapter_no": 1, "count": 2},
+            )
+            assert run_resp.status_code == 202, run_resp.text
+            assert run_resp.json()["chain_key"] == chain_key
+            job_id = run_resp.json()["job_id"]
+
+            job = (await client.get(f"/jobs/{job_id}")).json()
+            assert job["status"] == "done", job
+            return project_uuid, job["result"]
+
+
+async def test_continue_volume_injects_prior_chapter_into_second_write(
+    e2e_sm: async_sessionmaker[AsyncSession],
+) -> None:
+    """用例 1：continue_volume 两章 → 第二章写章请求含第一章 accepted 正文（前文注入）。"""
+    adapter = _RecordingChainAdapter()
+    project_uuid: uuid.UUID | None = None
+    try:
+        project_uuid, result = await _run_two_chapter_chain(adapter, e2e_sm, "continue_volume")
+        assert result["written"] == [1, 2]
+        assert result["completed"] is True
+
+        # 两次写章请求被记录（按章序）。
+        assert len(adapter.write_calls) == 2
+        first_input, first_text = adapter.write_calls[0]
+        second_input, _second_text = adapter.write_calls[1]
+
+        # 第一章正文标记（accepted 后即此文本）。
+        chapter1_mark = _CHAPTER_MARK.format(n=1)
+
+        # 核心断言：第二章写章请求上下文含第一章 accepted 正文（前文引子注入，不变量 #1/#5）。
+        assert chapter1_mark in second_input, second_input
+        # 续写上下文结构标记（build_continuation_context）也应在第二章请求中。
+        assert "前文正文" in second_input
+        # 第一章无前文 → 其请求不含「第1章标记」（首章占位降级，证明非凭空注入）。
+        assert chapter1_mark not in first_input
+
+        # DB 真源：两章 accepted + 两 digest 行。
+        async with e2e_sm() as verify:
+            accepted = (
+                (
+                    await verify.execute(
+                        select(Chapter).where(
+                            Chapter.project_id == project_uuid, Chapter.status == "accepted"
+                        )
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            assert {c.chapter_no for c in accepted} == {1, 2}
+            digests = (
+                (
+                    await verify.execute(
+                        select(ChapterDigest).where(ChapterDigest.project_id == project_uuid)
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            assert {d.chapter_no for d in digests} == {1, 2}
+    finally:
+        if project_uuid is not None:
+            await _cleanup(e2e_sm, project_uuid)
+
+
+async def test_draft_volume_does_not_inject_prior_chapter_into_second_write(
+    e2e_sm: async_sessionmaker[AsyncSession],
+) -> None:
+    """用例 2（回归守卫）：draft_volume 第二章写章请求**不含**第一章正文（区分两模式）。"""
+    adapter = _RecordingChainAdapter()
+    project_uuid: uuid.UUID | None = None
+    try:
+        project_uuid, result = await _run_two_chapter_chain(adapter, e2e_sm, "draft_volume")
+        assert result["written"] == [1, 2]
+
+        assert len(adapter.write_calls) == 2
+        _first_input, _first_text = adapter.write_calls[0]
+        second_input, _ = adapter.write_calls[1]
+
+        chapter1_mark = _CHAPTER_MARK.format(n=1)
+        # draft_volume 非续写：第二章写章请求不应含第一章正文（仅按记忆/digest 量产）。
+        assert chapter1_mark not in second_input
+        # 亦无续写上下文结构标记。
+        assert "前文正文（续写须无缝承接）" not in second_input
+    finally:
+        if project_uuid is not None:
+            await _cleanup(e2e_sm, project_uuid)
diff --git a/tests/test_teardown_ingest_e2e.py b/tests/test_teardown_ingest_e2e.py
new file mode 100644
index 0000000..71beebd
--- /dev/null
+++ b/tests/test_teardown_ingest_e2e.py
@@ -0,0 +1,216 @@
+"""F1 拆书 teardown 落库成 rules 端到端（真 pg + mock 网关零 token）。
+
+证明拆书入库闭环（不变量 #3：入库经白名单 gate；预览不写库）：
+1. teardown `generate` → 200 预览（BookTeardownResult）；断言 `rules` 表零新增（预览不写库）。
+2. teardown `ingest`（teardown 产物）→ 201；`rules` 表真落行（拆书结论拍平成可读规则条目）。
+
+镜像 `tests/test_t6_toolbox_e2e.py`：真 `Gateway` + 据 `req.output_schema` 分支返 parsed 的
+假适配器（绝不联网）+ 真 `SqlAlchemyLedgerSink`。注入缝 `get_tier_gateway_builder`（generate
+调 `build_gateway(spec.tier)`；rules ingest 无 continuity 预检故不调网关）。无 pg → skip。
+"""
+
+from __future__ import annotations
+
+import uuid
+from collections.abc import AsyncIterator, Awaitable, Callable
+from typing import Annotated
+
+import httpx
+import pytest
+from asgi_lifespan import LifespanManager
+from fastapi import Depends
+from sqlalchemy import delete, func, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+from ww_agents import BookTeardownResult
+from ww_db import get_session, get_sessionmaker
+from ww_db.models import Project, Rule, UsageLedger
+from ww_llm_gateway import Gateway, SqlAlchemyLedgerSink, resolve_route
+from ww_llm_gateway.adapters.base import (
+    Capabilities,
+    ProviderResult,
+    ProviderUsage,
+    StreamChunk,
+)
+from ww_llm_gateway.types import LlmRequest, Tier
+
+_PROVIDER = "deepseek"
+_USAGE = ProviderUsage(input_tokens=29, output_tokens=13)
+
+# teardown 假产物：themes/archetypes/structure/hooks 全非空（拍平成 4 条 rules）。
+_TEARDOWN = BookTeardownResult(
+    themes=["逆袭打脸", "守护苍生"],
+    archetypes=["扮猪吃虎的主角", "口嫌体正的女主"],
+    structure="开篇废柴受辱 → 奇遇得宝 → 步步攀升 → 终战封神",
+    hooks=["每章末留悬念", "三章一小高潮"],
+)
+
+
+class _FakeTeardownAdapter:
+    """实现 `ProviderAdapter`：仅 BookTeardownResult 一路（拆书 generate），绝不联网。"""
+
+    provider = _PROVIDER
+
+    def capabilities(self) -> Capabilities:
+        return Capabilities(structured_output=True)
+
+    async def complete(self, req: LlmRequest, model: str) -> ProviderResult:
+        if req.output_schema is BookTeardownResult:
+            return ProviderResult(text=_TEARDOWN.model_dump_json(), parsed=_TEARDOWN, usage=_USAGE)
+        raise AssertionError(f"unexpected output_schema in F1 fake adapter: {req.output_schema!r}")
+
+    async def stream(self, req: LlmRequest, model: str) -> AsyncIterator[StreamChunk]:
+        yield StreamChunk(usage=_USAGE)
+        raise AssertionError("teardown generation must not stream")
+
+
+@pytest.fixture
+async def e2e_sm() -> AsyncIterator[async_sessionmaker[AsyncSession]]:
+    """真 DB session 工厂；无 pg 时跳过。"""
+    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")
+    yield maker
+    await maker.kw["bind"].dispose()
+    get_sessionmaker.cache_clear()
+
+
+def _tier_builder_override(
+    adapter: _FakeTeardownAdapter,
+) -> Callable[[AsyncSession], Callable[[Tier], Awaitable[Gateway]]]:
+    def _override(
+        session: Annotated[AsyncSession, Depends(get_session)],
+    ) -> Callable[[Tier], Awaitable[Gateway]]:
+        async def _build(_tier: Tier) -> Gateway:
+            return Gateway(
+                adapters={adapter.provider: adapter},
+                ledger=SqlAlchemyLedgerSink(session),
+                resolver=resolve_route,
+            )
+
+        return _build
+
+    return _override
+
+
+async def _rule_count(sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> int:
+    async with sm() as s:
+        return int(
+            (
+                await s.execute(
+                    select(func.count()).select_from(Rule).where(Rule.project_id == project_uuid)
+                )
+            ).scalar_one()
+        )
+
+
+async def _cleanup(sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> None:
+    async with sm() as cleanup:
+        await cleanup.execute(delete(UsageLedger).where(UsageLedger.project_id == project_uuid))
+        await cleanup.execute(delete(Rule).where(Rule.project_id == project_uuid))
+        await cleanup.execute(delete(Project).where(Project.id == project_uuid))
+        await cleanup.commit()
+
+
+async def test_teardown_generate_preview_does_not_write_rules(
+    e2e_sm: async_sessionmaker[AsyncSession],
+) -> None:
+    """用例 1（负向）：teardown generate → 200 预览；rules 表零新增（不变量 #3 预览不写库）。"""
+    from ww_api.main import create_app
+    from ww_api.services.project_deps import get_tier_gateway_builder
+
+    app = create_app()
+    app.dependency_overrides[get_tier_gateway_builder] = _tier_builder_override(
+        _FakeTeardownAdapter()
+    )
+    transport = httpx.ASGITransport(app=app)
+    project_uuid: uuid.UUID | None = None
+    try:
+        async with LifespanManager(app):
+            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+                create_resp = await client.post("/projects", json={"title": "F1 拆书预览作品"})
+                assert create_resp.status_code == 201
+                project_id = create_resp.json()["id"]
+                project_uuid = uuid.UUID(project_id)
+
+                gen_resp = await client.post(
+                    f"/projects/{project_id}/skills/teardown/generate",
+                    json={"text": "第一章……（待拆解的样章正文）", "kind": "某爆款玄幻"},
+                )
+                assert gen_resp.status_code == 200, gen_resp.text
+                body = gen_resp.json()
+                assert body["tool_key"] == "teardown"
+                assert body["output_kind"] == "BookTeardownResult"
+                assert body["preview"]["themes"] == ["逆袭打脸", "守护苍生"]
+
+        # DB 真源：预览不写 rules 业务表（守不变量 #3）。
+        assert project_uuid is not None
+        assert await _rule_count(e2e_sm, project_uuid) == 0
+    finally:
+        if project_uuid is not None:
+            await _cleanup(e2e_sm, project_uuid)
+
+
+async def test_teardown_ingest_persists_rules(
+    e2e_sm: async_sessionmaker[AsyncSession],
+) -> None:
+    """用例 2：teardown ingest → 201；rules 表真落行（拆书结论拍平成可读规则条目）。"""
+    from ww_api.main import create_app
+    from ww_api.services.project_deps import get_tier_gateway_builder
+
+    app = create_app()
+    # ingest（rules 无 continuity 预检）不触达网关；override 防真凭据探测路径。
+    app.dependency_overrides[get_tier_gateway_builder] = _tier_builder_override(
+        _FakeTeardownAdapter()
+    )
+    transport = httpx.ASGITransport(app=app)
+    project_uuid: uuid.UUID | None = None
+    try:
+        async with LifespanManager(app):
+            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+                create_resp = await client.post("/projects", json={"title": "F1 拆书入库作品"})
+                assert create_resp.status_code == 201
+                project_id = create_resp.json()["id"]
+                project_uuid = uuid.UUID(project_id)
+
+                assert await _rule_count(e2e_sm, project_uuid) == 0
+
+                ingest_resp = await client.post(
+                    f"/projects/{project_id}/skills/teardown/ingest",
+                    json={
+                        "teardown": {
+                            "themes": _TEARDOWN.themes,
+                            "archetypes": _TEARDOWN.archetypes,
+                            "structure": _TEARDOWN.structure,
+                            "hooks": _TEARDOWN.hooks,
+                        }
+                    },
+                )
+                assert ingest_resp.status_code == 201, ingest_resp.text
+                ok = ingest_resp.json()
+                assert ok["table"] == "rules"
+                # 四个非空字段 → 四条 rules（确定性顺序）。
+                assert ok["created"] == ["themes", "archetypes", "structure", "hooks"]
+                assert ok["rejected_tables"] == []
+
+        # DB 真源：rules 行真落 pg（拍平成可读条目，内容含拆书结论）。
+        assert project_uuid is not None
+        assert await _rule_count(e2e_sm, project_uuid) == 4
+        async with e2e_sm() as verify:
+            rows = (
+                (await verify.execute(select(Rule).where(Rule.project_id == project_uuid)))
+                .scalars()
+                .all()
+            )
+            blob = "\n".join(r.content for r in rows)
+            # 拍平内容须含拆书结论（断真落行 + 形变保留）。
+            assert "逆袭打脸" in blob
+            assert "扮猪吃虎的主角" in blob
+            assert "开篇废柴受辱" in blob
+            assert "每章末留悬念" in blob
+    finally:
+        if project_uuid is not None:
+            await _cleanup(e2e_sm, project_uuid)
diff --git a/tests/test_templates_e2e.py b/tests/test_templates_e2e.py
new file mode 100644
index 0000000..ae39785
--- /dev/null
+++ b/tests/test_templates_e2e.py
@@ -0,0 +1,137 @@
+"""F3 提示词/模板库 端到端（真 pg；无网关——模板库不调 LLM）。
+
+证明单用户本地模板库 CRUD 闭环：创建→列出→删除真 pg 回环 + 输入校验（title/body 空 → 422）。
+owner_id 全程 stub（单用户原型）。无 pg → skip。
+
+镜像现有 `tests/test_*_e2e.py` 范式：真 `get_sessionmaker`（无 pg skip）+ `LifespanManager` +
+`ASGITransport`，DB 真源逐项断言。模板库不触达 LLM 网关，故无需 mock 适配器。
+"""
+
+from __future__ import annotations
+
+import uuid
+from collections.abc import AsyncIterator
+
+import httpx
+import pytest
+from asgi_lifespan import LifespanManager
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+from ww_db import get_sessionmaker
+from ww_db.models import PromptTemplate
+
+
+@pytest.fixture
+async def e2e_sm() -> AsyncIterator[async_sessionmaker[AsyncSession]]:
+    """真 DB session 工厂；无 pg 时跳过（每测试清缓存重建 engine、结束 dispose）。"""
+    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")
+    yield maker
+    await maker.kw["bind"].dispose()
+    get_sessionmaker.cache_clear()
+
+
+async def _cleanup(sm: async_sessionmaker[AsyncSession], ids: list[uuid.UUID]) -> None:
+    if not ids:
+        return
+    async with sm() as cleanup:
+        await cleanup.execute(delete(PromptTemplate).where(PromptTemplate.id.in_(ids)))
+        await cleanup.commit()
+
+
+async def _template_count(sm: async_sessionmaker[AsyncSession], template_id: uuid.UUID) -> int:
+    async with sm() as s:
+        return len(
+            (await s.execute(select(PromptTemplate).where(PromptTemplate.id == template_id)))
+            .scalars()
+            .all()
+        )
+
+
+async def test_templates_create_list_delete_roundtrip(
+    e2e_sm: async_sessionmaker[AsyncSession],
+) -> None:
+    """用例 1：POST → GET → DELETE 真 pg 回环（行真落 pg；删除后表无行）。"""
+    from ww_api.main import create_app
+
+    app = create_app()
+    transport = httpx.ASGITransport(app=app)
+    created_id: uuid.UUID | None = None
+    try:
+        async with LifespanManager(app):
+            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+                # 1) POST → 201，模板真落 pg。
+                create_resp = await client.post(
+                    "/templates",
+                    json={
+                        "title": "玄幻开篇模板",
+                        "body": "请按照黄金三章节奏，开篇即上钩……",
+                        "category": "开篇",
+                        "tool_key": "opening",
+                    },
+                )
+                assert create_resp.status_code == 201, create_resp.text
+                created = create_resp.json()
+                assert created["title"] == "玄幻开篇模板"
+                assert created["body"] == "请按照黄金三章节奏，开篇即上钩……"
+                assert created["category"] == "开篇"
+                assert created["tool_key"] == "opening"
+                created_id = uuid.UUID(created["id"])
+
+                # DB 真源：模板行真落 pg（owner stub）。
+                assert await _template_count(e2e_sm, created_id) == 1
+
+                # 2) GET → 列表含刚建模板。
+                list_resp = await client.get("/templates")
+                assert list_resp.status_code == 200
+                listed = list_resp.json()
+                assert created["id"] in [t["id"] for t in listed]
+                mine = next(t for t in listed if t["id"] == created["id"])
+                assert mine["title"] == "玄幻开篇模板"
+                assert mine["body"] == "请按照黄金三章节奏，开篇即上钩……"
+
+                # 3) DELETE → 204；DB 真源行被删。
+                del_resp = await client.delete(f"/templates/{created['id']}")
+                assert del_resp.status_code == 204
+                assert await _template_count(e2e_sm, created_id) == 0
+
+                # 删除后再列出不含该模板。
+                after = (await client.get("/templates")).json()
+                assert created["id"] not in [t["id"] for t in after]
+
+                # 4) 删除不存在 → 404 NOT_FOUND envelope。
+                missing = await client.delete(f"/templates/{uuid.uuid4()}")
+                assert missing.status_code == 404
+                assert missing.json()["error"]["code"] == "NOT_FOUND"
+    finally:
+        if created_id is not None:
+            await _cleanup(e2e_sm, [created_id])
+
+
+async def test_templates_create_blank_title_returns_422(
+    e2e_sm: async_sessionmaker[AsyncSession],
+) -> None:
+    """用例 2：title 空 → 422（schema min_length=1）；DB 无行落库。"""
+    from ww_api.main import create_app
+
+    app = create_app()
+    transport = httpx.ASGITransport(app=app)
+    async with LifespanManager(app):
+        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+            before = len((await client.get("/templates")).json())
+
+            resp = await client.post("/templates", json={"title": "", "body": "非空正文"})
+            assert resp.status_code == 422, resp.text
+
+            # body 空亦 422（守输入校验纪律）。
+            resp_body = await client.post("/templates", json={"title": "有标题", "body": ""})
+            assert resp_body.status_code == 422
+
+            # 两次 422 均未落库（列表数量不变）。
+            after = len((await client.get("/templates")).json())
+            assert after == before