codex-app-server/cloud/automations.py

"""自动化与分诊收件箱(P1 第 2 项)。

与 host.py 现有 automations / run-due 的关系(先说清楚，避免
两套调度打架)：

- host.py 的自动化存在 host state JSON 文件里(字段 name /
  task / repo / interval / last_run)，run-due 按
  now - last_run >= interval 判断到期，到期派发一个 host 任务
  跑一轮，跑完才回写 last_run。daemon tick 与 crontab 都只
  调这一个 run-due，它是 host 侧唯一的执行器。
- 本模块是云端控制面的另一套登记口径：自动化定义存在控制面
  数据库里(automations 表，带 project_id)，不是 host state
  文件。两者存储不相交，同一个自动化绝不会同时出现在两边，
  所以本批不存在双重执行。
- 云端执行器入口是本模块的 run_due_automation（残留批次 B
  落地，host.py 的 run-due 区不动）：它先走 claim_due_automation
  抢单，抢到后再过配额门、建云端任务、记 automation_runs 一
  行；抢不到的执行器绝不能建任务。这个 claim 门就是防重复
  执行的唯一关口，比 host run-due 的“跑完才写 last_run”更严
  （中途被杀也不会在当间隔内重跑，代价是崩掉的那次要等到
  下个间隔才补跑，见 claim 文档）。配额门与服务端建任务口
  径一致（见 _automation_quota_exceeded）：超配额时不建任务，
  产出记 skipped_quota（不是失败，不进失败重试循环），并记审计。
- 分诊收件箱是只读视图：到期自动化、失败或中断的任务、待
  处理的后续消息、最近的自动化产出记录。它不跑模型、不改
  任务状态，只把待处理项聚到一处。

通知接入(本批结论：遗留，不新写通知代码)：

- server 没有接 webhooks.NotificationService(它只有可注入
  sender 的占位实现，server docstring 也写明未接成)；真正
  的终态派发是 worker 走 cloud/push.py dispatch_event(订阅
  制，completed / failed / goal-unmet 等终态才发)。
- 自动化本身没有终态事件可派：定义创建不是完成，取消不是
  失败；硬在创建/取消时发通知是伪造成功。本批只记审计
  (automation_created / automation_cancelled)。等执行器
  落地后，它建的云端任务自然走 worker 现有终态派发(任务
  id 记在 automation_runs.task_id 里，派发与任务绑定，不用
  为自动化另起一套通知系统)。

本模块的 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

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
MAX_NAME_LEN = 100
MAX_PROMPT_LEN = 20000
MAX_REPO_URL_LEN = 1000
TRIAGE_LIMIT = 50

ACTIVE = "active"
CANCELLED = "cancelled"


def row_get(row, name):
    """行取值容错：sqlite3.Row 缺列报 IndexError、dict 缺键报
    KeyError、空行报 TypeError，一律回 None(老库行缺列时
    读自动化列表不能因此炸掉整页)。"""
    try:
        return row[name]
    except (KeyError, IndexError, TypeError):
        return None


AUTOMATION_COLUMNS = (
    "id", "project_id", "name", "prompt", "repo_url",
    "interval_seconds", "status", "last_run_at", "last_claimed_at",
    "created_by", "created_at", "cancelled_at",
    "first_run_at", "run_once")
RUN_COLUMNS = (
    "id", "automation_id", "project_id", "task_id", "status",
    "detail", "created_at")


def _row_dict(row, columns):
    """把行统一成可按名取值的形态：dict 与 sqlite3.Row 原样
    用；裸 sqlite3 连接给的是 tuple(按 SELECT 列顺序)，
    按建表列顺序 zip 成 dict——直接调模块的调用方不一定设
    row_factory，模块不能因此在取 row[id] 处炸掉。"""
    if row is None or isinstance(row, dict):
        return row
    try:
        row["id"]
        return row
    except TypeError:
        return dict(zip(columns, row))


def _existing_columns(conn, table):
    try:
        return {row[1] for row in conn.execute(
            "PRAGMA table_info(%s)" % table).fetchall()}
    except Exception:
        return set()


def ensure_automations_tables(conn):
    """幂等建自动化表与产出记录表，并补齐新增列。

    automations：一条定义；status 只取 active / cancelled，
    取消是软删除(行保留、审计可查，名字可重用)。
    automation_runs：一次执行产出；task_id 为空表示执行器
    抢单后尚未建任务(只应短暂存在，执行器建任务后补记)。
    调用方在写成功后 commit(server 的 handler 与 peer 模式
    同款：建表后由业务写操作一并 commit)。"""
    conn.execute(
        "CREATE TABLE IF NOT EXISTS automations("
        "id TEXT PRIMARY KEY, "
        "project_id TEXT, "
        "name TEXT, "
        "prompt TEXT, "
        "repo_url TEXT, "
        "interval_seconds INTEGER, "
        "status TEXT, "
        "last_run_at DOUBLE PRECISION, "
        "last_claimed_at DOUBLE PRECISION, "
        "created_by TEXT, "
        "created_at DOUBLE PRECISION, "
        "cancelled_at DOUBLE PRECISION, "
        "first_run_at DOUBLE PRECISION, "
        "run_once INTEGER)"
    )
    conn.execute(
        "CREATE TABLE IF NOT EXISTS automation_runs("
        "id TEXT PRIMARY KEY, "
        "automation_id TEXT, "
        "project_id TEXT, "
        "task_id TEXT, "
        "status TEXT, "
        "detail TEXT, "
        "created_at DOUBLE PRECISION)"
    )
    existing = _existing_columns(conn, "automations")
    for col, ddl in (
            ("repo_url", "TEXT"),
            ("last_claimed_at", "DOUBLE PRECISION"),
            ("cancelled_at", "DOUBLE PRECISION"),
            ("first_run_at", "DOUBLE PRECISION"),
            ("run_once", "INTEGER")):
        if col not in existing:
            try:
                conn.execute(
                    "ALTER TABLE automations ADD COLUMN %s %s"
                    % (col, ddl))
            except Exception:
                pass
    existing_runs = _existing_columns(conn, "automation_runs")
    for col, ddl in (
            ("task_id", "TEXT"),
            ("detail", "TEXT")):
        if col not in existing_runs:
            try:
                conn.execute(
                    "ALTER TABLE automation_runs ADD COLUMN %s %s"
                    % (col, ddl))
            except Exception:
                pass
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_automations_project"
        " ON automations(project_id, status)")
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_automation_runs"
        " ON automation_runs(automation_id, created_at)")
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_automation_runs_project"
        " ON automation_runs(project_id, created_at)")


def next_run_at(auto, now=None):
    """下次到期时刻：
    - 只跑一次且已跑过：None（不再到期，不是“下次”）；
    - 从未跑过且给了 first_run_at：就是该时刻（“明天 9 点”
      的首次运行就靠它，而不是建完即到期）；
    - 从未跑过且没给 first_run_at：创建时间（建完即到期，
      执行器下一轮即可认领，沿用旧语义）；
    - 跑过：上次执行 + 间隔（first_run_at 只管首次，之后
      不再看它）。"""
    last = row_get(auto, "last_run_at")
    if row_get(auto, "run_once") and last is not None:
        return None
    if last is None:
        first = row_get(auto, "first_run_at")
        if first is not None:
            return first
        return row_get(auto, "created_at")
    interval = row_get(auto, "interval_seconds") or 0
    return last + interval


def is_due(auto, now=None):
    """是否到期：仅 active 的定义可到期；now 缺省取当前时间。
    到期边界用 >=，与 host run-due 的 now - last_run >=
    interval 同款(含边界那一秒)。只跑一次且已跑过时永远
    不到期（先于 next_run_at 判断）；next_run_at 为 None
    的其它情形（行缺创建时间等脏数据）仍按旧代码认到期，
    不静默改口径。"""
    if row_get(auto, "status") != ACTIVE:
        return False
    if now is None:
        now = time.time()
    if row_get(auto, "run_once") and row_get(auto, "last_run_at") \
            is not None:
        return False
    nxt = next_run_at(auto)
    if nxt is None:
        return True
    return now >= nxt


def automation_to_dict(row, now=None):
    row = _row_dict(row, AUTOMATION_COLUMNS)
    nxt = next_run_at(row)
    return {
        "id": row["id"],
        "project_id": row["project_id"],
        "name": row["name"],
        "prompt": row["prompt"],
        "repo_url": row_get(row, "repo_url"),
        "interval_seconds": row["interval_seconds"],
        "status": row["status"],
        "last_run_at": row_get(row, "last_run_at"),
        "last_claimed_at": row_get(row, "last_claimed_at"),
        "created_by": row["created_by"],
        "created_at": row["created_at"],
        "cancelled_at": row_get(row, "cancelled_at"),
        "first_run_at": row_get(row, "first_run_at"),
        "run_once": bool(row_get(row, "run_once")),
        "next_run_at": nxt,
        "due": is_due(row, now),
    }


def _validate_fields(name, prompt, interval_seconds, repo_url,
                     first_run_at=None, run_once=False):
    if not name or not isinstance(name, str) or not name.strip():
        raise ValueError("name required")
    name = name.strip()
    if len(name) > MAX_NAME_LEN:
        raise ValueError("name too long")
    if not prompt or not isinstance(prompt, str) or not prompt.strip():
        raise ValueError("prompt required")
    prompt = prompt.strip()
    if len(prompt) > MAX_PROMPT_LEN:
        raise ValueError("prompt too long")
    if isinstance(interval_seconds, bool) \
            or not isinstance(interval_seconds, int):
        raise ValueError("interval_seconds must be an integer")
    if interval_seconds < MIN_INTERVAL_SECONDS:
        raise ValueError(
            "interval_seconds must be at least %d" % MIN_INTERVAL_SECONDS)
    if interval_seconds > MAX_INTERVAL_SECONDS:
        raise ValueError("interval_seconds too large")
    if repo_url is not None:
        if not isinstance(repo_url, str):
            raise ValueError("repo_url must be a string")
        repo_url = repo_url.strip() or None
        if repo_url is not None and len(repo_url) > MAX_REPO_URL_LEN:
            raise ValueError("repo_url too long")
    if first_run_at is not None:
        if isinstance(first_run_at, bool) or \
                not isinstance(first_run_at, (int, float)):
            raise ValueError("first_run_at must be a timestamp")
        first_run_at = float(first_run_at)
        if first_run_at <= 0:
            raise ValueError("first_run_at must be positive")
    if not isinstance(run_once, bool):
        raise ValueError("run_once must be a boolean")
    return name, prompt, interval_seconds, repo_url, \
        first_run_at, run_once


def create_automation(conn, project_id, name, prompt, created_by,
                      interval_seconds=86400, repo_url=None, now=None,
                      first_run_at=None, run_once=False):
    """创建自动化定义：同项目内 active 的名字唯一(取消过的
    名字可重用，与环境名唯一规则同款语义)。成功时 commit
    并返回 dict；校验失败抛 ValueError(调用方转 400)。
    first_run_at 为首次运行时刻（epoch 秒，过去则建完即
    到期）；run_once 为 True 时跑过一次后不再到期。"""
    (name, prompt, interval_seconds, repo_url,
     first_run_at, run_once) = _validate_fields(
        name, prompt, interval_seconds, repo_url,
        first_run_at, run_once)
    if now is None:
        now = time.time()
    ensure_automations_tables(conn)
    existing = conn.execute(
        "SELECT id FROM automations WHERE project_id = ?"
        " AND name = ? AND status = ?",
        (project_id, name, ACTIVE)).fetchone()
    if existing is not None:
        raise ValueError("automation name already exists in this project")
    auto_id = uuid.uuid4().hex
    conn.execute(
        "INSERT INTO automations(id, project_id, name, prompt,"
        " repo_url, interval_seconds, status, last_run_at,"
        " last_claimed_at, created_by, created_at, cancelled_at,"
        " first_run_at, run_once)"
        " VALUES(?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?, ?, NULL, ?, ?)",
        (auto_id, project_id, name, prompt, repo_url,
         interval_seconds, ACTIVE, created_by, now,
         first_run_at, 1 if run_once else 0))
    conn.commit()
    row = conn.execute(
        "SELECT * FROM automations WHERE id = ?", (auto_id,)).fetchone()
    return automation_to_dict(row, now)


def list_automations(conn, project_id, now=None):
    """列出某项目全部自动化定义(含已取消，取消的 due 为
    False)，按创建时间排序。调用方须先 ensure 表。"""
    if now is None:
        now = time.time()
    ensure_automations_tables(conn)
    rows = conn.execute(
        "SELECT * FROM automations WHERE project_id = ?"
        " ORDER BY created_at, id",
        (project_id,)).fetchall()
    return [automation_to_dict(r, now) for r in rows]


def get_automation(conn, automation_id):
    """取单条定义；不存在回 None(调用方转 404)。"""
    ensure_automations_tables(conn)
    row = conn.execute(
        "SELECT * FROM automations WHERE id = ?", (automation_id,)
    ).fetchone()
    if row is None:
        return None
    return automation_to_dict(row)


def cancel_automation(conn, automation_id, now=None):
    """取消自动化(软删除)：仅 active 的行能取消。

    返回 (ok, auto)：ok 为 True 时 auto 是取消后的 dict；
    行不存在时 (False, None)；已取消时 (False, 旧 dict)，
    调用方据此区分 404 与 409。成功时 commit。"""
    if now is None:
        now = time.time()
    ensure_automations_tables(conn)
    row = conn.execute(
        "SELECT * FROM automations WHERE id = ?", (automation_id,)
    ).fetchone()
    if row is None:
        return False, None
    row = _row_dict(row, AUTOMATION_COLUMNS)
    if row["status"] != ACTIVE:
        return False, automation_to_dict(row, now)
    cur = conn.execute(
        "UPDATE automations SET status = ?, cancelled_at = ?"
        " WHERE id = ? AND status = ?",
        (CANCELLED, now, automation_id, ACTIVE))
    conn.commit()
    if cur.rowcount != 1:
        return False, automation_to_dict(row, now)
    row = conn.execute(
        "SELECT * FROM automations WHERE id = ?", (automation_id,)
    ).fetchone()
    return True, automation_to_dict(row, now)


def update_automation(conn, automation_id, fields, now=None):
    """更新自动化定义（automation_update）：fields 只认
    name / prompt / interval_seconds / repo_url / first_run_at /
    run_once 六个键，其它键一律忽略（不报错、不存入，避免
    调用方拼错字段名后以为已改）；六个键全没给时抛
    ValueError（空更新不是成功）。校验与创建同款
    （_validate_fields，名字仍按“同项目 active 名字唯一”
    查重，改名撞名时 400）。

    返回 (ok, auto)：行不存在时 (False, None)；已取消时
    (False, 旧 dict)——取消是终态，改已取消的定义必须
    重建，不能靠 update 复活（与 cancel 的软删除语义
    一致）。成功时 commit。注意：更新不重置 last_run_at /
    last_claimed_at——改间隔只影响下一次到期判断（next_run
    由 last_run_at + 新间隔算），不立即触发一轮；想立即
    跑一轮应走执行器的 claim 路径，不是改定义。"""
    if now is None:
        now = time.time()
    ensure_automations_tables(conn)
    row = conn.execute(
        "SELECT * FROM automations WHERE id = ?", (automation_id,)
    ).fetchone()
    if row is None:
        return False, None
    row = _row_dict(row, AUTOMATION_COLUMNS)
    if row["status"] != ACTIVE:
        return False, automation_to_dict(row, now)
    if not isinstance(fields, dict):
        raise ValueError("fields must be an object")
    current = automation_to_dict(row, now)
    name = fields.get("name", current["name"])
    prompt = fields.get("prompt", current["prompt"])
    interval = fields.get("interval_seconds",
                          current["interval_seconds"])
    repo_url = fields.get("repo_url", current["repo_url"])
    first_run_at = fields.get("first_run_at",
                              current["first_run_at"])
    run_once = fields.get("run_once", current["run_once"])
    if not any(k in fields for k in (
            "name", "prompt", "interval_seconds", "repo_url",
            "first_run_at", "run_once")):
        raise ValueError("no updatable field given")
    (name, prompt, interval, repo_url,
     first_run_at, run_once) = _validate_fields(
        name, prompt, interval, repo_url, first_run_at, run_once)
    if name != current["name"]:
        clash = conn.execute(
            "SELECT id FROM automations WHERE project_id = ?"
            " AND name = ? AND status = ? AND id != ?",
            (row["project_id"], name, ACTIVE, automation_id)
        ).fetchone()
        if clash is not None:
            raise ValueError(
                "automation name already exists in this project")
    conn.execute(
        "UPDATE automations SET name = ?, prompt = ?,"
        " interval_seconds = ?, repo_url = ?, first_run_at = ?,"
        " run_once = ? WHERE id = ?",
        (name, prompt, interval, repo_url, first_run_at,
         1 if run_once else 0, automation_id))
    conn.commit()
    row = conn.execute(
        "SELECT * FROM automations WHERE id = ?", (automation_id,)
    ).fetchone()
    return True, automation_to_dict(row, now)


# 到期 SQL（due 与 claim 共用，绝不能只改一处）：
# - 从未跑过：没给 first_run_at 时即到期；给了则只到该时刻才到期；
# - 跑过：上次 + 间隔 <= now 才到期；
# - 只跑一次且已跑过：永远不到期（run_once=1 且 last_run_at 非空）。
_DUE_SQL = (
    " AND (run_once IS NULL OR run_once = 0 OR last_run_at IS NULL)"
    " AND ((last_run_at IS NULL AND (first_run_at IS NULL"
    " OR first_run_at <= ?))"
    " OR (last_run_at IS NOT NULL AND"
    " last_run_at + interval_seconds <= ?))")


def due_automations(conn, project_id, now=None):
    """列出某项目当前到期的 active 定义(纯读，不抢单)。
    SQL 层算到期：与 is_due 的 Python 算法同款，见 _DUE_SQL。"""
    if now is None:
        now = time.time()
    ensure_automations_tables(conn)
    rows = conn.execute(
        "SELECT * FROM automations WHERE project_id = ?"
        " AND status = ?" + _DUE_SQL +
        " ORDER BY created_at, id",
        (project_id, ACTIVE, now, now)).fetchall()
    return [automation_to_dict(r, now) for r in rows]


def claim_due_automation(conn, automation_id, now=None):
    """抢单：未来执行器的唯一入口。

    条件 UPDATE 只有一行能成功：同一自动化在同一间隔内，
    两个并发执行器只有一个拿到 rowcount 1，另一个拿到 0
    就必须放弃建任务——这是防重复执行的硬门。抢单即把
    last_run_at 写成 now(不是等执行完才写，与 host run-due
    相反)：好处是崩掉的执行不会在当间隔内被另一执行器重
    跑；代价是抢单后崩掉的那次产出会缺失，直到下个间隔才
    补跑。执行器抢单成功后必须调 record_automation_run 记
    产出，否则分诊收件箱里这次执行不可见。成功时 commit。
    返回 True 表示抢到，False 表示不存在、已取消或未到期。
    """
    if now is None:
        now = time.time()
    ensure_automations_tables(conn)
    cur = conn.execute(
        "UPDATE automations SET last_run_at = ?,"
        " last_claimed_at = ?"
        " WHERE id = ? AND status = ?" + _DUE_SQL,
        (now, now, automation_id, ACTIVE, now, now))
    conn.commit()
    return cur.rowcount == 1


def record_automation_run(conn, automation_id, project_id, task_id=None,
                          status="created", detail="", now=None):
    """记一次自动化产出：执行器建任务后调它，task_id 为建好
    的云端任务 id；status 取 created / completed / failed 等
    执行器自定的终态词(本模块不校验任务状态机，只原样存)。
    自动化不存在时抛 ValueError(执行器绝不能给不存在的
    定义记产出)。成功时 commit 并返回 run dict。"""
    if now is None:
        now = time.time()
    ensure_automations_tables(conn)
    auto = conn.execute(
        "SELECT id FROM automations WHERE id = ?",
        (automation_id,)).fetchone()
    if auto is None:
        raise ValueError("automation not found")
    run_id = uuid.uuid4().hex
    conn.execute(
        "INSERT INTO automation_runs(id, automation_id, project_id,"
        " task_id, status, detail, created_at)"
        " VALUES(?, ?, ?, ?, ?, ?, ?)",
        (run_id, automation_id, project_id, task_id, status,
         detail or "", now))
    conn.commit()
    return {
        "id": run_id,
        "automation_id": automation_id,
        "project_id": project_id,
        "task_id": task_id,
        "status": status,
        "detail": detail or "",
        "created_at": now,
    }


def _default_quota_usd_micros():
    """默认月配额（微美元）：解析收敛到 cloud/quota.py
    （同值同环境变量 CLOUD_DEFAULT_QUOTA_USD_MICROS，默认 $10）。"""
    return _quota_default()


def _unpriced_token_cap():
    """无价模型月 token 上限：与 cost.unpriced_token_cap 同值
    同环境变量（CLOUD_QUOTA_UNPRICED_TOKEN_CAP，默认 2000 万）。"""
    try:
        return max(0, int(os.environ.get(
            "CLOUD_QUOTA_UNPRICED_TOKEN_CAP", "20000000")))
    except (TypeError, ValueError):
        return 20000000


def _automation_quota_exceeded(conn, owner):
    """自动化建任务前的配额门：超限返回错误文本，否则 None。

    口径与服务端 _quota_status / worker _quota_exceeded 一致：
    按创建人算当月（UTC）已记账 USD + 无价模型 actual token；
    无配额行按默认额度（不是不限额），显式 NULL 行才不限额；
    查库失败 fail-closed（返回 quota_check_failed，调用方按跳
    过处理，绝不建任务）。cost_records 表不存在时视为零花费
    （CostStore 首次记账时才建表，不存在即从未记过账），其
    它查库异常才 fail-closed——全新库上不能因此误拒全部运行。
    取数与比较判定收敛到 cloud/quota.py（本函数只保留
    表存在探针与 PG 事务回滚处理，因其与自动化执行器的
    事务形态绑定，不宜塞进通用取数层）。
    """
    if not owner:
        return ("quota_exceeded: 自动化缺少创建人，无法查配额，"
                "拒绝建任务")
    try:
        quota = _quota_for_owner(conn, owner)
        if quota is None:
            return None
        month = datetime.now(timezone.utc).strftime("%Y-%m")
        try:
            conn.execute("SELECT COUNT(*) FROM cost_records").fetchone()
            table_exists = True
        except Exception:
            # PG 上失败语句会毒化当前事务：先 rollback，后面的
            # 任务 INSERT / 产出记录才能继续写；SQLite 上无害。
            try:
                conn.rollback()
            except Exception:
                pass
            table_exists = False
        if table_exists:
            used, unpriced_used = _quota_monthly_usage(
                conn, owner, month)
        else:
            used = 0
            unpriced_used = 0
    except Exception as exc:
        # 同上：先 rollback 再回 fail-closed，调用方仍能记
        # skipped_quota 产出（不记成失败重试循环）。
        try:
            conn.rollback()
        except Exception:
            pass
        return "quota_check_failed: %s" % exc
    decision = _quota_decide(quota, used, unpriced_used,
                             _unpriced_token_cap(), month)
    if decision is not None:
        return decision["message"]
    return None


def _is_integrity_error(exc):
    """链头唯一冲突判定：与 cloud/ledger.py _is_integrity_error
    同规则（sqlite3.IntegrityError，或 PG 适配层映射后的
    PostgresIntegrityError，按名判定，驱动类不跨模块 import）。

    只有这类错误值得在事务边界重试整段；其它错误重试无用，
    必须立即回滚抛错，不能被重试掩盖成“锁/冲突问题”。
    旧 _audit_row 已删除：本模块不再有任何 audit_log 直写，
    审计只走 _audit_on_conn（见模块文档）。
    """
    if isinstance(exc, sqlite3.IntegrityError):
        return True
    return type(exc).__name__ == "PostgresIntegrityError"


def run_due_automation(conn, automation_id, now=None):
    """云端自动化执行器的唯一入口：抢单 -> 配额门 -> 建任务 -> 记产出。

    抢不到（不存在、已取消、未到期、已被别的执行器抢走）时
    返回 None，不记产出、不建任务。抢到后先过配额门：创建人
    超配额时不建任务，产出记 skipped_quota（不是失败，不进
    失败重试循环）并记审计，返回该产出；配额内才 INSERT 云
    端任务（queued，executor/model 留 NULL 走 worker 默认，
    与服务端建任务缺省行为一致），产出记 created 并记审计。
    调用方（未来的定时执行器）只调本函数，绝不
    自己拼 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()
    ensure_automations_tables(conn)
    if not claim_due_automation(conn, automation_id, now):
        return None
    auto = conn.execute(
        "SELECT id, project_id, prompt, created_by FROM automations"
        " WHERE id = ?",
        (automation_id,)).fetchone()
    if auto is None:
        return None
    project_id = row_get(auto, "project_id")
    prompt = row_get(auto, "prompt")
    owner = row_get(auto, "created_by")
    gate = _automation_quota_exceeded(conn, owner)
    task_id = uuid.uuid4().hex
    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.rollback()
            except Exception:
                pass
            if not _is_integrity_error(exc):
                raise
    raise last_exc


