feat: 规范标准时间字段 created_at/updated_at（Unix 秒级、元数据、trigger）

SKILL.md

@@ -82,7 +82,7 @@ python {baseDir}/scripts/main.py version
 2. 按 `development/DEVELOPMENT.md` 完成开发与文档定制。
 3. 用户市场说明写入根 `README.md`；Agent 调用契约见 `references/CLI.md`、`references/SCHEMA.md`。
 4. 同步修改 `scripts/util/constants.py` 中的 `SKILL_SLUG` / `SKILL_VERSION`。
-5. 本地 SQLite 的中文表名/字段名与字段顺序规范见 `references/SCHEMA.md`（元数据表 `_jiangchang_*` + `CREATE TABLE` 物理顺序）。
+5. 本地 SQLite 的中文表名/字段名、标准时间字段与字段顺序规范见 `references/SCHEMA.md`（元数据表 `_jiangchang_*` + `CREATE TABLE` 物理顺序）。
 
 ## 技能命名与 slug
 

development/DEVELOPMENT.md

@@ -447,6 +447,8 @@ metadata:
   只做增删查改（模板默认表：`task_logs`）
 - `db/display_metadata.py`
   创建 `_jiangchang_tables` / `_jiangchang_columns` 并幂等写入中文展示名（见 `references/SCHEMA.md`）
+- `db/timestamp_columns.py`
+  标准时间字段 trigger 与维护约定（`created_at` / `updated_at`，Unix 秒级）
 
 不要在 `db` 层里：
 

development/REQUIREMENTS.md

@@ -149,7 +149,7 @@
 - `health`、`version` 命令执行正常
 - 主命令（如 `run`）在 mock / simulator 档位可重复验证
 - `python tests/run_tests.py -v` 必跑测试全部通过
-- `task_logs` 写入和查询符合 `references/SCHEMA.md`
+- `task_logs` 写入和查询符合 `references/SCHEMA.md`（含 `created_at` / `updated_at` Unix 秒级规范）
 - `init_db()` 已写入 `_jiangchang_tables` / `_jiangchang_columns`；用户可见表/字段具备中文 `display_name`
 - 字段展示顺序与 `PRAGMA table_info(task_logs)` 的 cid 一致；不依赖 `display_order`
 - `tests/test_display_metadata.py` 通过

development/RUNTIME.md

@@ -71,7 +71,7 @@ RPA 录屏成片（`RpaVideoSession`）、ffmpeg 路径、背景音乐探测均
 
 - `scripts/main.py`：唯一 CLI 入口
 - `scripts/cli/`：参数解析与命令分发
-- `scripts/db/`：SQLite 或本地持久化层
+- `scripts/db/`：SQLite 或本地持久化层（标准时间字段见 `references/SCHEMA.md`）
 - `scripts/service/`：业务用例与外部交互
 - `scripts/util/`：通用工具、常量、日志、路径薄封装
 

development/TESTING.md

@@ -29,6 +29,7 @@
 - `SKILL.md` YAML slug vs [`constants.SKILL_SLUG`](../scripts/util/constants.py)；
 - SQLite 骨架：`task_logs` 创建幂等与仓储读写；
 - 数据管理元数据：`_jiangchang_*` 幂等初始化、中文展示名、`PRAGMA table_info` 字段顺序（见 `test_display_metadata.py`）；
+- 标准时间字段：`created_at` / `updated_at` 默认值、trigger 维护与 `datetime_unix_seconds` 元数据（见 `test_timestamp_columns.py`）；
 - **adapter profile**：[`tests/test_adapter_profile_policy.py`](../tests/test_adapter_profile_policy.py) + [`adapter_test_utils`](../tests/adapter_test_utils.py) ——验证在未授权情况下绝不误判开启真实网络/RPA。
 - **发布打包守护**：[`tests/test_release_packaging_constraints.py`](../tests/test_release_packaging_constraints.py) ——`scripts/**/*.py` 单文件 < 1000 行、文本文件 UTF-8 without BOM。
 

references/SCHEMA.md

@@ -69,7 +69,7 @@ metadata:
 | `visible` | 是 | 是否展示 |
 | `searchable` | 是 | 是否可搜索 |
 | `editable` | 是 | 是否可编辑（与宿主权限结合） |
