fix: drop default task_logs; process in log, progress in task center (v1.0.81)

2026-08-14 20:53:25 +08:00
parent ff75aefcdd
commit 1384a70826
39 changed files with 387 additions and 696 deletions
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,12 @@
 - 小节正文写用户能看懂的变化（新能力、修复、注意事项），避免堆砌内部实现细节
 - 打 tag 前**必须**为即将发布的版本新增一节；找不到匹配小节时，本次发布不会更新 changelog 字段

+## 1.0.81
+
+- 模板默认不再创建或写入 `task_logs`：过程看运行日志，进度看任务中心，业务结果看业务表
+- `run` 录屏信息并入 `finish()`；去掉 CLI `logs` / `log-get`
+- 已安装技能的旧库不受影响
+
 ## 1.0.80

 - 沉淀数据管理「HTML 页面报告」契约：写入 `{skill_data}/html/*.html`，重新生成 Action id 固定为 `render-html`
--- a/SKILL.md
+++ b/SKILL.md
@@ -1,7 +1,7 @@
 ---
 name: 技能开发模板（通用业务版）
 description: "OpenClaw 通用业务技能开发模板，供复制后定制新业务 skill。定制步骤见 development/DEVELOPMENT.md。"
-version: 1.0.80
+version: 1.0.81
 author: 深圳匠厂科技有限公司
 metadata:
  openclaw:
@@ -44,7 +44,7 @@ allowed-tools:

 | 你要做什么 | Agent 怎么做 | 进任务中心？ |
 |------------|--------------|--------------|
-| 短查询 / 只读 / 秒级（`health` / `version` / `stats` / `list` / `get` / `logs` 等） | OpenClaw bash/exec + **共享 Python** 跑 CLI（见下方「共享 Python」） | 否（CLI 当场结束） |
+| 短查询 / 只读 / 秒级（`health` / `version` / `stats` / `list` / `get` 等） | OpenClaw bash/exec + **共享 Python** 跑 CLI（见下方「共享 Python」） | 否（CLI 当场结束） |
 | 已在 `actions.json` 声明且 `executionProfile: "sync"` 的能力 | 可用 `run_skill_action`，或同等效果的短 CLI（同一内核） | 否 |
 | 长耗时 / RPA / 浏览器 / 需要进度或取消；或 `executionProfile: "async"` | **必须** `run_skill_action`（先 `list_skill_actions` / `get_skill_action`） | **是**（仅因为该 Action 是 async） |
 | 用户只问用法 / 能做什么 | 读根目录 `README.md`；逐步操作再读 `TUTORIAL.md`；**不要**为答用法去列目录或默认打开 `references/` | — |
--- a/assets/README.md
+++ b/assets/README.md
@@ -3,7 +3,7 @@
 - `actions.json`：Skill Action manifest 最小示例（**可选**；不需要宿主入口时可删除）。每个 action **必须**显式声明 `executionProfile`（`sync` \| `async`）。默认仅 `health` / `version` / `config-path`，不放 toolbar，故无 `bind` 示例。
 - `html/example-report.html`：HTML 报告页最小壳（`{{TITLE}}` / `{{GENERATED_AT}}` / `{{BODY}}`）；渲染后须写入数据目录 `html/`，见 `references/ACTIONS.md`「HTML 页面报告」。
 - `examples/`：CLI 成功输出形状示例、以及 `render-html.action.example.json`（虚构路径与数据）。
- `schemas/`：轻量 JSON Schema（`skill-actions.schema.json` 为**新技能严格规范**；`task-log-record.schema.json` 等）。
+- `schemas/`：轻量 JSON Schema（`skill-actions.schema.json` 为**新技能严格规范**）。

 `skill-actions.schema.json` 使用 `additionalProperties: false`。正式支持 `bind.tables`（toolbar Action 必填）与可选 `bind.inputMapping`；placements 含 `toolbar` / `row` / `batch` / `cron` / `agent` / `skill-detail`（对齐匠厂宿主）。`concurrency`/`locks` 仍为预留。`placements` 与 `executionProfile` 正交——进任务中心只看 `async`，与入口无关。表级 Action 与 Agent 契约见 [`references/ACTIONS.md`](../references/ACTIONS.md)；开发深度规范见 [`development/SKILL_ACTION_RUNTIME.md`](../development/SKILL_ACTION_RUNTIME.md)。

--- a/assets/examples/README.md
+++ b/assets/examples/README.md
@@ -1,7 +1,6 @@
 # 示例 JSON

 - `version-response.json`：`python main.py version` 成功时 stdout 单行对象的形状参考。
- `log-get-response.json`：若技能有日志表，`log-get` 成功时单条对象的形状参考。
 - `skill-policy-snapshot.*.example.json`：Data Policy Snapshot 形状参考（见 `references/POLICY.md`）。
 - `skill-cell-presentation.example.json`：数据管理单元格呈现文档形状参考（见 `references/CELL_PRESENTATION.md`）。
 - `render-html.action.example.json`：HTML 报告「重新生成」Action 形状参考（固定 id `render-html`；见 `references/ACTIONS.md`）。
--- a/assets/examples/log-get-response.json
+++ b/assets/examples/log-get-response.json
@@ -1,12 +0,0 @@
-{
-  "id": 1,
-  "task_type": "demo_task",
-  "target_id": "demo_target_1",
-  "input_id": "demo_input_42",
-  "input_title": "示例任务输入标题",
-  "status": "success",
-  "error_msg": null,
-  "result_summary": "{\"processed\": 10, \"skipped\": 0}",
-  "created_at": "2026-04-01T10:00:00",
-  "updated_at": "2026-04-01T10:00:00"
-}
--- a/assets/schemas/task-log-record.schema.json
+++ b/assets/schemas/task-log-record.schema.json
@@ -1,21 +0,0 @@
-{
-  "$schema": "https://json-schema.org/draft/2020-12/schema",
-  "$id": "https://openclaw.local/skill-template/task-log-record.schema.json",
-  "title": "TaskLogRecord",
-  "description": "通用业务技能 log-get 返回的单条任务日志对象模板",
-  "type": "object",
-  "required": ["id", "task_type", "status", "created_at", "updated_at"],
-  "properties": {
-    "id": { "type": "integer" },
-    "task_type": { "type": "string" },
-    "target_id": { "type": ["string", "null"] },
-    "input_id": { "type": ["string", "null"] },
-    "input_title": { "type": ["string", "null"] },
-    "status": { "type": "string" },
-    "error_msg": { "type": ["string", "null"] },
-    "result_summary": { "type": ["string", "null"] },
-    "created_at": { "type": ["string", "null"] },
-    "updated_at": { "type": ["string", "null"] }
-  },
-  "additionalProperties": true
-}
--- a/development/ACCOUNT_MANAGER.md
+++ b/development/ACCOUNT_MANAGER.md
@@ -200,7 +200,7 @@ python scripts/main.py account open <account_id>
 3. 执行 `account list/get`，核对账号 ID、`profile_dir`、`session_status`。
 4. 在文件系统确认 `profile_dir` 是否存在、是否错用了其他用户目录。
 5. 执行 `lease list --active`，确认是否被其他任务占用或残留。
-6. 查看任务日志、错误码、失败截图和浏览器当前页面。
+6. 查看运行日志、任务中心终态、错误码、失败截图和浏览器当前页面。
 7. 判断问题属于账号不存在、租约冲突、登录失效、Profile 损坏、页面变化还是业务失败。

 仍需协助时必须提供：技能 slug、平台、脱敏账号标识 / 账号 ID、任务时间、错误码、账号与租约检查结果、`profile_dir` 是否存在、脱敏日志和截图。禁止发送密码、Cookie、Token 或验证码。
--- a/development/AI_RPA.md
+++ b/development/AI_RPA.md
@@ -81,7 +81,7 @@ Agent 与编排层必须确认：

 - 先验证业务成功标志，再记录成功；不得把“点击完成”当成“业务成功”。
 - 返回成功、部分成功、失败、停止或等待人工等明确状态。
- 保存必要的任务日志、错误码、失败截图和可选录屏。
+- 保存必要的运行日志、错误码、失败截图和可选录屏。
 - Agent 根据结构化结果解释完成数量、失败原因和下一步，不直接粘贴技术日志。

 ## 4. 合规与技术红线
--- a/development/CHECKLIST.md
+++ b/development/CHECKLIST.md
@@ -52,7 +52,7 @@ checklist_version: 1.0
 - [ ] S0-1 交互形态：纯本地 / API / 网页RPA / 桌面RPA / 手机RPA / 编排（调兄弟技能）
 - [ ] S0-2 目标站登录策略：required / optional / not_needed / 不适用
 - [ ] S0-3 是否依赖账号会话（account-manager 或等价 Profile）：是 / 否
- [ ] S0-4 持久化：无 / 仅任务日志 / 另有业务表
+- [ ] S0-4 持久化：无 / 另有业务表
 - [ ] S0-5 Skill Action：无 / 仅同步短任务 / 含异步长任务
 - [ ] S0-6 是否需要在任务中心查看进度或结果：是 / 否
 - [ ] S0-7 外部系统账号、数据、设备与操作权限的授权来源：
--- a/development/CONFIG.md
+++ b/development/CONFIG.md
@@ -125,7 +125,7 @@ HUMAN_WAIT_TIMEOUT=180                 # 单位：秒；来不及处理时可调

 - **模板默认 `OPENCLAW_RECORD_VIDEO=0`**，降低本地开发 / 调试成本（不启 ffmpeg、不生成 MP4）。
 - 需要**生产留证、合规记录、客户现场复现或排障**时，可在用户 `.env` 或进程环境变量中设为 `OPENCLAW_RECORD_VIDEO=1`。
- **默认关闭不代表可以删除录屏接入**：RPA / 长任务代码仍须保留 `RpaVideoSession`、`video.add_step` 与 video artifact 写入 `result_summary`（见 `RPA.md` §5.3）。
+- **默认关闭不代表可以删除录屏接入**：RPA / 长任务代码仍须保留 `RpaVideoSession`、`video.add_step` 与 video artifact 写入 `finish()`（见 `RPA.md` §5.3）。
 - **已落盘的用户 `.env` 不会被模板整体覆盖**；老用户保留原值，技能升级时仅通过 `merge_missing_env_keys` **追加** `.env.example` 中新增的 key。

 ## 🚫 红线：敏感信息不进 `.env`
--- a/development/DEVELOPMENT.md
+++ b/development/DEVELOPMENT.md
@@ -5,7 +5,7 @@
 本文默认你开发的是当前最常见的一类业务 skill：

 - 有明确的 `scripts/main.py` CLI 入口
- 可能需要读写本地 SQLite（例如任务日志 `task_logs`）
+- 可能需要读写本地 SQLite（业务表按需创建，不是默认必有）
 - 可能需要调用兄弟技能或外部 HTTP / RPA
 - 业务逻辑主要放在 `scripts/service/`

@@ -448,10 +448,10 @@ metadata:
  宿主 Action 入口 manifest（可选；见 `references/ACTIONS.md`）

 - `examples/`
-  比如 `version` 输出示例、`log-get` 输出示例
+  比如 `version` 输出示例

 - `schemas/`
-  比如 `skill-actions.schema.json`、日志记录、机读 JSON 的 schema
+  比如 `skill-actions.schema.json`、机读 JSON 的 schema

 不要把正式研发文档放到 `assets/`。
 用户市场四 Tab 进根 `README.md` / `TUTORIAL.md` / `DEMO.md` / `CHANGELOG.md`；Agent 编排资料进 `references/`；开发规范进 `development/`。
@@ -466,7 +466,7 @@ metadata:
 也就是说，`cli/app.py` 的职责是：

 1. 打印帮助
-2. 定义 `run / logs / log-get / health / version / init-db`（及业务子命令）
+2. 定义 `run / health / version / init-db`（及业务子命令）
 3. 把参数转交给 `service.task_service`（或薄适配后再进 domain service）

 不要在 `cli/app.py` 里直接写：
@@ -505,7 +505,7 @@ assets/actions.json             → placements / bind / executionProfile / entry

 `service` 层是**关键业务日志**的落点，须遵循 [`LOGGING.md`](LOGGING.md)：

- `cmd_run` 及主任务入口必须：`log.info("task_start ...")`、`log.info("task_log_saved ...")`；失败路径必须 `log.exception("task_failed ...")`。
+- `cmd_run` 及主任务入口必须：`log.info("task_start ...")`；失败路径必须 `log.exception("task_failed ...")`。
 - 外部 API、兄弟技能、RPA 操作须记录**开始 / 结束 / 耗时 / 状态**（`external_call_start` / `external_call_done elapsed_ms=...`）。
 - 长任务 / RPA 须配合 `activity.emit` 或 `RpaVideoSession.add_step` 输出用户可见进度；`cmd_run` 须 `job_context` + 各出口 `finish`（见 [`LOGGING.md`](LOGGING.md) §2.5）。
 - 使用 `from util.logging_config import get_skill_logger`，**不要**用 `print` 替代 logger 做排障日志。
@@ -526,10 +526,10 @@ assets/actions.json             → placements / bind / executionProfile / entry
 如果你的 skill 需要记录日志或本地状态，建议：

 - `db/connection.py`
-  只做连接和建表
+  只做连接和建表（模板默认只建 `_jiangchang_*` 元数据表；业务表按需 CREATE）

- `db/task_logs_repository.py`
-  只做增删查改（模板默认表：`task_logs`）
+- `db/<domain>_repository.py`
+  只做业务表增删查改（有业务表才建；模板不再默认提供任务审计表）
 - `db/display_metadata.py`
  创建 `_jiangchang_tables` / `_jiangchang_columns` 并幂等写入中文展示名（见 `references/SCHEMA.md`）
 - `db/display_metadata_validator.py`
@@ -667,18 +667,10 @@ python scripts/main.py <your-command> -h
 如果你的 skill 需要本地日志或数据库，再继续：

 ```bash
-python scripts/main.py <your-log-command>
-python scripts/main.py <your-detail-command> <id>
+python scripts/main.py <your-read-command>
 ```

