diff --git a/backend/docs/product-archive-runbook.md b/backend/docs/product-archive-runbook.md
new file mode 100644
index 0000000..8e4a13c
--- /dev/null
+++ b/backend/docs/product-archive-runbook.md
@@ -0,0 +1,140 @@
+# 产品档案与清理运行手册
+
+## 1. 服务启动
+
+产品档案要求三类持久卷全部存在：
+
+```bash
+docker compose up -d --build
+docker compose ps
+docker compose logs --tail=80 backend
+```
+
+`docker-compose.yml` 中必须挂载：
+
+- `wordcloud_products:/app/service_products`
+- `wordcloud_metadata:/app/service_metadata`
+- `wordcloud_orders:/app/service_orders`
+
+首次上线或更新后，先检查健康接口：
+
+```bash
+docker compose exec backend curl -fsS http://localhost:8000/api/health
+```
+
+## 2. 外部产品同步
+
+首期外部系统可以先使用创建/更新产品接口做幂等同步：
+
+```http
+POST /api/products
+Authorization: Bearer <生产订单管理口令>
+Content-Type: application/json
+
+{
+  "source": "external",
+  "external_product_id": "PROD-123",
+  "name": "产品名称",
+  "sku": "SKU-123",
+  "specification": "规格"
+}
+```
+
+同步以 `(source, external_product_id)` 合并，不按名称合并。后续产品接口字段稳定后，可以在现有模型上补适配层，不必改档案结构。
+
+产品列表和详情默认展示产品名称、SKU、规格、状态与归档数量；完整 `product_id` 只在产品详情页的“系统信息”折叠区暴露。
+
+## 3. 清理 dry-run
+
+物理清理默认关闭。上线后先只看候选，不删除数据：
+
+```bash
+docker compose exec backend python -m service.storage_metrics --max-age-days 30
+```
+
+输出 JSON 中重点检查：
+
+- `archive_protected_job_ids`: 已归档进产品档案的源任务，不会被临时清理。
+- `asset_referenced_job_ids`: 普通画布素材引用，只反映当前引用，不是永久保护。
+- `temporary_jobs`: 成功生成、30 天以上、未被产品档案保护的任务。
+- `pending_product_ids`: 已进入 30 天待清理期的产品。
+- `failed_cleanup` 产品不会出现在 `pending_product_ids`，需要人工检查后恢复或重新排队。
+- `reclaimable_bytes`: 预计可回收空间。
+
+服务启动和每 24 小时也会运行一次 `CleanupService.preview()`。没有设置清理开关时，它只计算候选，不删除。
+
+## 4. 启用物理清理
+
+确认 dry-run 报告没有误删对象后，再在部署环境中显式开启：
+
+```bash
+CLEANUP_APPLY_ENABLED=true docker compose up -d --build
+```
+
+或在既有运维体系中通过 secret/env manager 提供同名变量。开启后仍必须传入 `--apply` 才会真正删除：
+
+```bash
+docker compose exec backend python -m service.storage_metrics --max-age-days 30 --apply
+```
+
+同一 API 也可以触发，但必须同时满足：
+
+1. `CLEANUP_APPLY_ENABLED=true`
+2. `POST /api/maintenance/cleanup-run` body 为 `{"confirm": true}`
+3. 使用生产订单管理口令鉴权
+
+清理规则：
+
+- 只清理成功生成、超过 30 天、带位置库、且当前仍不在 `product_wordcloud_archives` 中的任务。
+- 每次执行前会重新检查归档引用；在预览和删除之间新归档的任务会被跳过。
+- 归档保护来自产品词云档案表，不来自普通素材引用。
+- 产品物理清理只在 `purge_after` 到期后发生；到期前可以恢复。
+
+## 5. 恢复待清理产品
+
+产品进入待清理状态后，30 天内可恢复：
+
+```http
+POST /api/products/{product_id}/restore
+Authorization: Bearer <生产订单管理口令>
+```
+
+前端产品档案页会显示：
+
+```text
+待清理 · 将于 YYYY/MM/DD 删除
+```
+
+点击“恢复产品”后，产品回到 `active`，原清理记录标记为 `restored`。
+
+删除按钮只是软删除，进入下一个 30 天可恢复窗口；真正的物理删除由清理服务执行。
+
+## 6. `failed_cleanup` 处置
+
+物理删除失败时，服务不会假装文件仍完整，也不会自动重试。产品进入：
+
+```text
+failed_cleanup
+```
+
+处置步骤：
+
+1. 打开产品档案详情，确认产品状态。
+2. 检查服务器上的产品目录：
+
+   ```bash
+   docker compose exec backend sh -lc 'find /app/service_products/<product_id> -maxdepth 2 -type f -printf "%p %s\n"'
+   ```
+
+3. 如果目录完整且可继续使用，调用恢复接口或点击“恢复产品”。
+4. 如果目录不完整，先人工恢复或确认业务上放弃该产品，再恢复后重新软删除。
+5. 不要直接修改 `product_archive.db`，也不要直接把失败状态改成 `pending_cleanup`。
+
+已放弃且确认可安全删除的失败记录，应由管理员按上述恢复/重排队流程处理；系统不会自动把它当作完整产品重试。
+
+## 7. 回滚与部署注意
+
+- 新版本上线前保留现有 `service_products`、`service_metadata`、`service_orders` 卷。
+- 若当前环境曾把产品档案写入旧的未挂载目录，先停写并复制目录到挂载卷，再启动新版本。
+- `CLEANUP_APPLY_ENABLED` 回到 `false` 后，调度器恢复为只 dry-run。
+- 回滚不会自动恢复已物理删除的数据；因此首次启用物理清理必须先审查 dry-run。
diff --git a/backend/tests/test_product_archive.py b/backend/tests/test_product_archive.py
index 959c89f..7045c05 100644
--- a/backend/tests/test_product_archive.py
+++ b/backend/tests/test_product_archive.py
@@ -6,6 +6,7 @@ import shutil
 from io import BytesIO
 from pathlib import Path
 from types import SimpleNamespace