-| `display_type` | 否 | 如 `text` / `textarea` |
+| `display_type` | 否 | 如 `text` / `textarea` / `datetime_unix_seconds`（标准时间字段见下文） |
 | `options_json` | 否 | 枚举选项 JSON |
 
 这两张表是**数据字典**，不是业务数据；宿主不会把它们当作普通业务表展示。
@@ -90,7 +90,36 @@ metadata:
 4. 状态、结果、错误等辅助字段
 5. created_at（如有）
 6. updated_at（如有）
-```
+
+## 标准时间字段（新技能统一规范）
+
+新技能业务表若需要时间审计字段，**统一**采用 Unix **秒级**时间戳，不要使用毫秒时间戳、ISO 字符串或 SQLite 文本 datetime。
+
+| 项目 | 规范 |
+|------|------|
+| 物理字段 | `created_at INTEGER`、`updated_at INTEGER` |
+| 存储格式 | Unix 秒级整数 |
+| 展示元数据 `display_type` | `datetime_unix_seconds` |
+| 中文名 | `created_at` → **创建时间**；`updated_at` → **更新时间** |
+| 用户编辑 | `editable = 0`（系统字段，宿主只展示格式化结果，不写回字符串） |
+| 字段顺序 | 位于业务字段末尾：`… → created_at → updated_at` |
+| 排序 | **不使用** `display_order`；顺序仅来自 `PRAGMA table_info` |
+
+### 自动生成与维护
+
+- **created_at**：`INTEGER NOT NULL DEFAULT (unixepoch())`；插入时可省略，由数据库生成。
+- **updated_at**：`INTEGER NOT NULL DEFAULT (unixepoch())`；插入时同样可省略。
+- **updated_at 更新**：模板默认表 `task_logs` 通过 SQLite trigger `{table}_set_updated_at` 在业务字段更新后自动刷新；若技能所有 UPDATE 都经过统一 repository，也可在 repository 层设置 `updated_at = unixepoch()`，**不要**叠加两套机制。
+
+实现参考：
+
+- DDL 与 trigger：`scripts/db/connection.py`、`scripts/db/timestamp_columns.py`
+- 中文元数据种子：`scripts/db/display_metadata.py`
+- 自检：`scripts/db/display_metadata_validator.py`
+
+宿主（匠厂）只负责把 `display_type = datetime_unix_seconds` 格式化为 `YYYY-MM-DD HH:mm:ss` 展示，**不修改**数据库原始值。
+
+> `datetime_unix_milliseconds`、`datetime_iso` 是宿主兼容能力，**不是**新技能模板推荐标准。
 
 ### 模板默认：`task_logs`
 
@@ -106,11 +135,13 @@ CREATE TABLE IF NOT EXISTS task_logs (
     status TEXT NOT NULL,
     error_msg TEXT,
     result_summary TEXT,
-    created_at INTEGER NOT NULL,
+    created_at INTEGER NOT NULL DEFAULT (unixepoch()),
-    updated_at INTEGER NOT NULL
+    updated_at INTEGER NOT NULL DEFAULT (unixepoch())
 );
 ```
 
+插入时可省略 `created_at`、`updated_at`；`updated_at` 在 UPDATE 时由 trigger `task_logs_set_updated_at` 自动维护（见 `scripts/db/timestamp_columns.py`）。
+
 对应中文展示（由 `init_db()` 写入元数据表）：
 
 | 物理字段 | 中文名 |