-如果你沿用了模板中的通用任务日志骨架，那么这里可以具体对应成：
-
-```bash
-python scripts/main.py logs
-python scripts/main.py log-get 1
-```
-
-除 `health` / `version` / `logs` 外，执行一次 `run`（或你的主任务命令）后，应确认**统一日志文件**已生成并可检索：
+执行一次 `run`（或你的主任务命令）后，应确认**统一日志文件**已生成并可检索：

 ```text
 {JIANGCHANG_DATA_ROOT}/{JIANGCHANG_USER_ID}/logs/jiangchang.log
@@ -887,7 +879,7 @@ uses: client-jiangchang/jiangchang-platform-kit/.github/workflows/reusable-relea
 - [ ] `release.ps1` 存在（发布前会自动跑 `python tests/run_tests.py -v`；失败不得打 tag）
 - [ ] `.github/workflows/release_skill.yaml` 存在
 - [ ] `python tests/run_tests.py -v` 全部通过（含 `test_development_policy_guard.py`）
- [ ] 主任务入口有关键日志事件词：`task_start`、`task_log_saved`；失败路径 `task_failed error_code=` + `logger.exception`/`warning`（见 [`LOGGING.md`](LOGGING.md) §0、§3）
+- [ ] 主任务入口有关键日志事件词：`task_start`；失败路径 `task_failed error_code=` + `logger.exception`/`warning`（见 [`LOGGING.md`](LOGGING.md) §0、§3）
 - [ ] RPA / 域模块（`scripts/service/<domain>/` 等）非零文件日志；`print(ERROR:)` 旁有 logger（`POLICY-LOGGING-006`～`008`）
 - [ ] 本地跑一次主任务后，统一日志能 grep 到 §3.4 最低可验收串（不只 `cli_start`）
 - [ ] 日志不泄露 password / token / cookie 等敏感明文（见 [`LOGGING.md`](LOGGING.md)）
--- a/development/LOGGING.md
+++ b/development/LOGGING.md
@@ -11,9 +11,9 @@
 | 通道 | 给谁看 | 能否单独当排障依据 |
 |------|--------|-------------------|
 | 统一文件日志 `jiangchang.log` | 运维 / 离线打包 | **必须能单独定位到失败阶段与 error_code** |
-| Activity / 任务中心 `emit` | 用户进度 | **不能**替代文件日志 |
+| Activity / 任务中心 `emit` / `finish` | 用户进度与终态 | **不能**替代文件日志 |
 | `print` / stdout | CLI 一眼结果 | **禁止**作为唯一失败记录 |
-| `task_logs` | 终态审计 | 摘要层；**不能**替代逐步 trace |
+| 业务表 | 用户可见业务结果 | 有业务才建；**不是**默认排障通道 |

 硬规定：

@@ -48,7 +48,7 @@ OpenClaw 技能常在客户电脑上异步执行，且大量依赖 RPA、浏览

 ## 2. 日志分层

-技能日志分四层，职责互不替代：
+技能日志分三层，职责互不替代：

 ### 2.1 统一文件日志（开发 / 运维排查）

@@ -122,17 +122,9 @@ RPA 等待用 `interruptible_sleep`，**不要**裸 `asyncio.sleep`（否则暂

 Journal 会写入 `type: lifecycle` 事件（`paused` / `resumed` / `stopping`），供宿主任务中心显示状态。

+### 2.3 业务表（可选，不是默认排障通道）

-### 2.3 `task_logs`（任务结果审计）
-
-**来源**：`db/task_logs_repository.save_task_log`
-
- 适合 `python scripts/main.py logs` / `log-get <id>` 查询。
- 记录任务最终状态、`result_summary`（可含 `video_path`、`video_log` 等）。
- 面向「这次任务成功还是失败、摘要是什么」，不是逐步 trace。
- **鉴权失败与未捕获异常也必须写入 `task_logs`**，保证 `logs` / `log-get` 排查链不断（见模板 `cmd_run` 示范）。
-
-表结构与字段见 [`references/SCHEMA.md`](../references/SCHEMA.md)。
+模板**默认不建**任务审计表。业务结果写入各技能自己的业务表；过程看 `jiangchang.log`，进度/终态看任务中心。

 ### 2.4 RPA video step / artifacts（用户可见动作与存证）

@@ -140,7 +132,7 @@ Journal 会写入 `type: lifecycle` 事件（`paused` / `resumed` / `stopping`

 - video artifact 是**标准可启用能力**，**不保证默认每次生成 MP4**（模板默认 `OPENCLAW_RECORD_VIDEO=0`）。
 - **`OPENCLAW_RECORD_VIDEO=0`**：不生成最终录屏，但 RPA / 长任务仍须保留 `RpaVideoSession`、`video.add_step` 与 Activity 进度同步。
- **`OPENCLAW_RECORD_VIDEO=1`**：`result_summary` 应包含 `video_path`、`raw_video`、`video_log`、`video_warnings` 等字段；CLI 可打印录屏路径与诊断。
+- **`OPENCLAW_RECORD_VIDEO=1`**：`finish()` 字段应包含 `video_path`、`raw_video`、`video_log`、`video_warnings` 等；CLI 可打印录屏路径与诊断。
 - 用户可见的中文步骤（「打开登录页」「点击提交」）。
 - 失败截图：`{数据目录}/rpa-artifacts/{batch_id}/...`（受 `OPENCLAW_ARTIFACTS_ON_FAILURE` 控制，默认开）

@@ -161,7 +153,6 @@ Journal 会写入 `type: lifecycle` 事件（`paused` / `resumed` / `stopping`
 | `entitlement_ok` / `entitlement_failed` | info / warning | 鉴权结果 |
 | `item_start` / `item_done` | info | 批量逐条时；单条可用 `task_start`/`task_success` 代替 |
 | `task_success` 或 `task_failed` | info / warning\|exception | 终态；失败须含 `error_code=` |
-| `task_log_saved` | info | 写入 `task_logs` 后 |

 ### 3.2 RPA / 桌面 / 浏览器额外最低集

@@ -198,7 +189,6 @@ rpa_start stage=open_window
 rpa_start stage=locate_control
 rpa_start stage=submit
 task_failed error_code=ELEMENT_NOT_FOUND ...
-task_log_saved status=failed
 ```

 成功路径则出现 `task_success` 而非 `task_failed`。若只有 `cli_start` → **不合格**。
@@ -224,7 +214,7 @@ task_log_saved status=failed

 | 字段 | 含义 |
 |------|------|
-| `task_type` | 任务类型（与 `task_logs.task_type` 对齐） |
+| `task_type` | 任务类型 |
 | `target_id` | 目标标识（平台、店铺、报表类型等） |
 | `input_id` | 输入记录 ID |
 | `batch_id` | 批次 / 录屏批次 |
@@ -251,7 +241,7 @@ batch_progress current=7 total=20 target_id=store-001

 | 级别 | 使用场景 |
 |------|----------|
-| **info** | 关键阶段、成功路径、外部调用开始/结束、批量进度、`task_log_saved` |
+| **info** | 关键阶段、成功路径、外部调用开始/结束、批量进度 |
 | **warning** | 可恢复问题、重试、跳过、人工介入、部分失败、鉴权失败（业务可预期） |
 | **exception** | 异常路径；**必须**用 `log.exception(...)` 保留 traceback |
 | **debug** | 低频诊断；**默认不依赖 debug 才能排查问题** |
@@ -264,7 +254,7 @@ batch_progress current=7 total=20 target_id=store-001

 ### 禁止记录原始值

-以下字段**不得**以明文出现在日志、Activity、task_logs 的 `error_msg` / `result_summary`、stdout：
+以下字段**不得**以明文出现在日志、Activity、`finish()` 字段、stdout：

 - `password`、`token`、`cookie`、`secret`、`api_key`、`authorization`
 - 验证码、动态口令全文
@@ -318,8 +308,6 @@ def cmd_run(target=None, input_id=None):
                "erp", "submit", elapsed_ms, "ok",
            )

-            tlr.save_task_log(..., status="success", ...)
-            log.info("task_log_saved task_type=%s status=success", task_type)
            finish(status="success", message="任务完成", skill=SKILL_SLUG)
            return 0
        except Exception:
@@ -328,7 +316,6 @@ def cmd_run(target=None, input_id=None):
                task_type, target, input_id,
            )
            emit("任务失败，已记录诊断信息", type="warn", skill=SKILL_SLUG, stage="run")
