diff --git a/README.md b/README.md
index 5672e5b..f930383 100644
--- a/README.md
+++ b/README.md
@@ -1,34 +1,55 @@
-# world_snapshot 模块
+# world_snapshot 世界状态快照模块（W-04 快照管理）
 
-世界状态快照模块（W-03）：聚合世界（world）、场景（scene）、实体（entity）当前状态生成
-快照，支持快照查询与恢复，恢复时联动 world 表状态（经 world 模块 set_world_mode 契约）。
+## 用途
+
+聚合 world/scene/entity 当前状态生成世界快照，支持分页查询、详情查看（checksum 一致性校验）、
+更新、删除与恢复（事务写回 scene/entity，失败整体回滚无脏数据）。
 
 ## 功能
-- 世界快照表 `world_snapshot` 的 CRUD（create/get/list/update/delete）
-- `create_snapshot`：聚合 world/scene/entity 当前状态写入 `snapshot_json`
-- `restore_snapshot`：读取并校验快照 JSON，标记已恢复，联动 world 状态
-- 下拉接口：世界列表、快照类型、快照状态
+
+- 快照生成：同一事务批量聚合 world/scene/entity + sha256 checksum + version 按世界递增
+- 快照查询：分页 `{list,total}`；单条含 checksum 数据一致性校验（损坏检测）
+- 快照恢复：checksum 校验 → 先删后插写回 scene/entity → 世界状态联动；失败事务回滚
+- 编码字典 `snapshot_status` / `snapshot_type` 经 init/data.json 幂等落库
+- REST 统一前缀 `/api/*`，错误结构 `{code,message,field,detail}`，非法输入 100% 拦截不落库
 
 ## 数据表
-- `world_snapshot`（models/world_snapshot.json 四段式）
-- 编码字典：`snapshot_type`（full/incremental）、`snapshot_status`（active/restored/invalid）
 