@@ -123,8 +154,8 @@ CREATE TABLE IF NOT EXISTS task_logs (
 | status | 状态 |
 | error_msg | 错误信息 |
 | result_summary | 结果摘要 |
-| created_at | 创建时间 |
+| created_at | 创建时间（`display_type = datetime_unix_seconds`，不可编辑） |
-| updated_at | 更新时间 |
+| updated_at | 更新时间（`display_type = datetime_unix_seconds`，不可编辑） |
 
 表中文名：**任务日志**。
 

scripts/db/connection.py

@@ -5,6 +5,7 @@ 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
 
 
@@ -27,11 +28,12 @@ def init_db() -> None:
                 status TEXT NOT NULL,
                 error_msg TEXT,
                 result_summary TEXT,
-                created_at INTEGER NOT NULL,
+                created_at INTEGER NOT NULL DEFAULT (unixepoch()),
-                updated_at INTEGER NOT NULL
+                updated_at INTEGER NOT NULL DEFAULT (unixepoch())
             )
             """
         )
+        init_task_logs_timestamp_maintenance(cur)
         init_display_metadata(cur)
         conn.commit()
     finally:

scripts/db/display_metadata.py

@@ -14,6 +14,8 @@ 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"})
 
 _JIANGCHANG_TABLES_DDL = """
 CREATE TABLE IF NOT EXISTS _jiangchang_tables (
@@ -59,8 +61,24 @@ TASK_LOGS_COLUMN_METADATA: tuple[dict[str, object], ...] = (
     {"column_name": "status", "display_name": "状态", "description": "任务当前状态", "visible": 1, "searchable": 1, "editable": 1},
     {"column_name": "error_msg", "display_name": "错误信息", "description": "任务失败时的错误说明", "visible": 1, "searchable": 1, "editable": 1},
     {"column_name": "result_summary", "display_name": "结果摘要", "description": "任务执行结果摘要", "visible": 1, "searchable": 1, "editable": 1},
-    {"column_name": "created_at", "display_name": "创建时间", "description": "记录创建时间", "visible": 1, "searchable": 1, "editable": 0},
+    {
-    {"column_name": "updated_at", "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,
+    },
 )
 
 
@@ -94,24 +112,26 @@ def upsert_column_metadata(
     visible: int = 1,
     searchable: int = 1,
     editable: int = 1,
+    display_type: str | None = None,
 ) -> None:
     # display_order 保留为 NULL：宿主按 PRAGMA table_info cid 展示字段，不读 display_order。
     cursor.execute(
         """
         INSERT INTO _jiangchang_columns (
             table_name, column_name, display_name, description,
-            display_order, visible, searchable, editable
+            display_order, visible, searchable, editable, display_type
         )
-        VALUES (?, ?, ?, ?, NULL, ?, ?, ?)
+        VALUES (?, ?, ?, ?, NULL, ?, ?, ?, ?)
         ON CONFLICT(table_name, column_name) DO UPDATE SET
             display_name = excluded.display_name,
             description = excluded.description,
             display_order = NULL,
             visible = excluded.visible,
             searchable = excluded.searchable,
-            editable = excluded.editable
+            editable = excluded.editable,
+            display_type = excluded.display_type
         """,
-        (table_name, column_name, display_name, description, visible, searchable, editable),
+        (table_name, column_name, display_name, description, visible, searchable, editable, display_type),
     )
 
 
@@ -135,6 +155,7 @@ def seed_task_logs_display_metadata(cursor) -> None:
             visible=int(column["visible"]),
             searchable=int(column["searchable"]),
             editable=int(column["editable"]),
+            display_type=str(column["display_type"]) if column.get("display_type") else None,
         )
 
 

scripts/db/display_metadata_validator.py

@@ -8,12 +8,15 @@ import sqlite3
 from dataclasses import dataclass, field
 
 from db.display_metadata import (
+    DATETIME_UNIX_SECONDS,
     METADATA_TABLES,
+    STANDARD_TIMESTAMP_COLUMNS,
     TASK_LOGS_COLUMN_METADATA,
     TASK_LOGS_TABLE,
     iter_user_tables,
     list_physical_column_names,
 )
+from db.timestamp_columns import has_updated_at_trigger
 from util.constants import SKILL_SLUG
 
 _TEMPLATE_PLACEHOLDER_SLUGS = frozenset({"your-skill-slug", "your_skill_slug"})
@@ -127,6 +130,116 @@ def validate_skill_frontmatter(skill_root: str, *, allow_template_placeholders:
     return report
 
 
+def _quote_identifier(name: str) -> str:
+    return '"' + name.replace('"', '""') + '"'
+
+
+def _table_column_info(cursor, table_name: str) -> dict[str, dict[str, object]]:
+    cursor.execute(f"PRAGMA table_info({_quote_identifier(table_name)})")
+    info: dict[str, dict[str, object]] = {}
+    for row in cursor.fetchall():
+        info[str(row[1])] = {
+            "cid": int(row[0]),
+            "name": str(row[1]),
+            "type": str(row[2] or ""),
+            "notnull": int(row[3] or 0),
+            "dflt_value": row[4],
+            "pk": int(row[5] or 0),
+        }
+    return info
+
+
+def _default_uses_unixepoch(default_value: object) -> bool:
+    if default_value is None:
+        return False
+    text = str(default_value).strip().lower()
+    return "unixepoch" in text
+
+
+def _expected_timestamp_display_name(column_name: str) -> str:
+    if column_name == "created_at":
+        return "创建时间"
+    if column_name == "updated_at":
+        return "更新时间"
+    return ""
+
+
+def validate_standard_timestamp_columns(
+    cursor,
+    table_name: str,
+    *,
+    physical_columns: list[str] | None = None,
+) -> tuple[list[str], list[str]]:
+    errors: list[str] = []
+    warnings: list[str] = []
+    ordered = physical_columns or list_physical_column_names(cursor, table_name)
+    column_info = _table_column_info(cursor, table_name)
+
+    for column_name in STANDARD_TIMESTAMP_COLUMNS:
+        if column_name not in ordered:
+            continue
+
+        physical = column_info.get(column_name)
+        if not physical:
+            errors.append(f"{table_name}.{column_name} 缺少物理字段定义")
+            continue
+
+        if "INT" not in str(physical["type"]).upper():
+            errors.append(
+                f"{table_name}.{column_name} 字段类型应为 INTEGER（Unix 秒级时间戳），当前为 {physical['type']!r}"
+            )
+
+        if column_name == "created_at" and not _default_uses_unixepoch(physical["dflt_value"]):
+            errors.append(
+                f"{table_name}.created_at 应设置数据库默认值 DEFAULT (unixepoch())，当前默认值为 {physical['dflt_value']!r}"
+            )
+        if column_name == "updated_at" and not _default_uses_unixepoch(physical["dflt_value"]):
+            errors.append(
+                f"{table_name}.updated_at 应设置数据库初始默认值 DEFAULT (unixepoch())，当前默认值为 {physical['dflt_value']!r}"
+            )
+
+        cursor.execute(
+            """
+            SELECT display_name, display_type, editable, searchable, display_order
+            FROM _jiangchang_columns
+            WHERE table_name = ? AND column_name = ?
+            """,
+            (table_name, column_name),
+        )
+        meta_row = cursor.fetchone()
+        if not meta_row:
+            errors.append(f"{table_name}.{column_name} 缺少 _jiangchang_columns 元数据")
+            continue
+
+        display_name, display_type, editable, searchable, display_order = meta_row
+        expected_name = _expected_timestamp_display_name(column_name)
+        if str(display_name or "").strip() != expected_name:
+            errors.append(
+                f"{table_name}.{column_name} 的 display_name 应为 {expected_name!r}，当前为 {display_name!r}"
+            )
+        if str(display_type or "").strip() != DATETIME_UNIX_SECONDS:
+            errors.append(
+                f"{table_name}.{column_name} 的 display_type 应为 {DATETIME_UNIX_SECONDS!r}，当前为 {display_type!r}"
+            )
+        if editable is None or int(editable) != 0:
+            errors.append(f"{table_name}.{column_name} 的 editable 应为 0（系统时间字段不可编辑）")
+        if display_order is not None:
+            errors.append(f"{table_name}.{column_name} 的 display_order 应为 NULL")
+        if searchable is not None and int(searchable) != 0:
+            warnings.append(f"{table_name}.{column_name} 的时间字段通常不必开启 searchable")
+
+    if "created_at" in ordered and "updated_at" in ordered:
+        if ordered.index("updated_at") < ordered.index("created_at"):
+            warnings.append(f"{table_name}: updated_at 不应出现在 created_at 之前")
+
+    if "updated_at" in ordered and not has_updated_at_trigger(cursor, table_name):
+        errors.append(
+            f"{table_name}.updated_at 缺少自动维护 trigger（期望 {table_name}_set_updated_at）"
+        )
+
+    return errors, warnings
+
+
 def column_order_warnings(table_name: str, ordered_names: list[str]) -> list[str]:
     warnings: list[str] = []
     if "id" in ordered_names and ordered_names.index("id") != 0:
@@ -208,6 +321,14 @@ def validate_display_metadata(conn: sqlite3.Connection, *, check_column_order: b
                     f"字段 {table_name}.{column_name} 的 display_name 应使用中文：{display_name!r}"
                 )
 
+        ts_errors, ts_warnings = validate_standard_timestamp_columns(
+            cur,
+            table_name,
+            physical_columns=physical_columns,
+        )
+        report.errors.extend(ts_errors)
+        report.warnings.extend(ts_warnings)
+
     cur.execute("SELECT table_name, column_name FROM _jiangchang_columns")
     for table_name, column_name in cur.fetchall():
         if table_name in METADATA_TABLES:

scripts/db/task_logs_repository.py

@@ -9,7 +9,6 @@ from __future__ import annotations
 from typing import Any, List, Optional, Tuple
 
 from db.connection import get_conn, init_db
-from util.timeutil import now_unix
 
 
 def save_task_log(
@@ -22,17 +21,16 @@ def save_task_log(
     result_summary: Optional[str] = None,
 ) -> int:
     init_db()
-    now = now_unix()
     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, created_at, updated_at)
+                (task_type, target_id, input_id, input_title, status, error_msg, result_summary)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+            VALUES (?, ?, ?, ?, ?, ?, ?)
             """,
-            (task_type, target_id, input_id, input_title or "", status, error_msg, result_summary, now, now),
+            (task_type, target_id, input_id, input_title or "", status, error_msg, result_summary),
         )
         new_id = int(cur.lastrowid)
         conn.commit()

scripts/db/timestamp_columns.py

@@ -0,0 +1,36 @@
+"""标准时间字段：Unix 秒级时间戳、默认值与 updated_at 自动维护。"""
+
+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
+FOR EACH ROW
+BEGIN
+    UPDATE task_logs
+    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"
+    cursor.execute(
+        """
+        SELECT 1 FROM sqlite_master
+        WHERE type = 'trigger' AND name = ? AND tbl_name = ?
+        LIMIT 1
+        """,
+        (trigger_name, table_name),
+    )
+    return cursor.fetchone() is not None

tests/README.md

@@ -74,6 +74,7 @@ python tests/run_tests.py test_cli_smoke
 | ``SKILL.md`` slug 与 ``constants.SKILL_SLUG`` | ``test_skill_metadata.py`` |
 | DB / ``task_logs`` 骨架冒烟 | ``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_`` 前缀，不自动当用例收集） |
 | **无 vendored ``jiangchang_skill_core``** | ``test_platform_import.py`` |
 | **runtime diagnostics 架构守护** | ``test_runtime_diagnostics_integration.py`` |

tests/test_timestamp_columns.py

@@ -0,0 +1,336 @@
+# -*- coding: utf-8 -*-
+"""标准时间字段：Unix 秒级时间戳、默认值、trigger 与元数据。"""
+from __future__ import annotations
+
+import sqlite3
+import time
+import unittest
+
+from _support import IsolatedDataRoot
+
+
+class TestTimestampColumns(unittest.TestCase):
+    def test_init_db_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
+            from db.display_metadata_validator import _table_column_info
+
+            init_db()
+            conn = get_conn()
+            try:
+                cur = conn.cursor()
+                physical = list_physical_column_names(cur, "task_logs")
+                self.assertEqual(physical[-2:], ["created_at", "updated_at"])
+
+                info = _table_column_info(cur, "task_logs")
+                for column_name in ("created_at", "updated_at"):
+                    column = info[column_name]
+                    self.assertIn("INT", str(column["type"]).upper())
+                    self.assertTrue(str(column["dflt_value"]).lower().find("unixepoch") >= 0)
+            finally:
+                conn.close()
+
+    def test_insert_without_timestamps_auto_generates_unix_seconds(self) -> None:
+        with IsolatedDataRoot():
+            from db.connection import get_conn, init_db
+
+            init_db()
+            conn = get_conn()
+            try:
+                cur = conn.cursor()
+                before = int(time.time())
+                cur.execute(
+                    "INSERT INTO task_logs (task_type, status) VALUES ('auto-ts', 'success')"
+                )
+                conn.commit()
+                cur.execute(
+                    "SELECT created_at, updated_at FROM task_logs ORDER BY id DESC LIMIT 1"
+                )
+                created_at, updated_at = cur.fetchone()
+                after = int(time.time())
+                self.assertIsInstance(created_at, int)
+                self.assertIsInstance(updated_at, int)
+                self.assertGreaterEqual(created_at, before)
+                self.assertLessEqual(created_at, after)
+                self.assertEqual(created_at, updated_at)
+            finally:
+                conn.close()
+
+    def test_update_business_field_keeps_created_at_and_refreshes_updated_at(self) -> None:
+        with IsolatedDataRoot():
+            from db.connection import get_conn, init_db
+
+            init_db()
+            conn = get_conn()
+            try:
+                cur = conn.cursor()
+                cur.execute(
+                    """
+                    INSERT INTO task_logs (task_type, status, created_at, updated_at)
+                    VALUES ('update-ts', 'success', 1000, 1000)
+                    """
+                )
+                log_id = int(cur.lastrowid)
+                conn.commit()
+
+                cur.execute("UPDATE task_logs SET status = 'failed' WHERE id = ?", (log_id,))
+                conn.commit()
+
+                cur.execute(
+                    "SELECT created_at, updated_at FROM task_logs WHERE id = ?",
+                    (log_id,),
+                )
+                created_at, updated_at = cur.fetchone()
+                self.assertEqual(created_at, 1000)
+                self.assertGreater(updated_at, 1000)
+            finally:
+                conn.close()
+
+    def test_reinit_does_not_rewrite_existing_timestamps(self) -> None:
+        with IsolatedDataRoot():
+            from db.connection import get_conn, init_db
+
+            init_db()
+            conn = get_conn()
+            try:
+                cur = conn.cursor()
+                cur.execute(
+                    """
+                    INSERT INTO task_logs (task_type, status, created_at, updated_at)
+                    VALUES ('stable-ts', 'success', 1234567890, 1234567891)
+                    """
+                )
+                conn.commit()
+            finally:
+                conn.close()
+
+            init_db()
+            conn = get_conn()
+            try:
+                cur = conn.cursor()
+                cur.execute(
+                    "SELECT created_at, updated_at FROM task_logs WHERE task_type = 'stable-ts'"
+                )
+                self.assertEqual(cur.fetchone(), (1234567890, 1234567891))
+            finally:
+                conn.close()
+
+    def test_timestamp_metadata_contract(self) -> None:
+        with IsolatedDataRoot():
+            from db.connection import get_conn, init_db
+            from db.display_metadata import DATETIME_UNIX_SECONDS
+
+            init_db()
+            conn = get_conn()
+            try:
+                cur = conn.cursor()
+                cur.execute(
+                    """
+                    SELECT column_name, display_name, display_type, editable, display_order
+                    FROM _jiangchang_columns
+                    WHERE table_name = 'task_logs'
+                      AND column_name IN ('created_at', 'updated_at')
+                    """
+                )
+                by_name = {
+                    name: (display_name, display_type, editable, display_order)
+                    for name, display_name, display_type, editable, display_order in cur.fetchall()
+                }
+                self.assertEqual(by_name["created_at"][0], "创建时间")
+                self.assertEqual(by_name["updated_at"][0], "更新时间")
+                self.assertEqual(by_name["created_at"][1], DATETIME_UNIX_SECONDS)
+                self.assertEqual(by_name["updated_at"][1], DATETIME_UNIX_SECONDS)
+                self.assertEqual(int(by_name["created_at"][2]), 0)
+                self.assertEqual(int(by_name["updated_at"][2]), 0)
+                self.assertIsNone(by_name["created_at"][3])
+                self.assertIsNone(by_name["updated_at"][3])
+            finally:
+                conn.close()
+
+    def test_metadata_upsert_keeps_timestamp_display_type(self) -> None:
+        with IsolatedDataRoot():
+            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()
+                cur.execute(
+                    """
+                    SELECT display_type FROM _jiangchang_columns
+                    WHERE table_name = 'task_logs' AND column_name = 'created_at'
+                    """
+                )
+                self.assertEqual(cur.fetchone()[0], DATETIME_UNIX_SECONDS)
+            finally:
+                conn.close()
+
+    def test_repository_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()
+                cur.execute(
+                    "SELECT created_at, updated_at FROM task_logs WHERE id = ?",
+                    (new_id,),
+                )
+                created_at, updated_at = cur.fetchone()
+                self.assertGreaterEqual(int(created_at), before)
+                self.assertLessEqual(int(created_at), after)
+                self.assertEqual(created_at, updated_at)
+            finally:
+                conn.close()
+
+    def test_validator_rejects_missing_display_type(self) -> None:
+        with IsolatedDataRoot():
+            from db.connection import get_conn, init_db
+            from db.display_metadata_validator import validate_template_database
+
+            init_db()
+            conn = get_conn()
+            try:
+                conn.execute(
+                    "UPDATE _jiangchang_columns SET display_type = NULL WHERE table_name = 'task_logs' AND column_name = 'created_at'"
+                )
+                conn.commit()
+                report = validate_template_database(conn)
+                self.assertFalse(report.ok)
+                self.assertTrue(any("display_type" in err for err in report.errors))
+            finally:
+                conn.close()
+
+    def test_validator_rejects_wrong_timestamp_type(self) -> None:
+        conn = sqlite3.connect(":memory:")
+        try:
+            from db.display_metadata import init_display_metadata
+            from db.display_metadata_validator import validate_display_metadata
+
+            conn.executescript(
+                """
+                CREATE TABLE demo_items (
+                    id INTEGER PRIMARY KEY,
+                    title TEXT,
+                    created_at TEXT
+                );
+                """
+            )
+            init_display_metadata(conn.cursor())
+            conn.execute(
+                """
+                INSERT INTO _jiangchang_tables (table_name, display_name, visible, readonly)
+                VALUES ('demo_items', '演示', 1, 0)
+                """
+            )
+            for column_name, display_name in (
+                ("id", "编号"),
+                ("title", "标题"),
+                ("created_at", "创建时间"),
+            ):
+                conn.execute(
+                    """
+                    INSERT INTO _jiangchang_columns
+                    (table_name, column_name, display_name, visible, searchable, editable, display_type)
+                    VALUES ('demo_items', ?, ?, 1, 0, 0, 'datetime_unix_seconds')
+                    """,
+                    (column_name, display_name),
+                )
+            conn.commit()
+            report = validate_display_metadata(conn)
+            self.assertFalse(report.ok)
+            self.assertTrue(any("created_at 字段类型应为 INTEGER" in err for err in report.errors))
+        finally:
+            conn.close()
+
+    def test_validator_rejects_editable_timestamp(self) -> None:
+        with IsolatedDataRoot():
+            from db.connection import get_conn, init_db
+            from db.display_metadata_validator import validate_template_database
+
+            init_db()
+            conn = get_conn()
+            try:
+                conn.execute(
+                    "UPDATE _jiangchang_columns SET editable = 1 WHERE table_name = 'task_logs' AND column_name = 'updated_at'"
+                )
+                conn.commit()
+                report = validate_template_database(conn)
+                self.assertFalse(report.ok)
+                self.assertTrue(any("editable 应为 0" in err for err in report.errors))
+            finally:
+                conn.close()
+
+    def test_validator_does_not_flag_amount_as_timestamp(self) -> None:
+        conn = sqlite3.connect(":memory:")
+        try:
+            from db.display_metadata import init_display_metadata
+            from db.display_metadata_validator import validate_display_metadata
+
+            conn.executescript(
+                """
+                CREATE TABLE demo_items (
+                    id INTEGER PRIMARY KEY,
+                    amount INTEGER,
+                    count INTEGER
+                );
+                """
+            )
+            init_display_metadata(conn.cursor())
+            conn.execute(
+                """
+                INSERT INTO _jiangchang_tables (table_name, display_name, visible, readonly)
+                VALUES ('demo_items', '演示', 1, 0)
+                """
+            )
+            for column_name, display_name in (
+                ("id", "编号"),
+                ("amount", "数量"),
+                ("count", "计数"),
+            ):
+                conn.execute(
+                    """
+                    INSERT INTO _jiangchang_columns
+                    (table_name, column_name, display_name, visible, searchable, editable)
+                    VALUES ('demo_items', ?, ?, 1, 1, 1)
+                    """,
+                    (column_name, display_name),
+                )
+            conn.commit()
+            report = validate_display_metadata(conn)
+            self.assertFalse(any("amount" in err and "display_type" in err for err in report.errors))
+            self.assertFalse(any("count" in err and "display_type" in err for err in report.errors))
+        finally:
+            conn.close()
+
+    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
+
+            init_db()
+            init_db()
+            conn = get_conn()
+            try:
+                self.assertTrue(has_updated_at_trigger(conn.cursor(), "task_logs"))
+                cur = conn.cursor()
+                cur.execute(
+                    "SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger' AND name = ?",
+                    (TASK_LOGS_UPDATED_AT_TRIGGER,),
+                )
+                self.assertEqual(int(cur.fetchone()[0]), 1)
+            finally:
+                conn.close()
+
+
+if __name__ == "__main__":
+    unittest.main()