audit: route automation audit rows through hash-chain pair (gap 4, option A)

Bypass closed: automations._audit_row (direct audit_log INSERT,
best-effort swallow) is deleted. run_due_automation now writes via
cloud.auth.audit_on_conn, the same audit_log + audit_ledger pair
AuthStore.audit() writes, sharing created_at. The audit pair runs
before record_automation_run's commit, so task row, run row and
audit pair commit or roll back together; chain-head IntegrityError
retries the whole block up to 5 times, other errors roll back and
raise (fail-closed). The claim stays committed, so a failed run is
retried next interval, never recorded as a failed run.
AuthStore.audit() now delegates to audit_on_conn; its pair, retry
count and fail-closed semantics are unchanged.

cloud/auth.py

@@ -40,6 +40,55 @@ _ACTION_MIN_ROLE = {
     "manage_project": "owner",
 }
 
+# audit_log 建表 SQL（AuthStore.__init__ 与 audit_on_conn 共用同一
+# 份，绝不能各写一份后漂移）。PG 适配层把其中的
+# INTEGER PRIMARY KEY AUTOINCREMENT 翻译成 identity 列，见
+# cloud/db.py；SQLite 下原样执行。
+_AUDIT_LOG_TABLE_SQL = (
+    "CREATE TABLE IF NOT EXISTS audit_log("
+    "id INTEGER PRIMARY KEY AUTOINCREMENT, "
+    "user_id TEXT, "
+    "project_id TEXT, "
+    "action TEXT, "
+    "detail TEXT, "
+    "created_at REAL)"
+)
+
+
+def audit_on_conn(conn, user_id, project_id, action, detail="",
+                  created_at=None):
+    """在调用方给定的连接上写一对审计记录（单次尝试）。
+
+    与 AuthStore.audit() 是同一对写：audit_log 行 + audit_ledger
+    链条目同 created_at（调用方不给时取当前时间），先建两张表
+    （幂等），再写两行。本函数不 commit、不 rollback、不重试：
+    事务边界属于调用方——调用方须在自己的事务里 commit，失败
+    时 rollback 整段业务写、链头唯一冲突时 rollback 后重试整段
+    （只重试本对写会丢掉调用方同事务里已写但未提交的业务行；
+    见 cloud/ledger.py append_entry 的警告）。AuthStore.audit()
+    即本函数 + “自有连接上 commit、唯一冲突重试 5 次”的包装。
+
+    存在理由：automations.run_due_automation 拿到的是调用方连接
+    （任务 INSERT 与产出记录都在该连接的事务里），不能像服务端
+    handler 那样另开 AuthStore（自有连接、自有事务）去写审计——
+    那样审计与任务不在同一事务，审计失败时任务已落盘，所谓
+    “整对回滚”只保审计对、不保业务对。本函数让自动化执行器
+    的审计行与任务行、产出记录行同事务提交、同事务回滚。
+    """
+    if created_at is None:
+        created_at = time.time()
+    created_at = float(created_at)
+    conn.execute(_AUDIT_LOG_TABLE_SQL)
+    _ledger_ensure(conn)
+    conn.execute(
+        "INSERT INTO audit_log(user_id, project_id, action,"
+        " detail, created_at)"
+        " VALUES(?, ?, ?, ?, ?)",
+        (user_id, project_id, action, detail, created_at),
+    )
+    _ledger_append(conn, user_id, project_id, action,
+                   detail=detail, created_at=created_at)
+
 
 class AuthStore:
     """基于 SQLite 的用户 / 项目成员 / 审计存储。"""
@@ -61,15 +110,7 @@ class AuthStore:
             "role TEXT, "
             "PRIMARY KEY(project_id, user_id))"
         )
-        self.conn.execute(
-            "CREATE TABLE IF NOT EXISTS audit_log("
-            "id INTEGER PRIMARY KEY AUTOINCREMENT, "
-            "user_id TEXT, "
-            "project_id TEXT, "
-            "action TEXT, "
-            "detail TEXT, "
-            "created_at REAL)"
-        )
+        self.conn.execute(_AUDIT_LOG_TABLE_SQL)
         # 防篡改账本（哈希链）：audit_log 本身不动，链写进
         # audit_ledger；建表失败即建库失败，不静默丢链。
         _ledger_ensure(self.conn)
