life-echo/docs/plans/2026-03-10-memoir-image-generation-implementation.md

# Memoir Image Generation Implementation Plan

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

**Goal:** 实现回忆录章节图片的完整闭环：解析 `{{{{IMAGE:...}}}}` 占位符，异步调用 Liblib 生成图片，上传腾讯云 COS，写回 `Chapter.images`，并在 Android 阅读页与 PDF 导出中正确渲染。

**Architecture:** 在后端新增一个小型 `memoir_images` 服务包，负责占位符解析、prompt 优化、provider 适配、COS 上传和状态合并。`process_memoir_segments` 只负责初始化图片元数据并派发子任务，`generate_chapter_images` 负责异步补图；Android 和 PDF 都改为消费 `Chapter.images` 对象数组，而不是依赖原始占位符文本。

**Tech Stack:** Python 3, FastAPI, SQLAlchemy, Celery, Redis, LangChain/OpenAI-compatible LLM, httpx, ReportLab, cos-python-sdk-v5, Kotlin, Jetpack Compose, Coil, JUnit4, Gradle

**Related Skills:** @superpowers:executing-plans, @superpowers:verification-before-completion, @superpowers:test-driven-development

---

## Preconditions

- 在独立 worktree 中执行这份计划，不要直接在主开发目录上动手。
- 设计稿参考：[2026-03-10-memoir-image-generation-design.md](/Users/timcook/Codes/hgtk/life-echo/docs/plans/2026-03-10-memoir-image-generation-design.md)
- `chapters.images` 已经是 JSON/JSONB 列，本计划不新增 SQL schema migration，只升级其内容结构。
- 后端测试统一从仓库根目录运行：`python -m unittest ... -v`
- Android 单测统一从 `app-android` 目录运行：`./gradlew :app:testDebugUnitTest ...`
- Android Compose 集成测试统一从 `app-android` 目录运行：`./gradlew :app:connectedDebugAndroidTest ...`

## Implementation Order

按顺序执行，不要并行跳步：

1. 先完成后端纯函数和服务抽象
2. 再接 Celery 任务与 API 出参
3. 然后改 PDF
4. 最后改 Android 数据模型和 UI

## Source of Truth Note

- 本期图片链路只以 `chapter.content` 里的 `{{{{IMAGE:...}}}}` 占位符为准。
- `api/agents/memory_agent.py` 中的 `image_suggestions` 这期不接入主流程，也不作为渲染或生成输入。
- 如果实现过程中发现 `image_suggestions` 与占位符信息重复，保持字段原样不动，后续单独做清理或删除。

## Task 1: Add placeholder parsing and initial image asset builders

**Files:**
- Create: `api/services/memoir_images/__init__.py`
- Create: `api/services/memoir_images/parser.py`
- Test: `api/tests/test_memoir_image_parser.py`

**Step 1: Write the failing test**

```python
import unittest

from api.services.memoir_images.parser import (
    build_initial_image_assets,
    parse_image_placeholders,
)


class MemoirImageParserTest(unittest.TestCase):
    def test_parse_image_placeholders_preserves_order_and_offsets(self):
        content = (
            "那条路我一直记得。\n\n"
            "{{{{IMAGE:南方小镇的青石板路}}}}\n\n"
            "奶奶总坐在门口。\n\n"
            "{{{{IMAGE:奶奶坐在院子里的藤椅上}}}}"
        )

        items = parse_image_placeholders(content, max_images=3)

        self.assertEqual([item["index"] for item in items], [0, 1])
        self.assertEqual(items[0]["description"], "南方小镇的青石板路")
        self.assertEqual(items[1]["placeholder"], "{{{{IMAGE:奶奶坐在院子里的藤椅上}}}}")
        self.assertLess(items[0]["start_offset"], items[1]["start_offset"])

    def test_build_initial_image_assets_marks_every_item_pending(self):
        placeholders = [
            {
                "index": 0,
                "description": "南方小镇的青石板路",
                "placeholder": "{{{{IMAGE:南方小镇的青石板路}}}}",
                "start_offset": 10,
            }
        ]

        assets = build_initial_image_assets(
            placeholders=placeholders,
            provider="liblib",
            style="watercolor",
            size="1024x1024",
            now_iso="2026-03-10T10:00:00Z",
        )

        self.assertEqual(assets[0]["status"], "pending")
        self.assertEqual(assets[0]["provider"], "liblib")
        self.assertEqual(assets[0]["url"], None)
```

**Step 2: Run test to verify it fails**

Run: `cd /Users/timcook/Codes/hgtk/life-echo && python -m unittest api.tests.test_memoir_image_parser -v`  
Expected: FAIL with `ModuleNotFoundError` or missing function errors

**Step 3: Write minimal implementation**

```python
import re
from typing import Any

PLACEHOLDER_RE = re.compile(r"\{\{\{\{IMAGE:(.*?)\}\}\}\}")


def parse_image_placeholders(content: str, max_images: int) -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []
    for match_index, match in enumerate(PLACEHOLDER_RE.finditer(content or "")):
        description = match.group(1).strip()
        if not description:
            continue
        items.append(
            {
                "index": len(items),
                "description": description,
                "placeholder": match.group(0),
                "start_offset": match.start(),
            }
        )
        if len(items) >= max_images:
            break
    return items


def build_initial_image_assets(...):
    return [
        {
            "index": item["index"],
            "placeholder": item["placeholder"],
            "description": item["description"],
            "prompt": None,
            "url": None,
            "status": "pending",
            "provider": provider,
            "style": style,
            "size": size,
            "error": None,
            "created_at": now_iso,
            "updated_at": now_iso,
        }
        for item in placeholders
    ]
```

**Step 4: Run test to verify it passes**

Run: `cd /Users/timcook/Codes/hgtk/life-echo && python -m unittest api.tests.test_memoir_image_parser -v`  
Expected: PASS

**Step 5: Commit**

```bash
git add api/services/memoir_images/__init__.py \
        api/services/memoir_images/parser.py \
        api/tests/test_memoir_image_parser.py
git commit -m "feat(api): add memoir image placeholder parsing"
```

## Task 2: Add image settings and prompt optimization service

**Files:**
- Create: `api/services/memoir_images/settings.py`
- Create: `api/services/memoir_images/prompting.py`
- Modify: `api/services/memoir_images/__init__.py`
- Test: `api/tests/test_memoir_image_prompting.py`

