perf(backend): 角色/世界观入库改批量——create_many 一次 flush 落全部行

角色入库端点与 toolbox world_entities 入库端点原逐卡/逐条 create（每行一次
SELECT/flush/refresh，N+1 round-trip）。新增 create_many：世界观用 add_all，
角色用一次 IN 查已有名 + 一次 flush（保持 (project_id,name) upsert 幂等与批内
重名去重、保序返回）。单条 create 薄封装 create_many，行为不变（既有调用方/E2E
复用）。复活并复用 CharacterWriteFields、新增 WorldEntityFields 作批量入参。
补两端点「多行一次落库」测试；既有幂等/冲突 gate 测试与 m5 真 pg E2E 全绿。

apps/api/tests/test_generation.py

@@ -28,7 +28,7 @@ from ww_agents import (
     WorldEntityCard,
     WorldGenResult,
 )
-from ww_core.domain.character_repo import CharacterWriteView
+from ww_core.domain.character_repo import CharacterWriteFields, CharacterWriteView
 from ww_core.domain.project_repo import ProjectCreate
 from ww_core.domain.rule_repo import RuleListItemView
 from ww_llm_gateway.types import LlmRequest, LlmResponse, ServedBy, Usage
@@ -120,6 +120,27 @@ class _FakeCharacterWriteRepo:
         )
         return CharacterWriteView(id=row_id, name=name, role=role)
 
+    async def create_many(
+        self, project_id: uuid.UUID, cards: list[CharacterWriteFields]
+    ) -> list[CharacterWriteView]:
+        # 镜像 Sql：逐卡 upsert（含批内重名去重），保序返回逐卡 View。
+        return [
+            await self.create(
+                project_id,
+                name=c.name,
+                role=c.role,
+                traits=list(c.traits),
+                appearance=c.appearance,
+                motive=c.motive,
+                backstory=c.backstory,
+                arc=c.arc,
+                speech_tics=list(c.speech_tics),
+                tags=list(c.tags),
+                relations=[dict(r) for r in c.relations],
+            )
+            for c in cards
+        ]
+
 
 class _FakeRulesReadRepo:
     """实现规则 repo 的读侧 `list_for_project`（带 id，供 list_rules 端点）。"""
@@ -352,6 +373,39 @@ async def test_ingest_characters_no_conflict_writes_and_201() -> None:
     assert session.commits == 1
 
 
+@pytest.mark.asyncio
+async def test_ingest_multiple_characters_all_land_in_one_batch() -> None:
+    # 批量入库：多张卡一次 create_many 全部落库，created 保序，单次 commit。
+    repo = FakeProjectRepo()
+    pid = await _seed_project(repo)
+    gateway = _SchemaRoutingGateway({ContinuityReview: _no_conflicts()})
+    char_repo = _FakeCharacterWriteRepo()
+    app, session = _make_app(project_repo=repo, gateway=gateway, char_repo=char_repo)
+
+    def _card(name: str) -> dict[str, Any]:
+        return {
+            "name": name,
+            "role": "配角",
+            "traits": [],
+            "backstory": "x",
+            "arc": "y",
+            "speech_tics": [],
+            "tags": [],
+            "relations": [],
+        }
+
+    async with _client(app) as client:
+        resp = await client.post(
+            f"/projects/{pid}/characters",
+            json={"cards": [_card("甲"), _card("乙"), _card("丙")]},
+        )
+
+    assert resp.status_code == 201
+    assert resp.json()["created"] == ["甲", "乙", "丙"]
+    assert [r["name"] for r in char_repo.rows] == ["甲", "乙", "丙"]
+    assert session.commits == 1
+
+
 @pytest.mark.asyncio
 async def test_ingest_characters_with_conflict_blocks_409() -> None:
     repo = FakeProjectRepo()

apps/api/tests/test_toolbox_endpoints.py

@@ -32,7 +32,7 @@ from ww_core.domain.outline_write_repo import OutlineWriteView
 from ww_core.domain.project_repo import ProjectCreate
 from ww_core.domain.repositories import OutlineView
 from ww_core.domain.rule_repo import RuleWriteView
-from ww_core.domain.world_entity_repo import WorldEntityWriteView
+from ww_core.domain.world_entity_repo import WorldEntityFields, WorldEntityWriteView
 from ww_llm_gateway.types import LlmRequest, LlmResponse, ServedBy, Usage
 from ww_shared import AppError, ErrorCode
 
@@ -75,6 +75,15 @@ class _FakeWorldWriteRepo:
         self.rows.append({"type": type, "name": name, "rules": list(rules)})
         return WorldEntityWriteView(id=uuid.uuid4(), type=type, name=name)
 
+    async def create_many(
+        self, project_id: uuid.UUID, entities: list[WorldEntityFields]
+    ) -> list[WorldEntityWriteView]:
+        # 镜像 Sql：批量插入，保序返回逐条 View。
+        return [
+            await self.create(project_id, type=e.type, name=e.name, rules=list(e.rules))
+            for e in entities
+        ]
+
 
 class _FakeOutlineWriteRepo:
     def __init__(self) -> None:
@@ -569,6 +578,30 @@ async def test_ingest_world_entities_no_conflict_writes_201() -> None:
     assert session.commits == 1
 
 
+@pytest.mark.asyncio
+async def test_ingest_multiple_world_entities_all_land_in_one_batch() -> None:
+    # 批量入库：多条实体一次 create_many 全部落库，created 保序，单次 commit。
+    repo = FakeProjectRepo()
+    pid = await _seed_project(repo)
+    gateway = _SchemaRoutingGateway({ContinuityReview: _no_conflicts()})
+    app, session, world_repo, _ = _make_app(project_repo=repo, gateway=gateway)
+    payload = {
+        "world_entities": [
+            {"type": "力量体系", "name": "吞噬系统", "rules": ["每日上限"]},
+            {"type": "势力", "name": "天机阁", "rules": ["中立"]},
+            {"type": "地点", "name": "无尽海", "rules": []},
+        ]
+    }
+
+    async with _client(app) as client:
+        resp = await client.post(f"/projects/{pid}/skills/golden-finger/ingest", json=payload)
+
+    assert resp.status_code == 201
+    assert resp.json()["created"] == ["吞噬系统", "天机阁", "无尽海"]
+    assert [r["name"] for r in world_repo.rows] == ["吞噬系统", "天机阁", "无尽海"]
+    assert session.commits == 1
+
+
 @pytest.mark.asyncio
 async def test_ingest_world_entities_conflict_blocks_409() -> None:
     repo = FakeProjectRepo()