@@ -361,19 +402,17 @@ class AuthStore:
         “audit_log 有行、链上无行”的静默缺口）。并发抢链头
         导致唯一冲突时在事务边界重试整对写入，最多 5 次；
         其它错误直接回滚抛错，不重试。
+
+        实现即 audit_on_conn() + 本连接上的 commit/重试包装，
+        对写本身与 #241 完全一致（同一函数、同一 created_at
+        语义），不是第二套审计写路径。
         """
         now = time.time()
         last_exc = None
         for _attempt in range(5):
             try:
-                self.conn.execute(
-                    "INSERT INTO audit_log(user_id, project_id, action,"
-                    " detail, created_at)"
-                    " VALUES(?, ?, ?, ?, ?)",
-                    (user_id, project_id, action, detail, now),
-                )
-                _ledger_append(self.conn, user_id, project_id, action,
-                               detail=detail, created_at=now)
+                audit_on_conn(self.conn, user_id, project_id, action,
+                              detail=detail, created_at=now)
                 self.conn.commit()
                 return
             except Exception as exc:

cloud/automations.py

@@ -38,13 +38,20 @@
   id 记在 automation_runs.task_id 里，派发与任务绑定，不用
   为自动化另起一套通知系统)。
 
-本模块只依赖标准库与同为标准库叶子的 cloud.quota，不
-import 其他 cloud 模块，避免循环 import；表迁移写法仿
-peer_messages(CREATE TABLE IF NOT EXISTS + PRAGMA 查列补齐，
-SQLite 与 Postgres 适配层同款)。
+本模块的 cloud 依赖只有两处叶子引用，避免循环 import：
+配额判定走同为标准库叶子的 cloud.quota；审计行上链
+（方案 A，见 run_due_automation）只走 cloud.auth.audit_on_conn——
+与 AuthStore.audit() 同一对写（audit_log 行 + audit_ledger
+链条目，同 created_at），不再有 audit_log 直写旁路。
+auth 只依赖 db / ledger 两个叶模块，三者都不 import
+automations / server / worker，不构成循环 import；其余 SQL
+（配额口径、表迁移）仍就地实现，不另 import 业务模块。
+表迁移写法仿 peer_messages(CREATE TABLE IF NOT
+EXISTS + PRAGMA 查列补齐，SQLite 与 Postgres 适配层同款）。
 """
 
 import os
+import sqlite3
 import time
 import uuid
 from datetime import datetime, timezone
@@ -53,6 +60,7 @@ from cloud.quota import decide as _quota_decide
 from cloud.quota import default_quota_usd_micros as _quota_default
 from cloud.quota import monthly_usage as _quota_monthly_usage
 from cloud.quota import quota_for_owner as _quota_for_owner
+from cloud.auth import audit_on_conn as _audit_on_conn
 
 MIN_INTERVAL_SECONDS = 60
 MAX_INTERVAL_SECONDS = 30 * 86400
@@ -595,51 +603,19 @@ def _automation_quota_exceeded(conn, owner):
     return None
 
 
-def _audit_row(conn, project_id, action, detail, now):
-    """best-effort 写审计行：失败不影响自动化运行本身。
+def _is_integrity_error(exc):
+    """链头唯一冲突判定：与 cloud/ledger.py _is_integrity_error
+    同规则（sqlite3.IntegrityError，或 PG 适配层映射后的
+    PostgresIntegrityError，按名判定，驱动类不跨模块 import）。
 