Reference: [2026-03-10-memoir-image-generation-design.md](/Users/timcook/Codes/hgtk/life-echo/docs/plans/2026-03-10-memoir-image-generation-design.md) §7.2

**Step 1: Write the failing test**

```python
import unittest
from unittest.mock import Mock

from api.services.memoir_images.prompting import MemoirImagePromptService
from api.services.memoir_images.settings import MemoirImageSettings


class MemoirImagePromptingTest(unittest.TestCase):
    def test_prompt_service_uses_category_style_and_plain_fallback_without_llm(self):
        settings = MemoirImageSettings(
            enabled=True,
            max_per_chapter=2,
            provider="liblib",
            default_style="watercolor",
            default_size="1024x1024",
            poll_interval_seconds=3,
            max_attempts=20,
        )
        service = MemoirImagePromptService(llm=None, settings=settings)

        result = service.build_prompt(
            chapter_title="童年的夏天",
            chapter_category="childhood",
            description="奶奶坐在院子里的藤椅上",
            context_excerpt="梧桐树下很安静，夏天总有蝉鸣。",
        )

        self.assertEqual(result["style"], "watercolor")
        self.assertEqual(result["size"], "1024x1024")
        self.assertIn("奶奶坐在院子里的藤椅上", result["prompt"])
        self.assertIn("childhood", result["prompt_context"])

    def test_prompt_service_parses_structured_llm_response(self):
        settings = MemoirImageSettings(
            enabled=True,
            max_per_chapter=2,
            provider="liblib",
            default_style="watercolor",
            default_size="1024x1024",
            poll_interval_seconds=3,
            max_attempts=20,
        )
        llm = Mock()
        llm.invoke.return_value.content = (
            '{"prompt":"A grandmother in a quiet courtyard, summer cicadas, soft watercolor",'
            '"style":"watercolor","size":"1024x1024"}'
        )
        service = MemoirImagePromptService(llm=llm, settings=settings)

        result = service.build_prompt(
            chapter_title="童年的夏天",
            chapter_category="childhood",
            description="奶奶坐在院子里的藤椅上",
            context_excerpt="梧桐树下很安静，夏天总有蝉鸣。",
        )

        self.assertEqual(result["prompt"], "A grandmother in a quiet courtyard, summer cicadas, soft watercolor")
        self.assertEqual(result["style"], "watercolor")
        self.assertEqual(result["size"], "1024x1024")
```

**Step 2: Run test to verify it fails**

Run: `cd /Users/timcook/Codes/hgtk/life-echo && python -m unittest api.tests.test_memoir_image_prompting -v`  
Expected: FAIL with missing module/class errors

**Step 3: Write minimal implementation**

```python
from dataclasses import dataclass
import json
import os


@dataclass(frozen=True)
class MemoirImageSettings:
    enabled: bool
    max_per_chapter: int
    provider: str
    default_style: str
    default_size: str
    poll_interval_seconds: int
    max_attempts: int

    @classmethod
    def from_env(cls) -> "MemoirImageSettings":
        return cls(
            enabled=os.getenv("MEMOIR_IMAGE_ENABLED", "").lower() in {"1", "true", "yes"},
            max_per_chapter=int(os.getenv("MEMOIR_IMAGE_MAX_PER_CHAPTER", "2")),
            provider=os.getenv("MEMOIR_IMAGE_PROVIDER", "liblib"),
            default_style=os.getenv("MEMOIR_IMAGE_STYLE_DEFAULT", "watercolor"),
            default_size=os.getenv("MEMOIR_IMAGE_SIZE_DEFAULT", "1024x1024"),
            poll_interval_seconds=int(os.getenv("MEMOIR_IMAGE_POLL_INTERVAL", "3")),
            max_attempts=int(os.getenv("MEMOIR_IMAGE_MAX_ATTEMPTS", "20")),
        )
```

```python
class MemoirImagePromptService:
    CATEGORY_STYLE_MAP = {
        "childhood": "watercolor",
        "family": "watercolor",
        "career_early": "realistic",
        "career_achievement": "realistic",
        "career_challenge": "realistic",
        "beliefs": "editorial illustration",
        "summary": "editorial illustration",
    }

    def build_prompt(...):
        style = self.CATEGORY_STYLE_MAP.get(chapter_category, self.settings.default_style)
        prompt_context = f"{chapter_category}: {chapter_title}"
        llm_input = {
            "chapter_title": chapter_title,
            "chapter_category": chapter_category,
            "description": description,
            "context_excerpt": context_excerpt,
            "default_style": style,
            "default_size": self.settings.default_size,
        }
        if self.llm:
            try:
                response = self.llm.invoke(
                    "Return JSON only with keys prompt, style, size. "
                    f"Convert the memoir scene into an image-generation prompt.\n{json.dumps(llm_input, ensure_ascii=False)}"
                )
                parsed = json.loads(response.content)
                return {
                    "prompt": parsed["prompt"],
                    "style": parsed.get("style", style),
                    "size": parsed.get("size", self.settings.default_size),
                    "prompt_context": prompt_context,
                }
            except Exception:
                pass
        return {
            "prompt": f"{description}\n\nScene context: {context_excerpt}",
            "style": style,
            "size": self.settings.default_size,
            "prompt_context": prompt_context,
        }
```

实现顺序固定如下：

1. 先实现 deterministic fallback：分类映射风格 + 默认尺寸
2. 再补 LLM 优化调用，要求只返回 JSON：`{"prompt": "...", "style": "...", "size": "..."}`
3. 再补 JSON 解析失败、LLM 不可用、字段缺失时的 fallback
4. 最后保留 `prompt_context`，用于日志、调试和任务排障

**Step 4: Run test to verify it passes**

Run: `cd /Users/timcook/Codes/hgtk/life-echo && python -m unittest api.tests.test_memoir_image_prompting -v`  
Expected: PASS

**Step 5: Commit**

```bash
git add api/services/memoir_images/__init__.py \
        api/services/memoir_images/settings.py \
        api/services/memoir_images/prompting.py \
        api/tests/test_memoir_image_prompting.py
git commit -m "feat(api): add memoir image prompt settings"
```

## Task 3: Add Liblib image provider adapter