+from datetime import datetime, timedelta, timezone
 
 import pytest
 from fastapi.testclient import TestClient
@@ -23,6 +24,10 @@ from service.product_archive import (  # noqa: E402
     validate_word_locations_db,
 )
 from service.product_archive_store import ProductArchiveStore  # noqa: E402
+from service.cleanup_service import CleanupService  # noqa: E402
+from service.metadata_store import MetadataStore  # noqa: E402
+from service.storage import Storage  # noqa: E402
+from service.schemas import JobStatus  # noqa: E402
 from service import app as service_app  # noqa: E402
 
 
@@ -464,6 +469,88 @@ def test_delete_rejects_failed_cleanup_product_until_restore(product_archive_cli
     assert soft_deleted.json()["status"] == "pending_cleanup"
 
 
+def test_only_visible_inserted_wordcloud_is_retained_after_30_day_cleanup(
+    product_archive_client, prepared_wordcloud_job, tmp_path, monkeypatch
+):
+    archived_at = datetime.now(timezone.utc)
+    created_at = archived_at - timedelta(days=31)
+    assets_dir = tmp_path / "e2e-assets"
+    assets_dir.mkdir(parents=True)
+    monkeypatch.setattr(service_app, "ASSETS_DIR", assets_dir)
+
+    source_specs = {
+        "job-visible": "visible",
+        "job-hidden": "hidden",
+        "job-detached": None,
+    }
+    source_assets: dict[str, dict[str, str]] = {}
+    source_paths: dict[str, Path] = {}
+    metadata_store = MetadataStore(tmp_path / "e2e-metadata" / "app.db")
+    storage = Storage(tmp_path / "e2e-workspace")
+
+    for job_id, placement in source_specs.items():
+        asset_id = f"asset_{job_id.replace('-', '_')}"
+        asset_dir = assets_dir / asset_id[:2] / asset_id
+        asset_dir.mkdir(parents=True)
+        (asset_dir / "meta.json").write_text(
+            json.dumps({"asset_id": asset_id, "type": "wordcloud", "job_id": job_id}),
+            encoding="utf-8",
+        )
+        source_assets[job_id] = asset_id
+        source_paths[job_id] = storage.job_root(job_id) / "output" / "word_locations.sqlite"
+        source_paths[job_id].parent.mkdir(parents=True, exist_ok=True)
+        with sqlite3.connect(source_paths[job_id]) as connection:
+            connection.execute("CREATE TABLE word_locations (id INTEGER PRIMARY KEY, name TEXT)")
+            connection.execute("INSERT INTO word_locations (name) VALUES (?)", (job_id,))
+        metadata_store.upsert_job(
+            JobStatus(
+                job_id=job_id,
+                status="success",
+                stage="done",
+                progress_percent=100,
+                message="done",
+                artifacts={"db": str(source_paths[job_id])},
+                created_at=created_at,
+                updated_at=created_at,
+            )
+        )
+
+    def resolve_job(job_id: str):
+        return SimpleNamespace(
+            status="success",
+            artifacts={"db": str(source_paths[job_id])},
+        )
+
+    monkeypatch.setattr(service_app, "_resolve_job_status", resolve_job)
+    product = create_product(product_archive_client)
+    response = archive_product_version(product_archive_client, product["product_id"], {
+        "layers": [
+            {"id": "shown", "visible": True},
+            {"id": "hidden", "visible": False},
+        ],
+        "elements": [
+            {"type": "sticker", "assetId": source_assets["job-visible"], "layerId": "shown"},
+            {"type": "sticker", "assetId": source_assets["job-hidden"], "layerId": "hidden"},
+        ],
+    })
+
+    assert response.status_code == 201
+    archived = response.json()["wordcloud_archives"]
+    assert [item["source_job_id"] for item in archived] == ["job-visible"]
+    visible_snapshot = Path(archived[0]["db_path"])
+    assert visible_snapshot.exists()
+
+    cleanup = CleanupService(storage, metadata_store, product_archive_client.product_store)
+    report = cleanup.apply(now=archived_at + timedelta(days=31))
+
+    assert "job-visible" not in report.deleted_job_ids
+    assert {"job-hidden", "job-detached"} <= set(report.deleted_job_ids)
+    assert not storage.job_root("job-hidden").exists()
+    assert not storage.job_root("job-detached").exists()
+    assert storage.job_root("job-visible").exists()
+    assert visible_snapshot.exists()
+
+
 def test_archive_failure_during_final_move_removes_staging_files(product_archive_client, prepared_wordcloud_job, monkeypatch):
     product = create_product(product_archive_client)
     service = service_app._product_archive_service()