-            tlr.save_task_log(..., status="failed", error_msg="...", ...)
            finish(status="failed", message="任务执行异常", skill=SKILL_SLUG)
            return 1
 ```
@@ -342,11 +329,10 @@ def cmd_run(target=None, input_id=None):
 客户现场或测试机排障，建议按以下顺序（由快到慢、由用户可见到运维细节）：

 1. **CLI stdout** — 是否有 `ERROR:<CODE>:`、`HINT:`、模板提示文案。
-2. **`python scripts/main.py logs`** — 最近任务是否写入 `task_logs`、status 与 `error_msg`。
-3. **`python scripts/main.py log-get <id>`** — 单条 JSON，含 `result_summary`（video 路径等）。
-4. **统一日志文件** — `{JIANGCHANG_DATA_ROOT}/{JIANGCHANG_USER_ID}/logs/jiangchang.log`（grep `trace_id`、`task_failed`）。
-5. **Run Journal** — `{JIANGCHANG_DATA_ROOT}/.jiangchang/runs/{job_id}.jsonl`（宿主 UI 进度来源）。
-6. **RPA artifacts** — `rpa-artifacts/` 失败截图；若 `OPENCLAW_RECORD_VIDEO=1` 还有录屏 MP4、`video_log`。未开启录屏时可能只有截图或日志，需要完整录屏请在用户 `.env` 中设置 `OPENCLAW_RECORD_VIDEO=1`。
+2. **任务中心 / Run Journal** — `{JIANGCHANG_DATA_ROOT}/.jiangchang/runs/{job_id}.jsonl`（进度与 `finish` 终态）。
+3. **统一日志文件** — `{JIANGCHANG_DATA_ROOT}/{JIANGCHANG_USER_ID}/logs/jiangchang.log`（grep `trace_id`、`task_failed`）。
+4. **业务表** — 宿主数据管理中该技能自己的业务结果（有才建）。
+5. **RPA artifacts** — `rpa-artifacts/` 失败截图；若 `OPENCLAW_RECORD_VIDEO=1` 还有录屏 MP4、`video_log`。未开启录屏时可能只有截图或日志，需要完整录屏请在用户 `.env` 中设置 `OPENCLAW_RECORD_VIDEO=1`。

 长时间无 stdout **不代表卡死**；先看 Run Journal / Activity，再看统一日志。

@@ -359,8 +345,8 @@ def cmd_run(target=None, input_id=None):
 | [`RUNTIME.md`](RUNTIME.md) | 统一日志能力来源、`util.logging_config` 薄封装、数据根约定 |
 | [`TESTING.md`](TESTING.md) | 日志相关 hard policy 测试（`test_development_policy_guard.py`） |
 | [`RPA.md`](RPA.md) | RPA step、`RpaVideoSession`、video / artifacts 标准 |
-| [`SCHEMA.md`](../references/SCHEMA.md) | `task_logs` 表结构与查询命令 |
-| [`CLI.md`](../references/CLI.md) | `logs` / `log-get` 命令与手工排查入口 |
+| [`SCHEMA.md`](../references/SCHEMA.md) | 业务表结构与展示元数据 |
+| [`CLI.md`](../references/CLI.md) | 最小命令与手工排查入口 |
 | [`CONFIG.md`](CONFIG.md) | 敏感配置不进日志；`.env` 边界 |
 | [`DEVELOPMENT.md`](DEVELOPMENT.md) | `service/` 层日志职责与发布前检查清单 |
 | [`SKILL_ACTION_RUNTIME.md`](SKILL_ACTION_RUNTIME.md) | 任务中心 / async Job 与 `job_context` 边界 |
--- a/development/POLICY_MATRIX.md
+++ b/development/POLICY_MATRIX.md
@@ -18,7 +18,7 @@
 | POLICY-TESTING-002 | `tests/integration/` 下 Python 测试文件若存在，默认应为 `.sample`，不得进入默认套件 | development/TESTING.md §7 真实联调测试的安全约束、§8 新 skill 的最小测试清单 | hard | 检查 `tests/integration/*.py` 无 `.sample` 后缀 | `tests/test_development_policy_guard.py::TestPolicyTesting002` |
 | POLICY-RPA-001 | 不得 import/use account-manager 内部 `rpa_helpers` / `inject_account_manager_scripts_path` / `get_account_credential` | development/ADAPTER.md §兄弟依赖；development/RPA.md §1.7 | hard | 扫描 `scripts/**/*.py` 禁止模式 | `tests/test_development_policy_guard.py::TestPolicyRpa001`（另见 `tests/test_no_rpa_helpers_import.py`） |
 | POLICY-RPA-002 | RPA 交付代码不得直接执行 ffmpeg 拼 MP4；应使用 `RpaVideoSession`（允许 `ffmpeg_path` 等诊断字段） | development/RPA.md §5.3 录屏成片标准 | hard | 扫描 `scripts/**/*.py` 禁止 `subprocess` / `os.system` / `os.popen` 直接调用 ffmpeg（`task_run_support.py` 保留豁免） | `tests/test_development_policy_guard.py::TestPolicyRpa002` |
-| POLICY-RPA-003 | 即使默认 `OPENCLAW_RECORD_VIDEO=0`，RPA / 长任务模板也必须保留录屏能力：`RpaVideoSession`、`video.add_step`、video artifact merge 到 `result_summary` | development/RPA.md §5.3；development/LOGGING.md；development/CONFIG.md | hard | 扫描 `scripts/service/*.py`（service 层整体须含上述接入） | `tests/test_development_policy_guard.py::TestPolicyRpa003` |
+| POLICY-RPA-003 | 即使默认 `OPENCLAW_RECORD_VIDEO=0`，RPA / 长任务模板也必须保留录屏能力：`RpaVideoSession`、`video.add_step`、video artifact merge 进 `finish()` 字段 | development/RPA.md §5.3；development/LOGGING.md；development/CONFIG.md | hard | 扫描 `scripts/service/*.py`（service 层整体须含上述接入） | `tests/test_development_policy_guard.py::TestPolicyRpa003` |
 | POLICY-RPA-004 | 网页 RPA：`.env.example` 若声明 `OPENCLAW_BROWSER_HEADLESS` 则默认必须为 `0`；`scripts/` 若使用 `launch_persistent_*` 则必须同时存在 `pick_web_account`（先 Profile 再开浏览器；实现上应走 ensure-web）；登录不全局强制，但不得跳过 account-manager | development/RPA.md §0.2；development/ADAPTER.md §兄弟依赖 | hard | 解析 `.env.example` + 扫描 `scripts/**/*.py` | `tests/test_development_policy_guard.py::TestPolicyRpa004` |
 | POLICY-RPA-005 | 必须保留 AI + RPA 总纲，明确 Agent / RPA 职责、授权范围、禁止破解/逆向/私有协议/安全验证绕过、人工介入、状态等待；真实业务默认动作节奏为 1～3 秒 | development/AI_RPA.md §4–§6；development/RPA.md §0.0/§0.3 | hard（文档与默认配置） / soft（真实行为） | 检查总纲关键边界；解析 `.env.example` 的 `STEP_DELAY_MIN=1.0` / `STEP_DELAY_MAX=3.0` | `tests/test_development_policy_guard.py::TestPolicyRpa005` |
 | POLICY-RPA-006 | 浏览器 RPA 必须保留 account-manager 生命周期标准；所有样例账号选择路径都须返回非空 `profile_dir + lease_token`，指定数字账号 ID 不得退化为无租约 `account get` | development/ACCOUNT_MANAGER.md；development/ADAPTER.md §兄弟依赖；development/RPA.md §0.2 | hard | 检查标准文档关键标记；扫描真实/仿真 `account_client.py` 的指定 ID 带租约实现并禁止空租约赋值 | `tests/test_development_policy_guard.py::TestPolicyRpa006` |
@@ -27,7 +27,7 @@
 | POLICY-DOCS-001 | 本矩阵存在且包含上述全部 `policy_id` | 本次规范化约定 | hard | 解析 `development/POLICY_MATRIX.md` | `tests/test_development_policy_guard.py::TestPolicyDocs001` |
 | POLICY-LOGGING-001 | CLI 入口必须调用 `setup_skill_logging` | development/LOGGING.md；development/RUNTIME.md | hard | 扫描 `scripts/cli/app.py` 含 `setup_skill_logging` 与 logger info | `tests/test_development_policy_guard.py::TestPolicyLogging001` |
 | POLICY-LOGGING-002 | service 主任务入口必须使用 `get_skill_logger` | development/LOGGING.md；development/DEVELOPMENT.md | hard | 扫描 `scripts/service/task_service.py` | `tests/test_development_policy_guard.py::TestPolicyLogging002` |
-| POLICY-LOGGING-003 | 主任务入口须含 `log.info`/`logger.info`、`log.exception`/`logger.exception`，并写入 `save_task_log` | development/LOGGING.md | hard | 扫描 `scripts/service/task_service.py` | `tests/test_development_policy_guard.py::TestPolicyLogging003` |
+| POLICY-LOGGING-003 | 主任务入口须含 `log.info`/`logger.info`、`log.exception`/`logger.exception` | development/LOGGING.md | hard | 扫描 `scripts/service/task_service.py` | `tests/test_development_policy_guard.py::TestPolicyLogging003` |
 | POLICY-LOGGING-004 | 长任务 / RPA 模板须使用 Activity 或 RPA video step 输出进度 | development/LOGGING.md；development/RPA.md | hard | 扫描 `scripts/service/task_service.py` 含 `emit(` / `activity.emit` / `video.add_step` | `tests/test_development_policy_guard.py::TestPolicyLogging004` |
 | POLICY-LOGGING-005 | 交付代码不得在 logging 调用行以 `key=value` 形式记录明显敏感字段 | development/LOGGING.md；development/CONFIG.md | hard | 扫描 `scripts/**/*.py` 的 logging 调用行（不含注释） | `tests/test_development_policy_guard.py::TestPolicyLogging005` |
 | POLICY-LOGGING-006 | `scripts/service/**/*.py` 业务域文件须有 `log/logger/_logger.(info|warning|error|exception)`；豁免薄工具：`__init__.py`、`task_run_support.py`、`entitlement_service.py`、`sibling_bridge.py` | development/LOGGING.md §0、§3 | hard | 扫描 service 树 | `tests/test_development_policy_guard.py::TestPolicyLogging006` |
--- a/development/REQUIREMENTS.md
+++ b/development/REQUIREMENTS.md
@@ -86,7 +86,7 @@

 - 当前用户在 `<业务场景>` 中需要反复处理 `<输入数据>`，人工操作耗时且易错。
 - 现有流程主要依赖人工处理，存在 `<效率低 / 容易出错 / 不可批量 / 不可追踪>` 等问题。
- 因此需要开发 `your-skill-slug`，用于自动完成 `<核心动作>`，输出 `<输出结果>`，并写入 `task_logs` 便于追踪。
+- 因此需要开发 `your-skill-slug`，用于自动完成 `<核心动作>`，输出 `<输出结果>`。

 填写提示：

@@ -104,7 +104,7 @@
 - 解析并校验 `<输入数据>` 的关键字段
 - 通过 adapter 完成 mock / simulator / real 档位下的 `<核心动作>`
 - 输出结构化结果（stdout JSON 或固定文本 + 机读字段）
- 写入 `task_logs`（及必要的业务表，见 `references/SCHEMA.md`）
+- 写入必要的业务表（见 `references/SCHEMA.md`；无业务表也可）
 - 支持失败可诊断（稳定错误码、`ERROR:` 前缀、必要截图/日志）
 - 完成 CLI 入口与 `health` / `version` / `init-db` 最小可运行验证
 - 完成正式环境发布与安装验证（如适用）
@@ -116,7 +116,7 @@
 模板写法：

 - 支持 `<命令1>`（如 `run`）
- 支持 `<命令2>`（如 `logs` / `log-get`）
+- 支持 `<命令2>`（如 `health` / 业务只读查询）
 - 支持读取 `<数据来源>`（兄弟 skill / 本地 DB / 外部 API）
 - 支持执行 `<核心业务动作>`
 - 支持记录 `<日志 / 状态 / 结果>`
@@ -130,7 +130,7 @@
 - 支持 `python scripts/main.py init-db`（幂等建库；宿主安装/更新后可静默调用）
 - 支持 `python scripts/main.py <your-main-command>`
 - 支持核心业务编排（HTTP / 批处理 / 可选 RPA，按四档 adapter 选型）
- 支持写入任务日志（`task_logs`）
+- 支持失败可诊断（统一文件日志 + 任务中心 `finish`）

 ### 可后续扩展

@@ -207,9 +207,9 @@
 - `health`、`version`、`init-db` 命令执行正常
 - `python tests/run_tests.py -v` 必跑测试全部通过（mock / 离线门禁；**mock 通 ≠ 完成**）
 - 主命令在业务默认档（模板为 `real_rpa`）真实跑通；不得只 mock 通就交活
- `task_logs` 写入和查询符合 `references/SCHEMA.md`（含 `created_at` / `updated_at` Unix 秒级规范）
+- 业务表（若有）写入和查询符合 `references/SCHEMA.md`（含 `created_at` / `updated_at` Unix 秒级规范）
 - `init_db()` 已写入 `_jiangchang_tables` / `_jiangchang_columns`；用户可见表/字段具备中文 `display_name`
- 字段展示顺序与 `PRAGMA table_info(task_logs)` 的 cid 一致；不依赖 `display_order`
+- 字段展示顺序与 `PRAGMA table_info(<业务表>)` 的 cid 一致；不依赖 `display_order`
 - `tests/test_display_metadata.py` 通过
 - 真实联调样例仍放 `tests/integration/`，默认套件不包含真实外联（与业务 `.env` 默认 `real_rpa` 不冲突）
 - `SKILL.md` 的 `developer_ids` 已配置为开发者本人匠厂用户 ID（开发期不公开时，否则本人无法在市场安装自测；取 ID 步骤见 `development/DEVELOPMENT.md` §6）
@@ -259,19 +259,18 @@
 ## 2. 业务背景

 当前用户在 `<业务场景>` 中需要反复处理 `<输入数据>`，人工操作耗时且易错。
-因此需要开发 `your-skill-slug`，自动完成 `<核心动作>`，输出 `<输出结果>`，并写入 task_logs。
+因此需要开发 `your-skill-slug`，自动完成 `<核心动作>`，输出 `<输出结果>`。

 ## 3. 开发目标

 - 解析并校验输入
 - 通过 adapter 完成 mock / simulator / real 档位动作
- 输出结构化结果并写入 task_logs
+- 输出结构化结果；业务结果写入业务表（有才建）
 - 支持失败可诊断

 ## 4. 功能范围

 - 支持 health / version / init-db / run
- 支持 logs / log-get

 ## 5. 非功能要求

@@ -288,7 +287,7 @@

 - 成功 JSON / 摘要
 - ERROR: 稳定错误码
- task_logs 记录
+- 任务中心终态（`finish`）

 ## 7. 依赖的兄弟技能或外部系统

@@ -303,7 +302,7 @@
 ## 9. 验收标准

 - 命令可运行；tests/run_tests.py -v 通过
- task_logs 符合 SCHEMA
+- 业务表（若有）符合 SCHEMA；过程看运行日志，进度看任务中心
 - `developer_ids` 已配置为「你的用户ID」（匠厂设置中复制的正整数）
 - 正式环境安装验证通过；已声明的宿主入口（新建任务 / 数据管理 / 定时任务 / 任务中心 / 技能详情等）按 DEVELOPMENT §15 自测
 ## 10. 开发注意事项
--- a/development/RPA.md
+++ b/development/RPA.md
@@ -421,13 +421,13 @@ skill 退出/抛错统一用 `ERROR:` 前缀 + 稳定码，方便宿主与上层
 - **skill-template 默认 `OPENCLAW_RECORD_VIDEO=0`**（开发调试更快，不启 ffmpeg）。
 - **`OPENCLAW_RECORD_VIDEO=1` 时启用录屏**：字幕、旁白、背景音、MP4 成片由 platform-kit 统一处理。
 - 使用 platform-kit 的 **`RpaVideoSession`**；**skill 不要自行合成视频**（不要自己调 ffmpeg 拼 MP4）。
- **默认关闭 ≠ 可以删除录屏接入**：RPA / 长任务 **service 层**（`scripts/service/*.py`）**必须保留** `RpaVideoSession`、`video.add_step`、video artifact merge 到 `result_summary`（见 `POLICY-RPA-003`）。
+- **默认关闭 ≠ 可以删除录屏接入**：RPA / 长任务 **service 层**（`scripts/service/*.py`）**必须保留** `RpaVideoSession`、`video.add_step`、video artifact merge 进 `finish()`（见 `POLICY-RPA-003`）。
 - `OPENCLAW_RECORD_VIDEO=0` 时 session **无副作用**（不启 ffmpeg、不写字幕文件、不生成最终 MP4），但 **`RpaVideoSession.add_step` 仍可同步 Activity 进度**。
 - 生产留证 / 合规 / 客户现场复现时，由用户在 `.env` 或进程环境变量中开启 `OPENCLAW_RECORD_VIDEO=1`。
 - **ffmpeg 是唯一录屏器**（Windows：`gdigrab` + `desktop`）。
 - **最终视频**（`OPENCLAW_RECORD_VIDEO=1` 时）：`{skill_data_dir}/videos/{skill_slug}_{yyyyMMdd_HHmmss}_{batch_id}.mp4`
 - **中间产物**：`rpa-artifacts/{batch_id}/capture.mp4`、`subtitles/`、`logs/` 等。
- 任务完成后 CLI / `result_summary` 应包含 video 字段（开启录屏时有路径；关闭时仍保留 `video` 结构与诊断占位）：`video_path`、`raw_video`、`video_log`、`video_warnings`、`music_path`、`voiceover_path`、`audio_warnings`（见 `scripts/service/task_run_support.py`）。
+- 任务完成后 CLI / `finish()` 应包含 video 字段（开启录屏时有路径；关闭时仍保留 `video` 结构与诊断占位）：`video_path`、`raw_video`、`video_log`、`video_warnings`、`music_path`、`voiceover_path`、`audio_warnings`（见 `scripts/service/task_run_support.py`）。

 模板最小示范见 `scripts/service/task_service.py` 的 `_run_template_demo()`。

--- a/development/TESTING.md
+++ b/development/TESTING.md
@@ -22,12 +22,12 @@

 必跑套件要像一个紧张的守门员：**快、确定、离线**。典型覆盖：

- CLI：导入 [`cli.app`](../scripts/cli/app.py) 走解析链路、`health`（runtime diagnostics）/ `version` / `init-db` / `logs` / `log-get` 冒烟；
+- CLI：导入 [`cli.app`](../scripts/cli/app.py) 走解析链路、`health`（runtime diagnostics）/ `version` / `init-db` 冒烟；
 - 架构守护：无 `scripts/jiangchang_skill_core/`、`platform-kit>=1.2.0` 导入来源、文档/runtime 标准（见 `test_platform_import.py` 等）；
 - **真实 subprocess**：[`tests/test_entrypoint_subprocess.py`](../tests/test_entrypoint_subprocess.py) 再调用一遍 `python scripts/main.py`，防路径漂移；
 - 运行时：`runtime_paths` 与 **`JIANGCHANG_*` 隔离**；
 - `SKILL.md` YAML slug vs [`constants.SKILL_SLUG`](../scripts/util/constants.py)；
- SQLite 骨架：`task_logs` 创建幂等与仓储读写；
+- SQLite 骨架：`init_db` 幂等创建元数据表，默认不建业务表；
 - 数据管理元数据：`_jiangchang_*` 幂等初始化、中文展示名、`PRAGMA table_info` 字段顺序、editable 权限（见 `test_display_metadata.py`、`test_data_management_contract.py`）；
 - Skill Action manifest：`assets/actions.json` 结构、CLI 映射、placements 与 inputSchema 守护（见 `test_actions_manifest.py`）；
 - 标准时间字段：`created_at` / `updated_at` 默认值、trigger 维护与 `datetime_unix_seconds` 元数据（见 `test_timestamp_columns.py`）；
@@ -198,7 +198,7 @@ Golden fixture 流程同理（[`tests/samples/test_golden_cases.py.sample`](../t
 [`tests/test_development_policy_guard.py`](../tests/test_development_policy_guard.py) 中的 `POLICY-LOGGING-*` 规则会检查 logging **hard policy**（见 [`POLICY_MATRIX.md`](POLICY_MATRIX.md)）：

 - CLI 是否调用 `setup_skill_logging`（`POLICY-LOGGING-001`）
- `task_service.py` 是否使用 `get_skill_logger`、含 `log.info` / `log.exception`、写入 `save_task_log`（`POLICY-LOGGING-002` / `003`）
+- `task_service.py` 是否使用 `get_skill_logger`、含 `log.info` / `log.exception`（`POLICY-LOGGING-002` / `003`）
 - 长任务模板是否使用 Activity 或 RPA video step（`POLICY-LOGGING-004`）
 - 交付代码是否在 logging 调用中拼接明显敏感字段（`POLICY-LOGGING-005`）
 - **域模块**（`scripts/service/**`，豁免薄工具）是否有文件日志调用（`POLICY-LOGGING-006`）
@@ -242,7 +242,7 @@ Golden fixture 流程同理（[`tests/samples/test_golden_cases.py.sample`](../t
 ## 12. RPA / video 测试标准

 - 模板 `.env.example` 默认 **`OPENCLAW_RECORD_VIDEO=0`**；默认必跑测试**不应启动真实 ffmpeg**（patch session 或显式设 `OPENCLAW_RECORD_VIDEO=0`）。
- **但测试必须守护录屏接入能力**（`POLICY-RPA-003` 扫描 `scripts/service/*.py`）：`RpaVideoSession` 被构造、`video.add_step` 被调用、video artifact 结构 merge 进 `result_summary`（`video` / `video_path` 等字段），与默认是否录屏无关。
+- **但测试必须守护录屏接入能力**（`POLICY-RPA-003` 扫描 `scripts/service/*.py`）：`RpaVideoSession` 被构造、`video.add_step` 被调用、video artifact 结构 merge 进 `finish()`（`video` / `video_path` 等字段），与默认是否录屏无关。
 - 断言 `title` / `closing_title` 是**中文**业务文案。
 - 断言 step 文案贴近用户动作，不是技术日志（如「准备执行示例任务」而非「enter cmd_run」）。
 - 若需测试**真实录屏**（ffmpeg 成片），必须显式设 `OPENCLAW_RECORD_VIDEO=1`，并放在 `tests/integration/` 或手动触发场景，**不得**进入默认 unittest 套件。
--- a/examples/real_browser_rpa/README.md
+++ b/examples/real_browser_rpa/README.md
@@ -62,7 +62,7 @@ real_browser_rpa/
 4. 修改 `util/constants.py` 中的 `TARGET_PLATFORM`、`DEFAULT_START_URL`、selector 常量
 5. 替换 `task_rpa.py` 中的登录态 marker、搜索框/结果解析逻辑（**必须 F12 实测**）
 6. 替换 `human_verification.py` 中的验证码 selector（若目标站点 DOM 不同）
-7. 在 `task_service.py` 中接入真实 DB、任务日志、RpaVideoSession
+7. 在 `task_service.py` 中接入真实 DB、RpaVideoSession
 8. 在新技能根目录补齐 `scripts/main.py` CLI；`SKILL.md` 与根 `README.md` 按模板主目录规则编写，**不要**复制本 README

 ## Copy map（复制到新技能 `scripts/`）
@@ -116,7 +116,6 @@ real_browser_rpa/
 - `_parse_item_locator()` 业务字段解析
 - 用户可见中文文案
 - 数据库写入逻辑
- 任务日志逻辑
 - 录屏/截图业务标题

 ### 禁止照抄的
@@ -175,7 +174,7 @@ real_browser_rpa/
 | `human_verification.py` | `SLIDER_SELECTORS`、`SMS_SELECTORS`、`DIALOG_TEXT_CONTAINERS` |
 | `task_rpa.py` | `LOGGED_IN_SELECTORS`、`LOGGED_OUT_SELECTOR`、`_parse_item_locator()` |
 | `account_client.py` | `PLACEHOLDER_PLATFORM`、`LEASE_HOLDER`、账号 setup 文案 |
-| `task_service.py` | 接入 DB、任务日志、entitlement、视频留痕 |
+| `task_service.py` | 接入 DB、entitlement、视频留痕 |

 ## 依赖说明

--- a/examples/real_browser_rpa/scripts/service/task_service.py
+++ b/examples/real_browser_rpa/scripts/service/task_service.py
@@ -74,7 +74,6 @@ async def run_keyword_search_task(

    真实 skill 中还应：
    - 写 SQLite 入库（items 表）
-    - 写 scrape_tasks / task_logs 任务日志
    - 使用 RpaVideoSession 录制步骤留痕
    - 失败时 capture_failure 截图
    """
--- a/examples/simulator_browser_rpa/scripts/service/task_service.py
+++ b/examples/simulator_browser_rpa/scripts/service/task_service.py
@@ -51,7 +51,7 @@ async def run_batch_submit(
    """
    最小编排示例。

-    真实 skill 中还应：写 task_logs、entitlement、RpaVideoSession 等。
+    真实 skill 中还应：写业务表、entitlement、RpaVideoSession 等。
    账号 pick / lease 在 adapter 内完成，编排层不直接碰 account-manager。
    """
    del force  # 示例占位；真实 skill 可用于跳过确认
--- a/references/CLI.md
+++ b/references/CLI.md
@@ -24,7 +24,7 @@ python {baseDir}/scripts/main.py <command> …

 | 类型 | 本技能命令（自填） | Agent |
 |------|--------------------|-------|
-| **短 CLI 白名单** | （例）`health` / `version` / `logs` / `render-html` / `<your-read-or-prep>` | 可用 bash/exec + 共享 `python` |
+| **短 CLI 白名单** | （例）`health` / `version` / `render-html` / `<your-read-or-prep>` | 可用 bash/exec + 共享 `python` |
 | **Action 强制黑名单** | （例）`<your-long-or-rpa-command>` | **禁止** bash + `process poll`；必须 `run_skill_action` |

 细节与 Action 清单见 [`ACTIONS.md`](ACTIONS.md)。若提供 HTML 报告，「重新生成」须有 Action id **`render-html`**（可同时进短 CLI 白名单）；见 ACTIONS.md「HTML 页面报告」。
@@ -48,13 +48,12 @@ python {baseDir}/scripts/main.py init-db
 - **`policy`（可选）**：仅当 `actions.json` 顶层启用 Data Policy 时需要；向 stdout 输出 Snapshot JSON（见 `references/POLICY.md`）。模板默认不实现；多数技能保持 L0 即可。
 - **`run` / RPA / 长任务**：CLI 仍是同一内核；**Agent 禁止**用 bash 干等长任务结束。须走宿主 Skill Action（`run_skill_action`），且该 Action 一般为 `executionProfile: "async"` → **任务中心**。长时间无 stdout **不代表卡死**；进度看任务中心 / Run Journal。
 - **排查顺序**（由快到慢）：
-  1. `python {baseDir}/scripts/main.py logs`
-  2. `python {baseDir}/scripts/main.py log-get <log_id>`
-  3. 统一日志文件 `{JIANGCHANG_DATA_ROOT}/{JIANGCHANG_USER_ID}/logs/jiangchang.log`（含 `trace_id`、`task_failed` 等）
-  4. Run Journal `{JIANGCHANG_DATA_ROOT}/.jiangchang/runs/{job_id}.jsonl`
-  5. RPA artifacts / `video_log` / 失败截图（开发规范见 `development/RPA.md`）；未开启录屏时可能只有截图或日志
+  1. 任务中心 / Run Journal `{JIANGCHANG_DATA_ROOT}/.jiangchang/runs/{job_id}.jsonl`
+  2. 统一日志文件 `{JIANGCHANG_DATA_ROOT}/{JIANGCHANG_USER_ID}/logs/jiangchang.log`（含 `trace_id`、`task_failed` 等）
+  3. 业务表（宿主数据管理；有才建）
+  4. RPA artifacts / `video_log` / 失败截图（开发规范见 `development/RPA.md`）；未开启录屏时可能只有截图或日志
 - **录屏默认关闭**：模板 `.env.example` 为 `OPENCLAW_RECORD_VIDEO=0`；若未开启录屏，任务**可能不会**打印录屏路径。
- 需要录屏时在用户 `.env` 或进程环境变量中设置 **`OPENCLAW_RECORD_VIDEO=1`**；开启后任务完成时 CLI 应打印录屏路径、录屏日志、视频/音频诊断（见 `task_run_support._print_video_summary`），且 `result_summary` 含 video 字段。
+- 需要录屏时在用户 `.env` 或进程环境变量中设置 **`OPENCLAW_RECORD_VIDEO=1`**；开启后任务完成时 CLI 应打印录屏路径、录屏日志、视频/音频诊断（见 `task_run_support._print_video_summary`），且 `finish()` 含 video 字段。

 ## config-path（配置路径）

@@ -82,12 +81,8 @@ python {baseDir}/scripts/main.py init-db
 ```bash
 python {baseDir}/scripts/main.py run
 python {baseDir}/scripts/main.py run -t <target> -i <input_id>
-python {baseDir}/scripts/main.py logs
-python {baseDir}/scripts/main.py logs --task-type demo --status failed
-python {baseDir}/scripts/main.py log-get <log_id>
 ```

 | 命令 | 用途 | Agent 注意 |
 |------|------|------------|
 | `run` | 执行一次任务（常为长任务 / RPA） | **走 `run_skill_action`**，不要 bash poll |
-| `logs` / `log-get` | 查任务日志 | 可用短 CLI |
--- a/references/SCHEMA.md
+++ b/references/SCHEMA.md
@@ -66,7 +66,7 @@ metadata:
 ```

 - 数据库文件：英文 slug，例如 `your-skill-slug.db`（默认 `{skill-slug}.db`）。
- 业务表、字段：**英文 snake_case** 物理名，例如 `demo_items`、`task_logs`、`created_at`。
+- 业务表、字段：**英文 snake_case** 物理名，例如 `demo_items`、`created_at`。
 - **禁止**用中文作为 SQLite 真实表名或字段名。

 ## 外部来源只读与本地数据可写
@@ -442,7 +442,7 @@ readonly = 0

 - **created_at**：`INTEGER NOT NULL DEFAULT (unixepoch())`；插入时可省略，由数据库生成。
 - **updated_at**：`INTEGER NOT NULL DEFAULT (unixepoch())`；插入时同样可省略。
- **updated_at 更新**：模板默认表 `task_logs` 通过 SQLite trigger `{table}_set_updated_at` 在业务字段更新后自动刷新。
+- **updated_at 更新**：业务表通过 SQLite trigger `{table}_set_updated_at` 在业务字段更新后自动刷新（`init_updated_at_trigger`）。

 实现参考：

@@ -450,51 +450,18 @@ readonly = 0
 - 中文元数据种子：`scripts/db/display_metadata.py`
 - 自检：`scripts/db/display_metadata_validator.py`

-### 模板默认：`task_logs`
+### 系统自动写入表的只读约定

-`task_logs` 是技能内部**系统自动写入**的任务审计日志（由 `task_logs_repository` INSERT/UPDATE），不是供用户在宿主数据管理中手工维护的业务表。
-
-权限设计：
+技能内部**系统自动写入**、不供用户在宿主数据管理中手工维护的表，应声明：

 | 层级 | 值 | 说明 |
 |------|-----|------|
 | 表级 `readonly` | `1` | 宿主数据管理只读展示，不提供新增/编辑/删除表单 |
-| 所有业务字段 `editable` | `0` | 含 `task_type`、`status`、`error_msg`、`result_summary` 等 |
-| 技能内部 repository | 正常 INSERT/UPDATE | 元数据权限只约束宿主 UI，不限制技能代码写库 |
+| 相关业务字段 `editable` | `0` | 元数据权限只约束宿主 UI，不限制技能代码写库 |

-物理定义（权威顺序）：
+模板**默认不建**此类审计表。过程看 `jiangchang.log`，进度/终态看任务中心，业务结果看业务表。

-```sql
-CREATE TABLE IF NOT EXISTS task_logs (
-    id INTEGER PRIMARY KEY AUTOINCREMENT,
-    task_type TEXT NOT NULL,
-    target_id TEXT,
-    input_id TEXT,
-    input_title TEXT,
-    status TEXT NOT NULL,
-    error_msg TEXT,
-    result_summary TEXT,
-    created_at INTEGER NOT NULL DEFAULT (unixepoch()),
-    updated_at INTEGER NOT NULL DEFAULT (unixepoch())
-);
-```
-
-对应中文展示（由 `init_db()` 写入元数据表）：
-
-| 物理字段 | 中文名 | editable |
-|----------|--------|----------|
-| id | 编号 | 0 |
-| task_type | 任务类型 | 0 |
-| target_id | 目标编号 | 0 |
-| input_id | 输入编号 | 0 |
-| input_title | 输入标题 | 0 |
-| status | 状态 | 0 |
-| error_msg | 错误信息 | 0 |
-| result_summary | 结果摘要 | 0 |
-| created_at | 创建时间 | 0 |
-| updated_at | 更新时间 | 0 |
-
-表中文名：**任务日志**；表级 `readonly = 1`（宿主只读；技能 repository 仍可写入）。
+表级 `readonly = 1` 时，宿主数据管理只读展示；技能 repository 仍可写入。

 ### 通用示例：`demo_items`（测试 fixture 参考）

@@ -617,7 +584,7 @@ seed_column_metadata_batch(

 ## 中文命名质量

- 使用普通用户能理解的业务用语：`task_logs` → **任务日志**，`error_msg` → **错误信息**。
+- 使用普通用户能理解的业务用语：`error_msg` → **错误信息**，`created_at` → **创建时间**。
 - 禁止：`Task Logs`、`input编号`、`errormsg`、直接把物理名当 `display_name`。
 - 不宜向用户暴露的实现细节（如 `payload_json`、`selector`）应设 `visible = 0` 或给出可理解中文名。

@@ -629,17 +596,6 @@ seed_column_metadata_batch(
 4. 初始化脚本必须幂等。
 5. 元数据中的表、字段必须在库中真实存在；可见项不得缺少中文 `display_name`。

-## 不同业务的 `task_logs` 映射
-
-| 业务场景 | task_type | target_id | input_id |
-|----------|-----------|-----------|----------|
-| 发布类 | publish | 账号 ID | 内容 ID |
-| 工资代发 | disburse | 付款账户 | 批次 ID |
-| 对账 | reconcile | 平台标识 | 对账批次 ID |
-| 发票验真 | verify | 税务地区 | 发票批次 ID |
-
-业务特有字段优先放入 `result_summary`（JSON 字符串），避免随意加列；若确需新列，必须同步元数据并保持合理物理顺序。
-
 ## 开发自检

 - `init_db()` 后运行 `tests/test_display_metadata.py` 与 `tests/test_data_management_contract.py`。
@@ -649,5 +605,5 @@ seed_column_metadata_batch(
 ## 模板原则

 - 模板不做复杂历史迁移框架；新 skill 从当前 schema 起步。
- 元数据 DDL 与 `task_logs` 中文种子以 `scripts/db/display_metadata.py` 为单一权威来源。
+- 元数据 DDL 与种子 helper 以 `scripts/db/display_metadata.py` 为单一权威来源。
 - 模板定义「如何声明能力」；具体技能定义「声明什么业务能力」；宿主负责「如何通用渲染和执行」。
--- a/scripts/cli/app.py
+++ b/scripts/cli/app.py
@@ -13,8 +13,6 @@ from service.task_service import (
    cmd_config_path,
    cmd_health,
    cmd_init_db,
-    cmd_log_get,
-    cmd_logs,
    cmd_run,
    cmd_version,
 )
@@ -58,8 +56,6 @@ def _handle_run(args: argparse.Namespace) -> int:
 def _print_full_usage() -> None:
    print("通用业务技能模板（main.py）可用命令：")
    print("  python main.py run [target [input_id]] [-t target] [-i input_id]")
-    print("  python main.py logs [--limit N] [--status s] [--task-type t] [--target-id tid]")
-    print("  python main.py log-get <log_id>")
    print("  python main.py health")
    print("  python main.py config-path")
    print("  python main.py version")
@@ -69,7 +65,7 @@ def _print_full_usage() -> None:
 def build_parser() -> ZhArgumentParser:
    p = ZhArgumentParser(
        prog="main.py",
-        description="通用业务技能模板：任务执行命令骨架、日志查询、健康检查、版本输出。",
+        description="通用业务技能模板：任务执行命令骨架、健康检查、版本输出。",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    sub = p.add_subparsers(dest="cmd", required=True, parser_class=ZhArgumentParser)
@@ -80,19 +76,6 @@ def build_parser() -> ZhArgumentParser:
    sp.add_argument("run_tail", nargs="*", metavar="位置参数")
    sp.set_defaults(handler=_handle_run)

-    sp = sub.add_parser("logs", help="查看任务日志")
-    sp.add_argument("--limit", type=int, default=10)
-    sp.add_argument("--status", default=None)
-    sp.add_argument("--task-type", default=None, dest="task_type")
-    sp.add_argument("--target-id", default=None, dest="target_id")
-    sp.set_defaults(handler=lambda a: cmd_logs(
-        limit=a.limit, status=a.status, task_type=a.task_type, target_id=a.target_id
-    ))
-
-    sp = sub.add_parser("log-get", help="按 log_id 查看单条任务日志(JSON)")
-    sp.add_argument("log_id")
-    sp.set_defaults(handler=lambda a: cmd_log_get(a.log_id))
-
    sp = sub.add_parser("health", help="健康检查")
    sp.set_defaults(handler=lambda _a: cmd_health())

@@ -116,7 +99,7 @@ def main(argv: Optional[List[str]] = None) -> int:
        _print_full_usage()
        return 1
    if len(argv) == 2 and argv[0] not in {
-        "run", "logs", "log-get", "health", "config-path", "version", "init-db",
+        "run", "health", "config-path", "version", "init-db",
        "-h", "--help",
    }:
        return cmd_run(target=argv[0], input_id=argv[1])
--- a/scripts/db/connection.py
+++ b/scripts/db/connection.py
@@ -1,11 +1,13 @@
-"""SQLite 连接与任务日志表迁移模板。"""
+"""SQLite 连接与幂等初始化模板。
+
+模板默认只建数据管理元数据表。业务表由复制后的技能自行 CREATE，并同步展示元数据。
+"""

 from __future__ import annotations

 import sqlite3

 from db.display_metadata import init_display_metadata
-from db.timestamp_columns import init_task_logs_timestamp_maintenance
 from util.runtime_paths import get_db_path


@@ -17,23 +19,6 @@ def init_db() -> None:
    conn = get_conn()
    try:
        cur = conn.cursor()
-        cur.execute(
-            """
-            CREATE TABLE IF NOT EXISTS task_logs (
-                id INTEGER PRIMARY KEY AUTOINCREMENT,
-                task_type TEXT NOT NULL,
-                target_id TEXT,
-                input_id TEXT,
-                input_title TEXT,
-                status TEXT NOT NULL,
-                error_msg TEXT,
-                result_summary TEXT,
-                created_at INTEGER NOT NULL DEFAULT (unixepoch()),
-                updated_at INTEGER NOT NULL DEFAULT (unixepoch())
-            )
-            """
-        )
-        init_task_logs_timestamp_maintenance(cur)
        init_display_metadata(cur)
        conn.commit()
    finally:
--- a/scripts/db/display_metadata.py
+++ b/scripts/db/display_metadata.py
@@ -13,7 +13,6 @@ INTERNAL_METADATA_TABLES = frozenset({"_jiangchang_tables", "_jiangchang_columns
 LEGACY_METADATA_TABLES = frozenset({"_schema_meta"})
 METADATA_TABLES = INTERNAL_METADATA_TABLES | LEGACY_METADATA_TABLES

-TASK_LOGS_TABLE = "task_logs"
 DATETIME_UNIX_SECONDS = "datetime_unix_seconds"
 STANDARD_TIMESTAMP_COLUMNS = frozenset({"created_at", "updated_at"})

@@ -46,44 +45,6 @@ CREATE TABLE IF NOT EXISTS _jiangchang_columns (
 )
 """

-TASK_LOGS_TABLE_METADATA = {
-    "table_name": TASK_LOGS_TABLE,
-    "display_name": "任务日志",
-    "description": "记录技能任务的执行状态和结果（系统自动写入，宿主数据管理只读展示）",
-    "visible": 1,
-    "readonly": 1,
-}
-
-TASK_LOGS_COLUMN_METADATA: tuple[dict[str, object], ...] = (
-    {"column_name": "id", "display_name": "编号", "description": "任务日志唯一编号", "visible": 1, "searchable": 1, "editable": 0},
-    {"column_name": "task_type", "display_name": "任务类型", "description": "任务的业务类型", "visible": 1, "searchable": 1, "editable": 0},
-    {"column_name": "target_id", "display_name": "目标编号", "description": "任务目标对象编号", "visible": 1, "searchable": 1, "editable": 0},
-    {"column_name": "input_id", "display_name": "输入编号", "description": "输入对象编号", "visible": 1, "searchable": 1, "editable": 0},
-    {"column_name": "input_title", "display_name": "输入标题", "description": "输入对象标题", "visible": 1, "searchable": 1, "editable": 0},
-    {"column_name": "status", "display_name": "状态", "description": "任务当前状态", "visible": 1, "searchable": 1, "editable": 0},
-    {"column_name": "error_msg", "display_name": "错误信息", "description": "任务失败时的错误说明", "visible": 1, "searchable": 1, "editable": 0},
-    {"column_name": "result_summary", "display_name": "结果摘要", "description": "任务执行结果摘要", "visible": 1, "searchable": 1, "editable": 0},
-    {
-        "column_name": "created_at",
-        "display_name": "创建时间",
-        "description": "记录创建时间（Unix 秒级时间戳，数据库自动生成）",
-        "visible": 1,
-        "searchable": 0,
-        "editable": 0,
-        "display_type": DATETIME_UNIX_SECONDS,
-    },
-    {
-        "column_name": "updated_at",
-        "display_name": "更新时间",
-        "description": "记录最后更新时间（Unix 秒级时间戳，自动维护）",
-        "visible": 1,
-        "searchable": 0,
-        "editable": 0,
-        "display_type": DATETIME_UNIX_SECONDS,
-    },
-)
-
-
 def _ensure_jiangchang_tables_columns(cursor) -> None:
    """旧库补列：CREATE IF NOT EXISTS 不会给已有表加新字段。"""
    cursor.execute("PRAGMA table_info('_jiangchang_tables')")
@@ -195,24 +156,6 @@ def upsert_column_metadata(
    )


-def seed_task_logs_display_metadata(cursor) -> None:
-    meta = TASK_LOGS_TABLE_METADATA
-    upsert_table_metadata(
-        cursor,
-        table_name=str(meta["table_name"]),
-        display_name=str(meta["display_name"]),
-        description=str(meta["description"]) if meta.get("description") else None,
-        visible=int(meta["visible"]),
-        readonly=int(meta["readonly"]),
-    )
-    seed_column_metadata_batch(
-        cursor,
-        table_name=TASK_LOGS_TABLE,
-        columns=TASK_LOGS_COLUMN_METADATA,
-        valid_column_names=[str(column["column_name"]) for column in TASK_LOGS_COLUMN_METADATA],
-    )
-
-
 def seed_column_metadata_batch(
    cursor,
    *,
@@ -307,7 +250,6 @@ def table_allows_manual_create(cursor, table_name: str) -> bool:

 def init_display_metadata(cursor) -> None:
    init_display_metadata_tables(cursor)
-    seed_task_logs_display_metadata(cursor)


 def iter_user_tables(cursor) -> Iterable[str]:
--- a/scripts/db/display_metadata_validator.py
+++ b/scripts/db/display_metadata_validator.py
@@ -12,8 +12,6 @@ from db.display_metadata import (
    DATETIME_UNIX_SECONDS,
    METADATA_TABLES,
    STANDARD_TIMESTAMP_COLUMNS,
-    TASK_LOGS_COLUMN_METADATA,
-    TASK_LOGS_TABLE,
    iter_user_tables,
    list_editable_business_columns,
    list_physical_column_names,
@@ -516,11 +514,5 @@ def validate_display_metadata(conn: sqlite3.Connection, *, check_column_order: b


 def validate_template_database(conn: sqlite3.Connection) -> ValidationReport:
-    report = validate_display_metadata(conn)
-    physical = list_physical_column_names(conn.cursor(), TASK_LOGS_TABLE)
-    expected = [str(item["column_name"]) for item in TASK_LOGS_COLUMN_METADATA]
-    if physical != expected:
-        report.errors.append(
-            f"task_logs 物理字段顺序应为 {expected}，当前为 {physical}"
-        )
-    return report
+    """模板自检入口：校验展示元数据完整性。模板默认无业务表。"""
+    return validate_display_metadata(conn)
--- a/scripts/db/task_logs_repository.py
+++ b/scripts/db/task_logs_repository.py
@@ -1,86 +0,0 @@
-"""task_logs 表读写模板。
-
-通用业务日志仓储：记录每次任务执行的 task_type / target / input / 状态 / 结果摘要。
-不假设任何特定业务（发布、对账、审核等都可复用）。
-"""
-
-from __future__ import annotations
-
-from typing import Any, List, Optional, Tuple
-
-from db.connection import get_conn, init_db
-
-
-def save_task_log(
-    task_type: str,
-    target_id: Optional[str] = None,
-    input_id: Optional[str] = None,
-    input_title: Optional[str] = None,
-    status: str = "success",
-    error_msg: Optional[str] = None,
-    result_summary: Optional[str] = None,
-) -> int:
-    init_db()
-    conn = get_conn()
-    try:
-        cur = conn.cursor()
-        cur.execute(
-            """
-            INSERT INTO task_logs
-                (task_type, target_id, input_id, input_title, status, error_msg, result_summary)
-            VALUES (?, ?, ?, ?, ?, ?, ?)
-            """,
-            (task_type, target_id, input_id, input_title or "", status, error_msg, result_summary),
-        )
-        new_id = int(cur.lastrowid)
-        conn.commit()
-    finally:
-        conn.close()
-    return new_id
-
-
-def list_task_logs(
-    limit: int,
-    status: Optional[str] = None,
-    task_type: Optional[str] = None,
-    target_id: Optional[str] = None,
-) -> List[Tuple[Any, ...]]:
-    init_db()
-    conn = get_conn()
-    try:
-        cur = conn.cursor()
-        sql = (
-            "SELECT id, task_type, target_id, input_id, input_title, status, error_msg, result_summary, "
-            "created_at, updated_at FROM task_logs WHERE 1=1 "
-        )
-        params: List[Any] = []
-        if status:
-            sql += "AND status = ? "
-            params.append(status)
-        if task_type:
-            sql += "AND task_type = ? "
-            params.append(task_type)
-        if target_id:
-            sql += "AND target_id = ? "
-            params.append(target_id)
-        sql += "ORDER BY created_at DESC, id DESC LIMIT ?"
-        params.append(int(limit))
-        cur.execute(sql, tuple(params))
-        return list(cur.fetchall())
-    finally:
-        conn.close()
-
-
-def get_task_log_by_id(log_id: int) -> Optional[Tuple[Any, ...]]:
-    init_db()
-    conn = get_conn()
-    try:
-        cur = conn.cursor()
-        cur.execute(
-            "SELECT id, task_type, target_id, input_id, input_title, status, error_msg, result_summary, "
-            "created_at, updated_at FROM task_logs WHERE id = ?",
-            (int(log_id),),
-        )
-        return cur.fetchone()
-    finally:
-        conn.close()
--- a/scripts/db/timestamp_columns.py
+++ b/scripts/db/timestamp_columns.py
@@ -1,30 +1,44 @@
-"""标准时间字段：Unix 秒级时间戳、默认值与 updated_at 自动维护。"""
+"""标准时间字段：Unix 秒级时间戳、默认值与 updated_at 自动维护。
+
+业务表若包含 created_at / updated_at，在 CREATE TABLE 后调用
+``init_updated_at_trigger(cursor, table_name)``。模板默认不建业务表。
+"""

 from __future__ import annotations

 STANDARD_TIMESTAMP_COLUMNS = frozenset({"created_at", "updated_at"})
 DATETIME_UNIX_SECONDS = "datetime_unix_seconds"
-TASK_LOGS_UPDATED_AT_TRIGGER = "task_logs_set_updated_at"

-_TASK_LOGS_UPDATED_AT_TRIGGER_DDL = f"""
-CREATE TRIGGER IF NOT EXISTS {TASK_LOGS_UPDATED_AT_TRIGGER}
-AFTER UPDATE ON task_logs
+
+def updated_at_trigger_name(table_name: str) -> str:
+    return f"{table_name}_set_updated_at"
+
+
+def _quote_identifier(name: str) -> str:
+    return '"' + name.replace('"', '""') + '"'
+
+
+def init_updated_at_trigger(cursor, table_name: str) -> None:
+    """幂等创建 {table}_set_updated_at：业务字段 UPDATE 后刷新 updated_at。"""
+    trigger = updated_at_trigger_name(table_name)
+    quoted = _quote_identifier(table_name)
+    quoted_trigger = _quote_identifier(trigger)
+    cursor.execute(
+        f"""
+CREATE TRIGGER IF NOT EXISTS {quoted_trigger}
+AFTER UPDATE ON {quoted}
 FOR EACH ROW
 BEGIN
-    UPDATE task_logs
+    UPDATE {quoted}
    SET updated_at = unixepoch()
    WHERE id = NEW.id AND updated_at = OLD.updated_at;
 END
 """
-
-
-def init_task_logs_timestamp_maintenance(cursor) -> None:
-    """幂等创建 task_logs.updated_at 自动维护 trigger。"""
-    cursor.execute(_TASK_LOGS_UPDATED_AT_TRIGGER_DDL)
+    )


 def has_updated_at_trigger(cursor, table_name: str) -> bool:
-    trigger_name = f"{table_name}_set_updated_at"
+    trigger_name = updated_at_trigger_name(table_name)
    cursor.execute(
        """
        SELECT 1 FROM sqlite_master
--- a/scripts/service/task_run_support.py
+++ b/scripts/service/task_run_support.py
@@ -10,7 +10,7 @@ def build_video_info(
    compose_context_before: Optional[dict[str, Any]] = None,
    compose_context_after: Optional[dict[str, Any]] = None,
 ) -> dict[str, Any]:
-    """将 RpaVideoSession.summary() 转为 task log / CLI 可用的 video 块。"""
+    """将 RpaVideoSession.summary() 转为 finish() / CLI 可用的 video 块。"""
    return {
        "enabled": video_summary.get("enabled", False),
        "path": video_summary.get("path"),
@@ -31,7 +31,7 @@ def merge_video_into_result_summary(
    payload: dict[str, Any],
    video_info: Optional[dict[str, Any]],
 ) -> dict[str, Any]:
-    """把 video artifact 字段写入 result_summary JSON 对象。"""
+    """把 video artifact 字段写入 finish() 结果对象（video / video_path 等）。"""
    if not video_info:
        return payload
    payload = dict(payload)
--- a/scripts/service/task_service.py
+++ b/scripts/service/task_service.py
@@ -1,6 +1,7 @@
-"""任务编排、日志查询模板。
+"""任务编排模板。

-编排层：job_context / emit / checkpoint / finish、鉴权、任务日志与错误转换。
+编排层：job_context / emit / checkpoint / finish、鉴权与错误转换。
+过程看统一文件日志；进度/终态看任务中心；业务结果写入业务表（有才建）。
 领域业务请放在 ``service/<domain>_service.py``，由本模块调用；多入口（CLI/Action/cron）复用同一内核。
 复制后在 cmd_run（或 cmd_sync_*）中接线到真实 domain service，勿按 toolbar/cron/agent 分叉业务。
 """
@@ -11,12 +12,11 @@ import asyncio
 import json
 import os
 import uuid
-from typing import Any, Dict, Optional
+from typing import Any, Optional

 from jiangchang_skill_core import collect_runtime_diagnostics, config, format_runtime_health_lines
 from jiangchang_skill_core.activity import JobStopped, emit, finish, job_context

-from db import task_logs_repository as tlr
 from jiangchang_skill_core.rpa.video_session import RpaVideoSession
 from service.entitlement_service import check_entitlement
 from service.task_run_support import (
@@ -27,7 +27,6 @@ from service.task_run_support import (
 from util.constants import LOG_LOGGER_NAME, PLATFORM_KIT_MIN_VERSION, SKILL_SLUG, SKILL_VERSION
 from util.logging_config import get_skill_logger, setup_skill_logging
 from util.runtime_paths import get_db_path, get_skill_data_dir, get_skill_root
-from util.timeutil import unix_to_iso
 from db.connection import init_db


@@ -84,21 +83,6 @@ def cmd_run(target: Optional[str] = None, input_id: Optional[str] = None) -> int
                    task_type,
                )
                emit("鉴权失败，已记录诊断信息", type="warn", skill=SKILL_SLUG, stage="auth")
-                tlr.save_task_log(
-                    task_type=task_type,
-                    target_id=target,
-                    input_id=input_id,
-                    input_title="模板示例任务",
-                    status="failed",
-                    error_msg=reason,
-                    result_summary=json.dumps({"stage": "auth", "error": reason}, ensure_ascii=False),
-                )
-                log.info(
-                    "task_log_saved task_type=%s target_id=%s input_id=%s stage=auth status=failed",
-                    task_type,
-                    target,
-                    input_id,
-                )
                finish(
                    status="failed",
                    message=reason,
@@ -129,21 +113,6 @@ def cmd_run(target: Optional[str] = None, input_id: Optional[str] = None) -> int
                },
                video_info,
            )
-            tlr.save_task_log(
-                task_type=task_type,
-                target_id=target,
-                input_id=input_id,
-                input_title="模板示例任务",
-                status="failed",
-                error_msg="模板仓库未实现真实业务",
-                result_summary=json.dumps(summary_payload, ensure_ascii=False),
-            )
-            log.info(
-                "task_log_saved task_type=%s target_id=%s input_id=%s status=failed",
-                task_type,
-                target,
-                input_id,
-            )

            log.warning(
                "task_failed task_type=%s stage=demo_finish error_code=TEMPLATE_STUB",
@@ -153,10 +122,8 @@ def cmd_run(target: Optional[str] = None, input_id: Optional[str] = None) -> int
                status="failed",
                message="模板仓库未实现真实业务",
                skill=SKILL_SLUG,
-                template_demo=True,
-                target=target,
-                input_id=input_id,
                error_code="TEMPLATE_STUB",
+                **summary_payload,
            )
            print("ERROR:TEMPLATE_STUB 这是模板仓库，请复制后在 scripts/service/task_service.py 中实现 cmd_run 的真实业务逻辑。")
            return rc
@@ -170,32 +137,6 @@ def cmd_run(target: Optional[str] = None, input_id: Optional[str] = None) -> int
                input_id,
            )
            emit("任务失败，已记录诊断信息", type="warn", skill=SKILL_SLUG, stage="run")
-            try:
-                tlr.save_task_log(
-                    task_type=task_type,
-                    target_id=target,
-                    input_id=input_id,
-                    input_title="模板示例任务",
-                    status="failed",
-                    error_msg="任务执行异常，详见统一日志",
-                    result_summary=json.dumps(
-                        {"stage": "run", "error": "unexpected_exception"},
-                        ensure_ascii=False,
-                    ),
-                )
-                log.info(
-                    "task_log_saved task_type=%s target_id=%s input_id=%s stage=run status=failed",
-                    task_type,
-                    target,
-                    input_id,
-                )
-            except Exception:
-                log.exception(
-                    "task_log_save_failed task_type=%s target_id=%s input_id=%s",
-                    task_type,
-                    target,
-                    input_id,
-                )
            finish(
                status="failed",
                message="任务执行异常，详见统一日志",
@@ -205,67 +146,6 @@ def cmd_run(target: Optional[str] = None, input_id: Optional[str] = None) -> int
            return 1


-def cmd_logs(
-    limit: int = 10,
-    status: Optional[str] = None,
-    task_type: Optional[str] = None,
-    target_id: Optional[str] = None,
-) -> int:
-    if limit <= 0:
-        limit = 10
-    rows = tlr.list_task_logs(limit, status, task_type, target_id)
-    if not rows:
-        print("暂无任务日志")
-        return 0
-
-    sep_line = "_" * 39
-    for idx, r in enumerate(rows):
-        rid, ttype, tid, iid, ititle, st, err, rsum, cat, uat = r
-        print(f"id：{rid}")
-        print(f"task_type：{ttype or ''}")
-        print(f"target_id：{tid or ''}")
-        print(f"input_id：{iid or ''}")
-        print(f"input_title：{ititle or ''}")
-        print(f"status：{st or ''}")
-        print(f"error_msg：{err or ''}")
-        print(f"result_summary：{rsum or ''}")
-        print(f"created_at：{unix_to_iso(cat) or str(cat or '')}")
-        print(f"updated_at：{unix_to_iso(uat) or str(uat or '')}")
-        if idx != len(rows) - 1:
-            print(sep_line)
-            print()
-    return 0
-
-
-def cmd_log_get(log_id: str) -> int:
-    if not str(log_id).isdigit():
-        print("❌ log_id 必须是数字")
-        return 1
-    row = tlr.get_task_log_by_id(int(log_id))
-    if not row:
-        print("❌ 没有这条任务日志")
-        return 1
-    rid, ttype, tid, iid, ititle, st, err, rsum, cat, uat = row
-    print(
-        json.dumps(
-            {
-                "id": int(rid),
-                "task_type": ttype,
-                "target_id": tid,
-                "input_id": iid,
-                "input_title": ititle,
-                "status": st,
-                "error_msg": err,
-                "result_summary": rsum,
-                "created_at": unix_to_iso(cat),
-                "updated_at": unix_to_iso(uat),
-            },
-            ensure_ascii=False,
-        )
-    )
-    return 0
-
-
 def cmd_config_path() -> int:
    example_path = os.path.join(get_skill_root(), ".env.example")
    env_path = config.get_env_file_path() or ""
--- a/scripts/util/constants.py
+++ b/scripts/util/constants.py
@@ -1,6 +1,6 @@
 """技能标识、版本与平台公共库约束（复制后请修改 slug/version/logger）。"""

 SKILL_SLUG = "your-skill-slug"
-SKILL_VERSION = "1.0.80"
+SKILL_VERSION = "1.0.81"
 LOG_LOGGER_NAME = "openclaw.skill.your_skill_slug"
 PLATFORM_KIT_MIN_VERSION = "1.2.11"
--- a/tests/README.md
+++ b/tests/README.md
@@ -75,7 +75,7 @@ python tests/run_tests.py test_cli_smoke
 | **subprocess 真实入口** ``python scripts/main.py`` | ``test_entrypoint_subprocess.py`` |
 | 运行路径与 **JIANGCHANG_*** 隔离 | ``test_runtime_paths.py`` |
 | ``SKILL.md`` slug 与 ``constants.SKILL_SLUG`` | ``test_skill_metadata.py`` |
-| DB / ``task_logs`` 骨架冒烟 | ``test_db_smoke.py`` |
+| DB / 元数据骨架冒烟 | ``test_db_smoke.py`` |
 | 数据管理元数据 / 字段顺序 | ``test_display_metadata.py`` |
 | 标准时间字段（Unix 秒 / trigger / 元数据） | ``test_timestamp_columns.py`` |
 | **adapter profile / 外呼授权策略** | ``test_adapter_profile_policy.py`` + ``adapter_test_utils.py``（非 ``test_`` 前缀，不自动当用例收集） |
--- a/tests/_support.py
+++ b/tests/_support.py
@@ -79,3 +79,63 @@ class IsolatedDataRoot:
        if self._tmp and os.path.isdir(self._tmp):
            shutil.rmtree(self._tmp, ignore_errors=True)
        self._tmp = None
+
+
+DEMO_ITEMS_TABLE = "demo_items"
+
+DEMO_ITEMS_DDL = """
+CREATE TABLE IF NOT EXISTS demo_items (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    title TEXT NOT NULL,
+    status TEXT NOT NULL DEFAULT 'draft',
+    created_at INTEGER NOT NULL DEFAULT (unixepoch()),
+    updated_at INTEGER NOT NULL DEFAULT (unixepoch())
+)
+"""
+
+DEMO_ITEMS_COLUMN_METADATA: tuple[dict[str, object], ...] = (
+    {"column_name": "id", "display_name": "编号", "description": "主键", "visible": 1, "searchable": 1, "editable": 0},
+    {"column_name": "title", "display_name": "标题", "description": "标题", "visible": 1, "searchable": 1, "editable": 1},
+    {"column_name": "status", "display_name": "状态", "description": "状态", "visible": 1, "searchable": 1, "editable": 1},
+    {
+        "column_name": "created_at",
+        "display_name": "创建时间",
+        "description": "记录创建时间（Unix 秒级时间戳，数据库自动生成）",
+        "visible": 1,
+        "searchable": 0,
+        "editable": 0,
+        "display_type": "datetime_unix_seconds",
+    },
+    {
+        "column_name": "updated_at",
+        "display_name": "更新时间",
+        "description": "记录最后更新时间（Unix 秒级时间戳，自动维护）",
+        "visible": 1,
+        "searchable": 0,
+        "editable": 0,
+        "display_type": "datetime_unix_seconds",
+    },
+)
+
+
+def seed_demo_items(cursor, *, readonly: int = 0) -> None:
+    """测试夹具：创建 demo_items 物理表、时间戳 trigger 与展示元数据。"""
+    from db.display_metadata import seed_column_metadata_batch, upsert_table_metadata
+    from db.timestamp_columns import init_updated_at_trigger
+
+    cursor.execute(DEMO_ITEMS_DDL)
+    init_updated_at_trigger(cursor, DEMO_ITEMS_TABLE)
+    upsert_table_metadata(
+        cursor,
+        table_name=DEMO_ITEMS_TABLE,
+        display_name="示例记录",
+        description="模板测试用业务表示例，不是默认交付表",
+        visible=1,
+        readonly=readonly,
+    )
+    seed_column_metadata_batch(
+        cursor,
+        table_name=DEMO_ITEMS_TABLE,
+        columns=DEMO_ITEMS_COLUMN_METADATA,
+        valid_column_names=[str(column["column_name"]) for column in DEMO_ITEMS_COLUMN_METADATA],
+    )
--- a/tests/test_actions_manifest.py
+++ b/tests/test_actions_manifest.py
@@ -100,7 +100,7 @@ def _load_bindable_business_tables() -> set[str]:
    - SQLite 物理表存在（排除 sqlite_* 与 _jiangchang_* 元数据表）
    - ``_jiangchang_tables`` 中有对应行且 ``visible=1``

-    失败时抛出 RuntimeError（带诊断），不得静默返回空集供测试 skip。
+    模板默认无业务表，允许返回空集。失败时抛出 RuntimeError（带诊断）。
    """
    from db.connection import init_db
    from db.display_metadata import METADATA_TABLES
@@ -136,12 +136,6 @@ def _load_bindable_business_tables() -> set[str]:
            finally:
                conn.close()
            bindable = physical & visible_meta
-            if not bindable:
-                raise RuntimeError(
-                    "isolated init_db produced no bindable tables "
-                    f"(physical={sorted(physical)}, visible_meta={sorted(visible_meta)}, "
-                    f"db_path={db_path!r})"
-                )
            return bindable
    except RuntimeError:
        raise
@@ -312,15 +306,14 @@ class TestActionsManifest(unittest.TestCase):
        self.assertEqual(errors, [], msg="\n".join(errors))

    def test_bindable_business_tables_helper_loads_template_db(self) -> None:
-        """helper 本身必须可用：即便默认 manifest 无 toolbar bind，也要验证隔离库可读。"""
+        """helper 本身必须可用：模板默认无业务表，隔离库仍应可读且不含元数据表。"""
        bindable = _load_bindable_business_tables()
-        self.assertIn("task_logs", bindable)
+        self.assertEqual(bindable, set())
        self.assertNotIn("_jiangchang_tables", bindable)
        self.assertNotIn("_jiangchang_columns", bindable)

    def test_bind_tables_exist_in_physical_and_visible_metadata(self) -> None:
        bindable = _load_bindable_business_tables()
-        self.assertTrue(bindable, msg="bindable table set must be non-empty after init_db")
        manifest = _load_actions_manifest()
        missing: list[str] = []
        for action in manifest["actions"]:
@@ -356,7 +349,7 @@ class TestActionsManifest(unittest.TestCase):
                    "entrypoint": {"type": "cli", "command": "health", "args": []},
                }
                if placement == "toolbar":
-                    action["bind"] = {"tables": ["task_logs"]}
+                    action["bind"] = {"tables": ["demo_items"]}
                manifest = {
                    "schemaVersion": 1,
                    "skill": "your-skill-slug",
--- a/tests/test_cli_smoke.py
+++ b/tests/test_cli_smoke.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-"""通用 CLI 冒烟：用法、health、version、logs、log-get（不触网、不深测 run 占位）。"""
+"""通用 CLI 冒烟：用法、health、version、init-db（不触网、不深测 run 占位）。"""
 from __future__ import annotations

 import io
@@ -78,20 +78,13 @@ class TestCliSmoke(unittest.TestCase):
            self.assertTrue(payload2["ok"])
            self.assertEqual(payload2["db_path"], payload["db_path"])

-    def test_logs_empty_returns_zero(self) -> None:
-        with IsolatedDataRoot():
-            buf = io.StringIO()
-            with redirect_stdout(buf), redirect_stderr(io.StringIO()):
-                rc = main(["logs"])
-            self.assertEqual(rc, 0)
-            self.assertIn("暂无", buf.getvalue())
-
-    def test_log_get_non_numeric_returns_nonzero(self) -> None:
+    def test_unknown_logs_subcommand_is_rejected(self) -> None:
        buf = io.StringIO()
-        with redirect_stdout(buf), redirect_stderr(io.StringIO()):
-            rc = main(["log-get", "not-a-number"])
-        self.assertNotEqual(rc, 0)
-        self.assertIn("数字", buf.getvalue())
+        err = io.StringIO()
+        with redirect_stdout(buf), redirect_stderr(err):
+            with self.assertRaises(SystemExit) as caught:
+                main(["logs"])
+        self.assertNotEqual(caught.exception.code, 0)

    def test_run_placeholder_returns_nonzero_without_network(self) -> None:
        """占位 run：RpaVideoSession 示范 + 模板提示；OPENCLAW_RECORD_VIDEO=0 时不启 ffmpeg。"""
--- a/tests/test_db_smoke.py
+++ b/tests/test_db_smoke.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-"""task_logs 表：init 幂等、写入与查询（全程隔离数据目录）。"""
+"""init_db：幂等创建元数据表，默认不建业务表。"""
 from __future__ import annotations

 import unittest
@@ -8,44 +8,30 @@ from _support import IsolatedDataRoot


 class TestDbSmoke(unittest.TestCase):
-    def test_init_db_idempotent_and_crud_smoke(self) -> None:
+    def test_init_db_idempotent_metadata_only(self) -> None:
        with IsolatedDataRoot():
-            from db import task_logs_repository as tlr
            from db.connection import get_conn, init_db
+            from db.display_metadata import METADATA_TABLES, iter_user_tables

            init_db()
-            init_db()  # 幂等
+            init_db()

            conn = get_conn()
            try:
                cur = conn.cursor()
-                cur.execute("SELECT display_name FROM _jiangchang_tables WHERE table_name = 'task_logs'")
-                self.assertEqual(cur.fetchone()[0], "任务日志")
+                cur.execute(
+                    "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'"
+                )
+                tables = {str(row[0]) for row in cur.fetchall()}
+                self.assertIn("_jiangchang_tables", tables)
+                self.assertIn("_jiangchang_columns", tables)
+                self.assertNotIn("task_logs", tables)
+                self.assertEqual(list(iter_user_tables(cur)), [])
+                leftover = tables - METADATA_TABLES
+                self.assertEqual(leftover, set())
            finally:
                conn.close()

-            new_id = tlr.save_task_log(
-                task_type="demo_task",
-                target_id="tgt-1",
-                input_id="42",
-                input_title="t",
-                status="success",
-                error_msg=None,
-                result_summary=None,
-            )
-            self.assertIsInstance(new_id, int)
-            self.assertGreater(new_id, 0)
-
-            rows = tlr.list_task_logs(limit=10)
-            self.assertTrue(any(r[0] == new_id for r in rows))
-
-            row = tlr.get_task_log_by_id(new_id)
-            self.assertIsNotNone(row)
-            self.assertEqual(int(row[0]), new_id)
-
-            missing = tlr.get_task_log_by_id(999_999)
-            self.assertIsNone(missing)
-

 if __name__ == "__main__":
    unittest.main()
--- a/tests/test_development_policy_guard.py
+++ b/tests/test_development_policy_guard.py
@@ -876,7 +876,7 @@ class TestPolicyLogging002(unittest.TestCase):


 class TestPolicyLogging003(unittest.TestCase):
-    def test_task_service_logs_start_failure_and_task_logs(self) -> None:
+    def test_task_service_logs_start_and_failure(self) -> None:
        skill_root = get_skill_root()
        path = os.path.join(skill_root, "scripts", "service", "task_service.py")
        text = _read_text(path)
@@ -886,8 +886,6 @@ class TestPolicyLogging003(unittest.TestCase):
            missing.append("log.info or logger.info")
        if not re.search(r"\b(log|logger)\.exception\s*\(", text):
            missing.append("log.exception or logger.exception")
-        if "save_task_log" not in text:
-            missing.append("save_task_log")
        self.assertEqual(
            missing,
            [],
--- a/tests/test_display_metadata.py
+++ b/tests/test_display_metadata.py
@@ -6,7 +6,13 @@ import os
 import sqlite3
 import unittest

-from _support import IsolatedDataRoot, get_skill_root
+from _support import (
+    DEMO_ITEMS_COLUMN_METADATA,
+    DEMO_ITEMS_TABLE,
+    IsolatedDataRoot,
+    get_skill_root,
+    seed_demo_items,
+)


 class TestDisplayMetadata(unittest.TestCase):
@@ -20,14 +26,18 @@ class TestDisplayMetadata(unittest.TestCase):
            conn = get_conn()
            try:
                cur = conn.cursor()
-                cur.execute("SELECT COUNT(*) FROM _jiangchang_tables WHERE table_name = 'task_logs'")
-                self.assertEqual(int(cur.fetchone()[0]), 1)
-                cur.execute("SELECT COUNT(*) FROM _jiangchang_columns WHERE table_name = 'task_logs'")
-                self.assertEqual(int(cur.fetchone()[0]), 10)
+                cur.execute("SELECT COUNT(*) FROM _jiangchang_tables")
+                self.assertEqual(int(cur.fetchone()[0]), 0)
+                cur.execute("SELECT COUNT(*) FROM _jiangchang_columns")
+                self.assertEqual(int(cur.fetchone()[0]), 0)
+                cur.execute(
+                    "SELECT 1 FROM sqlite_master WHERE type='table' AND name='task_logs'"
+                )
+                self.assertIsNone(cur.fetchone())
            finally:
                conn.close()

-    def test_task_logs_has_chinese_table_and_column_names(self) -> None:
+    def test_demo_items_fixture_has_chinese_names(self) -> None:
        with IsolatedDataRoot():
            from db.connection import get_conn, init_db

@@ -35,32 +45,35 @@ class TestDisplayMetadata(unittest.TestCase):
            conn = get_conn()
            try:
                cur = conn.cursor()
+                seed_demo_items(cur)
+                conn.commit()
                cur.execute(
-                    "SELECT display_name, readonly FROM _jiangchang_tables WHERE table_name = 'task_logs'"
+                    "SELECT display_name, readonly FROM _jiangchang_tables WHERE table_name = ?",
+                    (DEMO_ITEMS_TABLE,),
                )
                display_name, readonly = cur.fetchone()
-                self.assertEqual(display_name, "任务日志")
-                self.assertEqual(int(readonly), 1)
+                self.assertEqual(display_name, "示例记录")
+                self.assertEqual(int(readonly), 0)

                cur.execute(
                    """
                    SELECT column_name, display_name, editable
                    FROM _jiangchang_columns
-                    WHERE table_name = 'task_logs'
+                    WHERE table_name = ?
                    ORDER BY column_name
-                    """
+                    """,
+                    (DEMO_ITEMS_TABLE,),
                )
                by_name = {row[0]: (row[1], row[2]) for row in cur.fetchall()}
                self.assertEqual(by_name["status"][0], "状态")
-                self.assertEqual(int(by_name["status"][1]), 0)
-                self.assertEqual(by_name["error_msg"][0], "错误信息")
-                self.assertEqual(int(by_name["error_msg"][1]), 0)
+                self.assertEqual(int(by_name["status"][1]), 1)
+                self.assertEqual(by_name["title"][0], "标题")
                self.assertEqual(by_name["created_at"][0], "创建时间")
                self.assertEqual(int(by_name["created_at"][1]), 0)
            finally:
                conn.close()

-    def test_task_logs_not_manual_create_in_host_ui(self) -> None:
+    def test_readonly_table_not_manual_create_in_host_ui(self) -> None:
        with IsolatedDataRoot():
            from db.connection import get_conn, init_db
            from db.display_metadata import table_allows_manual_create
@@ -68,33 +81,34 @@ class TestDisplayMetadata(unittest.TestCase):
            init_db()
            conn = get_conn()
            try:
-                self.assertFalse(table_allows_manual_create(conn.cursor(), "task_logs"))
+                cur = conn.cursor()
+                seed_demo_items(cur, readonly=1)
+                conn.commit()
+                self.assertFalse(table_allows_manual_create(cur, DEMO_ITEMS_TABLE))
            finally:
                conn.close()

-    def test_task_logs_physical_order_matches_pragma_cid(self) -> None:
+    def test_demo_items_physical_order_matches_pragma_cid(self) -> None:
        with IsolatedDataRoot():
-            from db.connection import init_db
-            from db.display_metadata import TASK_LOGS_COLUMN_METADATA, list_physical_column_names
-            from db.connection import get_conn
+            from db.connection import get_conn, init_db
+            from db.display_metadata import list_physical_column_names

            init_db()
            conn = get_conn()
            try:
                cur = conn.cursor()
-                physical = list_physical_column_names(cur, "task_logs")
-                expected = [str(item["column_name"]) for item in TASK_LOGS_COLUMN_METADATA]
-                self.assertEqual(
-                    physical,
-                    expected,
-                )
-                cur.execute("PRAGMA table_info('task_logs')")
+                seed_demo_items(cur)
+                conn.commit()
+                physical = list_physical_column_names(cur, DEMO_ITEMS_TABLE)
+                expected = [str(item["column_name"]) for item in DEMO_ITEMS_COLUMN_METADATA]
+                self.assertEqual(physical, expected)
+                cur.execute("PRAGMA table_info('demo_items')")
                cid_order = [row[1] for row in sorted(cur.fetchall(), key=lambda row: int(row[0]))]
                self.assertEqual(cid_order, expected)
            finally:
                conn.close()

-    def test_init_db_sets_task_logs_display_order_null(self) -> None:
+    def test_seed_sets_display_order_null(self) -> None:
        with IsolatedDataRoot():
            from db.connection import get_conn, init_db

@@ -102,12 +116,15 @@ class TestDisplayMetadata(unittest.TestCase):
            conn = get_conn()
            try:
                cur = conn.cursor()
+                seed_demo_items(cur)
+                conn.commit()
                cur.execute(
                    """
                    SELECT column_name, display_order
                    FROM _jiangchang_columns
-                    WHERE table_name = 'task_logs'
-                    """
+                    WHERE table_name = ?
+                    """,
+                    (DEMO_ITEMS_TABLE,),
                )
                for column_name, display_order in cur.fetchall():
                    self.assertIsNone(
@@ -117,33 +134,32 @@ class TestDisplayMetadata(unittest.TestCase):
            finally:
                conn.close()

-    def test_reinit_clears_stale_display_order(self) -> None:
+    def test_reseed_clears_stale_display_order(self) -> None:
        with IsolatedDataRoot():
            from db.connection import get_conn, init_db

            init_db()
            conn = get_conn()
            try:
+                cur = conn.cursor()
+                seed_demo_items(cur)
                conn.execute(
                    """
                    UPDATE _jiangchang_columns
                    SET display_order = 99
-                    WHERE table_name = 'task_logs' AND column_name = 'id'
-                    """
+                    WHERE table_name = ? AND column_name = 'id'
+                    """,
+                    (DEMO_ITEMS_TABLE,),
                )
                conn.commit()
-            finally:
-                conn.close()
-
-            init_db()
-            conn = get_conn()
-            try:
-                cur = conn.cursor()
+                seed_demo_items(cur)
+                conn.commit()
                cur.execute(
                    """
                    SELECT display_order FROM _jiangchang_columns
-                    WHERE table_name = 'task_logs' AND column_name = 'id'
-                    """
+                    WHERE table_name = ? AND column_name = 'id'
+                    """,
+                    (DEMO_ITEMS_TABLE,),
                )
                self.assertIsNone(cur.fetchone()[0])
            finally:
@@ -158,22 +174,26 @@ class TestDisplayMetadata(unittest.TestCase):
            conn = get_conn()
            try:
                cur = conn.cursor()
+                seed_demo_items(cur)
+                conn.commit()
                cur.execute(
                    """
                    UPDATE _jiangchang_columns
                    SET display_order = 1
-                    WHERE table_name = 'task_logs' AND column_name = 'updated_at'
-                    """
+                    WHERE table_name = ? AND column_name = 'updated_at'
+                    """,
+                    (DEMO_ITEMS_TABLE,),
                )
                cur.execute(
                    """
                    UPDATE _jiangchang_columns
                    SET display_order = 99
-                    WHERE table_name = 'task_logs' AND column_name = 'id'
-                    """
+                    WHERE table_name = ? AND column_name = 'id'
+                    """,
+                    (DEMO_ITEMS_TABLE,),
                )
                conn.commit()
-                physical = list_physical_column_names(cur, "task_logs")
+                physical = list_physical_column_names(cur, DEMO_ITEMS_TABLE)
                self.assertEqual(physical[0], "id")
                self.assertEqual(physical[-1], "updated_at")
            finally:
@@ -188,7 +208,7 @@ class TestDisplayMetadata(unittest.TestCase):
            conn = get_conn()
            try:
                tables = list(iter_user_tables(conn.cursor()))
-                self.assertEqual(tables, ["task_logs"])
+                self.assertEqual(tables, [])
            finally:
                conn.close()

@@ -271,13 +291,15 @@ class TestDisplayMetadata(unittest.TestCase):
            init_db()
            conn = get_conn()
            try:
+                seed_demo_items(conn.cursor())
                conn.execute(
-                    "DELETE FROM _jiangchang_columns WHERE table_name = 'task_logs' AND column_name = 'status'"
+                    "DELETE FROM _jiangchang_columns WHERE table_name = ? AND column_name = 'status'",
+                    (DEMO_ITEMS_TABLE,),
                )
                conn.commit()
                report = validate_display_metadata(conn)
                self.assertFalse(report.ok)
-                self.assertTrue(any("task_logs.status" in err for err in report.errors))
+                self.assertTrue(any("demo_items.status" in err for err in report.errors))
            finally:
                conn.close()

@@ -289,12 +311,14 @@ class TestDisplayMetadata(unittest.TestCase):
            init_db()
            conn = get_conn()
            try:
+                seed_demo_items(conn.cursor())
                conn.execute(
                    """
                    INSERT INTO _jiangchang_columns
                    (table_name, column_name, display_name, visible, searchable, editable)
-                    VALUES ('task_logs', 'not_exists', '不存在', 1, 1, 1)
-                    """
+                    VALUES (?, 'not_exists', '不存在', 1, 1, 1)
+                    """,
+                    (DEMO_ITEMS_TABLE,),
                )
                conn.commit()
                report = validate_display_metadata(conn)
@@ -498,12 +522,11 @@ metadata:
        finally:
            conn.close()

-    def test_schema_md_documents_task_logs_readonly_semantics(self) -> None:
+    def test_schema_md_documents_readonly_and_seed_helpers(self) -> None:
        path = os.path.join(get_skill_root(), "references", "SCHEMA.md")
        with open(path, encoding="utf-8") as f:
            text = f.read()
        self.assertIn("readonly = 1", text)
-        self.assertIn("task_logs", text)
        self.assertIn("seed_column_metadata_batch", text)
        self.assertIn("valid_column_names", text)

--- a/tests/test_timestamp_columns.py
+++ b/tests/test_timestamp_columns.py
@@ -6,11 +6,11 @@ import sqlite3
 import time
 import unittest

-from _support import IsolatedDataRoot
+from _support import DEMO_ITEMS_TABLE, IsolatedDataRoot, seed_demo_items


 class TestTimestampColumns(unittest.TestCase):
-    def test_init_db_creates_integer_timestamp_columns_with_defaults(self) -> None:
+    def test_seed_creates_integer_timestamp_columns_with_defaults(self) -> None:
        with IsolatedDataRoot():
            from db.connection import get_conn, init_db
            from db.display_metadata import list_physical_column_names
@@ -20,10 +20,12 @@ class TestTimestampColumns(unittest.TestCase):
            conn = get_conn()
            try:
                cur = conn.cursor()
-                physical = list_physical_column_names(cur, "task_logs")
+                seed_demo_items(cur)
+                conn.commit()
+                physical = list_physical_column_names(cur, DEMO_ITEMS_TABLE)
                self.assertEqual(physical[-2:], ["created_at", "updated_at"])

-                info = _table_column_info(cur, "task_logs")
+                info = _table_column_info(cur, DEMO_ITEMS_TABLE)
                for column_name in ("created_at", "updated_at"):
                    column = info[column_name]
                    self.assertIn("INT", str(column["type"]).upper())
@@ -39,13 +41,14 @@ class TestTimestampColumns(unittest.TestCase):
            conn = get_conn()
            try:
                cur = conn.cursor()
+                seed_demo_items(cur)
                before = int(time.time())
                cur.execute(
-                    "INSERT INTO task_logs (task_type, status) VALUES ('auto-ts', 'success')"
+                    "INSERT INTO demo_items (title, status) VALUES ('auto-ts', 'draft')"
                )
                conn.commit()
                cur.execute(
-                    "SELECT created_at, updated_at FROM task_logs ORDER BY id DESC LIMIT 1"
+                    "SELECT created_at, updated_at FROM demo_items ORDER BY id DESC LIMIT 1"
                )
                created_at, updated_at = cur.fetchone()
                after = int(time.time())
@@ -65,21 +68,22 @@ class TestTimestampColumns(unittest.TestCase):
            conn = get_conn()
            try:
                cur = conn.cursor()
+                seed_demo_items(cur)
                cur.execute(
                    """
-                    INSERT INTO task_logs (task_type, status, created_at, updated_at)
-                    VALUES ('update-ts', 'success', 1000, 1000)
+                    INSERT INTO demo_items (title, status, created_at, updated_at)
+                    VALUES ('update-ts', 'draft', 1000, 1000)
                    """
                )
-                log_id = int(cur.lastrowid)
+                row_id = int(cur.lastrowid)
                conn.commit()

-                cur.execute("UPDATE task_logs SET status = 'failed' WHERE id = ?", (log_id,))
+                cur.execute("UPDATE demo_items SET status = 'published' WHERE id = ?", (row_id,))
                conn.commit()

                cur.execute(
-                    "SELECT created_at, updated_at FROM task_logs WHERE id = ?",
-                    (log_id,),
+                    "SELECT created_at, updated_at FROM demo_items WHERE id = ?",
+                    (row_id,),
                )
                created_at, updated_at = cur.fetchone()
                self.assertEqual(created_at, 1000)
@@ -95,10 +99,11 @@ class TestTimestampColumns(unittest.TestCase):
            conn = get_conn()
            try:
                cur = conn.cursor()
+                seed_demo_items(cur)
                cur.execute(
                    """
-                    INSERT INTO task_logs (task_type, status, created_at, updated_at)
-                    VALUES ('stable-ts', 'success', 1234567890, 1234567891)
+                    INSERT INTO demo_items (title, status, created_at, updated_at)
+                    VALUES ('stable-ts', 'draft', 1234567890, 1234567891)
                    """
                )
                conn.commit()
@@ -110,7 +115,7 @@ class TestTimestampColumns(unittest.TestCase):
            try:
                cur = conn.cursor()
                cur.execute(
-                    "SELECT created_at, updated_at FROM task_logs WHERE task_type = 'stable-ts'"
+                    "SELECT created_at, updated_at FROM demo_items WHERE title = 'stable-ts'"
                )
                self.assertEqual(cur.fetchone(), (1234567890, 1234567891))
            finally:
@@ -125,13 +130,16 @@ class TestTimestampColumns(unittest.TestCase):
            conn = get_conn()
            try:
                cur = conn.cursor()
+                seed_demo_items(cur)
+                conn.commit()
                cur.execute(
                    """
                    SELECT column_name, display_name, display_type, editable, display_order
                    FROM _jiangchang_columns
-                    WHERE table_name = 'task_logs'
+                    WHERE table_name = ?
                      AND column_name IN ('created_at', 'updated_at')
-                    """
+                    """,
+                    (DEMO_ITEMS_TABLE,),
                )
                by_name = {
                    name: (display_name, display_type, editable, display_order)
@@ -153,37 +161,43 @@ class TestTimestampColumns(unittest.TestCase):
            from db.connection import get_conn, init_db
            from db.display_metadata import DATETIME_UNIX_SECONDS

-            init_db()
            init_db()
            conn = get_conn()
            try:
                cur = conn.cursor()
+                seed_demo_items(cur)
+                seed_demo_items(cur)
+                conn.commit()
                cur.execute(
                    """
                    SELECT display_type FROM _jiangchang_columns
-                    WHERE table_name = 'task_logs' AND column_name = 'created_at'
-                    """
+                    WHERE table_name = ? AND column_name = 'created_at'
+                    """,
+                    (DEMO_ITEMS_TABLE,),
                )
                self.assertEqual(cur.fetchone()[0], DATETIME_UNIX_SECONDS)
            finally:
                conn.close()

-    def test_repository_insert_uses_database_timestamps(self) -> None:
+    def test_insert_uses_database_timestamps(self) -> None:
        with IsolatedDataRoot():
-            from db import task_logs_repository as tlr
            from db.connection import get_conn, init_db

            init_db()
-            before = int(time.time())
-            new_id = tlr.save_task_log(task_type="repo-ts", status="success")
-            after = int(time.time())
-
            conn = get_conn()
            try:
                cur = conn.cursor()
+                seed_demo_items(cur)
+                before = int(time.time())
                cur.execute(
-                    "SELECT created_at, updated_at FROM task_logs WHERE id = ?",
-                    (new_id,),
+                    "INSERT INTO demo_items (title, status) VALUES ('repo-ts', 'draft')"
+                )
+                row_id = int(cur.lastrowid)
+                conn.commit()
+                after = int(time.time())
+                cur.execute(
+                    "SELECT created_at, updated_at FROM demo_items WHERE id = ?",
+                    (row_id,),
                )
                created_at, updated_at = cur.fetchone()
                self.assertGreaterEqual(int(created_at), before)
@@ -200,8 +214,11 @@ class TestTimestampColumns(unittest.TestCase):
            init_db()
            conn = get_conn()
            try:
+                seed_demo_items(conn.cursor())
                conn.execute(
-                    "UPDATE _jiangchang_columns SET display_type = NULL WHERE table_name = 'task_logs' AND column_name = 'created_at'"
+                    "UPDATE _jiangchang_columns SET display_type = NULL "
+                    "WHERE table_name = ? AND column_name = 'created_at'",
+                    (DEMO_ITEMS_TABLE,),
                )
                conn.commit()
                report = validate_template_database(conn)
@@ -260,8 +277,11 @@ class TestTimestampColumns(unittest.TestCase):
            init_db()
            conn = get_conn()
            try:
+                seed_demo_items(conn.cursor())
                conn.execute(
-                    "UPDATE _jiangchang_columns SET editable = 1 WHERE table_name = 'task_logs' AND column_name = 'updated_at'"
+                    "UPDATE _jiangchang_columns SET editable = 1 "
+                    "WHERE table_name = ? AND column_name = 'updated_at'",
+                    (DEMO_ITEMS_TABLE,),
                )
                conn.commit()
                report = validate_template_database(conn)
@@ -315,17 +335,19 @@ class TestTimestampColumns(unittest.TestCase):
    def test_trigger_creation_is_idempotent(self) -> None:
        with IsolatedDataRoot():
            from db.connection import get_conn, init_db
-            from db.timestamp_columns import TASK_LOGS_UPDATED_AT_TRIGGER, has_updated_at_trigger
+            from db.timestamp_columns import has_updated_at_trigger, updated_at_trigger_name

-            init_db()
            init_db()
            conn = get_conn()
            try:
-                self.assertTrue(has_updated_at_trigger(conn.cursor(), "task_logs"))
                cur = conn.cursor()
+                seed_demo_items(cur)
+                seed_demo_items(cur)
+                conn.commit()
+                self.assertTrue(has_updated_at_trigger(cur, DEMO_ITEMS_TABLE))
                cur.execute(
                    "SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger' AND name = ?",
-                    (TASK_LOGS_UPDATED_AT_TRIGGER,),
+                    (updated_at_trigger_name(DEMO_ITEMS_TABLE),),
                )
                self.assertEqual(int(cur.fetchone()[0]), 1)
            finally:
--- a/tests/test_video_service.py
+++ b/tests/test_video_service.py
@@ -1,9 +1,8 @@
 # -*- coding: utf-8 -*-
-"""RpaVideoSession 模板示范与 result_summary / CLI 输出测试。"""
+"""RpaVideoSession 模板示范与 finish() video 字段 / CLI 输出测试。"""
 from __future__ import annotations

 import io
-import json
 import os
 import re
 import unittest
@@ -131,7 +130,7 @@ class TestPrintVideoSummary(unittest.TestCase):


 class TestMergeVideoIntoResultSummary(unittest.TestCase):
-    def test_result_summary_includes_video_fields(self) -> None:
+    def test_finish_payload_includes_video_fields(self) -> None:
        from service.task_run_support import merge_video_into_result_summary

        video_info = {
@@ -202,9 +201,17 @@ class TestTemplateRunVideoSession(unittest.TestCase):
                    return None

            with patch.object(task_service, "RpaVideoSession", _CapturingFakeSession):
+                finish_kwargs: list[dict] = []
+                original_finish = task_service.finish
+
+                def _capture_finish(*args, **kwargs):
+                    finish_kwargs.append(kwargs)
+                    return original_finish(*args, **kwargs)
+
                buf = io.StringIO()
-                with redirect_stdout(buf), redirect_stderr(io.StringIO()):
-                    rc = task_service.cmd_run(target="demo-target", input_id="42")
+                with patch.object(task_service, "finish", side_effect=_capture_finish):
+                    with redirect_stdout(buf), redirect_stderr(io.StringIO()):
+                        rc = task_service.cmd_run(target="demo-target", input_id="42")

            self.assertEqual(rc, 1)
            self.assertTrue(created, "RpaVideoSession should be constructed")
@@ -220,39 +227,39 @@ class TestTemplateRunVideoSession(unittest.TestCase):
            out = buf.getvalue()
            self.assertIn("模板", out)

-            from db import task_logs_repository as tlr
-
-            rows = tlr.list_task_logs(1)
-            self.assertTrue(rows)
-            summary = json.loads(rows[0][7])
+            self.assertTrue(finish_kwargs, "finish() should be called")
+            summary = finish_kwargs[-1]
            self.assertIn("video", summary)
            self.assertIn("video_path", summary)


-class TestCmdRunTaskLogCoverage(unittest.TestCase):
-    def test_cmd_run_entitlement_failure_writes_task_log(self) -> None:
+class TestCmdRunFinishCoverage(unittest.TestCase):
+    def test_cmd_run_entitlement_failure_calls_finish(self) -> None:
        with IsolatedDataRoot(user_id="_entitlement_fail"):
            from service import task_service

+            finish_kwargs: list[dict] = []
+            original_finish = task_service.finish
+
+            def _capture_finish(*args, **kwargs):
+                finish_kwargs.append(kwargs)
+                return original_finish(*args, **kwargs)
+
            with patch.object(task_service, "check_entitlement", return_value=(False, "mock denied")):
-                buf = io.StringIO()
-                with redirect_stdout(buf), redirect_stderr(io.StringIO()):
-                    rc = task_service.cmd_run(target="t1", input_id="99")
+                with patch.object(task_service, "finish", side_effect=_capture_finish):
+                    buf = io.StringIO()
+                    with redirect_stdout(buf), redirect_stderr(io.StringIO()):
+                        rc = task_service.cmd_run(target="t1", input_id="99")

            self.assertEqual(rc, 1)
-            from db import task_logs_repository as tlr
+            self.assertTrue(finish_kwargs)
+            payload = finish_kwargs[-1]
+            self.assertEqual(payload.get("status"), "failed")
+            self.assertEqual(payload.get("error_code"), "ENTITLEMENT_DENIED")
+            self.assertEqual(payload.get("stage"), "auth")
+            self.assertIn("mock denied", payload.get("message") or "")

-            rows = tlr.list_task_logs(1)
-            self.assertTrue(rows)
-            _rid, _ttype, tid, iid, _ititle, status, err, rsum, _cat, _uat = rows[0]
-            self.assertEqual(status, "failed")
-            self.assertEqual(tid, "t1")
-            self.assertEqual(iid, "99")
-            self.assertIn("mock denied", err or "")
-            summary = json.loads(rsum or "{}")
-            self.assertEqual(summary.get("stage"), "auth")
-
-    def test_cmd_run_exception_writes_task_log(self) -> None:
+    def test_cmd_run_exception_calls_finish(self) -> None:
        with IsolatedDataRoot(user_id="_exception_run"):
            os.environ["OPENCLAW_RECORD_VIDEO"] = "0"
            config.reset_cache()
@@ -262,25 +269,26 @@ class TestCmdRunTaskLogCoverage(unittest.TestCase):
            async def _boom(*_args, **_kwargs):
                raise RuntimeError("boom")

+            finish_kwargs: list[dict] = []
+            original_finish = task_service.finish
+
+            def _capture_finish(*args, **kwargs):
+                finish_kwargs.append(kwargs)
+                return original_finish(*args, **kwargs)
+
            with patch.object(task_service, "check_entitlement", return_value=(True, "")):
                with patch.object(task_service, "_run_template_demo", side_effect=_boom):
-                    buf = io.StringIO()
-                    with redirect_stdout(buf), redirect_stderr(io.StringIO()):
-                        rc = task_service.cmd_run(target="t2", input_id="88")
+                    with patch.object(task_service, "finish", side_effect=_capture_finish):
+                        buf = io.StringIO()
+                        with redirect_stdout(buf), redirect_stderr(io.StringIO()):
+                            rc = task_service.cmd_run(target="t2", input_id="88")

            self.assertEqual(rc, 1)
-            from db import task_logs_repository as tlr
-
-            rows = tlr.list_task_logs(1)
-            self.assertTrue(rows)
-            _rid, _ttype, tid, iid, _ititle, status, err, rsum, _cat, _uat = rows[0]
-            self.assertEqual(status, "failed")
-            self.assertEqual(tid, "t2")
-            self.assertEqual(iid, "88")
-            self.assertIn("任务执行异常", err or "")
-            summary = json.loads(rsum or "{}")
-            self.assertEqual(summary.get("stage"), "run")
-            self.assertEqual(summary.get("error"), "unexpected_exception")
+            self.assertTrue(finish_kwargs)
+            payload = finish_kwargs[-1]
+            self.assertEqual(payload.get("status"), "failed")
+            self.assertEqual(payload.get("stage"), "run")
+            self.assertIn("任务执行异常", payload.get("message") or "")

    def test_get_task_logger_fallback_when_not_initialized(self) -> None:
        from service import task_service