-    旧实现显式插 TEXT uuid 当 id：SQLite 下能跑，但生产 PG 的
-    audit_log.id 是 BIGINT GENERATED BY DEFAULT AS IDENTITY，
-    插 TEXT 直接报错、审计行全丢（best-effort 吞掉后无声无息）。
-    修法：INSERT 一律不带 id，让两端自增/identity 生成；
-    表不存在时先按 PG 口径建、再按 SQLite 口径建、逐个重试，
-    哪边建表成功就用哪边——automations 按设计不 import 其它
-    cloud 模块，所以用“先插、失败再按口径重试”做方言探测，
-    不靠 import 判断后端。
+    只有这类错误值得在事务边界重试整段；其它错误重试无用，
+    必须立即回滚抛错，不能被重试掩盖成“锁/冲突问题”。
+    旧 _audit_row 已删除：本模块不再有任何 audit_log 直写，
+    审计只走 _audit_on_conn（见模块文档）。
     """
-    params = ("system", project_id, action, detail, now)
-    try:
-        conn.execute(
-            "INSERT INTO audit_log(user_id, project_id, action,"
-            " detail, created_at) VALUES(?, ?, ?, ?, ?)", params)
-        conn.commit()
-        return
-    except Exception:
-        try:
-            conn.rollback()
-        except Exception:
-            pass
-    for ddl in (
-            "CREATE TABLE IF NOT EXISTS audit_log("
-            "id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, "
-            "user_id TEXT, project_id TEXT, action TEXT, detail TEXT, "
-            "created_at DOUBLE PRECISION)",
-            "CREATE TABLE IF NOT EXISTS audit_log("
-            "id INTEGER PRIMARY KEY AUTOINCREMENT, "
-            "user_id TEXT, project_id TEXT, action TEXT, detail TEXT, "
-            "created_at REAL)"):
-        try:
-            conn.execute(ddl)
-            conn.execute(
-                "INSERT INTO audit_log(user_id, project_id, action,"
-                " detail, created_at) VALUES(?, ?, ?, ?, ?)", params)
-            conn.commit()
-            return
-        except Exception:
-            try:
-                conn.rollback()
-            except Exception:
-                pass
+    if isinstance(exc, sqlite3.IntegrityError):
+        return True
+    return type(exc).__name__ == "PostgresIntegrityError"
 
 
 def run_due_automation(conn, automation_id, now=None):
@@ -651,9 +627,22 @@ def run_due_automation(conn, automation_id, now=None):
     失败重试循环）并记审计，返回该产出；配额内才 INSERT 云
     端任务（queued，executor/model 留 NULL 走 worker 默认，
     与服务端建任务缺省行为一致），产出记 created 并记审计。
-    任务 INSERT 与产出记录由 record_automation_run 的 commit
-    一并落盘。调用方（未来的定时执行器）只调本函数，绝不
+    调用方（未来的定时执行器）只调本函数，绝不
     自己拼 claim + INSERT，否则配额门可被绕过。
+
+    审计失败语义（方案 A，旧 best-effort _audit_row 已删除）：
+    审计对写（audit_log 行 + audit_ledger 链条目，同 created_at
+    = now，走 cloud.auth.audit_on_conn）先于产出记录的 commit
+    执行，任务行、产出记录行、审计对行同事务提交：审计失败
+    时三者一起回滚，异常抛给执行器——不记“任务已建、审计缺
+    失”的半截状态，也不把审计失败记成 failed 产出、不进失败
+    重试循环。抢单已单独提交（last_run_at 已写），按 claim 的
+    文档，这次执行等下个间隔再跑，与“抢单后崩掉等下个间隔
+    补跑”一致。链头唯一冲突时整段回滚重试，最多 5 次（与
+    AuthStore.audit 同数）；其它错误立即回滚抛错，不重试。
+    审计行 user_id 仍为 "system"，detail 格式不变
+    （created："<自动化id> -> <任务id>"；skipped：
+    "<自动化id>: <配额门文本>"）。
     """
     if now is None:
         now = time.time()
@@ -670,46 +659,62 @@ def run_due_automation(conn, automation_id, now=None):
     prompt = row_get(auto, "prompt")
     owner = row_get(auto, "created_by")
     gate = _automation_quota_exceeded(conn, owner)
-    if gate is not None:
-        run = record_automation_run(
-            conn, automation_id, project_id, task_id=None,
-            status="skipped_quota", detail=gate, now=now)
-        _audit_row(conn, project_id, "automation_run_quota_skipped",
-                   "%s: %s" % (automation_id, gate), now)
-        return run
     task_id = uuid.uuid4().hex
