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/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):