**Files:**
- Create: `api/services/memoir_images/provider.py`
- Modify: `api/services/memoir_images/__init__.py`
- Test: `api/tests/test_memoir_image_provider.py`

**Step 1: Write the failing test**

```python
import unittest
from unittest.mock import Mock

from api.services.memoir_images.provider import LiblibImageProvider


class MemoirImageProviderTest(unittest.TestCase):
    def test_submit_generation_handles_sync_provider_response(self):
        http_client = Mock()
        http_client.post.return_value.json.return_value = {
            "status": "succeeded",
            "image_url": "https://provider.example.com/1.png",
        }
        provider = LiblibImageProvider(http_client=http_client, api_key="test-key", base_url="https://example.com")

        job = provider.submit_generation(prompt="foo", size="1024x1024", style="watercolor")

        self.assertEqual(job["status"], "completed")
        self.assertEqual(job["image_url"], "https://provider.example.com/1.png")

    def test_submit_generation_handles_async_provider_response(self):
        http_client = Mock()
        http_client.post.return_value.json.return_value = {
            "task_id": "job-123",
            "status": "queued",
        }
        provider = LiblibImageProvider(http_client=http_client, api_key="test-key", base_url="https://example.com")

        job = provider.submit_generation(prompt="foo", size="1024x1024", style="watercolor")

        self.assertEqual(job["status"], "processing")
        self.assertEqual(job["job_id"], "job-123")

    def test_poll_until_complete_returns_completed_job(self):
        http_client = Mock()
        http_client.get.return_value.json.side_effect = [
            {"status": "queued"},
            {"status": "succeeded", "image_url": "https://provider.example.com/1.png"},
        ]
        provider = LiblibImageProvider(http_client=http_client, api_key="test-key", base_url="https://example.com")

        job = provider.poll_until_complete(
            {"status": "processing", "job_id": "job-123"},
            poll_interval_seconds=0,
            max_attempts=2,
        )

        self.assertEqual(job["status"], "completed")
        self.assertEqual(job["image_url"], "https://provider.example.com/1.png")

    def test_download_image_fetches_binary_payload(self):
        http_client = Mock()
        http_client.get.return_value.content = b"png-bytes"
        provider = LiblibImageProvider(http_client=http_client, api_key="test-key", base_url="https://example.com")

        payload = provider.download_image({"image_url": "https://provider.example.com/1.png"})

        self.assertEqual(payload, b"png-bytes")
```

**Step 2: Run test to verify it fails**

Run: `cd /Users/timcook/Codes/hgtk/life-echo && python -m unittest api.tests.test_memoir_image_provider -v`  
Expected: FAIL with missing module/class errors

**Step 3: Write minimal implementation**

```python
import os
import httpx
import time


class LiblibImageProvider:
    def __init__(self, http_client=None, api_key: str | None = None, base_url: str | None = None):
        self.http_client = http_client or httpx.Client(timeout=60)
        self.api_key = api_key or os.getenv("LIBLIB_API_KEY", "")
        self.base_url = (base_url or os.getenv("LIBLIB_BASE_URL", "")).rstrip("/")

    def submit_generation(self, prompt: str, size: str, style: str) -> dict:
        response = self.http_client.post(
            f"{self.base_url}/v1/images/generations",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={"prompt": prompt, "size": size, "style": style},
        )
        data = response.json()
        if data.get("image_url"):
            return {"status": "completed", "image_url": data["image_url"], "job_id": None}
        return {"status": "processing", "job_id": data.get("task_id"), "image_url": None}

    def poll_until_complete(self, job: dict, poll_interval_seconds: int, max_attempts: int) -> dict:
        for _ in range(max_attempts):
            response = self.http_client.get(
                f"{self.base_url}/v1/images/generations/{job['job_id']}",
                headers={"Authorization": f"Bearer {self.api_key}"},
            )
            data = response.json()
            if data.get("image_url"):
                return {"status": "completed", "image_url": data["image_url"], "job_id": job["job_id"]}
            time.sleep(poll_interval_seconds)
        raise TimeoutError(f"Liblib image generation timed out for {job['job_id']}")

    def download_image(self, job: dict) -> bytes:
        response = self.http_client.get(job["image_url"])
        return response.content
```

**Step 4: Run test to verify it passes**

Run: `cd /Users/timcook/Codes/hgtk/life-echo && python -m unittest api.tests.test_memoir_image_provider -v`  
Expected: PASS

**Step 5: Commit**

```bash
git add api/services/memoir_images/__init__.py \
        api/services/memoir_images/provider.py \
        api/tests/test_memoir_image_provider.py
git commit -m "feat(api): add liblib memoir image provider"
```

## Task 4: Add Tencent COS storage service and dependency

**Files:**
- Modify: `api/requirements.txt`
- Create: `api/services/memoir_images/storage.py`
- Modify: `api/services/memoir_images/__init__.py`
- Test: `api/tests/test_memoir_image_storage.py`

**Step 1: Write the failing test**

```python
import unittest
from unittest.mock import Mock, patch

from api.services.memoir_images.storage import TencentCosStorageService


class MemoirImageStorageTest(unittest.TestCase):
    @patch("api.services.memoir_images.storage.CosS3Client")
    def test_upload_bytes_returns_persistent_cos_url(self, client_cls):
        client = Mock()
        client_cls.return_value = client
        storage = TencentCosStorageService(
            secret_id="id",
            secret_key="key",
            region="ap-shanghai",
            bucket="memoir-1250000000",
            base_url="https://memoir-1250000000.cos.ap-shanghai.myqcloud.com",
        )

        url = storage.upload_bytes(
            image_bytes=b"png-bytes",
            key="memoirs/u1/c1/0-demo.png",
            content_type="image/png",
        )

        self.assertEqual(
            url,
            "https://memoir-1250000000.cos.ap-shanghai.myqcloud.com/memoirs/u1/c1/0-demo.png",
        )
        client.put_object.assert_called_once()
```

**Step 2: Run test to verify it fails**

Run: `cd /Users/timcook/Codes/hgtk/life-echo && python -m unittest api.tests.test_memoir_image_storage -v`  
Expected: FAIL with missing module/class errors

**Step 3: Write minimal implementation**