-    # 任务列补齐（与 ensure_automations_tables 的补列惯例同款）：
-    # 裸库/老库的 tasks 表可能没有 executor 等列，INSERT 前补齐，
-    # 否则建任务直接报错；PG 适配层会翻译 PRAGMA 查询。
-    existing_tasks = _existing_columns(conn, "tasks")
-    for col, ddl in (
-            ("executor", "TEXT"),
-            ("model", "TEXT"),
-            ("effort", "TEXT"),
-            ("environment_id", "TEXT"),
-            ("parent_task_id", "TEXT"),
-            ("coordinator_part", "TEXT")):
-        if col not in existing_tasks:
+    last_exc = None
+    for _attempt in range(5):
+        try:
+            if gate is not None:
+                _audit_on_conn(
+                    conn, "system", project_id,
+                    "automation_run_quota_skipped",
+                    "%s: %s" % (automation_id, gate), created_at=now)
+                return record_automation_run(
+                    conn, automation_id, project_id, task_id=None,
+                    status="skipped_quota", detail=gate, now=now)
+            # 任务列补齐（与 ensure_automations_tables 的补列惯例
+            # 同款）：裸库/老库的 tasks 表可能没有 executor 等列，
+            # INSERT 前补齐，否则建任务直接报错；PG 适配层会翻译
+            # PRAGMA 查询。放在重试循环内：回滚会撤销未提交的
+            # 补列，重试时须重新查列补齐，不能只补一次。
+            existing_tasks = _existing_columns(conn, "tasks")
+            for col, ddl in (
+                    ("executor", "TEXT"),
+                    ("model", "TEXT"),
+                    ("effort", "TEXT"),
+                    ("environment_id", "TEXT"),
+                    ("parent_task_id", "TEXT"),
+                    ("coordinator_part", "TEXT")):
+                if col not in existing_tasks:
+                    try:
+                        conn.execute(
+                            "ALTER TABLE tasks ADD COLUMN %s %s"
+                            % (col, ddl))
+                    except Exception:
+                        pass
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, prompt, status,"
+                " idempotency_key, depends_on, priority, created_by,"
+                " created_at, updated_at, last_heartbeat, executor,"
+                " model, effort, environment_id, parent_task_id,"
+                " coordinator_part)"
+                " VALUES(?, ?, ?, 'queued', NULL, '[]', 0, ?, ?, ?,"
+                " NULL, NULL, NULL, NULL, NULL, NULL, NULL)",
+                (task_id, project_id, prompt, owner, now, now))
+            _audit_on_conn(
+                conn, "system", project_id, "automation_run_created",
+                "%s -> %s" % (automation_id, task_id), created_at=now)
+            return record_automation_run(
+                conn, automation_id, project_id, task_id=task_id,
+                status="created", detail="", now=now)
+        except Exception as exc:
+            last_exc = exc
             try:
-                conn.execute(
-                    "ALTER TABLE tasks ADD COLUMN %s %s" % (col, ddl))
+                conn.rollback()
             except Exception:
                 pass
-    conn.execute(
-        "INSERT INTO tasks(id, project_id, prompt, status,"
-        " idempotency_key, depends_on, priority, created_by,"
-        " created_at, updated_at, last_heartbeat, executor,"
-        " model, effort, environment_id, parent_task_id,"
-        " coordinator_part)"
-        " VALUES(?, ?, ?, 'queued', NULL, '[]', 0, ?, ?, ?,"
-        " NULL, NULL, NULL, NULL, NULL, NULL, NULL)",
-        (task_id, project_id, prompt, owner, now, now))
-    run = record_automation_run(
-        conn, automation_id, project_id, task_id=task_id,
-        status="created", detail="", now=now)
-    _audit_row(conn, project_id, "automation_run_created",
-               "%s -> %s" % (automation_id, task_id), now)
-    return run
+            if not _is_integrity_error(exc):
+                raise
+    raise last_exc
 
 
 def list_automation_runs(conn, automation_id, limit=TRIAGE_LIMIT):