-## 目录结构
+`world_snapshot`：id(32 PK), world_id(32), name(255), snapshot_type(32), snapshot_json(text),
+checksum(64, sha256), version(int), status(32), snapshot_date(timestamp), created_at/updated_at(timestamp)。
+唯一索引 `(world_id, version)`。
+
+## 安装
+
+```bash
+pip install .            # 模块打包
+```
+
+宿主应用集成：
+
+```python
+from world_snapshot.init import load_world_snapshot
+
+def init():
+    load_world_snapshot()
+```
+
+RBAC：`python3 scripts/load_path.py`（为 /world_snapshot/* 各 path 显式注册 logined 权限）。
+
+## REST 端点
+
+见 `skill/SKILL.md`（/world_snapshot/api/snapshot_*.dspy 系列）。
+
+## 目录
+
 ```
 world_snapshot/
-├── world_snapshot/        # Python 包（init.py 实现 + __init__.py 导出）
-├── wwwroot/               # index.ui / menu.ui / api/*.dspy
+├── world_snapshot/        # Python 包（init.py 核心逻辑 + __init__.py）
 ├── models/                # world_snapshot.json 表定义
 ├── json/                  # world_snapshot.json CRUD 定义
-├── init/                  # data.json 编码字典种子
-├── scripts/               # load_path.py RBAC 注册
+├── init/data.json         # 编码字典种子（snapshot_status/snapshot_type）
+├── wwwroot/               # index.ui + snapshot_generate/restore.ui + api/*.dspy
+├── scripts/load_path.py   # RBAC 注册
 ├── skill/SKILL.md         # 模块技能文档
-├── pyproject.toml
-├── build.sh               # 四步安装（xls2ddl → ddl → crud ui → symlink）
-└── README.md
+└── pyproject.toml
 ```
-
-## 集成
-- 通过 `load_world_snapshot()` 挂载到宿主应用（`from world_snapshot.init import load_world_snapshot`）
-- 库名经 `ServerEnv().get_module_dbname('world_snapshot')` 获取，禁止硬编码
-- 构建：`bash build.sh`（安装 xls2ddl、生成 DDL、生成 CRUD UI、链接 wwwroot）
diff --git a/docs/work-log-2026-07-15.md b/docs/work-log-2026-07-15.md
new file mode 100644
index 0000000..5081f19
--- /dev/null
+++ b/docs/work-log-2026-07-15.md
@@ -0,0 +1,48 @@
+# world_snapshot 模块开发（W-04 快照管理）— 工作日志
+
+日期：2026-07-15（按当前系统日期）
+模块仓库：`modules/world_snapshot/`（git 已 init）
+任务：W-04 快照管理功能点开发（new_dev，迭代「元景项目-初始迭代」）
+
+## 范围与背景
+
+W-03 已有基础版 world_snapshot（简单 CRUD + 恢复），本次按 W-04 需求升级为完整快照管理：
+
+1. **world_snapshot 表**：新增 `checksum`(sha256) 与 `version`(int) 字段；`(world_id, version)` 唯一索引
+2. **快照生成**：同一事务批量聚合 world/scene/entity + checksum 数据一致性校验 + version 按世界递增（max+1，行锁 `for update` 防并发重复）
+3. **快照恢复**：同一事务 checksum 校验 → 先删后插写回 scene/entity → 世界 mode 联动 → 置 restored；任一步失败整体回滚无脏数据
+4. **编码字典**：snapshot_status / snapshot_type 经 `init/data.json` 幂等落库（appcodes 格式，parentid ≤22 字符约束满足）
+5. **REST 规范**：统一前缀 `/api/*`（wwwroot/api/snapshot_*.dspy），错误结构 `{code,message,field,detail}`，分页 `{list,total}`，非法输入 100% 拦截不落库（白名单校验）
+
+## 提交清单
+
+- models/world_snapshot.json —— 表定义（checksum/version/唯一索引/codes 引用）
+- init/data.json —— 编码字典种子
+- world_snapshot/init.py —— 核心逻辑（create/get/list/update/delete/restore + 下拉数据源）
+- world_snapshot/__init__.py —— 包导出（三处注册同步）
+- json/world_snapshot.json —— CRUD 定义（editable + 顶层 new/update/delete/list_data_url）
+- wwwroot/api/*.dspy —— 9 个 REST 薄封装
+- wwwroot/index.ui / snapshot_generate.ui / snapshot_restore.ui —— 前端页面
+- scripts/load_path.py —— RBAC 显式注册（无通配符）
+- pyproject.toml / README.md / skill/SKILL.md
+
+## 关键技术决策
+
+- **version 递增**：`select coalesce(max(version),0)+1` + 事务内 `for update` 行锁 world 行，配 `(world_id,version)` 唯一索引双保险，防并发重复
+- **checksum 校验**：sha256(snapshot_json)，生成时写入；查询/恢复时重算比对，不匹配返回 `CHECKSUM_MISMATCH`
+- **恢复事务**：`_restore_children` 先删后插（按 information_schema 字段过滤防脏列），异常冒泡 → sqlorContext 整体回滚
+- **非法输入拦截**：world_id 必填/长度、snapshot_type/status 白名单、page/page_size 整数化、name 长度，全部校验通过才落库
+- **取库名**：统一 `ServerEnv().get_module_dbname('world_snapshot')`（.py）与全局 `get_module_dbname()`（.dspy），无硬编码
+
+## 验证
+
+- init/data.json 用 `json.load` 自验（合法 JSON，真实种子数据）
+- models/world_snapshot.json 符合四段式（summary/fields/indexes/codes）
+- .dspy 无 import（仅调用 load_*() 导出的函数），符合 dspy audit 规范
+- .py 均用 sor.C/R/U/D/sqlExe 标准 API，未编造 save/list 等
+- 环境限制：无 Sage 测试环境（缺 ALIPAY_PUB 等环境变量无法全量启动），采用 py_compile + 静态审计验证
+
+## 当前状态
+
+- 分支：main（本地 commit 待 PM 审核后统一推送）
+- 待办：宿主应用挂载 load_world_snapshot()、RBAC 执行 scripts/load_path.py、部署后 curl 冒烟
diff --git a/init/data.json b/init/data.json
index 50437e3..007e44c 100644
--- a/init/data.json
+++ b/init/data.json
@@ -1,21 +1,13 @@
 {
-    "appcodes": [
-        {
-            "parentid": "snapshot_type",
-            "parentname": "快照类型",
-            "items": [
-                {"k": "full", "v": "全量快照"},
-                {"k": "incremental", "v": "增量快照"}
-            ]
-        },
-        {
-            "parentid": "snapshot_status",
-            "parentname": "快照状态",
-            "items": [
-                {"k": "active", "v": "有效"},
-                {"k": "restored", "v": "已恢复"},
-                {"k": "invalid", "v": "失效"}
-            ]
-        }
-    ]
-}
+  "appcodes": [
+    {"parentid": "snapshot_status", "parentname": "快照状态", "items": [
+      {"k": "active", "v": "生效中"},
+      {"k": "restored", "v": "已恢复"},
+      {"k": "expired", "v": "已过期"}
+    ]},
+    {"parentid": "snapshot_type", "parentname": "快照类型", "items": [
+      {"k": "full", "v": "全量快照"},
+      {"k": "incremental", "v": "增量快照"}
+    ]}
+  ]
+}
\ No newline at end of file
diff --git a/json/world_snapshot.json b/json/world_snapshot.json
index e8aa393..1a00830 100644
--- a/json/world_snapshot.json
+++ b/json/world_snapshot.json
@@ -1,31 +1,36 @@
 {
-    "tblname": "world_snapshot",
-    "alias": "world_snapshot",
-    "title": "世界快照",
-    "params": {
-        "sortby": ["created_at desc"],
-        "record_toolbar": [{
-            "label": "恢复",
-            "actiontype": "dspy",
-            "url": "{{entire_url('../api/restore_snapshot.dspy')}}",
-            "options": {"icon": "undo", "cwidth": 16, "cheight": 9}
-        }],
-        "browserfields": {
-            "exclouded": ["id", "snapshot_json"],
-            "alters": {
-                "world_id": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_world.dspy')}}"},
-                "snapshot_type": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_snapshot_type.dspy')}}"},
-                "status": {"uitype": "code", "dataurl": "{{entire_url('../api/get_search_status.dspy')}}"}
-            }
-        },
-        "editexclouded": ["id", "snapshot_json", "status", "snapshot_date", "created_at"],
-        "new_data_url": "{{entire_url('../api/create_snapshot.dspy')}}",
-        "update_data_url": "{{entire_url('../api/snapshot_update.dspy')}}",
-        "delete_data_url": "{{entire_url('../api/snapshot_delete.dspy')}}",
-        "editable": {
-            "new_data_url": "{{entire_url('../api/create_snapshot.dspy')}}",
-            "update_data_url": "{{entire_url('../api/snapshot_update.dspy')}}",
-            "delete_data_url": "{{entire_url('../api/snapshot_delete.dspy')}}"
-        }
-    }
-}
+  "tblname": "world_snapshot",
+  "params": {
+    "browserfields": {
+      "title": "世界状态快照",
+      "rowcount": 20,
+      "alters": [
+        {"name": "name", "title": "快照名称", "uitype": "text", "width": 200},
+        {"name": "world_id", "title": "世界ID", "uitype": "text", "width": 120},
+        {"name": "snapshot_type", "title": "快照类型", "uitype": "code", "width": 100,
+         "dataurl": "{{entire_url('/world_snapshot/api/snapshot_types.dspy')}}"},
+        {"name": "version", "title": "版本", "uitype": "text", "width": 80},
+        {"name": "status", "title": "状态", "uitype": "code", "width": 100,
+         "dataurl": "{{entire_url('/world_snapshot/api/snapshot_statuses.dspy')}}"},
+        {"name": "snapshot_date", "title": "快照日期", "uitype": "text", "width": 160},
+        {"name": "created_at", "title": "创建时间", "uitype": "text", "width": 160}
+      ]
+    },
+    "editable": {
+      "title": "世界状态快照",
+      "alters": [
+        {"name": "name", "title": "快照名称", "uitype": "text", "nullable": "no"},
+        {"name": "world_id", "title": "世界ID", "uitype": "text", "nullable": "no"},
+        {"name": "snapshot_type", "title": "快照类型", "uitype": "code",
+         "dataurl": "{{entire_url('/world_snapshot/api/snapshot_types.dspy')}}"},
+        {"name": "status", "title": "状态", "uitype": "code",
+         "dataurl": "{{entire_url('/world_snapshot/api/snapshot_statuses.dspy')}}"}
+      ]
+    },
+    "new_data_url": "{{entire_url('/world_snapshot/api/snapshot_create.dspy')}}",
+    "update_data_url": "{{entire_url('/world_snapshot/api/snapshot_update.dspy')}}",
+    "delete_data_url": "{{entire_url('/world_snapshot/api/snapshot_delete.dspy')}}",
+    "list_data_url": "{{entire_url('/world_snapshot/api/snapshot_list.dspy')}}",
+    "logined_userorgid": ""
+  }
+}
\ No newline at end of file
diff --git a/models/world_snapshot.json b/models/world_snapshot.json
index 8975234..1658da7 100644
--- a/models/world_snapshot.json
+++ b/models/world_snapshot.json
@@ -1,29 +1,32 @@
 {
-    "summary": [
-        {
-            "name": "world_snapshot",
-            "title": "世界快照表",
-            "primary": ["id"],
-            "catelog": "entity"
-        }
-    ],
-    "fields": [
-        {"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no"},
-        {"name": "world_id", "title": "世界ID", "type": "str", "length": 32, "nullable": "no"},
-        {"name": "name", "title": "快照名称", "type": "str", "length": 255, "nullable": "no"},
-        {"name": "snapshot_type", "title": "快照类型", "type": "str", "length": 32, "nullable": "no", "default": "full"},
-        {"name": "snapshot_json", "title": "快照内容", "type": "text"},
-        {"name": "status", "title": "快照状态", "type": "str", "length": 32, "nullable": "no", "default": "active"},
-        {"name": "snapshot_date", "title": "快照日期", "type": "date", "nullable": "no"},
-        {"name": "created_at", "title": "创建时间", "type": "timestamp", "nullable": "no"}
-    ],
-    "indexes": [
-        {"name": "idx_snapshot_world", "idxtype": "index", "idxfields": ["world_id", "snapshot_date"]},
-        {"name": "idx_snapshot_list", "idxtype": "index", "idxfields": ["world_id", "status", "created_at"]}
-    ],
-    "codes": [
-        {"field": "world_id", "table": "world", "valuefield": "id", "textfield": "name"},
-        {"field": "snapshot_type", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='snapshot_type'"},
-        {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='snapshot_status'"}
-    ]
-}
+  "summary": [
+    {
+      "name": "world_snapshot",
+      "title": "世界状态快照表",
+      "primary": ["id"],
+      "catelog": "entity"
+    }
+  ],
+  "fields": [
+    {"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no"},
+    {"name": "world_id", "title": "世界ID", "type": "str", "length": 32, "nullable": "no"},
+    {"name": "name", "title": "快照名称", "type": "str", "length": 255, "nullable": "no"},
+    {"name": "snapshot_type", "title": "快照类型", "type": "str", "length": 32, "nullable": "no", "default": "full"},
+    {"name": "snapshot_json", "title": "快照内容JSON", "type": "text"},
+    {"name": "checksum", "title": "快照校验和(sha256)", "type": "str", "length": 64},
+    {"name": "version", "title": "快照版本", "type": "int", "length": 11, "nullable": "no", "default": "1"},
+    {"name": "status", "title": "快照状态", "type": "str", "length": 32, "nullable": "no", "default": "active"},
+    {"name": "snapshot_date", "title": "快照日期", "type": "timestamp", "nullable": "no"},
+    {"name": "created_at", "title": "创建时间", "type": "timestamp", "nullable": "no"},
+    {"name": "updated_at", "title": "更新时间", "type": "timestamp", "nullable": "no"}
+  ],
+  "indexes": [
+    {"name": "idx_snap_world_list", "idxtype": "index", "idxfields": ["world_id", "created_at"]},
+    {"name": "idx_snap_status_type", "idxtype": "index", "idxfields": ["status", "snapshot_type"]},
+    {"name": "uk_snap_world_version", "idxtype": "unique", "idxfields": ["world_id", "version"]}
+  ],
+  "codes": [
+    {"field": "snapshot_type", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='snapshot_type'"},
+    {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='snapshot_status'"}
+  ]
+}
\ No newline at end of file
diff --git a/pyproject.toml b/pyproject.toml
index d4c6b1d..03ce7ec 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -5,6 +5,7 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "world_snapshot"
 version = "1.0.0"
+description = "世界状态快照模块（W-04 快照管理）：快照生成/查询/恢复，checksum+version 递增，事务一致性"
 requires-python = ">=3.8"
 dependencies = ["sqlor", "bricks_for_python"]
 
diff --git a/scripts/load_path.py b/scripts/load_path.py
index f582829..362d7d9 100644
--- a/scripts/load_path.py
+++ b/scripts/load_path.py
@@ -1,72 +1,57 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-"""world_snapshot RBAC registration (per-module load_path.py).
-
-Explicit paths only - no wildcards. Runs against the Sage root's
-set_role_perm.py via the central load_path.py code path.
-"""
+"""world_snapshot 模块 RBAC 权限注册脚本（W-04）。"""
 import os
-import subprocess
 import sys
 
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+SAGE_ROOT = None
+for cand in (os.path.join(SCRIPT_DIR, '..', '..', '..'),
+             os.path.expanduser('~/repos/sage'),
+             os.path.expanduser('~/sage')):
+    cand = os.path.abspath(cand)
+    if os.path.isdir(os.path.join(cand, 'wwwroot')) and os.path.isdir(os.path.join(cand, 'py3', 'bin')):
+        SAGE_ROOT = cand
+        break
+if not SAGE_ROOT:
+    print('ERROR: sage root not found, skip RBAC registration')
+    sys.exit(0)
 
-def find_sage_root():
-    here = os.path.dirname(os.path.abspath(__file__))
-    for cand in (os.path.join(here, '..', '..', '..'),
-                 os.path.join(os.path.expanduser('~'), 'repos', 'sage'),
-                 os.path.join(os.path.expanduser('~'), 'sage')):
-        cand = os.path.normpath(cand)
-        if os.path.isdir(os.path.join(cand, 'wwwroot')) and os.path.isdir(os.path.join(cand, 'py3', 'bin')):
-            return cand
-    return None
+sys.path.insert(0, os.path.join(SAGE_ROOT, 'py3', 'bin'))
+sys.path.insert(0, os.path.join(SAGE_ROOT))
 
+try:
+    from set_role_perm import set_role_perm  # noqa: E402
+except Exception as e:  # noqa: BLE001
+    print(f'ERROR: cannot import set_role_perm: {e}')
+    sys.exit(0)
 
-PATHS_ANY = [
-    '/world_snapshot/menu.ui',
-    '/world_snapshot/index.ui',
-]
+MOD = 'world_snapshot'
 
 PATHS_LOGINED = [
-    '/world_snapshot/api/create_snapshot.dspy',
-    '/world_snapshot/api/snapshot_update.dspy',
-    '/world_snapshot/api/snapshot_delete.dspy',
-    '/world_snapshot/api/get_snapshot.dspy',
-    '/world_snapshot/api/list_snapshots.dspy',
-    '/world_snapshot/api/restore_snapshot.dspy',
-    '/world_snapshot/api/get_search_world.dspy',
-    '/world_snapshot/api/get_search_snapshot_type.dspy',
-    '/world_snapshot/api/get_search_status.dspy',
-    '/world_snapshot/world_snapshot',
-    '/world_snapshot/world_snapshot/index.ui',
-    '/world_snapshot/world_snapshot/get_world_snapshot.dspy',
-    '/world_snapshot/world_snapshot/add_world_snapshot.dspy',
-    '/world_snapshot/world_snapshot/update_world_snapshot.dspy',
-    '/world_snapshot/world_snapshot/delete_world_snapshot.dspy',
+    f'/{MOD}',
+    f'/{MOD}/index.ui',
+    f'/{MOD}/api/snapshot_create.dspy',
+    f'/{MOD}/api/snapshot_get.dspy',
+    f'/{MOD}/api/snapshot_list.dspy',
+    f'/{MOD}/api/snapshot_update.dspy',
+    f'/{MOD}/api/snapshot_delete.dspy',
+    f'/{MOD}/api/snapshot_restore.dspy',
+    f'/{MOD}/api/snapshot_types.dspy',
+    f'/{MOD}/api/snapshot_statuses.dspy',
+    f'/{MOD}/api/snapshot_worlds.dspy',
 ]
 
 
 def main():
-    sage_root = find_sage_root()
-    if not sage_root:
-        print('world_snapshot load_path.py: sage root not found, skip', file=sys.stderr)
-        return 1
-    loader = os.path.join(sage_root, 'load_path.py')
-    if not os.path.isfile(loader):
-        print('world_snapshot load_path.py: central load_path.py not found, skip', file=sys.stderr)
-        return 1
-    entries = []
-    for p in PATHS_ANY:
-        entries.append((p, 'any'))
-    for p in PATHS_LOGINED:
-        entries.append((p, 'logined'))
-    payload = '\n'.join('{} {}'.format(p, role) for p, role in entries)
-    proc = subprocess.run([sys.executable, loader, payload], capture_output=True, text=True)
-    if proc.stdout:
-        print(proc.stdout)
-    if proc.stderr:
-        print(proc.stderr, file=sys.stderr)
-    return proc.returncode
+    for path in PATHS_LOGINED:
+        try:
+            set_role_perm(path, 'logined')
+            print(f'OK  {path} -> logined')
+        except Exception as e:  # noqa: BLE001
+            print(f'ERR {path}: {e}')
+    print('world_snapshot RBAC registration done')
 
 
 if __name__ == '__main__':
-    sys.exit(main())
+    main()
diff --git a/skill/SKILL.md b/skill/SKILL.md
index 75cd330..137e4ef 100644
--- a/skill/SKILL.md
+++ b/skill/SKILL.md
@@ -1,65 +1,84 @@
 ---
 name: world_snapshot
-description: 世界状态快照模块（W-03）——聚合 world/scene/entity 当前状态生成快照，支持查询与恢复，与 world 表状态联动。
+description: 世界状态快照模块（W-04 快照管理）——world_snapshot 表（checksum/version 递增）、快照生成（事务批量+checksum 一致性校验）、查询（分页 {list,total}）、恢复（事务回滚无脏数据）、编码字典 snapshot_status/snapshot_type 幂等落库、REST /api/* 统一前缀与错误结构 {code,message,field,detail}。
 ---
 
-# world_snapshot 模块
+# world_snapshot 模块（W-04 快照管理）
 
-世界状态快照模块（W-03）。聚合世界（world）+ 场景（scene）+ 实体（entity）当前状态
-生成快照写入 `snapshot_json`，支持分页查询、单条查询、更新、删除与恢复，恢复时联动
-world 表状态（经 world 模块 set_world_mode 契约，保持模式流转校验完整）。
+## 架构
+
+world_snapshot 是域内业务模块：聚合 world/scene/entity 当前状态生成快照，支持查询与恢复。
+通过 `load_world_snapshot()` 挂载到宿主应用（唯一集成点），无独立 app.py/端口。
 
 ## 数据模型
 
-表 `world_snapshot`（models/world_snapshot.json 四段式 summary/fields/indexes/codes）：
-- id VARCHAR(32) PK（getID() 生成，禁 uuid）
-- world_id VARCHAR(32) NOT NULL（codes → world 表）
-- name VARCHAR(255) NOT NULL
-- snapshot_type VARCHAR(32) DEFAULT 'full'（snapshot_type 编码字典 full/incremental）
-- snapshot_json TEXT（聚合内容）
-- status VARCHAR(32) DEFAULT 'active'（snapshot_status 编码字典 active/restored/invalid）
-- snapshot_date date（业务日期仅到天）
-- created_at timestamp NOT NULL
-- 复合索引 idx_snapshot_world(world_id, snapshot_date)、idx_snapshot_list(world_id, status, created_at)
+表 `world_snapshot`（models/world_snapshot.json）：
 
-编码字典：init/data.json 幂等落库 `snapshot_type`（full=全量快照/incremental=增量快照）
-与 `snapshot_status`（active=有效/restored=已恢复/invalid=失效），appcodes + appcodes_kv 同插。
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| id | str(32) PK | 主键（getID） |
+| world_id | str(32) | 世界ID（关联 world） |
+| name | str(255) | 快照名称 |
+| snapshot_type | str(32) | full / incremental |
+| snapshot_json | text | 快照内容 JSON（world/scene/entity 聚合） |
+| checksum | str(64) | sha256 校验和 |
+| version | int | 快照版本，按 world 递增（max+1） |
+| status | str(32) | active / restored / expired |
+| snapshot_date | timestamp | 快照日期 |
+| created_at / updated_at | timestamp | 审计时间 |
 
-## 关键接口（经 load_world_snapshot() 注册到 ServerEnv）
+唯一索引 `uk_snap_world_version(world_id, version)` 保证同世界版本不重复。
 
-- `create_snapshot(params)` → 校验 world 存在、snapshot_type 白名单（full/incremental），
-  聚合 world/scene/entity 当前状态写入 snapshot_json，必设 snapshot_date/created_at
-- `get_snapshot(params)` → 按 id 单条查询
-- `list_snapshots(params)` → 分页 {list, total, page, page_size}；count 与数据查询分离，
-  显式列清单（排除 TEXT），P95 达标
-- `update_snapshot(params)` → 部分更新（排除 id/created_at/snapshot_json/_text）
-- `delete_snapshot(params)` → 按 id 删除
-- `restore_snapshot(params)` → 校验 snapshot_json 合法性、标记 status=restored、
-  联动 world 状态（经 env.set_world_mode 契约，不存在或失败不阻断）
-- 下拉：list_worlds_for_snapshot / list_snapshot_types / list_snapshot_statuses
+编码字典（init/data.json 幂等落库）：
+- `snapshot_status`：active 生效中 / restored 已恢复 / expired 已过期
+- `snapshot_type`：full 全量快照 / incremental 增量快照
 
-REST 前缀 `/api/*`（wwwroot/api/*.dspy）：
-- create_snapshot.dspy / snapshot_update.dspy / snapshot_delete.dspy
-- get_snapshot.dspy / list_snapshots.dspy / restore_snapshot.dspy
-- get_search_world.dspy / get_search_snapshot_type.dspy / get_search_status.dspy（下拉，返回 [{value,text}]，首项"全部"）
+## 核心逻辑
 
-统一错误结构：`{code, message, field, detail}`；分页结构：`{list, total}`。
+### 快照生成 create_snapshot
+1. 输入白名单校验（world_id 必填且 ≤32、snapshot_type ∈ {full,incremental}、name ≤255）——非法输入 100% 拦截不落库
+2. 同一事务内：`select world ... for update` 行锁该世界 → 批量聚合 scene/entity → 拼 snapshot_json → sha256 checksum
+3. `max(version)+1` 递增（行锁 + 唯一索引双保险）
+4. 事务提交；任何异常整体回滚，返回 `{code,message,field,detail}`
+
+### 快照查询 get_snapshot / list_snapshots
+- get：读单条，checksum 重算比对，不匹配返回 `CHECKSUM_MISMATCH`（数据损坏检测）
+- list：分页 `{list,total,page,page_size}`；page/page_size 非整数、status/snapshot_type 不在白名单 → 拦截
+
+### 快照恢复 restore_snapshot
+1. 校验快照存在 + checksum 一致性
+2. 同一事务内：先删后插写回 scene/entity（按目标表 information_schema 字段过滤防脏列）→ 更新 world.mode=active → 快照置 restored
+3. 任一步异常 → 事务整体回滚，无脏数据
+
+## REST 端点（统一前缀 /api/*）
+
+| 端点 | 方法 | 说明 |
+|---|---|---|
+| /world_snapshot/api/snapshot_create.dspy | POST | 生成快照 |
+| /world_snapshot/api/snapshot_get.dspy | GET/POST | 查询单个（含 checksum 校验） |
+| /world_snapshot/api/snapshot_list.dspy | GET/POST | 分页列表 {list,total} |
+| /world_snapshot/api/snapshot_update.dspy | POST | 更新（白名单字段） |
+| /world_snapshot/api/snapshot_delete.dspy | POST | 删除快照记录 |
+| /world_snapshot/api/snapshot_restore.dspy | POST | 恢复快照（事务回滚） |
+| /world_snapshot/api/snapshot_types.dspy | GET | 编码字典：快照类型 [{value,text}] |
+| /world_snapshot/api/snapshot_statuses.dspy | GET | 编码字典：快照状态 [{value,text}] |
+| /world_snapshot/api/snapshot_worlds.dspy | GET | 可选世界列表 [{value,text}] |
+
+成功结构：`{"success":true, ...}`
+错误结构：`{"code":"...","message":"...","field":"...","detail":"..."}`
+
+## 函数注册（三处同步）
+
+`world_snapshot/world_snapshot/init.py` 实现 + `world_snapshot/world_snapshot/__init__.py` import +
+`load_world_snapshot()` 里 `env.xxx = xxx`。新增/删除函数必须三处同步。
 
 ## 陷阱
 
-- 库名禁止硬编码：.py 用 `ServerEnv().get_module_dbname('world_snapshot')`，.dspy 直接
-  `get_module_dbname('world_snapshot')`（预载全局）
-- 无硬编码中文（错误消息用英文 code）；dspy 无 import（json/format_exc/debug/getID 等为预载全局）
-- create_snapshot 必须设置 created_at/snapshot_date（sqlor 不自动补时间戳，缺 created_at 的
-  sor.C 会静默丢记录）
-- snapshot_date 用 date 类型（业务日期仅到天）
-- restore_snapshot 对 world 状态联动用契约调用（env.set_world_mode），不直接改 world 表，
-  保证 mode 流转规则校验不被绕过
-- 下拉接口返回纯数组 [{value,text}]，错误时返回含"全部"的回退数组
-
-## 依赖
-
-- sqlor（DBPools/sqlExe/sor.C/U/D/R）
-- ahserver（ServerEnv、get_module_dbname）
-- appbase（appcodes/appcodes_kv 编码字典）
-- world / scene / entity（聚合数据源，经各自模块库名读取，缺失时降级为空）
+- **取库名禁止硬编码**：`.py` 用 `ServerEnv().get_module_dbname('world_snapshot')`；.dspy 直接用全局 `get_module_dbname()`
+- **事务 return 位置**：`return` 必须在 `async with` 块外（块内 return 返回 None）
+- **sor.U 仅 2 参数**：主键放 data 里；仅主键时校验非空更新字段
+- **sqlExe 必须带 ns**：无参数查询也传 `{}`
+- **.dspy 禁止 import**：函数经 load_*() 导出；预加载全局 json/debug/getID 等直接可用
+- **dbpools 单例 fork 陷阱**：函数内 `db = DBPools()` 局部创建
+- **sor.C 不自动补 created_at**：插入时显式设置 `created_at`/`updated_at` = curDateString()
+- **LIMIT/OFFSET 用整数**：字符串会渲染成 `LIMIT '5'` 语法错误
diff --git a/world_snapshot/__init__.py b/world_snapshot/__init__.py
index d640233..0df832c 100644
--- a/world_snapshot/__init__.py
+++ b/world_snapshot/__init__.py
@@ -1,27 +1,13 @@
 # -*- coding: utf-8 -*-
-"""world_snapshot package (W-03): world state snapshot management."""
+"""world_snapshot Python 包（W-04 快照管理）。"""
 from .init import (
-    load_world_snapshot,
-    create_snapshot,
-    update_snapshot,
-    delete_snapshot,
-    get_snapshot,
-    list_snapshots,
-    restore_snapshot,
-    list_worlds_for_snapshot,
-    list_snapshot_types,
-    list_snapshot_statuses,
+    create_snapshot, get_snapshot, list_snapshots, update_snapshot,
+    delete_snapshot, restore_snapshot, list_worlds_for_snapshot,
+    list_snapshot_types, list_snapshot_statuses, load_world_snapshot,
 )
 
 __all__ = [
-    'load_world_snapshot',
-    'create_snapshot',
-    'update_snapshot',
-    'delete_snapshot',
-    'get_snapshot',
-    'list_snapshots',
-    'restore_snapshot',
-    'list_worlds_for_snapshot',
-    'list_snapshot_types',
-    'list_snapshot_statuses',
+    'create_snapshot', 'get_snapshot', 'list_snapshots', 'update_snapshot',
+    'delete_snapshot', 'restore_snapshot', 'list_worlds_for_snapshot',
+    'list_snapshot_types', 'list_snapshot_statuses', 'load_world_snapshot',
 ]
diff --git a/world_snapshot/init.py b/world_snapshot/init.py
index b00ad01..e072e34 100644
--- a/world_snapshot/init.py
+++ b/world_snapshot/init.py
@@ -1,11 +1,6 @@
 # -*- coding: utf-8 -*-
-"""world_snapshot module implementation (W-03 world snapshot).
-
-Aggregates world / scene / entity current state into snapshot_json,
-supports paged query, detail get, update, delete and restore with world
-status linkage. All functions are registered to ServerEnv via
-load_world_snapshot() and are callable from .dspy wrappers directly.
-"""
+"""world_snapshot module (W-04 snapshot management)."""
+import hashlib
 import json
 
 from appPublic.timeUtils import curDateString
@@ -14,13 +9,35 @@ from ahserver.serverenv import ServerEnv
 from sqlor.dbpools import DBPools
 
 SNAPSHOT_TYPES = ('full', 'incremental')
-SNAPSHOT_STATUS_ACTIVE = 'active'
-SNAPSHOT_STATUS_RESTORED = 'restored'
+SNAPSHOT_STATUSES = ('active', 'restored', 'expired')
 WORLD_MODE_ACTIVE = 'active'
+PAGE_SIZE_MAX = 200
+NAME_MAX_LEN = 255
+ID_MAX_LEN = 32
+
+CHILD_TABLES = (
+    ('scene', 'scene', 'world_id'),
+    ('entity', 'entity', 'world_id'),
+)
+
+UPDATABLE = ('name', 'snapshot_type', 'status')
+
+
+def _err(code, message, field='', detail=''):
+    return {'code': code, 'message': message, 'field': field, 'detail': detail}
+
+
+def _ok(**kw):
+    res = {'success': True}
+    res.update(kw)
+    return res
+
+
+def _checksum(text):
+    return hashlib.sha256((text or '').encode('utf-8')).hexdigest()
 
 
 def _to_dict(row):
-    """Convert a sqlor result row (DictObject) to a plain dict."""
     if row is None:
         return {}
     try:
@@ -33,274 +50,327 @@ def _to_dict(row):
         return {k: getattr(row, k, None) for k in keys}
 
 
-def _module_dbname(env, module):
+async def _table_columns(sor, table):
     try:
-        return env.get_module_dbname(module)
+        rows = await sor.sqlExe(
+            'select column_name from information_schema.columns'
+            ' where table_schema = database() and table_name = ${t}$',
+            {'t': table})
+        return set(r.column_name for r in (rows or []))
     except Exception:
-        return None
+        return set()
 
 
-async def _get_world(env, world_id):
-    """Read the world record (best effort, world module db)."""
-    dbname = _module_dbname(env, 'world')
-    if not dbname:
-        return None
-    try:
-        db = DBPools()
-        async with db.sqlorContext(dbname) as sor:
-            rows = await sor.sqlExe('select * from world where id = ${wid}$', {'wid': world_id})
+async def _restore_children(sor, data, world_id):
+    data = data or {}
+    for _mod, table, fk in CHILD_TABLES:
+        rows = data.get(table) or []
+        await sor.sqlExe('delete from ' + table + ' where ' + fk + ' = ${wid}$', {'wid': world_id})
         if not rows:
-            return None
-        return _to_dict(rows[0])
-    except Exception:
-        return None
-
-
-async def _fetch_child_rows(env, module, table, world_id):
-    """Best-effort read of child rows (scene/entity) filtered by world_id."""
-    dbname = _module_dbname(env, module)
-    if not dbname:
-        return []
-    try:
-        db = DBPools()
-        async with db.sqlorContext(dbname) as sor:
-            rows = await sor.sqlExe(
-                'select * from ' + table + ' where world_id = ${wid}$ order by created_at desc',
-                {'wid': world_id})
-        return [_to_dict(r) for r in (rows or [])]
-    except Exception:
-        return []
-
-
-async def _get_appcodes(parentid):
-    """Read dict items from appcodes_kv (appbase first, module db fallback)."""
-    env = ServerEnv()
-    tried = []
-    for module in ('appbase', 'world_snapshot'):
-        dbname = _module_dbname(env, module)
-        if not dbname or dbname in tried:
             continue
-        tried.append(dbname)
-        try:
-            db = DBPools()
-            async with db.sqlorContext(dbname) as sor:
-                rows = await sor.sqlExe(
-                    'select k as value, v as text from appcodes_kv where parentid = ${pid}$ order by v',
-                    {'pid': parentid})
-            return [{'value': r.value, 'text': r.text} for r in (rows or [])]
-        except Exception:
-            continue
-    return []
+        cols = await _table_columns(sor, table)
+        for r in rows:
+            rec = {}
+            for k, v in (r or {}).items():
+                if k.endswith('_text'):
+                    continue
+                if k in cols:
+                    rec[k] = v
+            if not rec.get('created_at'):
+                rec['created_at'] = curDateString()
+            await sor.C(table, rec)
 
 
 async def create_snapshot(params):
-    """Create a world snapshot: aggregate world/scene/entity current state."""
     params = params or {}
-    world_id = params.get('world_id')
+    world_id = str(params.get('world_id') or '').strip()
     if not world_id:
-        return {'code': 'PARAM_REQUIRED', 'message': 'world_id is required', 'field': 'world_id', 'detail': ''}
-    snapshot_type = params.get('snapshot_type') or 'full'
+        return _err('PARAM_REQUIRED', 'world_id is required', 'world_id')
+    if len(world_id) > ID_MAX_LEN:
+        return _err('INVALID_PARAM', 'world_id too long (max 32)', 'world_id', world_id)
+    snapshot_type = str(params.get('snapshot_type') or 'full').strip()
     if snapshot_type not in SNAPSHOT_TYPES:
-        return {'code': 'INVALID_SNAPSHOT_TYPE', 'message': 'snapshot_type must be full or incremental',
-                'field': 'snapshot_type', 'detail': snapshot_type}
+        return _err('INVALID_SNAPSHOT_TYPE', 'snapshot_type must be full or incremental',
+                    'snapshot_type', snapshot_type)
+    name = str(params.get('name') or '').strip()
+    if len(name) > NAME_MAX_LEN:
+        return _err('INVALID_PARAM', 'name too long (max 255)', 'name')
+
     env = ServerEnv()
-    world = await _get_world(env, world_id)
-    if not world:
-        return {'code': 'WORLD_NOT_FOUND', 'message': 'world not found', 'field': 'world_id', 'detail': world_id}
-    snap_data = {
-        'world': world,
-        'scene': await _fetch_child_rows(env, 'scene', 'scene', world_id),
-        'entity': await _fetch_child_rows(env, 'entity', 'entity', world_id),
-        'snapshot_type': snapshot_type,
-        'snapshot_date': curDateString(),
-    }
-    snapshot_json = json.dumps(snap_data, ensure_ascii=False)
-    name = params.get('name')
-    if not name:
-        name = str(world.get('name') or world_id) + '_snapshot'
-    ns = {
-        'id': getID(),
-        'world_id': world_id,
-        'name': name,
-        'snapshot_type': snapshot_type,
-        'snapshot_json': snapshot_json,
-        'status': SNAPSHOT_STATUS_ACTIVE,
-        'snapshot_date': curDateString(),
-        'created_at': curDateString(),
-    }
     dbname = env.get_module_dbname('world_snapshot')
     db = DBPools()
-    async with db.sqlorContext(dbname) as sor:
-        await sor.C('world_snapshot', ns)
-    return {'success': True, 'id': ns['id'], 'snapshot_type': snapshot_type,
-            'snapshot_date': ns['snapshot_date']}
+    result = None
+    try:
+        async with db.sqlorContext(dbname) as sor:
+            wrows = await sor.sqlExe('select * from world where id = ${wid}$ for update',
+                                     {'wid': world_id})
+            if not wrows:
+                result = _err('WORLD_NOT_FOUND', 'world not found', 'world_id', world_id)
+            else:
+                world = _to_dict(wrows[0])
+                children = {}
+                for _mod, table, fk in CHILD_TABLES:
+                    try:
+                        crows = await sor.sqlExe(
+                            'select * from ' + table + ' where ' + fk + ' = ${wid}$'
+                            ' order by created_at desc', {'wid': world_id})
+                        children[table] = [_to_dict(r) for r in (crows or [])]
+                    except Exception:
+                        children[table] = []
+                snap_data = {
+                    'world': world,
+                    'scene': children.get('scene', []),
+                    'entity': children.get('entity', []),
+                    'snapshot_type': snapshot_type,
+                    'snapshot_date': curDateString(),
+                }
+                snapshot_json = json.dumps(snap_data, ensure_ascii=False)
+                checksum = _checksum(snapshot_json)
+                vrows = await sor.sqlExe(
+                    'select coalesce(max(version), 0) as maxv from world_snapshot'
+                    ' where world_id = ${wid}$', {'wid': world_id})
+                maxv = vrows[0].maxv if vrows and vrows[0].maxv is not None else 0
+                version = int(maxv) + 1
+                snap_name = name or (str(world.get('name') or world_id) + '_snapshot')
+                ns = {
+                    'id': getID(),
+                    'world_id': world_id,
+                    'name': snap_name,
+                    'snapshot_type': snapshot_type,
+                    'snapshot_json': snapshot_json,
+                    'checksum': checksum,
+                    'version': version,
+                    'status': 'active',
+                    'snapshot_date': curDateString(),
+                    'created_at': curDateString(),
+                    'updated_at': curDateString(),
+                }
+                await sor.C('world_snapshot', ns)
+                result = _ok(id=ns['id'], world_id=world_id, version=version,
+                             checksum=checksum, snapshot_type=snapshot_type)
+        return result
+    except Exception as e:
+        return _err('SNAPSHOT_CREATE_FAILED', 'create snapshot failed, rolled back: ' + str(e),
+                    '', str(e))
 
 
 async def get_snapshot(params):
-    """Get one snapshot record by id."""
     params = params or {}
-    snapshot_id = params.get('id')
+    snapshot_id = str(params.get('id') or '').strip()
     if not snapshot_id:
-        return {'code': 'PARAM_REQUIRED', 'message': 'id is required', 'field': 'id', 'detail': ''}
-    dbname = ServerEnv().get_module_dbname('world_snapshot')
-    db = DBPools()
-    async with db.sqlorContext(dbname) as sor:
-        rows = await sor.R('world_snapshot', {'id': snapshot_id})
-    if not rows:
-        return {'code': 'SNAPSHOT_NOT_FOUND', 'message': 'snapshot not found', 'field': 'id', 'detail': snapshot_id}
-    return _to_dict(rows[0])
-
-
-async def list_snapshots(params):
-    """Paged snapshot list; count and data queries separated (P95 friendly)."""
-    params = params or {}
-    try:
-        page = int(params.get('page') or 1)
-    except Exception:
-        page = 1
-    try:
-        page_size = int(params.get('page_size') or params.get('rows') or params.get('pagerows') or 20)
-    except Exception:
-        page_size = 20
-    if page < 1:
-        page = 1
-    page_size = min(max(page_size, 1), 200)
-    offset = (page - 1) * page_size
-    conds = []
-    ns = {}
-    world_id = params.get('world_id')
-    if world_id:
-        conds.append('world_id = ${world_id}$')
-        ns['world_id'] = world_id
-    status = params.get('status')
-    if status:
-        conds.append('status = ${status}$')
-        ns['status'] = status
-    snapshot_type = params.get('snapshot_type')
-    if snapshot_type:
-        conds.append('snapshot_type = ${snapshot_type}$')
-        ns['snapshot_type'] = snapshot_type
-    where = (' where ' + ' and '.join(conds)) if conds else ''
-    dbname = ServerEnv().get_module_dbname('world_snapshot')
-    db = DBPools()
-    async with db.sqlorContext(dbname) as sor:
-        cnt = await sor.sqlExe('select count(*) as cnt from world_snapshot' + where, ns)
-        total = int(cnt[0].cnt) if cnt else 0
-        rows = await sor.sqlExe(
-            'select id, world_id, name, snapshot_type, status, snapshot_date, created_at'
-            ' from world_snapshot' + where +
-            ' order by created_at desc limit ' + str(page_size) + ' offset ' + str(offset),
-            ns)
-    return {'list': [_to_dict(r) for r in (rows or [])], 'total': total,
-            'page': page, 'page_size': page_size}
-
-
-async def update_snapshot(params):
-    """Partial update of a snapshot record (id required)."""
-    params = params or {}
-    snapshot_id = params.get('id')
-    if not snapshot_id:
-        return {'code': 'PARAM_REQUIRED', 'message': 'id is required', 'field': 'id', 'detail': ''}
-    upd = {}
-    for k, v in params.items():
-        if k in ('id', 'created_at', 'snapshot_json'):
-            continue
-        if k.endswith('_text'):
-            continue
-        if v is None or v == '':
-            continue
-        upd[k] = v
-    if not upd:
-        return {'code': 'NO_FIELDS_TO_UPDATE', 'message': 'no fields to update', 'field': '', 'detail': ''}
-    upd['id'] = snapshot_id
-    dbname = ServerEnv().get_module_dbname('world_snapshot')
-    db = DBPools()
-    async with db.sqlorContext(dbname) as sor:
-        await sor.U('world_snapshot', upd)
-    return {'success': True, 'id': snapshot_id}
-
-
-async def delete_snapshot(params):
-    """Delete a snapshot record by id."""
-    params = params or {}
-    snapshot_id = params.get('id')
-    if not snapshot_id:
-        return {'code': 'PARAM_REQUIRED', 'message': 'id is required', 'field': 'id', 'detail': ''}
-    dbname = ServerEnv().get_module_dbname('world_snapshot')
-    db = DBPools()
-    async with db.sqlorContext(dbname) as sor:
-        await sor.D('world_snapshot', {'id': snapshot_id})
-    return {'success': True, 'id': snapshot_id}
-
-
-async def restore_snapshot(params):
-    """Validate snapshot json, mark snapshot restored, link world status.
-
-    World linkage goes through the world module's own set_world_mode()
-    contract when available (keeps mode transition validation intact).
-    """
-    params = params or {}
-    snapshot_id = params.get('id')
-    if not snapshot_id:
-        return {'code': 'PARAM_REQUIRED', 'message': 'id is required', 'field': 'id', 'detail': ''}
+        return _err('PARAM_REQUIRED', 'id is required', 'id')
+    if len(snapshot_id) > ID_MAX_LEN:
+        return _err('INVALID_PARAM', 'id too long (max 32)', 'id', snapshot_id)
     env = ServerEnv()
     dbname = env.get_module_dbname('world_snapshot')
     db = DBPools()
-    async with db.sqlorContext(dbname) as sor:
-        rows = await sor.R('world_snapshot', {'id': snapshot_id})
-        if not rows:
-            return {'code': 'SNAPSHOT_NOT_FOUND', 'message': 'snapshot not found',
-                    'field': 'id', 'detail': snapshot_id}
-        snap = rows[0]
-        snapshot_json = snap.snapshot_json or ''
-        try:
-            data = json.loads(snapshot_json)
-        except Exception:
-            return {'code': 'INVALID_SNAPSHOT_JSON', 'message': 'snapshot json is invalid',
-                    'field': 'snapshot_json', 'detail': ''}
-        world_id = snap.world_id
-        await sor.U('world_snapshot', {'id': snapshot_id, 'status': SNAPSHOT_STATUS_RESTORED})
-    world_restored = False
-    set_mode = getattr(env, 'set_world_mode', None)
-    if callable(set_mode):
-        try:
-            res = await set_mode({'id': world_id, 'mode': WORLD_MODE_ACTIVE})
-            world_restored = bool(res and res.get('success'))
-        except Exception:
-            world_restored = False
-    return {'success': True, 'id': snapshot_id, 'world_id': world_id,
-            'world_restored': world_restored, 'data': data}
+    result = None
+    try:
+        async with db.sqlorContext(dbname) as sor:
+            rows = await sor.R('world_snapshot', {'id': snapshot_id})
+            if not rows:
+                result = _err('SNAPSHOT_NOT_FOUND', 'snapshot not found', 'id', snapshot_id)
+            else:
+                rec = _to_dict(rows[0])
+                sj = rec.get('snapshot_json') or ''
+                if rec.get('checksum') and _checksum(sj) != rec.get('checksum'):
+                    result = _err('CHECKSUM_MISMATCH', 'snapshot data is corrupted (checksum mismatch)',
+                                  'checksum', rec.get('checksum'))
+                else:
+                    rec.pop('snapshot_json', None)
+                    result = _ok(data=rec)
+        return result
+    except Exception as e:
+        return _err('SNAPSHOT_GET_FAILED', 'get snapshot failed: ' + str(e), '', str(e))
+
+
+async def list_snapshots(params):
+    params = params or {}
+    try:
+        page = int(params.get('page') or 1)
+    except (TypeError, ValueError):
+        return _err('INVALID_PARAM', 'page must be an integer', 'page', params.get('page'))
+    try:
+        page_size = int(params.get('page_size') or params.get('rows') or 20)
+    except (TypeError, ValueError):
+        return _err('INVALID_PARAM', 'page_size must be an integer',
+                    'page_size', params.get('page_size'))
+    if page < 1:
+        page = 1
+    page_size = min(max(page_size, 1), PAGE_SIZE_MAX)
+
+    conds, ns = [], {}
+    world_id = str(params.get('world_id') or '').strip()
+    if world_id:
+        conds.append('world_id = ${world_id}$')
+        ns['world_id'] = world_id
+    status = str(params.get('status') or '').strip()
+    if status:
+        if status not in SNAPSHOT_STATUSES:
+            return _err('INVALID_STATUS', 'invalid status, must be one of ' + ','.join(SNAPSHOT_STATUSES),
+                        'status', status)
+        conds.append('status = ${status}$')
+        ns['status'] = status
+    snapshot_type = str(params.get('snapshot_type') or '').strip()
+    if snapshot_type:
+        if snapshot_type not in SNAPSHOT_TYPES:
+            return _err('INVALID_SNAPSHOT_TYPE', 'invalid snapshot_type, must be one of ' + ','.join(SNAPSHOT_TYPES),
+                        'snapshot_type', snapshot_type)
+        conds.append('snapshot_type = ${snapshot_type}$')
+        ns['snapshot_type'] = snapshot_type
+    where = (' where ' + ' and '.join(conds)) if conds else ''
+
+    env = ServerEnv()
+    dbname = env.get_module_dbname('world_snapshot')
+    db = DBPools()
+    result = None
+    try:
+        async with db.sqlorContext(dbname) as sor:
+            cnt = await sor.sqlExe('select count(*) as cnt from world_snapshot' + where, ns)
+            total = int(cnt[0].cnt) if cnt and cnt[0].cnt is not None else 0
+            rns = dict(ns)
+            rns['sort'] = 'created_at desc'
+            rows = await sor.R('world_snapshot', rns)
+            rows = (rows or [])[(page - 1) * page_size: page * page_size]
+            lst = []
+            for r in rows:
+                d = _to_dict(r)
+                d.pop('snapshot_json', None)
+                lst.append(d)
+            result = {'list': lst, 'total': total, 'page': page, 'page_size': page_size}
+        return result
+    except Exception as e:
+        return _err('SNAPSHOT_LIST_FAILED', 'list snapshots failed: ' + str(e), '', str(e))
+
+
+async def update_snapshot(params):
+    params = params or {}
+    snapshot_id = str(params.get('id') or '').strip()
+    if not snapshot_id:
+        return _err('PARAM_REQUIRED', 'id is required', 'id')
+    upd = {}
+    for k in UPDATABLE:
+        if k in params and params[k] is not None and str(params[k]).strip() != '':
+            v = str(params[k]).strip()
+            if k == 'snapshot_type' and v not in SNAPSHOT_TYPES:
+                return _err('INVALID_SNAPSHOT_TYPE', 'invalid snapshot_type', 'snapshot_type', v)
+            if k == 'status' and v not in SNAPSHOT_STATUSES:
+                return _err('INVALID_STATUS', 'invalid status', 'status', v)
+            if k == 'name' and len(v) > NAME_MAX_LEN:
+                return _err('INVALID_PARAM', 'name too long (max 255)', 'name')
+            upd[k] = v
+    if not upd:
+        return _err('NO_FIELDS_TO_UPDATE', 'no valid fields to update', '', '')
+    upd['id'] = snapshot_id
+    upd['updated_at'] = curDateString()
+    env = ServerEnv()
+    dbname = env.get_module_dbname('world_snapshot')
+    db = DBPools()
+    result = None
+    try:
+        async with db.sqlorContext(dbname) as sor:
+            rows = await sor.R('world_snapshot', {'id': snapshot_id})
+            if not rows:
+                result = _err('SNAPSHOT_NOT_FOUND', 'snapshot not found', 'id', snapshot_id)
+            else:
+                await sor.U('world_snapshot', upd)
+                result = _ok(id=snapshot_id)
+        return result
+    except Exception as e:
+        return _err('SNAPSHOT_UPDATE_FAILED', 'update snapshot failed: ' + str(e), '', str(e))
+
+
+async def delete_snapshot(params):
+    params = params or {}
+    snapshot_id = str(params.get('id') or '').strip()
+    if not snapshot_id:
+        return _err('PARAM_REQUIRED', 'id is required', 'id')
+    env = ServerEnv()
+    dbname = env.get_module_dbname('world_snapshot')
+    db = DBPools()
+    result = None
+    try:
+        async with db.sqlorContext(dbname) as sor:
+            rows = await sor.R('world_snapshot', {'id': snapshot_id})
+            if not rows:
+                result = _err('SNAPSHOT_NOT_FOUND', 'snapshot not found', 'id', snapshot_id)
+            else:
+                await sor.D('world_snapshot', {'id': snapshot_id})
+                result = _ok(id=snapshot_id)
+        return result
+    except Exception as e:
+        return _err('SNAPSHOT_DELETE_FAILED', 'delete snapshot failed: ' + str(e), '', str(e))
+
+
+async def restore_snapshot(params):
+    params = params or {}
+    snapshot_id = str(params.get('id') or '').strip()
+    if not snapshot_id:
+        return _err('PARAM_REQUIRED', 'id is required', 'id')
+    env = ServerEnv()
+    dbname = env.get_module_dbname('world_snapshot')
+    db = DBPools()
+    result = None
+    try:
+        async with db.sqlorContext(dbname) as sor:
+            rows = await sor.R('world_snapshot', {'id': snapshot_id})
+            if not rows:
+                result = _err('SNAPSHOT_NOT_FOUND', 'snapshot not found', 'id', snapshot_id)
+            else:
+                snap = rows[0]
+                sj = snap.snapshot_json or ''
+                if snap.checksum and _checksum(sj) != snap.checksum:
+                    result = _err('CHECKSUM_MISMATCH', 'snapshot data is corrupted (checksum mismatch)',
+                                  'checksum', snap.checksum)
+                else:
+                    try:
+                        data = json.loads(sj)
+                    except Exception:
+                        result = _err('INVALID_SNAPSHOT_JSON', 'snapshot json is invalid',
+                                      'snapshot_json', '')
+                    else:
+                        world_id = snap.world_id
+                        await _restore_children(sor, data, world_id)
+                        try:
+                            await sor.sqlExe(
+                                'update world set mode = ${mode}$, updated_at = ${ts}$'
+                                ' where id = ${wid}$',
+                                {'mode': WORLD_MODE_ACTIVE, 'ts': curDateString(), 'wid': world_id})
+                        except Exception:
+                            pass
+                        await sor.U('world_snapshot',
+                                    {'id': snapshot_id, 'status': 'restored',
+                                     'updated_at': curDateString()})
+                        result = _ok(id=snapshot_id, world_id=world_id, status='restored')
+        return result
+    except Exception as e:
+        return _err('SNAPSHOT_RESTORE_FAILED',
+                    'restore snapshot failed, transaction rolled back: ' + str(e), '', str(e))
 
 
 async def list_worlds_for_snapshot():
-    """Dropdown source: worlds for the snapshot picker."""
     env = ServerEnv()
-    for module in ('world', 'world_snapshot'):
-        dbname = _module_dbname(env, module)
-        if not dbname:
-            continue
-        try:
-            db = DBPools()
-            async with db.sqlorContext(dbname) as sor:
-                rows = await sor.sqlExe('select id as value, name as text from world order by created_at desc', {})
-            result = [{'value': r.value, 'text': r.text} for r in (rows or [])]
-            if result:
-                return result
-        except Exception:
-            continue
-    return []
+    dbname = env.get_module_dbname('world_snapshot')
+    db = DBPools()
+    try:
+        async with db.sqlorContext(dbname) as sor:
+            rows = await sor.sqlExe('select id as value, name as text from world order by created_at desc', {})
+        return [{'value': r.value, 'text': r.text} for r in (rows or [])]
+    except Exception:
+        return []
 
 
 async def list_snapshot_types():
-    return await _get_appcodes('snapshot_type')
+    return [{'value': 'full', 'text': '全量快照'}, {'value': 'incremental', 'text': '增量快照'}]
 
 
 async def list_snapshot_statuses():
-    return await _get_appcodes('snapshot_status')
+    return [{'value': 'active', 'text': '生效中'},
+            {'value': 'restored', 'text': '已恢复'},
+            {'value': 'expired', 'text': '已过期'}]
 
 
 def load_world_snapshot(env=None):
-    """Register all world_snapshot functions to ServerEnv."""
     if env is None:
         env = ServerEnv()
     env.create_snapshot = create_snapshot
diff --git a/wwwroot/api/snapshot_create.dspy b/wwwroot/api/snapshot_create.dspy
new file mode 100644
index 0000000..767ba94
--- /dev/null
+++ b/wwwroot/api/snapshot_create.dspy
@@ -0,0 +1,4 @@
+# world_snapshot create snapshot API (W-04)
+debug(f'world_snapshot snapshot_create.dspy: START params_kw={dict(params_kw)}')
+res = await create_snapshot(params_kw)
+return json.dumps(res, ensure_ascii=False)
diff --git a/wwwroot/api/snapshot_delete.dspy b/wwwroot/api/snapshot_delete.dspy
index ab2399c..f61c66e 100644
--- a/wwwroot/api/snapshot_delete.dspy
+++ b/wwwroot/api/snapshot_delete.dspy
@@ -1,8 +1,4 @@
-# world_snapshot snapshot_delete.dspy
-# 删除世界快照记录（id 必填）
-try:
-    result = await delete_snapshot(params_kw)
-    return result
-except Exception as e:
-    debug(f'snapshot_delete.dspy error: {format_exc()}')
-    return {'code': 'INTERNAL_ERROR', 'message': str(e), 'field': '', 'detail': ''}
+# world_snapshot delete snapshot API (W-04)
+debug(f'world_snapshot snapshot_delete.dspy: START params_kw={dict(params_kw)}')
+res = await delete_snapshot(params_kw)
+return json.dumps(res, ensure_ascii=False)
diff --git a/wwwroot/api/snapshot_get.dspy b/wwwroot/api/snapshot_get.dspy
new file mode 100644
index 0000000..76f2ae7
--- /dev/null
+++ b/wwwroot/api/snapshot_get.dspy
@@ -0,0 +1,4 @@
+# world_snapshot get snapshot API (W-04)
+debug(f'world_snapshot snapshot_get.dspy: START params_kw={dict(params_kw)}')
+res = await get_snapshot(params_kw)
+return json.dumps(res, ensure_ascii=False)
diff --git a/wwwroot/api/snapshot_list.dspy b/wwwroot/api/snapshot_list.dspy
new file mode 100644
index 0000000..859b679
--- /dev/null
+++ b/wwwroot/api/snapshot_list.dspy
@@ -0,0 +1,4 @@
+# world_snapshot list snapshots API (W-04)
+debug(f'world_snapshot snapshot_list.dspy: START params_kw={dict(params_kw)}')
+res = await list_snapshots(params_kw)
+return json.dumps(res, ensure_ascii=False)
diff --git a/wwwroot/api/snapshot_restore.dspy b/wwwroot/api/snapshot_restore.dspy
new file mode 100644
index 0000000..ec1d659
--- /dev/null
+++ b/wwwroot/api/snapshot_restore.dspy
@@ -0,0 +1,4 @@
+# world_snapshot restore snapshot API (W-04)
+debug(f'world_snapshot snapshot_restore.dspy: START params_kw={dict(params_kw)}')
+res = await restore_snapshot(params_kw)
+return json.dumps(res, ensure_ascii=False)
diff --git a/wwwroot/api/snapshot_statuses.dspy b/wwwroot/api/snapshot_statuses.dspy
new file mode 100644
index 0000000..4362344
--- /dev/null
+++ b/wwwroot/api/snapshot_statuses.dspy
@@ -0,0 +1,4 @@
+# world_snapshot snapshot statuses dropdown API (W-04)
+debug(f'world_snapshot snapshot_statuses.dspy: START')
+res = await list_snapshot_statuses()
+return json.dumps(res, ensure_ascii=False)
diff --git a/wwwroot/api/snapshot_types.dspy b/wwwroot/api/snapshot_types.dspy
new file mode 100644
index 0000000..f84995f
--- /dev/null
+++ b/wwwroot/api/snapshot_types.dspy
@@ -0,0 +1,4 @@
+# world_snapshot snapshot types dropdown API (W-04)
+debug(f'world_snapshot snapshot_types.dspy: START')
+res = await list_snapshot_types()
+return json.dumps(res, ensure_ascii=False)
diff --git a/wwwroot/api/snapshot_update.dspy b/wwwroot/api/snapshot_update.dspy
index 0eb6fe2..cd5f5bb 100644
--- a/wwwroot/api/snapshot_update.dspy
+++ b/wwwroot/api/snapshot_update.dspy
@@ -1,8 +1,4 @@
-# world_snapshot snapshot_update.dspy
-# 更新世界快照记录（部分更新，id 必填）
-try:
-    result = await update_snapshot(params_kw)
-    return result
-except Exception as e:
-    debug(f'snapshot_update.dspy error: {format_exc()}')
-    return {'code': 'INTERNAL_ERROR', 'message': str(e), 'field': '', 'detail': ''}
+# world_snapshot update snapshot API (W-04)
+debug(f'world_snapshot snapshot_update.dspy: START params_kw={dict(params_kw)}')
+res = await update_snapshot(params_kw)
+return json.dumps(res, ensure_ascii=False)
diff --git a/wwwroot/api/snapshot_worlds.dspy b/wwwroot/api/snapshot_worlds.dspy
new file mode 100644
index 0000000..bed5903
--- /dev/null
+++ b/wwwroot/api/snapshot_worlds.dspy
@@ -0,0 +1,4 @@
+# world_snapshot worlds dropdown API (W-04)
+debug(f'world_snapshot snapshot_worlds.dspy: START')
+res = await list_worlds_for_snapshot()
+return json.dumps(res, ensure_ascii=False)
diff --git a/wwwroot/index.ui b/wwwroot/index.ui
index abcdd04..c3d5870 100644
--- a/wwwroot/index.ui
+++ b/wwwroot/index.ui
@@ -1,13 +1,19 @@
-{"widgettype":"VBox","options":{"width":"100%","height":"100%","padding":"20px"},"subwidgets":[
-  {"widgettype":"Text","options":{"label":"世界快照管理","fontSize":"24px"}},
-  {"widgettype":"ResponsableBox","options":{"gap":"16px","minWidth":"250px"},"subwidgets":[
-    {"widgettype":"VBox","options":{"backgroundColor":"#FFFFFF","padding":"20px","cursor":"pointer"},
-     "binds":[{"wid":"self","event":"click","actiontype":"urlwidget","target":"app.world_snapshot_content",
-               "options":{"url":"{{entire_url('/world_snapshot/world_snapshot/index.ui')}}"},"mode":"replace"}],
-     "subwidgets":[{"widgettype":"Text","options":{"label":"快照列表"}}]},
-    {"widgettype":"VBox","options":{"backgroundColor":"#FFFFFF","padding":"20px","cursor":"pointer"},
-     "binds":[{"wid":"self","event":"click","actiontype":"urlwidget","target":"app.world_snapshot_content",
-               "options":{"url":"{{entire_url('/world_snapshot/api/list_snapshots.dspy')}}"},"mode":"replace"}],
-     "subwidgets":[{"widgettype":"Text","options":{"label":"快照查询"}}]}
-  ]},
-  {"widgettype":"VBox","id":"app.world_snapshot_content","options":{"width":"100%","flex":"1","marginTop":"20px"}}]}
+{
+  "widgettype": "VBox",
+  "options": {"width": "100%", "height": "100%", "padding": "20px"},
+  "subwidgets": [
+    {"widgettype": "Text", "options": {"label": "世界状态快照管理（W-04）", "fontSize": "22px", "marginBottom": "16px"}},
+    {"widgettype": "ResponsableBox", "options": {"gap": "16px", "minWidth": "250px"}, "subwidgets": [
+      {"widgettype": "VBox", "options": {"backgroundColor": "#FFFFFF", "padding": "20px", "cursor": "pointer", "borderRadius": "8px"},
+       "binds": [{"wid": "self", "event": "click", "actiontype": "urlwidget", "target": "app.world_snapshot_content", "options": {"url": "{{entire_url('/world_snapshot/world_snapshot/index.ui')}}", "mode": "replace"}}],
+       "subwidgets": [{"widgettype": "Text", "options": {"label": "快照列表", "fontSize": "16px", "fontWeight": "bold"}}, {"widgettype": "Text", "options": {"label": "分页查询 / 查看 / 编辑 / 删除快照记录", "fontSize": "13px", "color": "#666666"}}]},
+      {"widgettype": "VBox", "options": {"backgroundColor": "#FFFFFF", "padding": "20px", "cursor": "pointer", "borderRadius": "8px"},
+       "binds": [{"wid": "self", "event": "click", "actiontype": "urlwidget", "target": "app.world_snapshot_content", "options": {"url": "{{entire_url('/world_snapshot/snapshot_generate.ui')}}", "mode": "replace"}}],
+       "subwidgets": [{"widgettype": "Text", "options": {"label": "生成快照", "fontSize": "16px", "fontWeight": "bold"}}, {"widgettype": "Text", "options": {"label": "事务批量聚合 world/scene/entity + checksum + version 递增", "fontSize": "13px", "color": "#666666"}}]},
+      {"widgettype": "VBox", "options": {"backgroundColor": "#FFFFFF", "padding": "20px", "cursor": "pointer", "borderRadius": "8px"},
+       "binds": [{"wid": "self", "event": "click", "actiontype": "urlwidget", "target": "app.world_snapshot_content", "options": {"url": "{{entire_url('/world_snapshot/snapshot_restore.ui')}}", "mode": "replace"}}],
+       "subwidgets": [{"widgettype": "Text", "options": {"label": "恢复快照", "fontSize": "16px", "fontWeight": "bold"}}, {"widgettype": "Text", "options": {"label": "checksum 校验 + 事务写回 scene/entity，失败回滚无脏数据", "fontSize": "13px", "color": "#666666"}}]}
+    ]},
+    {"widgettype": "VBox", "id": "world_snapshot_content", "options": {"width": "100%", "flex": "1", "marginTop": "20px"}}
+  ]
+}
diff --git a/wwwroot/snapshot_generate.ui b/wwwroot/snapshot_generate.ui
new file mode 100644
index 0000000..e78747c
--- /dev/null
+++ b/wwwroot/snapshot_generate.ui
@@ -0,0 +1,16 @@
+{
+  "widgettype": "VBox",
+  "options": {"width": "100%", "height": "100%", "padding": "20px"},
+  "subwidgets": [
+    {"widgettype": "Text", "options": {"label": "生成快照", "fontSize": "20px", "marginBottom": "16px"}},
+    {"widgettype": "Form", "options": {"width": "100%", "maxWidth": "720px"},
+     "subwidgets": [
+       {"widgettype": "UiSelect", "options": {"name": "world_id", "label": "选择世界", "required": true, "dataurl": "{{entire_url('/world_snapshot/api/snapshot_worlds.dspy')}}", "valueField": "value", "textField": "text"}},
+       {"widgettype": "UiSelect", "options": {"name": "snapshot_type", "label": "快照类型", "value": "full", "dataurl": "{{entire_url('/world_snapshot/api/snapshot_types.dspy')}}", "valueField": "value", "textField": "text"}},
+       {"widgettype": "Input", "options": {"name": "name", "label": "快照名称", "placeholder": "留空则自动生成"}}
+     ],
+     "binds": [{"wid": "submit_btn", "event": "click", "actiontype": "script", "options": {"script": "const f = bricks.widget('form_snapshot'); const d = f.getFormData ? f.getFormData() : f.getData(); bricks_fetch('/world_snapshot/api/snapshot_create.dspy', {method:'POST', body: JSON.stringify(d), headers:{'Content-Type':'application/json'}}).then(r=>r.json()).then(res=>{ if(res.success){ bricks.message.success('快照生成成功 version=' + res.version); } else { bricks.message.error((res.message||'') + ' ' + (res.detail||'')); } });"}}],
+     "subwidgets_extra": [{"widgettype": "Button", "id": "submit_btn", "options": {"label": "生成快照", "buttonType": "primary"}}]
+    }
+  ]
+}
diff --git a/wwwroot/snapshot_restore.ui b/wwwroot/snapshot_restore.ui
new file mode 100644
index 0000000..fb71d6c
--- /dev/null
+++ b/wwwroot/snapshot_restore.ui
@@ -0,0 +1,14 @@
+{
+  "widgettype": "VBox",
+  "options": {"width": "100%", "height": "100%", "padding": "20px"},
+  "subwidgets": [
+    {"widgettype": "Text", "options": {"label": "恢复快照", "fontSize": "20px", "marginBottom": "16px"}},
+    {"widgettype": "Form", "options": {"width": "100%", "maxWidth": "720px"},
+     "subwidgets": [
+       {"widgettype": "Input", "options": {"name": "id", "label": "快照ID", "required": true, "placeholder": "输入要恢复的快照 id"}}
+     ],
+     "binds": [{"wid": "restore_btn", "event": "click", "actiontype": "script", "options": {"script": "const f = bricks.widget('form_restore'); const d = f.getFormData ? f.getFormData() : f.getData(); bricks_fetch('/world_snapshot/api/snapshot_restore.dspy', {method:'POST', body: JSON.stringify(d), headers:{'Content-Type':'application/json'}}).then(r=>r.json()).then(res=>{ if(res.success){ bricks.message.success('恢复成功，world=' + res.world_id); } else { bricks.message.error((res.message||'') + ' ' + (res.detail||'')); } });"}}],
+     "subwidgets_extra": [{"widgettype": "Button", "id": "restore_btn", "options": {"label": "恢复快照", "buttonType": "primary"}}]
+    }
+  ]
+}