```python
import os
from qcloud_cos import CosConfig, CosS3Client


class TencentCosStorageService:
    def __init__(self, secret_id: str, secret_key: str, region: str, bucket: str, base_url: str):
        self.bucket = bucket
        self.base_url = base_url.rstrip("/")
        config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key)
        self.client = CosS3Client(config)

    def upload_bytes(self, image_bytes: bytes, key: str, content_type: str) -> str:
        self.client.put_object(
            Bucket=self.bucket,
            Body=image_bytes,
            Key=key,
            ContentType=content_type,
        )
        return f"{self.base_url}/{key}"

    @classmethod
    def from_env(cls) -> "TencentCosStorageService":
        return cls(
            secret_id=os.getenv("TENCENT_COS_SECRET_ID", ""),
            secret_key=os.getenv("TENCENT_COS_SECRET_KEY", ""),
            region=os.getenv("TENCENT_COS_REGION", ""),
            bucket=os.getenv("TENCENT_COS_BUCKET", ""),
            base_url=os.getenv("TENCENT_COS_BASE_URL", ""),
        )
```

并在 `api/requirements.txt` 中添加：

```text
cos-python-sdk-v5>=1.9.30
```

**Step 4: Run test to verify it passes**

Run: `cd /Users/timcook/Codes/hgtk/life-echo/api && source .venv/bin/activate && pip install -r requirements.txt && cd .. && python -m unittest api.tests.test_memoir_image_storage -v`  
Expected: PASS

**Step 5: Commit**

```bash
git add api/requirements.txt \
        api/services/memoir_images/__init__.py \
        api/services/memoir_images/storage.py \
        api/tests/test_memoir_image_storage.py
git commit -m "feat(api): add tencent cos storage for memoir images"
```

## Task 5: Initialize chapter image assets during chapter creation and API serialization

**Files:**
- Modify: `api/database/models.py`
- Modify: `api/tasks/memoir_tasks.py`
- Modify: `api/tasks/__init__.py`
- Modify: `api/routers/chapters.py`
- Modify: `api/scripts/reprocess_user_memoir.py`
- Test: `api/tests/test_memoir_image_bootstrap.py`

**Step 1: Write the failing test**

```python
import unittest
from unittest.mock import patch

from api.tasks.memoir_tasks import initialize_chapter_images


class MemoirImageBootstrapTest(unittest.TestCase):
    @patch("api.tasks.memoir_tasks.generate_chapter_images.delay")
    def test_initialize_chapter_images_sets_pending_assets_and_enqueues_task(self, delay_mock):
        chapter = type(
            "ChapterStub",
            (),
            {
                "id": "chapter-1",
                "title": "童年的夏天",
                "category": "childhood",
                "content": "那条路我一直记得。\n\n{{{{IMAGE:南方小镇的青石板路}}}}",
                "images": [],
            },
        )()

        assets = initialize_chapter_images(chapter)

        self.assertEqual(len(assets), 1)
        self.assertEqual(assets[0]["status"], "pending")
        delay_mock.assert_called_once_with("chapter-1")
```

**Step 2: Run test to verify it fails**

Run: `cd /Users/timcook/Codes/hgtk/life-echo && python -m unittest api.tests.test_memoir_image_bootstrap -v`  
Expected: FAIL with missing function errors

**Step 3: Write minimal implementation**

```python
# Runtime modules in this backend use the existing import root (`services.*`, `database.*`),
# not `api.services.*`. Keep tests importing through `api.*` when running from repo root.
from services.memoir_images.parser import (
    build_initial_image_assets,
    parse_image_placeholders,
)
from services.memoir_images.prompting import MemoirImagePromptService
from services.memoir_images.settings import MemoirImageSettings


def initialize_chapter_images(chapter) -> list[dict]:
    settings = MemoirImageSettings.from_env()
    if not settings.enabled:
        chapter.images = []
        return chapter.images

    prompt_service = MemoirImagePromptService(llm=None, settings=settings)
    placeholders = parse_image_placeholders(chapter.content, settings.max_per_chapter)
    style = prompt_service.CATEGORY_STYLE_MAP.get(chapter.category, settings.default_style)
    chapter.images = build_initial_image_assets(
        placeholders=placeholders,
        provider=settings.provider,
        style=style,
        size=settings.default_size,
        now_iso=datetime.now(timezone.utc).isoformat(),
    )
    if chapter.images:
        generate_chapter_images.delay(chapter.id)
    return chapter.images
```

同时完成这些接线：

- `api/tasks/memoir_tasks.py` 在章节创建和章节更新后都调用 `initialize_chapter_images(chapter)`
- `api/scripts/reprocess_user_memoir.py` 使用同一 helper 初始化 `images`
- `api/routers/chapters.py` 继续返回 `chapter.images or []`，但现在它是对象数组
- `api/database/models.py` 将 `Chapter.images` 注释改成“图片元数据对象列表”
- `api/tasks/__init__.py` 导出 `generate_chapter_images`

**Step 4: Run test to verify it passes**

Run: `cd /Users/timcook/Codes/hgtk/life-echo && python -m unittest api.tests.test_memoir_image_bootstrap -v`  
Expected: PASS

**Step 5: Commit**

```bash
git add api/database/models.py \
        api/tasks/memoir_tasks.py \
        api/tasks/__init__.py \
        api/routers/chapters.py \
        api/scripts/reprocess_user_memoir.py \
        api/tests/test_memoir_image_bootstrap.py
git commit -m "feat(api): initialize memoir chapter image assets"
```

## Task 6: Implement the asynchronous `generate_chapter_images` Celery task

**Files:**
- Modify: `api/tasks/memoir_tasks.py`
- Modify: `api/services/memoir_images/provider.py`
- Modify: `api/services/memoir_images/prompting.py`
- Modify: `api/services/memoir_images/storage.py`
- Test: `api/tests/test_generate_chapter_images_task.py`

**Step 1: Write the failing test**