cloud/tests/test_audit_automations_chain.py

@@ -0,0 +1,260 @@
+"""自动化执行器审计上链（缺口 4，方案 A）单测。
+
+钉住的不是“写了审计行”，而是“审计行与链条目成对、与业务行
+同事务”：
+
+- created / skipped 两路：audit_log 行与 audit_ledger 行同
+  created_at、同 user_id/action/detail，verify_chain ok。
+- 链头唯一冲突：整段（任务 + 产出 + 审计对）回滚重试一次后
+  成功，最终各恰好一行，不出现任务已建、审计缺失的半截态。
+- 非唯一冲突的审计失败：fail-closed 抛错，任务 + 产出 + 审计
+  对全部回滚（0 行），只有已单独提交的抢单 last_run_at 留下，
+  下个间隔可补跑——不是吞错、不是 failed 产出。
+- audit_on_conn 与 AuthStore.audit() 是同一对写：前者写的行
+  能作为后者的链前驱，prev 衔接、verify_chain ok。
+- 同间隔内第二次跑不写任何新行。
+"""
+
+import os
+import re
+import sqlite3
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+from cloud.auth import AuthStore, audit_on_conn
+from cloud.automations import (
+    create_automation,
+    ensure_automations_tables,
+    run_due_automation,
+)
+from cloud.ledger import GENESIS_HASH, verify_chain
+from cloud.server import init_db
+
+
+def _count(conn, sql, params=()):
+    return conn.execute(sql, params).fetchone()[0]
+
+
+class _ProxyConn:
+    """把除指定拦截外的调用原样委托给真实连接的代理。
+
+    run_due_automation 只用 conn.execute / commit / rollback，
+    代理这三者即足以模拟链写失败；其余属性（如 row_factory）
+    不用代理——调用方不读它们。
+    """
+
+    def __init__(self, real):
+        self._real = real
+
+    def execute(self, sql, params=()):
+        return self._real.execute(sql, params)
+
+    def commit(self):
+        return self._real.commit()
+
+    def rollback(self):
+        return self._real.rollback()
+
+
+class _FlakyLedgerConn(_ProxyConn):
+    """首个 audit_ledger INSERT 报唯一冲突，之后放行。"""
+
+    def __init__(self, real):
+        super().__init__(real)
+        self.flaked = False
+
+    def execute(self, sql, params=()):
+        if (not self.flaked
+                and sql.strip().upper().startswith(
+                    "INSERT INTO AUDIT_LEDGER")):
+            self.flaked = True
+            raise sqlite3.IntegrityError(
+                "UNIQUE constraint failed: audit_ledger.entry_hash")
+        return super().execute(sql, params)
+
+
+class _BrokenLedgerConn(_ProxyConn):
+    """全部 audit_ledger INSERT 报非唯一冲突错误。"""
+
+    def execute(self, sql, params=()):
+        if sql.strip().upper().startswith("INSERT INTO AUDIT_LEDGER"):
+            raise RuntimeError("ledger storage boom")
+        return super().execute(sql, params)
+
+
+class AutomationChainTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        self.conn = sqlite3.connect(self.db_path)
+        self.conn.row_factory = sqlite3.Row
+        ensure_automations_tables(self.conn)
+
+    def tearDown(self):
+        self.conn.close()
+        self.tmp.cleanup()
+
+    def _auto(self, owner="owner1"):
+        return create_automation(self.conn, "proj", "daily", "do thing",
+                                 owner, interval_seconds=60)
+
+    def _set_quota(self, user_id, micros):
+        self.conn.execute(
+            "CREATE TABLE IF NOT EXISTS user_quotas("
+            "user_id TEXT PRIMARY KEY, quota_usd_micros BIGINT, "
+            " updated_at DOUBLE PRECISION)")
+        self.conn.execute(
+            "INSERT OR REPLACE INTO user_quotas(user_id,"
+            " quota_usd_micros, updated_at) VALUES(?, ?, ?)",
+            (user_id, micros, time.time()))
+        self.conn.commit()
+
+    def _pair_counts(self):
+        # 缺表计 0：SQLite 的 DDL 可事务回滚，首跑审计失败回滚时
+        # 会连 audit_on_conn 刚建的 audit_log / audit_ledger 表一起
+        # 撤销——下次跑由 CREATE IF NOT EXISTS 重建，不是丢数据。
+        out = []
+        for table in ("tasks", "automation_runs", "audit_log",
+                      "audit_ledger"):
+            try:
+                out.append(_count(
+                    self.conn, "SELECT COUNT(*) FROM %s" % table))
+            except sqlite3.OperationalError:
+                out.append(0)
+        return tuple(out)
+
+    def _check_pair(self, action, detail):
+        audit = self.conn.execute(
+            "SELECT user_id, project_id, action, detail, created_at"
+            " FROM audit_log WHERE action = ?", (action,)).fetchone()
+        self.assertIsNotNone(audit)
+        self.assertEqual(audit[0], "system")
+        self.assertEqual(audit[1], "proj")
+        self.assertEqual(audit[3], detail)
+        entry = self.conn.execute(
+            "SELECT user_id, project_id, task_id, action, detail,"
+            " created_at, prev_hash, entry_hash FROM audit_ledger"
+            " WHERE action = ?", (action,)).fetchone()
+        self.assertIsNotNone(entry)
+        self.assertEqual(entry[0], "system")
+        self.assertEqual(entry[1], "proj")
+        self.assertEqual(entry[3], action)
+        self.assertEqual(entry[4], detail)
+        self.assertEqual(entry[5], audit[4])
+        result = verify_chain(self.conn)
+        self.assertTrue(result["ok"], result)
+        return audit, entry
+
+    def test_created_run_writes_chained_pair(self):
+        auto = self._auto()
+        run = run_due_automation(self.conn, auto["id"])
+        self.assertIsNotNone(run)
+        self.assertEqual(run["status"], "created")
+        self.assertEqual(self._pair_counts(), (1, 1, 1, 1))
+        self._check_pair("automation_run_created",
+                         "%s -> %s" % (auto["id"], run["task_id"]))
+
+    def test_skipped_run_writes_chained_pair(self):
+        auto = self._auto()
+        self._set_quota("owner1", 0)
+        run = run_due_automation(self.conn, auto["id"])
+        self.assertIsNotNone(run)
+        self.assertEqual(run["status"], "skipped_quota")
+        self.assertIsNone(run["task_id"])
+        self.assertEqual(self._pair_counts(), (0, 1, 1, 1))
+        audit, _entry = self._check_pair(
+            "automation_run_quota_skipped",
+            "%s: %s" % (auto["id"], run["detail"]))
+        self.assertIn("quota", audit[3])
+
+    def test_chain_head_race_retries_whole_block_once(self):
+        auto = self._auto()
+        proxy = _FlakyLedgerConn(self.conn)
+        run = run_due_automation(proxy, auto["id"])
+        self.assertTrue(proxy.flaked)
+        self.assertIsNotNone(run)
+        self.assertEqual(run["status"], "created")
+        # 重试后仍恰好各一行：回滚掉了第一次的任务 + 审计行，
+        # 没有“任务两行、审计一行”或反过来的半截态。
+        self.assertEqual(self._pair_counts(), (1, 1, 1, 1))
+        self._check_pair("automation_run_created",
+                         "%s -> %s" % (auto["id"], run["task_id"]))
+        # 抢单在重试循环外、已单独提交，不因回滚丢失。
+        claimed = self.conn.execute(
+            "SELECT last_run_at FROM automations WHERE id = ?",
+            (auto["id"],)).fetchone()[0]
+        self.assertIsNotNone(claimed)
+
+    def test_non_integrity_audit_failure_rolls_back_all(self):
+        auto = self._auto()
+        proxy = _BrokenLedgerConn(self.conn)
+        with self.assertRaises(RuntimeError):
+            run_due_automation(proxy, auto["id"])
+        # 任务、产出、审计对全部回滚：不是吞错后留半截。
+        self.assertEqual(self._pair_counts(), (0, 0, 0, 0))
+        # 抢单已提交：下个间隔可补跑，这次不记 failed 产出。
+        claimed = self.conn.execute(
+            "SELECT last_run_at FROM automations WHERE id = ?",
+            (auto["id"],)).fetchone()[0]
+        self.assertIsNotNone(claimed)
+
+    def test_second_run_same_interval_writes_nothing_more(self):
+        auto = self._auto()
+        run = run_due_automation(self.conn, auto["id"])
+        self.assertIsNotNone(run)
+        self.assertIsNone(run_due_automation(self.conn, auto["id"]))
+        self.assertEqual(self._pair_counts(), (1, 1, 1, 1))
+
+    def test_audit_on_conn_chains_with_auth_store_audit(self):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            audit_on_conn(conn, "u1", "p1", "manual_action", "d",
+                          created_at=123.0)
+            conn.commit()
+        finally:
+            conn.close()
+        store = AuthStore(self.db_path)
+        try:
+            store.audit("u2", "p1", "second_action", "e")
+            result = verify_chain(store.conn)
+            self.assertTrue(result["ok"], result)
+            entries = store.conn.execute(
+                "SELECT action, prev_hash FROM audit_ledger"
+                " ORDER BY id").fetchall()
+            self.assertEqual(len(entries), 2)
+            self.assertEqual(entries[0][0], "manual_action")
+            self.assertEqual(entries[0][1], GENESIS_HASH)
+            first_hash = store.conn.execute(
+                "SELECT entry_hash FROM audit_ledger ORDER BY id"
+                " LIMIT 1").fetchone()[0]
+            self.assertEqual(entries[1][1], first_hash)
+        finally:
+            store.conn.close()
+
+
+class NoAuditBypassTestCase(unittest.TestCase):
+    """生产代码里 audit_log 的 INSERT 只允许出现在 auth.py。
+
+    这是缺口 4 的“无旁路”自证：将来任何模块再起 audit_log
+    直写（哪怕只是 best-effort），本测试即红，须先走
+    audit_on_conn / AuthStore.audit 才能合入。测试固件里的
+    直写（test_tamper_ledger 的篡改模拟、test_chat_history 的
+    旧行模拟）不在 cloud/*.py 内，不在本断言范围。
+    """
+
+    def test_only_auth_writes_audit_log(self):
+        cloud_dir = Path(__file__).resolve().parents[1]
+        offenders = []
+        for path in sorted(cloud_dir.glob("*.py")):
+            text = path.read_text(encoding="utf-8")
+            if re.search(r"INSERT\s+INTO\s+audit_log", text,
+                         re.IGNORECASE):
+                offenders.append(path.name)
+        self.assertEqual(offenders, ["auth.py"])
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_automation_run_due.py

