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

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_`` 前缀，不自动当用例收集） |

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],
+    )

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",

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。"""

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()

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,
             [],

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)
 

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:

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