```python
import unittest
from unittest.mock import Mock, patch

from api.tasks.memoir_tasks import generate_chapter_images


class GenerateChapterImagesTaskTest(unittest.TestCase):
    @patch("api.tasks.memoir_tasks.SessionLocal")
    @patch("api.tasks.memoir_tasks.TencentCosStorageService")
    @patch("api.tasks.memoir_tasks.LiblibImageProvider")
    @patch("api.tasks.memoir_tasks.MemoirImagePromptService")
    def test_generate_chapter_images_marks_successful_item_completed(
        self,
        prompt_service_cls,
        provider_cls,
        storage_cls,
        session_local_cls,
    ):
        chapter = type(
            "ChapterStub",
            (),
            {
                "id": "chapter-1",
                "user_id": "user-1",
                "title": "童年的夏天",
                "category": "childhood",
                "content": "那条路我一直记得。\n\n{{{{IMAGE:南方小镇的青石板路}}}}",
                "images": [
                    {
                        "index": 0,
                        "placeholder": "{{{{IMAGE:南方小镇的青石板路}}}}",
                        "description": "南方小镇的青石板路",
                        "status": "pending",
                        "url": None,
                    }
                ],
            },
        )()

        db = Mock()
        db.get.return_value = chapter
        session_local_cls.return_value = db
        prompt_service_cls.return_value.build_prompt.return_value = {
            "prompt": "A serene southern China town",
            "style": "watercolor",
            "size": "1024x1024",
            "prompt_context": "childhood: 童年的夏天",
        }
        provider_cls.return_value.submit_generation.return_value = {
            "status": "completed",
            "image_url": "https://provider.example.com/1.png",
        }
        provider_cls.return_value.download_image.return_value = b"png-bytes"
        storage_cls.return_value.upload_bytes.return_value = "https://cos.example.com/memoirs/u1/c1/0.png"

        generate_chapter_images.run("chapter-1")

        self.assertEqual(chapter.images[0]["status"], "completed")
        self.assertEqual(chapter.images[0]["url"], "https://cos.example.com/memoirs/u1/c1/0.png")
        self.assertEqual(chapter.images[0]["prompt"], "A serene southern China town")
        db.commit.assert_called()

    @patch("api.tasks.memoir_tasks.SessionLocal")
    @patch("api.tasks.memoir_tasks.TencentCosStorageService")
    @patch("api.tasks.memoir_tasks.LiblibImageProvider")
    @patch("api.tasks.memoir_tasks.MemoirImagePromptService")
    def test_generate_chapter_images_skips_completed_items_for_idempotency(
        self,
        prompt_service_cls,
        provider_cls,
        storage_cls,
        session_local_cls,
    ):
        chapter = type(
            "ChapterStub",
            (),
            {
                "id": "chapter-1",
                "user_id": "user-1",
                "title": "童年的夏天",
                "category": "childhood",
                "content": "那条路我一直记得。",
                "images": [
                    {
                        "index": 0,
                        "placeholder": "{{{{IMAGE:南方小镇的青石板路}}}}",
                        "description": "南方小镇的青石板路",
                        "status": "completed",
                        "url": "https://cos.example.com/already-there.png",
                    }
                ],
            },
        )()

        db = Mock()
        db.get.return_value = chapter
        session_local_cls.return_value = db

        generate_chapter_images.run("chapter-1")

        provider_cls.return_value.submit_generation.assert_not_called()
        storage_cls.return_value.upload_bytes.assert_not_called()
```

**Step 2: Run test to verify it fails**

Run: `cd /Users/timcook/Codes/hgtk/life-echo && python -m unittest api.tests.test_generate_chapter_images_task -v`  
Expected: FAIL with missing task or missing state transition logic

**Step 3: Write minimal implementation**

```python
@shared_task(bind=True, max_retries=3, default_retry_delay=30)
def generate_chapter_images(self, chapter_id: str):
    db = SessionLocal()
    try:
        chapter = db.get(Chapter, chapter_id)
        if not chapter or not chapter.images:
            return {"status": "no_images"}

        prompt_service = MemoirImagePromptService(llm_service.get_llm(), MemoirImageSettings.from_env())
        provider = LiblibImageProvider()
        storage = TencentCosStorageService.from_env()

        for item in chapter.images:
            if item.get("status") == "completed" and item.get("url"):
                continue
            if item.get("status") not in {"pending", "failed"}:
                continue

            item["status"] = "processing"
            db.commit()

            try:
                prompt_data = prompt_service.build_prompt(...)
                job = provider.submit_generation(...)
                if job["status"] != "completed":
                    job = provider.poll_until_complete(...)
                image_bytes = provider.download_image(job)
                key = build_cos_key(chapter.user_id, chapter.id, item["index"], prompt_data["prompt"])
                item["url"] = storage.upload_bytes(image_bytes, key, "image/png")
                item["prompt"] = prompt_data["prompt"]
                item["style"] = prompt_data["style"]
                item["size"] = prompt_data["size"]
                item["status"] = "completed"
                item["error"] = None
            except Exception as exc:
                item["status"] = "failed"
                item["error"] = str(exc)
            item["updated_at"] = datetime.now(timezone.utc).isoformat()

        db.commit()
        return {"status": "success"}
    finally:
        db.close()
```

同时补上这个 helper，不要把 key 生成逻辑散落在任务里：

```python
import hashlib


def build_cos_key(user_id: str, chapter_id: str, index: int, prompt: str) -> str:
    short_hash = hashlib.sha1(prompt.encode("utf-8")).hexdigest()[:10]
    return f"memoirs/{user_id}/{chapter_id}/{index}-{short_hash}.png"
```

约束：

- key 路径固定为 `memoirs/{user_id}/{chapter_id}/{index}-{short_hash}.png`
- hash 只基于最终 `prompt`，保证同 prompt 重试时 key 稳定
- 不要直接把原始 prompt 文本拼进文件名

**Step 4: Run test to verify it passes**

Run: `cd /Users/timcook/Codes/hgtk/life-echo && python -m unittest api.tests.test_generate_chapter_images_task -v`  
Expected: PASS

**Step 5: Commit**

```bash
git add api/tasks/memoir_tasks.py \
        api/services/memoir_images/provider.py \
        api/services/memoir_images/prompting.py \
        api/services/memoir_images/storage.py \
        api/tests/test_generate_chapter_images_task.py
git commit -m "feat(api): generate memoir chapter images asynchronously"
```

## Task 7: Upgrade PDF export to embed completed images and strip placeholders

**Files:**
- Modify: `api/services/pdf_service.py`
- Test: `api/tests/test_pdf_service_images.py`

**Step 1: Write the failing test**