@@ -5,13 +5,14 @@ import tempfile
 import time
 import unittest
 
+from cloud.auth import audit_on_conn
 from cloud.automations import (
-    _audit_row,
     create_automation,
     ensure_automations_tables,
     list_automation_runs,
     run_due_automation,
 )
+from cloud.ledger import GENESIS_HASH, compute_entry_hash
 from cloud.server import init_db
 
 
@@ -49,25 +50,48 @@ def _set_quota(db_path, user_id, micros):
         conn.close()
 
 
-class AuditRowPgIdentityTestCase(unittest.TestCase):
+class AuditPairPgIdentityTestCase(unittest.TestCase):
     """模拟 PG：id 是 BIGINT identity，显式插 TEXT uuid 会报错。
 
-    假连接只接受不带 id 的 INSERT；旧实现（带 TEXT id）在此
-    假连接上应失败、新实现应成功且审计行落盘。
+    假连接只接受不带 id 的 INSERT；旧 _audit_row（带 TEXT id）
+    在此假连接上应失败。接替它的 audit_on_conn 须同时做到：
+    audit_log 行不带 id、链条目与该行同 created_at、空链时
+    prev 为 GENESIS、entry_hash 按规范载荷重算一致；且本函数
+    不 commit（事务属于调用方，见 run_due_automation 文档）。
+
+    假连接类名须为 PgConnection：ledger.ensure_schema 靠类名
+    判定 PG 连接以跳过 SQLite 触发器，与真实 PG 适配层同款。
     """
 