def list_automation_runs(conn, automation_id, limit=TRIAGE_LIMIT):
    """列某自动化的最近产出记录，按时间倒序。"""
    ensure_automations_tables(conn)
    rows = conn.execute(
        "SELECT * FROM automation_runs WHERE automation_id = ?"
        " ORDER BY created_at DESC, id LIMIT ?",
        (automation_id, limit)).fetchall()
    out = []
    for raw in rows:
        r = _row_dict(raw, RUN_COLUMNS)
        out.append({
            "id": r["id"],
            "automation_id": r["automation_id"],
            "project_id": r["project_id"],
            "task_id": row_get(r, "task_id"),
            "status": r["status"],
            "detail": row_get(r, "detail"),
            "created_at": r["created_at"],
        })
    return out


def triage_inbox(conn, project_id, now=None):
    """分诊收件箱(只读，不写业务行、不改任务状态)：

    - automations_due：当前到期的自动化定义。
    - failed_tasks：失败或中断的任务(最需要人工看的项)，
      按更新时间倒序。
    - pending_followups：仍 pending 的后续消息(含任务间
      消息排队的那部分，带 sender_task_id)，按创建时间排序；
      followups 表不存在时该段为空(不因此炸掉整页)。
    - recent_runs：最近的自动化产出记录，带对应任务的当前
      状态(任务删了则 task_status 为 None，记录本身保留)。

    调用方(server handler)须先 ensure 自动化表与 followups
    表；本函数内 followups 缺表时只吞缺表错，其它错照常抛
    出(调用方转 502，不伪造一页空收件箱)。"""
    if now is None:
        now = time.time()
    ensure_automations_tables(conn)
    due = due_automations(conn, project_id, now)
    task_rows = conn.execute(
        "SELECT id, status, prompt, created_at, updated_at,"
        " error_text FROM tasks WHERE project_id = ?"
        " AND status IN ('failed', 'interrupted')"
        " ORDER BY updated_at DESC, id LIMIT ?",
        (project_id, TRIAGE_LIMIT)).fetchall()
    failed = []
    for raw in task_rows:
        r = _row_dict(raw, ("id", "status", "prompt", "created_at",
                            "updated_at", "error_text"))
        failed.append({
            "id": r["id"],
            "status": r["status"],
            "prompt": (r["prompt"] or "")[:300],
            "created_at": r["created_at"],
            "updated_at": r["updated_at"],
            "error_text": (row_get(r, "error_text") or "")[:300],
        })
    try:
        fu_rows = conn.execute(
            "SELECT f.id, f.task_id, f.body, f.mode, f.status,"
            " f.sender_task_id, f.created_at FROM task_followups f"
            " JOIN tasks t ON t.id = f.task_id"
            " WHERE f.project_id = ? AND f.status = 'pending'"
            " ORDER BY f.created_at, f.id LIMIT ?",
            (project_id, TRIAGE_LIMIT)).fetchall()
        pending = []
        for raw in fu_rows:
            r = _row_dict(raw, ("id", "task_id", "body", "mode",
                                "status", "sender_task_id",
                                "created_at"))
            pending.append({
                "id": r["id"],
                "task_id": r["task_id"],
                "body": (r["body"] or "")[:300],
                "mode": r["mode"],
                "status": r["status"],
                "sender_task_id": row_get(r, "sender_task_id"),
                "created_at": r["created_at"],
            })
    except Exception as exc:
        msg = str(exc)
        if "no such table" in msg or "does not exist" in msg:
            pending = []
        else:
            raise
    run_rows = conn.execute(
        "SELECT r.id, r.automation_id, r.task_id, r.status,"
        " r.detail, r.created_at, t.status AS task_status"
        " FROM automation_runs r"
        " LEFT JOIN tasks t ON t.id = r.task_id"
        " WHERE r.project_id = ?"
        " ORDER BY r.created_at DESC, r.id LIMIT ?",
        (project_id, TRIAGE_LIMIT)).fetchall()
    runs = []
    for raw in run_rows:
        r = _row_dict(raw, ("id", "automation_id", "task_id", "status",
                            "detail", "created_at", "task_status"))
        runs.append({
            "id": r["id"],
            "automation_id": r["automation_id"],
            "task_id": row_get(r, "task_id"),
            "status": r["status"],
            "detail": row_get(r, "detail"),
            "created_at": r["created_at"],
            "task_status": row_get(r, "task_status"),
        })
    return {
        "project_id": project_id,
        "automations_due": due,
        "failed_tasks": failed,
        "pending_followups": pending,
        "recent_runs": runs,
        "counts": {
            "automations_due": len(due),
            "failed_tasks": len(failed),
            "pending_followups": len(pending),
            "recent_runs": len(runs),
        },
    }