```python
import unittest
from unittest.mock import AsyncMock, patch

from api.services.pdf_service import PDFService


class PDFServiceImagesTest(unittest.IsolatedAsyncioTestCase):
    @patch("api.services.pdf_service.httpx.AsyncClient")
    async def test_generate_pdf_embeds_completed_images_and_removes_placeholders(self, async_client_cls):
        png_bytes = (
            b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
            b"\x08\x02\x00\x00\x00\x90wS\xde\x00\x00\x00\x0cIDATx\x9cc```\x00\x00"
            b"\x00\x04\x00\x01\xf6\x178U\x00\x00\x00\x00IEND\xaeB`\x82"
        )
        client = AsyncMock()
        client.__aenter__.return_value.get.return_value.content = png_bytes
        async_client_cls.return_value = client

        service = PDFService()
        book = type("BookStub", (), {"title": "我的回忆录"})()
        chapter = type(
            "ChapterStub",
            (),
            {
                "title": "童年的夏天",
                "content": "那条路我一直记得。\n\n{{{{IMAGE:南方小镇的青石板路}}}}\n\n奶奶常坐在那里。",
                "images": [
                    {
                        "index": 0,
                        "placeholder": "{{{{IMAGE:南方小镇的青石板路}}}}",
                        "url": "https://cos.example.com/0.png",
                        "status": "completed",
                    }
                ],
            },
        )()

        pdf_bytes = await service.generate_pdf(book, [chapter])

        self.assertGreater(len(pdf_bytes), 100)
        self.assertNotIn(b"IMAGE:", pdf_bytes)
```

**Step 2: Run test to verify it fails**

Run: `cd /Users/timcook/Codes/hgtk/life-echo && python -m unittest api.tests.test_pdf_service_images -v`  
Expected: FAIL because placeholders are still emitted as raw text and image embedding does not exist

**Step 3: Write minimal implementation**

```python
from reportlab.platypus import Image as ReportLabImage
from io import BytesIO
import httpx
import re

PLACEHOLDER_RE = re.compile(r"\{\{\{\{IMAGE:.*?\}\}\}\}|\{\{IMAGE:.*?\}\}", re.DOTALL)


def strip_image_placeholders(text: str) -> str:
    return PLACEHOLDER_RE.sub("", text or "").strip()


def split_content_blocks(content: str, images: list[dict]) -> list[dict]:
    blocks = []
    remaining = content
    for image in sorted(images or [], key=lambda item: item.get("index", 0)):
        placeholder = image.get("placeholder")
        if not placeholder or placeholder not in remaining:
            continue
        before, remaining = remaining.split(placeholder, 1)
        cleaned_before = strip_image_placeholders(before)
        if cleaned_before:
            blocks.append({"type": "text", "value": cleaned_before})
        if image.get("status") == "completed" and image.get("url"):
            blocks.append({"type": "image", "url": image["url"]})
    cleaned_remaining = strip_image_placeholders(remaining)
    if cleaned_remaining:
        blocks.append({"type": "text", "value": cleaned_remaining})
    return blocks


async def _fetch_image_bytes(self, url: str) -> bytes:
    async with httpx.AsyncClient(timeout=30) as client:
        response = await client.get(url)
        response.raise_for_status()
        return response.content
```

并在 `generate_pdf()` 中：

- 使用 `split_content_blocks()` 构建 story
- 文本块仍走 `Paragraph`
- 图片块下载后用 `ReportLabImage(BytesIO(image_bytes), width=5 * inch, height=3.75 * inch)`
- 对未完成图片和失败图片直接跳过，但要继续调用 `strip_image_placeholders()` 清理任何残留占位符文本

**Step 4: Run test to verify it passes**

Run: `cd /Users/timcook/Codes/hgtk/life-echo && python -m unittest api.tests.test_pdf_service_images -v`  
Expected: PASS

**Step 5: Commit**

```bash
git add api/services/pdf_service.py \
        api/tests/test_pdf_service_images.py
git commit -m "feat(api): embed memoir chapter images in pdf export"
```

## Task 8: Migrate Android models to structured chapter images and parse content blocks

**Files:**
- Modify: `app-android/app/src/main/java/com/huaga/life_echo/network/models/MemoirModels.kt`
- Modify: `app-android/app/src/main/java/com/huaga/life_echo/data/mock/MockDataProvider.kt`
- Modify: `app-android/app/src/main/java/com/huaga/life_echo/ui/screens/MyMemoirScreen.kt`
- Modify: `app-android/app/src/main/java/com/huaga/life_echo/ui/components/memoir/ChapterCard.kt`
- Create: `app-android/app/src/main/java/com/huaga/life_echo/ui/components/memoir/MemoirContentBlocks.kt`
- Test: `app-android/app/src/test/java/com/huaga/life_echo/ui/components/memoir/MemoirContentBlocksTest.kt`

**Step 1: Write the failing test**

```kotlin
package com.huaga.life_echo.ui.components.memoir

import com.huaga.life_echo.network.models.ChapterImageDto
import org.junit.Assert.assertEquals
import org.junit.Assert.assertTrue
import org.junit.Test

class MemoirContentBlocksTest {

    @Test
    fun splitMemoirContent_insertsImageBlockForCompletedImage_andDropsFailedPlaceholder() {
        val blocks = splitMemoirContent(
            content = "那条路我一直记得。\n\n{{{{IMAGE:南方小镇的青石板路}}}}\n\n后来就下雨了。",
            images = listOf(
                ChapterImageDto(
                    index = 0,
                    placeholder = "{{{{IMAGE:南方小镇的青石板路}}}}",
                    description = "南方小镇的青石板路",
                    prompt = null,
                    url = "https://cos.example.com/0.png",
                    status = "completed",
                    provider = "liblib",
                    style = "watercolor",
                    size = "1024x1024",
                    error = null,
                    created_at = null,
                    updated_at = null,
                )
            )
        )

        assertEquals(MemoirContentBlock.Text::class, blocks[0]::class)
        assertEquals(MemoirContentBlock.Image::class, blocks[1]::class)
        assertTrue((blocks[1] as MemoirContentBlock.Image).image.url!!.contains("cos.example.com"))
    }
}
```

**Step 2: Run test to verify it fails**

Run: `cd /Users/timcook/Codes/hgtk/life-echo/app-android && ./gradlew :app:testDebugUnitTest --tests "com.huaga.life_echo.ui.components.memoir.MemoirContentBlocksTest" -i`  
Expected: FAIL because `ChapterImageDto` and `splitMemoirContent()` do not exist

**Step 3: Write minimal implementation**

```kotlin
@Serializable
data class ChapterImageDto(
    val index: Int,
    val placeholder: String,
    val description: String,
    val prompt: String? = null,
    val url: String? = null,
    val status: String,
    val provider: String? = null,
    val style: String? = null,
    val size: String? = null,
    val error: String? = null,
    val created_at: String? = null,
    val updated_at: String? = null,
)
```

```kotlin
sealed interface MemoirContentBlock {
    data class Text(val content: String) : MemoirContentBlock
    data class Image(val image: ChapterImageDto) : MemoirContentBlock
}