-    class _PgLikeConn:
+    class _FakeCursor:
+        def __init__(self, row=None):
+            self._row = row
+            self.lastrowid = 1
+
+        def fetchone(self):
+            return self._row
+
+    class PgConnection:
         def __init__(self):
-            self.rows = []
+            self.audit_rows = []
+            self.ledger_rows = []
             self.committed = False
 
         def execute(self, sql, params=()):
-            if sql.strip().upper().startswith("INSERT INTO AUDIT_LOG(ID"):
+            upper = sql.strip().upper()
+            if upper.startswith("INSERT INTO AUDIT_LOG(ID"):
                 raise ValueError("column \"id\" is of type bigint")
-            if sql.strip().upper().startswith("INSERT INTO AUDIT_LOG"):
-                self.rows.append(params)
-                return None
-            raise ValueError("unexpected sql in fake PG conn")
+            if upper.startswith("INSERT INTO AUDIT_LOG"):
+                self.audit_rows.append(params)
+                return AuditPairPgIdentityTestCase._FakeCursor()
+            if upper.startswith("SELECT ENTRY_HASH FROM AUDIT_LEDGER"):
+                return AuditPairPgIdentityTestCase._FakeCursor(None)
+            if upper.startswith("INSERT INTO AUDIT_LEDGER"):
+                self.ledger_rows.append(params)
+                return AuditPairPgIdentityTestCase._FakeCursor()
+            if upper.startswith("CREATE TABLE"):
+                return AuditPairPgIdentityTestCase._FakeCursor()
+            raise ValueError("unexpected sql in fake PG conn: %r" % sql)
 
         def commit(self):
             self.committed = True