private val imagePlaceholderRegex =
    Regex("""\{\{\{\{IMAGE:.*?\}\}\}\}|\{\{IMAGE:.*?\}\}""", setOf(RegexOption.DOT_MATCHES_ALL))

private fun stripImagePlaceholders(text: String): String =
    text.replace(imagePlaceholderRegex, "").trim()

fun splitMemoirContent(content: String, images: List<ChapterImageDto>): List<MemoirContentBlock> {
    var remaining = content
    val blocks = mutableListOf<MemoirContentBlock>()
    images.sortedBy { it.index }.forEach { image ->
        val placeholder = image.placeholder
        if (!remaining.contains(placeholder)) return@forEach
        val parts = remaining.split(placeholder, limit = 2)
        val before = stripImagePlaceholders(parts.first())
        if (before.isNotBlank()) blocks += MemoirContentBlock.Text(before)
        if (image.status == "completed" && !image.url.isNullOrBlank()) blocks += MemoirContentBlock.Image(image)
        remaining = parts.getOrElse(1) { "" }
    }
    val trailingText = stripImagePlaceholders(remaining)
    if (trailingText.isNotBlank()) blocks += MemoirContentBlock.Text(trailingText)
    return blocks
}
```

同时完成这些模型迁移：

- `ChapterDto.images` 改为 `List<ChapterImageDto>`
- `ChapterContentDto.images` 改为 `List<ChapterImageDto>`
- `MockDataProvider.kt` 中的假数据改成对象数组
- `ChapterCard.kt` 不再用 `images.isNotEmpty()` 判断可渲染图片，改成 `images.any { it.status == "completed" && !it.url.isNullOrBlank() }`
- `splitMemoirContent()` 在尾部文本落块前必须清理剩余占位符，避免超过 `max_per_chapter` 时裸占位符出现在阅读页

**Step 4: Run test to verify it passes**

Run: `cd /Users/timcook/Codes/hgtk/life-echo/app-android && ./gradlew :app:testDebugUnitTest --tests "com.huaga.life_echo.ui.components.memoir.MemoirContentBlocksTest" -i`  
Expected: PASS

**Step 5: Commit**

```bash
cd app-android
git add app/src/main/java/com/huaga/life_echo/network/models/MemoirModels.kt \
        app/src/main/java/com/huaga/life_echo/data/mock/MockDataProvider.kt \
        app/src/main/java/com/huaga/life_echo/ui/screens/MyMemoirScreen.kt \
        app/src/main/java/com/huaga/life_echo/ui/components/memoir/ChapterCard.kt \
        app/src/main/java/com/huaga/life_echo/ui/components/memoir/MemoirContentBlocks.kt \
        app/src/test/java/com/huaga/life_echo/ui/components/memoir/MemoirContentBlocksTest.kt
git commit -m "feat(android): add memoir chapter image models and parsing"
```

## Task 9: Render chapter images in Android reading views with loading and failure states

**Files:**
- Create: `app-android/app/src/main/java/com/huaga/life_echo/ui/components/memoir/MemoirInlineImage.kt`
- Create: `app-android/app/src/main/java/com/huaga/life_echo/ui/components/memoir/MemoirImageViewerDialog.kt`
- Modify: `app-android/app/src/main/java/com/huaga/life_echo/ui/components/memoir/ChapterReadingView.kt`
- Modify: `app-android/app/src/main/java/com/huaga/life_echo/ui/components/memoir/FullTextReadingView.kt`
- Modify: `app-android/app/src/main/java/com/huaga/life_echo/utils/TextUtils.kt`
- Test: `app-android/app/src/androidTest/java/com/huaga/life_echo/ui/components/memoir/ChapterReadingImageBlocksTest.kt`

**Step 1: Write the failing test**

```kotlin
package com.huaga.life_echo.ui.components.memoir

import androidx.activity.ComponentActivity
import androidx.compose.ui.test.assertExists
import androidx.compose.ui.test.junit4.createAndroidComposeRule
import androidx.compose.ui.test.onNodeWithTag
import com.huaga.life_echo.network.models.ChapterContentDto
import com.huaga.life_echo.network.models.ChapterImageDto
import org.junit.Rule
import org.junit.Test

class ChapterReadingImageBlocksTest {

    @get:Rule
    val composeRule = createAndroidComposeRule<ComponentActivity>()

    @Test
    fun chapterReadingView_showsLoadingCard_forProcessingImage() {
        val chapter = ChapterContentDto(
            id = "chapter-1",
            title = "童年的夏天",
            content = "那条路我一直记得。\n\n{{{{IMAGE:南方小镇的青石板路}}}}",
            orderIndex = 0,
            status = "completed",
            category = "childhood",
            pageCount = null,
            updatedAt = 0L,
            quotes = emptyList(),
            images = listOf(
                ChapterImageDto(
                    index = 0,
                    placeholder = "{{{{IMAGE:南方小镇的青石板路}}}}",
                    description = "南方小镇的青石板路",
                    prompt = null,
                    url = null,
                    status = "processing",
                    provider = "liblib",
                    style = "watercolor",
                    size = "1024x1024",
                    error = null,
                    created_at = null,
                    updated_at = null,
                )
            ),
        )

        composeRule.setContent { ChapterReadingView(chapter = chapter) }

        composeRule.onNodeWithTag("memoir-image-loading-0").assertExists()
    }

    @Test
    fun chapterReadingView_showsFailureCard_forFailedImage_withoutRawPlaceholderText() {
        val chapter = ChapterContentDto(
            id = "chapter-1",
            title = "童年的夏天",
            content = "那条路我一直记得。\n\n{{{{IMAGE:南方小镇的青石板路}}}}",
            orderIndex = 0,
            status = "completed",
            category = "childhood",
            pageCount = null,
            updatedAt = 0L,
            quotes = emptyList(),
            images = listOf(
                ChapterImageDto(
                    index = 0,
                    placeholder = "{{{{IMAGE:南方小镇的青石板路}}}}",
                    description = "南方小镇的青石板路",
                    prompt = null,
                    url = null,
                    status = "failed",
                    provider = "liblib",
                    style = "watercolor",
                    size = "1024x1024",
                    error = "provider timeout",
                    created_at = null,
                    updated_at = null,
                )
            ),
        )

        composeRule.setContent { ChapterReadingView(chapter = chapter) }

        composeRule.onNodeWithTag("memoir-image-error-0").assertExists()
    }
}
```

**Step 2: Run test to verify it fails**

Run: `cd /Users/timcook/Codes/hgtk/life-echo/app-android && ./gradlew :app:connectedDebugAndroidTest -Pandroid.testInstrumentationRunnerArguments.class=com.huaga.life_echo.ui.components.memoir.ChapterReadingImageBlocksTest`  
Expected: FAIL because there is no image composable, no test tags, and placeholders are still treated as plain text

**Step 3: Write minimal implementation**

```kotlin
@Composable
fun MemoirInlineImage(
    image: ChapterImageDto,
    onClick: () -> Unit,
    modifier: Modifier = Modifier,
) {
    when (image.status) {
        "completed" -> AsyncImage(
            model = image.url,
            contentDescription = image.description,
            modifier = modifier
                .fillMaxWidth()
                .clip(RoundedCornerShape(16.dp))
                .clickable(onClick = onClick)
                .testTag("memoir-image-${image.index}")
        )
        "pending", "processing" -> Box(
            modifier = modifier
                .fillMaxWidth()
                .height(220.dp)
                .clip(RoundedCornerShape(16.dp))
                .background(LightPurple.copy(alpha = 0.15f))
                .testTag("memoir-image-loading-${image.index}")
        )
        "failed" -> Column(
            modifier = modifier
                .fillMaxWidth()
                .clip(RoundedCornerShape(16.dp))
                .background(LightPurple.copy(alpha = 0.10f))
                .padding(16.dp)
                .testTag("memoir-image-error-${image.index}")
        ) {
            Text(text = "图片生成失败")
            Text(text = image.description)
        }
        else -> Unit
    }
}
```

并完成这些 UI 接线：

- `ChapterReadingView.kt` 使用 `splitMemoirContent()` 顺序渲染文本块和图片块
- `FullTextReadingView.kt` 复用相同的 block renderer
- `MemoirImageViewerDialog.kt` 提供点击放大预览
- `TextUtils.removeImagePlaceholders()` 改为仅服务卡片摘要等兜底文本，不再承担阅读页主渲染逻辑
- 失败态必须显示非阻塞错误卡片，不显示原始占位符，也不让正文断裂

**Step 4: Run test to verify it passes**

Run: `cd /Users/timcook/Codes/hgtk/life-echo/app-android && ./gradlew :app:connectedDebugAndroidTest -Pandroid.testInstrumentationRunnerArguments.class=com.huaga.life_echo.ui.components.memoir.ChapterReadingImageBlocksTest`  
Expected: PASS

**Step 5: Commit**

```bash
cd app-android
git add app/src/main/java/com/huaga/life_echo/ui/components/memoir/MemoirInlineImage.kt \
        app/src/main/java/com/huaga/life_echo/ui/components/memoir/MemoirImageViewerDialog.kt \
        app/src/main/java/com/huaga/life_echo/ui/components/memoir/ChapterReadingView.kt \
        app/src/main/java/com/huaga/life_echo/ui/components/memoir/FullTextReadingView.kt \
        app/src/main/java/com/huaga/life_echo/utils/TextUtils.kt \
        app/src/androidTest/java/com/huaga/life_echo/ui/components/memoir/ChapterReadingImageBlocksTest.kt
git commit -m "feat(android): render memoir images in reading views"
```

## Final Verification Checklist

Run these only after all tasks above are complete.

### Backend targeted tests

```bash
cd /Users/timcook/Codes/hgtk/life-echo
python -m unittest \
  api.tests.test_memoir_image_parser \
  api.tests.test_memoir_image_prompting \
  api.tests.test_memoir_image_provider \
  api.tests.test_memoir_image_storage \
  api.tests.test_memoir_image_bootstrap \
  api.tests.test_generate_chapter_images_task \
  api.tests.test_pdf_service_images \
  -v
```

Expected: all PASS

### Android targeted tests

```bash
cd /Users/timcook/Codes/hgtk/life-echo/app-android
./gradlew :app:testDebugUnitTest --tests "com.huaga.life_echo.ui.components.memoir.*" -i
./gradlew :app:connectedDebugAndroidTest -Pandroid.testInstrumentationRunnerArguments.class=com.huaga.life_echo.ui.components.memoir.ChapterReadingImageBlocksTest
```

Expected: all PASS

### Manual smoke test

1. 生成一个包含至少两个 `{{{{IMAGE:...}}}}` 占位符的章节
2. 验证章节文本先出现，`Chapter.images` 先写入 `pending` 项
3. 等待子任务完成，刷新 Android 目录页和阅读页，确认图片出现在正确位置
4. 人为让 provider 返回失败，确认 Android 阅读页显示失败卡片，正文仍可读且用户看不到原始占位符
5. 导出 PDF，确认图片嵌入成功，PDF 文本中没有 `IMAGE:` 字样

## Documentation Follow-up

在代码完成并验证通过后，再更新这两个文档：

- `api/README.md`
- `api/docs/本地开发环境配置.md`

需要补充的内容：

- 新增环境变量：`MEMOIR_IMAGE_*`、`LIBLIB_*`、`TENCENT_COS_*`
- 图片任务的异步行为说明
- 本地调试 COS 和 provider 的最小配置

## Deferred Work

本计划不实现以下内容，除非现有任务全部完成并且用户明确要求继续：

- `Book.cover_image_url` 自动生成
- 多 provider 切换
- iOS 客户端适配
- PDF 导出接口的传输格式重构