@@ -75,15 +99,33 @@ class AuditRowPgIdentityTestCase(unittest.TestCase):
         def rollback(self):
             pass
 
-    def test_insert_without_id_succeeds_on_pg_like_conn(self):
-        conn = self._PgLikeConn()
-        _audit_row(conn, "p1", "automation_run_created",
-                   "auto1 -> task1", 123.0)
-        self.assertTrue(conn.committed)
-        self.assertEqual(len(conn.rows), 1)
+    def test_pair_without_id_succeeds_on_pg_like_conn(self):
+        conn = self.PgConnection()
+        audit_on_conn(conn, "system", "p1", "automation_run_created",
+                      "auto1 -> task1", created_at=123.0)
+        # 调用方拥有事务：本函数自己不 commit。
+        self.assertFalse(conn.committed)
+        self.assertEqual(len(conn.audit_rows), 1)
         # 参数里没有 uuid：第一位是 system，第二位是 project_id。
-        self.assertEqual(conn.rows[0][0], "system")
-        self.assertEqual(conn.rows[0][1], "p1")
+        self.assertEqual(conn.audit_rows[0][0], "system")
+        self.assertEqual(conn.audit_rows[0][1], "p1")
+        self.assertEqual(conn.audit_rows[0][4], 123.0)
+        self.assertEqual(len(conn.ledger_rows), 1)
+        ledger = conn.ledger_rows[0]
+        # audit_ledger 列序：user_id, project_id, task_id, action,
+        # detail, created_at, prev_hash, entry_hash。
+        self.assertEqual(ledger[0], "system")
+        self.assertEqual(ledger[1], "p1")
+        self.assertIsNone(ledger[2])
+        self.assertEqual(ledger[3], "automation_run_created")
+        self.assertEqual(ledger[4], "auto1 -> task1")
+        self.assertEqual(ledger[5], 123.0)
+        self.assertEqual(ledger[6], GENESIS_HASH)
+        self.assertEqual(
+            ledger[7],
+            compute_entry_hash(GENESIS_HASH, "system", "p1", None,
+                               "automation_run_created",
+                               "auto1 -> task1", 123.0))
 
 
 class RunDueAutomationTestCase(unittest.TestCase):