codex-app-server/cloud/resident.py

"""常驻监听任务（resident watch）。

与现有三样东西的关系，先说清楚，避免把旧语义冒充成常驻：

- 任务状态机（cloud/server.py）：queued -> running ->
  completed / failed / interrupted / merged / merge-failed，
  其中 /pause 只是把进行中回合任务标 interrupted 并置
  paused_by_user=1（见 server._pause_task），不是一个可续跑
  的“已暂停常驻态”。本模块绝不复用那条路：常驻态存在
  resident_watches 表里，与任务表互不干扰。
- 云端自动化（cloud/automations.py）：定义层只有 active /
  cancelled，到期靠外部触发器调 run_due_automation 抢单建任务，
  服务端本身不起定时器。本模块复用它的两道口径而不是另起
  一套：抢单用条件 UPDATE（同一间隔内只有一个执行器能拿到
  rowcount 1），配额门直接调 automations._automation_quota_
  exceeded（与 server / worker 同口径，改口径时只改那一处）。
- goal 循环（host.py run_goal_loop，max_rounds=0 不限轮次）：
  目标是“做完一个目标”，结束条件是模型宣告达成；本模块的
  目标是“一直监听外部变化”，结束条件只能是用户结束（或
  配额耗尽转暂停，见下），两者不是一回事。

状态机（本模块自有，与任务状态机不冲突）：
    running -> paused -> running（恢复）
    running / paused -> ended（终态）
    ended -> 任何操作一律拒绝（恢复必须新建任务，见 end_watch）

暂停保留现场：暂停只改 status / paused_reason / next_poll_at，
last_content_hash、last_poll_at、last_change_at、last_change_text
一律不动；暂停期间 due 为 False，驱动绝不轮询。恢复时
next_poll_at 置为 now（立刻复查一次），用旧游标比对：暂停
期间网站变了，恢复后第一次轮询就会把那次变化报出来，而不是
丢掉——“从游标续跑”就是这个意思。

结束是终态：ended 后 next_poll_at 为 NULL，驱动的 due 查询
只认 status='running'，所以结束后绝不会静默重启；ended 行
保留在库里供审计与查询，名字可重用（与自动化 cancelled 的
名字重用规则同款）。

变化才花钱：轮询本身是廉价抓取 + sha256 比对，不跑模型。
只有三类轮询产出：baseline（首次抓取，只建游标，不上报）、
unchanged（游标不变，不建任务）、changed（游标变了，才建
一个 queued 任务，prompt = 用户的处理提示 + 变化摘录）。
每次轮询都在 resident_polls 留一行，可查。

配额边界：配额门只在“变化要建模型任务”那一刻过，不在
轮询那一刻过（轮询不花模型额度）。配额耗尽时不结束、不丢
变化：状态转 paused、paused_reason='quota_exceeded'，游标不
推进（恢复后复查仍会报出同一变化），轮询记
changed_quota_paused。配额检查本身失败（fail-closed 文本）
时不转暂停，按瞬时失败退避处理，游标同样不推进。

驱动：服务端本身不起定时器（与自动化同款结论）。真正的
驱动是 cloud/worker.py 的 run_loop 轮次边界调
process_due_resident_watches（见 worker 里的调用点），或
任何外部触发器直接调本模块的 run_due_resident_watch /
process_due_resident_watches；两者都走 claim 门，两个驱动
同时跑也不会在同一间隔内重复轮询。

本模块只依赖标准库 + cloud.automations 的两条复用函数
（_audit_row、_automation_quota_exceeded），不 import
cloud 其它模块，避免循环 import。
"""

import hashlib
import ipaddress
import re
import time
import urllib.request
import uuid
from urllib.parse import urlsplit

from cloud.automations import _audit_row as _auto_audit_row
from cloud.automations import _automation_quota_exceeded

MIN_INTERVAL_SECONDS = 60
MAX_INTERVAL_SECONDS = 30 * 86400
MAX_NAME_LEN = 100
MAX_PROMPT_LEN = 20000
MAX_URL_LEN = 2000
MIN_FETCH_TIMEOUT_SECONDS = 1
MAX_FETCH_TIMEOUT_SECONDS = 60
DEFAULT_FETCH_TIMEOUT_SECONDS = 10
MAX_FETCH_BYTES = 1000000
MAX_CHANGE_TEXT_LEN = 2000
MAX_POLL_DETAIL_LEN = 500
POLL_LIMIT = 50

RUNNING = "running"
PAUSED = "paused"
ENDED = "ended"

# 变化判定口径（v1）：对抓取文本做空白折叠后取 sha256。
# 动态页面（广告、时间戳）本身会抖动，v1 不做选择器/区域
# 提取，抖动风险写在验证摘要的已知偏差里，不假装已解决。
_WS_RE = re.compile(r"\s+")


def row_get(row, name):
    """行取值容错：与 automations.row_get 同款语义。"""
    try:
        return row[name]
    except (KeyError, IndexError, TypeError):
        return None


WATCH_COLUMNS = (
    "id", "project_id", "name", "target_url", "prompt",
    "interval_seconds", "fetch_timeout_seconds", "status",
    "last_poll_at", "last_claimed_at", "next_poll_at",
    "last_content_hash", "last_change_at", "last_change_text",
    "consecutive_failures", "paused_reason", "ended_at",
    "created_by", "created_at", "updated_at")

POLL_COLUMNS = (
    "id", "watch_id", "project_id", "status", "observed_hash",
    "detail", "task_id", "created_at")


def _row_dict(row, columns):
    if row is None or isinstance(row, dict):
        return row
    try:
        row["id"]
        return row
    except (KeyError, IndexError, TypeError):
        # sqlite3.Row 缺列报 IndexError（不是 TypeError）：只有
        # SELECT * 的行能原样用，单列查询（如 claim 取 interval）
        # 走 zip 回落，columns 必须与 SELECT 列顺序一致。
        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_resident_tables(conn):
    """幂等建常驻监听表，补齐新增列（仿 automations 写法）。"""
    conn.execute(
        "CREATE TABLE IF NOT EXISTS resident_watches("
        "id TEXT PRIMARY KEY, "
        "project_id TEXT, "
        "name TEXT, "
        "target_url TEXT, "
        "prompt TEXT, "
        "interval_seconds INTEGER, "
        "fetch_timeout_seconds INTEGER, "
        "status TEXT, "
        "last_poll_at DOUBLE PRECISION, "
        "last_claimed_at DOUBLE PRECISION, "
        "next_poll_at DOUBLE PRECISION, "
        "last_content_hash TEXT, "
        "last_change_at DOUBLE PRECISION, "
        "last_change_text TEXT, "
        "consecutive_failures INTEGER, "
        "paused_reason TEXT, "
        "ended_at DOUBLE PRECISION, "
        "created_by TEXT, "
        "created_at DOUBLE PRECISION, "
        "updated_at DOUBLE PRECISION)")
    conn.execute(
        "CREATE TABLE IF NOT EXISTS resident_polls("
        "id TEXT PRIMARY KEY, "
        "watch_id TEXT, "
        "project_id TEXT, "
        "status TEXT, "
        "observed_hash TEXT, "
        "detail TEXT, "
        "task_id TEXT, "
        "created_at DOUBLE PRECISION)")
    existing = _existing_columns(conn, "resident_watches")
    for col, ddl in (
            ("fetch_timeout_seconds", "INTEGER"),
            ("next_poll_at", "DOUBLE PRECISION"),
            ("last_content_hash", "TEXT"),
            ("last_change_at", "DOUBLE PRECISION"),
            ("last_change_text", "TEXT"),
            ("consecutive_failures", "INTEGER"),
            ("paused_reason", "TEXT"),
            ("ended_at", "DOUBLE PRECISION")):
        if col not in existing:
            try:
                conn.execute(
                    "ALTER TABLE resident_watches ADD COLUMN %s %s"
                    % (col, ddl))
            except Exception:
                pass
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_resident_watches_project"
        " ON resident_watches(project_id, status)")
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_resident_polls_watch"
        " ON resident_polls(watch_id, created_at)")


def _backoff_seconds(failures):
    """失败退避：30、60、120……秒，上限 3600 秒。"""
    if failures <= 0:
        return 0
    return min(3600, 30 * (2 ** (failures - 1)))


def watch_to_dict(row, now=None):
    row = _row_dict(row, WATCH_COLUMNS)
    if now is None:
        now = time.time()
    status = row["status"]
    nxt = row_get(row, "next_poll_at")
    due = bool(status == RUNNING and (nxt is None or nxt <= now))
    return {
        "id": row["id"],
        "project_id": row["project_id"],
        "name": row["name"],
        "target_url": row["target_url"],
        "prompt": row["prompt"],
        "interval_seconds": row["interval_seconds"],
        "fetch_timeout_seconds": row_get(row, "fetch_timeout_seconds")
        or DEFAULT_FETCH_TIMEOUT_SECONDS,
        "status": status,
        "last_poll_at": row_get(row, "last_poll_at"),
        "last_claimed_at": row_get(row, "last_claimed_at"),
        "next_poll_at": nxt,
        "last_content_hash": row_get(row, "last_content_hash"),
        "last_change_at": row_get(row, "last_change_at"),
        "last_change_text": row_get(row, "last_change_text"),
        "consecutive_failures": row_get(row, "consecutive_failures") or 0,
        "paused_reason": row_get(row, "paused_reason"),
        "ended_at": row_get(row, "ended_at"),
        "created_by": row["created_by"],
        "created_at": row["created_at"],
        "updated_at": row_get(row, "updated_at"),
        "due": due,
    }


def validate_target_url(url):
    """目标 URL 校验：仅 http/https、无 userinfo、主机非空、
    字面 localhost / 私有 / 回环 / 链路本地 IP 拒绝（与
    websearch._is_blocked_fetch_url 的字面 IP 口径同款，
    DNS 解析后的私有地址不在 v1 覆盖内，见摘要已知偏差）。"""
    if not url or not isinstance(url, str) or not url.strip():
        raise ValueError("target_url required")
    url = url.strip()
    if len(url) > MAX_URL_LEN:
        raise ValueError("target_url too long")
    parts = urlsplit(url)
    if parts.scheme not in ("http", "https"):
        raise ValueError("target_url must be http or https")
    if parts.username or parts.password:
        raise ValueError("target_url must not contain credentials")
    host = (parts.hostname or "").strip()
    if not host:
        raise ValueError("target_url host required")
    if host == "localhost":
        raise ValueError("localhost URLs are blocked")
    try:
        addr = ipaddress.ip_address(host)
    except ValueError:
        addr = None
    if addr is not None and (addr.is_private or addr.is_loopback
                             or addr.is_link_local or addr.is_reserved
                             or addr.is_multicast):
        raise ValueError("private / loopback URLs are blocked")
    return url


def _validate_fields(name, prompt, target_url, interval_seconds,
                     fetch_timeout_seconds):
    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")
    target_url = validate_target_url(target_url)
    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 fetch_timeout_seconds is None:
        fetch_timeout_seconds = DEFAULT_FETCH_TIMEOUT_SECONDS
    if isinstance(fetch_timeout_seconds, bool) \
            or not isinstance(fetch_timeout_seconds, int):
        raise ValueError("fetch_timeout_seconds must be an integer")
    if not MIN_FETCH_TIMEOUT_SECONDS <= fetch_timeout_seconds \
            <= MAX_FETCH_TIMEOUT_SECONDS:
        raise ValueError(
            "fetch_timeout_seconds must be between %d and %d"
            % (MIN_FETCH_TIMEOUT_SECONDS, MAX_FETCH_TIMEOUT_SECONDS))
    return name, prompt, target_url, interval_seconds, fetch_timeout_seconds


def create_watch(conn, project_id, name, target_url, prompt, created_by,
                 interval_seconds=3600, fetch_timeout_seconds=None,
                 now=None):
    """新建常驻监听：建完即到期（next_poll_at=now），驱动下
    一轮即可认领；同项目内未结束（running/paused）的名字唯一，
    已结束的名字可重用。创建本身不走配额门（轮询不花模型额度，
    配额只在变化建任务时过，见 run_due_resident_watch）。"""
    (name, prompt, target_url, interval_seconds,
     fetch_timeout_seconds) = _validate_fields(
        name, prompt, target_url, interval_seconds, fetch_timeout_seconds)
    if now is None:
        now = time.time()
    ensure_resident_tables(conn)
    clash = conn.execute(
        "SELECT id FROM resident_watches WHERE project_id = ?"
        " AND name = ? AND status != ?",
        (project_id, name, ENDED)).fetchone()
    if clash is not None:
        raise ValueError(
            "resident watch name already exists in this project")
    watch_id = uuid.uuid4().hex
    conn.execute(
        "INSERT INTO resident_watches(id, project_id, name, target_url,"
        " prompt, interval_seconds, fetch_timeout_seconds, status,"
        " last_poll_at, last_claimed_at, next_poll_at,"
        " last_content_hash, last_change_at, last_change_text,"
        " consecutive_failures, paused_reason, ended_at,"
        " created_by, created_at, updated_at)"
        " VALUES(?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?,"
        " NULL, NULL, NULL, 0, NULL, NULL, ?, ?, ?)",
        (watch_id, project_id, name, target_url, prompt,
         interval_seconds, fetch_timeout_seconds, RUNNING, now,
         created_by, now, now))
    conn.commit()
    row = conn.execute(
        "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
    return watch_to_dict(row, now)


def list_watches(conn, project_id, now=None):
    if now is None:
        now = time.time()
    ensure_resident_tables(conn)
    rows = conn.execute(
        "SELECT * FROM resident_watches WHERE project_id = ?"
        " ORDER BY created_at, id",
        (project_id,)).fetchall()
    return [watch_to_dict(r, now) for r in rows]


def get_watch(conn, watch_id, now=None):
    if now is None:
        now = time.time()
    ensure_resident_tables(conn)
    row = conn.execute(
        "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
    if row is None:
        return None
    return watch_to_dict(row, now)


def list_polls(conn, watch_id, limit=POLL_LIMIT):
    ensure_resident_tables(conn)
    rows = conn.execute(
        "SELECT * FROM resident_polls WHERE watch_id = ?"
        " ORDER BY created_at DESC, id LIMIT ?",
        (watch_id, limit)).fetchall()
    out = []
    for raw in rows:
        r = _row_dict(raw, POLL_COLUMNS)
        out.append({
            "id": r["id"],
            "watch_id": r["watch_id"],
            "project_id": r["project_id"],
            "status": r["status"],
            "observed_hash": row_get(r, "observed_hash"),
            "detail": row_get(r, "detail") or "",
            "task_id": row_get(r, "task_id"),
            "created_at": r["created_at"],
        })
    return out


def _transition(conn, watch_id, from_statuses, to_status, now,
                extra_sets=(), extra_vals=()):
    """条件状态迁移：返回 (ok, dict)。行不存在时 (False, None)；
    状态不对时 (False, 旧 dict)，调用方据此区分 404 与 409。"""
    ensure_resident_tables(conn)
    row = conn.execute(
        "SELECT * FROM resident_watches WHERE id = ?",
        (watch_id,)).fetchone()
    if row is None:
        return False, None
    row = _row_dict(row, WATCH_COLUMNS)
    if row["status"] not in from_statuses:
        return False, watch_to_dict(row, now)
    placeholders = ",".join("?" for _ in from_statuses)
    sets = ["status = ?", "updated_at = ?"] + list(extra_sets)
    vals = [to_status, now] + list(extra_vals) + [watch_id] \
        + list(from_statuses)
    cur = conn.execute(
        "UPDATE resident_watches SET %s WHERE id = ? AND status IN (%s)"
        % (", ".join(sets), placeholders), tuple(vals))
    conn.commit()
    if cur.rowcount != 1:
        return False, watch_to_dict(row, now)
    row = conn.execute(
        "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
    return True, watch_to_dict(row, now)


def pause_watch(conn, watch_id, now=None, reason="user_paused"):
    """用户暂停：仅 running 能暂停；保留全部现场，只把
    next_poll_at 清掉，驱动即停。"""
    if now is None:
        now = time.time()
    return _transition(conn, watch_id, (RUNNING,), PAUSED, now,
                       extra_sets=("paused_reason = ?",
                                   "next_poll_at = NULL"),
                       extra_vals=(reason,))


def resume_watch(conn, watch_id, now=None):
    """恢复：仅 paused 能恢复；现场不动，next_poll_at=now
    立刻复查一次（用旧游标比对，暂停期间的变化不丢）。"""
    if now is None:
        now = time.time()
    return _transition(conn, watch_id, (PAUSED,), RUNNING, now,
                       extra_sets=("paused_reason = NULL",
                                   "next_poll_at = ?"),
                       extra_vals=(now,))


def end_watch(conn, watch_id, now=None):
    """结束：running / paused -> ended，终态。ended 后
    pause / resume / end 一律 (False, 旧 dict)，调用方转 409；
    恢复已结束的监听必须新建任务，不复活本行。"""
    if now is None:
        now = time.time()
    return _transition(conn, watch_id, (RUNNING, PAUSED), ENDED, now,
                       extra_sets=("ended_at = ?",
                                   "next_poll_at = NULL"),
                       extra_vals=(now,))


def update_watch(conn, watch_id, fields, now=None):
    """更新定义：只认 name / prompt / target_url /
    interval_seconds / fetch_timeout_seconds；ended 的行 409
    （终态不能改，改它必须重建）；改间隔不重置游标、
    不立即触发：已排好的 next_poll_at 不变，下一次轮询
    完成后才按新间隔重算（想立即复查请暂停后恢复）。"""
    if now is None:
        now = time.time()
    ensure_resident_tables(conn)
    row = conn.execute(
        "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
    if row is None:
        return False, None
    row = _row_dict(row, WATCH_COLUMNS)
    if row["status"] == ENDED:
        return False, watch_to_dict(row, now)
    if not isinstance(fields, dict):
        raise ValueError("fields must be an object")
    if not any(k in fields for k in (
            "name", "prompt", "target_url", "interval_seconds",
            "fetch_timeout_seconds")):
        raise ValueError("no updatable field given")
    current = watch_to_dict(row, now)
    (name, prompt, target_url, interval, timeout) = _validate_fields(
        fields.get("name", current["name"]),
        fields.get("prompt", current["prompt"]),
        fields.get("target_url", current["target_url"]),
        fields.get("interval_seconds", current["interval_seconds"]),
        fields.get("fetch_timeout_seconds",
                   current["fetch_timeout_seconds"]))
    if name != current["name"]:
        clash = conn.execute(
            "SELECT id FROM resident_watches WHERE project_id = ?"
            " AND name = ? AND status != ? AND id != ?",
            (row["project_id"], name, ENDED, watch_id)).fetchone()
        if clash is not None:
            raise ValueError(
                "resident watch name already exists in this project")
    conn.execute(
        "UPDATE resident_watches SET name = ?, prompt = ?,"
        " target_url = ?, interval_seconds = ?,"
        " fetch_timeout_seconds = ?, updated_at = ? WHERE id = ?",
        (name, prompt, target_url, interval, timeout, now, watch_id))
    conn.commit()
    row = conn.execute(
        "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
    return True, watch_to_dict(row, now)


def due_watches(conn, now=None):
    """全部项目当前到期的 running 监听（纯读，不抢单），
    供驱动列出待处理项；执行入口是 claim + run_due。"""
    if now is None:
        now = time.time()
    ensure_resident_tables(conn)
    rows = conn.execute(
        "SELECT * FROM resident_watches WHERE status = ?"
        " AND (next_poll_at IS NULL OR next_poll_at <= ?)"
        " ORDER BY next_poll_at, created_at, id",
        (RUNNING, now)).fetchall()
    return [watch_to_dict(r, now) for r in rows]


def claim_due_watch(conn, watch_id, now=None):
    """抢单：条件 UPDATE 只有一行能成功。抢到即把
    last_poll_at / last_claimed_at 写成 now，并把 next_poll_at
    预置为 now + interval（崩在抓取中途时，下个间隔仍会重试，
    不会因认领锁死而永久停摆；抓取完成后 run_due 会按成败
    重算 next_poll_at）。抢不到时返回 False，调用方必须放弃。"""
    if now is None:
        now = time.time()
    ensure_resident_tables(conn)
    row = conn.execute(
        "SELECT interval_seconds FROM resident_watches WHERE id = ?",
        (watch_id,)).fetchone()
    if row is None:
        return False
    interval = row_get(_row_dict(row, ("interval_seconds",)),
                       "interval_seconds") or MIN_INTERVAL_SECONDS
    cur = conn.execute(
        "UPDATE resident_watches SET last_poll_at = ?,"
        " last_claimed_at = ?, next_poll_at = ?"
        " WHERE id = ? AND status = ?"
        " AND (next_poll_at IS NULL OR next_poll_at <= ?)",
        (now, now, now + interval, watch_id, RUNNING, now))
    conn.commit()
    return cur.rowcount == 1


def _content_hash(text):
    normalized = _WS_RE.sub(" ", text or "").strip()
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()


def default_fetch(url, timeout):
    """默认廉价抓取：urllib，超时可配，读满 1MB 即断，
    解码失败按 replace 处理。调用方可注入 fetcher 替换它
    （测试与未来选择器提取都走注入点，不改本函数签名）。"""
    req = urllib.request.Request(url, headers={
        "User-Agent": "codex-app-server-resident-watch/1.0"})
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        raw = resp.read(MAX_FETCH_BYTES + 1)[:MAX_FETCH_BYTES]
    return raw.decode("utf-8", errors="replace")


def _record_poll(conn, watch_id, project_id, status, observed_hash,
                 detail, task_id, now):
    poll_id = uuid.uuid4().hex
    conn.execute(
        "INSERT INTO resident_polls(id, watch_id, project_id, status,"
        " observed_hash, detail, task_id, created_at)"
        " VALUES(?, ?, ?, ?, ?, ?, ?, ?)",
        (poll_id, watch_id, project_id, status, observed_hash,
         (detail or "")[:MAX_POLL_DETAIL_LEN], task_id, now))
    return {
        "id": poll_id,
        "watch_id": watch_id,
        "project_id": project_id,
        "status": status,
        "observed_hash": observed_hash,
        "detail": (detail or "")[:MAX_POLL_DETAIL_LEN],
        "task_id": task_id,
        "created_at": now,
    }


def _create_change_task(conn, watch, excerpt, now):
    """变化建任务：列补齐口径与 automations.run_due_automation
    同款；任务 prompt = 用户处理提示 + 变化摘录。"""
    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
    task_id = uuid.uuid4().hex
    prompt = ("%s\n\n[常驻监听变化] %s %s\n变化摘录：\n%s"
              % (watch["prompt"], watch["name"], watch["target_url"],
                 excerpt))
    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, watch["project_id"], prompt, watch["created_by"],
         now, now))
    return task_id


def run_due_resident_watch(conn, watch_id, fetcher=None, now=None):
    """单条常驻监听的驱动唯一入口：抢单 -> 廉价抓取 -> hash
    比对 -> 记轮询 -> 仅变化时建任务。

    抢不到返回 None（不存在、已暂停/已结束、未到期、已被
    别的驱动抢走），不记轮询、不建任务。抓取失败记 failed
    并退避，游标不动。首次抓取记 baseline，只建游标。未变
    化记 unchanged，不建任务。变化时先过配额门：配额耗尽
    转 paused（paused_reason='quota_exceeded'，游标不推进），
    记 changed_quota_paused；配额检查失败按瞬时失败退避，
    游标同样不推进；配额内才建 queued 任务并推进游标。"""
    if now is None:
        now = time.time()
    ensure_resident_tables(conn)
    if not claim_due_watch(conn, watch_id, now):
        return None
    row = conn.execute(
        "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
    watch = watch_to_dict(row, now)
    interval = watch["interval_seconds"]
    timeout = watch["fetch_timeout_seconds"]
    fetch = fetcher or default_fetch
    try:
        text = fetch(watch["target_url"], timeout)
    except Exception as exc:
        failures = watch["consecutive_failures"] + 1
        nxt = now + interval + _backoff_seconds(failures)
        conn.execute(
            "UPDATE resident_watches SET consecutive_failures = ?,"
            " next_poll_at = ?, updated_at = ? WHERE id = ?",
            (failures, nxt, now, watch_id))
        poll = _record_poll(conn, watch_id, watch["project_id"],
                            "failed", None, str(exc)[:MAX_POLL_DETAIL_LEN],
                            None, now)
        conn.commit()
        _auto_audit_row(conn, watch["project_id"],
                        "resident_watch_poll_failed",
                        "%s: %s" % (watch_id, str(exc)[:200]), now)
        return poll
    if not isinstance(text, str):
        text = str(text)
    observed = _content_hash(text)
    excerpt = text[:MAX_CHANGE_TEXT_LEN]
    old_hash = watch["last_content_hash"]
    if old_hash is None:
        conn.execute(
            "UPDATE resident_watches SET last_content_hash = ?,"
            " consecutive_failures = 0, next_poll_at = ?,"
            " updated_at = ? WHERE id = ?",
            (observed, now + interval, now, watch_id))
        poll = _record_poll(conn, watch_id, watch["project_id"],
                            "baseline", observed, "first poll", None, now)
        conn.commit()
        return poll
    if observed == old_hash:
        conn.execute(
            "UPDATE resident_watches SET consecutive_failures = 0,"
            " next_poll_at = ?, updated_at = ? WHERE id = ?",
            (now + interval, now, watch_id))
        poll = _record_poll(conn, watch_id, watch["project_id"],
                            "unchanged", observed, "", None, now)
        conn.commit()
        return poll
    # 真实变化：配额门只在这里过。
    gate = _automation_quota_exceeded(conn, watch["created_by"])
    if gate is not None and "quota_check_failed" not in gate:
        conn.execute(
            "UPDATE resident_watches SET status = ?,"
            " paused_reason = ?, next_poll_at = NULL, updated_at = ?"
            " WHERE id = ?",
            (PAUSED, "quota_exceeded", now, watch_id))
        poll = _record_poll(conn, watch_id, watch["project_id"],
                            "changed_quota_paused", observed, gate,
                            None, now)
        conn.commit()
        _auto_audit_row(conn, watch["project_id"],
                        "resident_watch_paused_quota",
                        "%s: %s" % (watch_id, gate[:200]), now)
        return poll
    if gate is not None:
        failures = watch["consecutive_failures"] + 1
        nxt = now + interval + _backoff_seconds(failures)
        conn.execute(
            "UPDATE resident_watches SET consecutive_failures = ?,"
            " next_poll_at = ?, updated_at = ? WHERE id = ?",
            (failures, nxt, now, watch_id))
        poll = _record_poll(conn, watch_id, watch["project_id"],
                            "changed_quota_check_failed", observed, gate,
                            None, now)
        conn.commit()
        return poll
    task_id = _create_change_task(conn, watch, excerpt, now)
    conn.execute(
        "UPDATE resident_watches SET last_content_hash = ?,"
        " last_change_at = ?, last_change_text = ?,"
        " consecutive_failures = 0, next_poll_at = ?,"
        " updated_at = ? WHERE id = ?",
        (observed, now, excerpt, now + interval, now, watch_id))
    poll = _record_poll(conn, watch_id, watch["project_id"],
                        "changed", observed, excerpt[:MAX_POLL_DETAIL_LEN],
                        task_id, now)
    conn.commit()
    _auto_audit_row(conn, watch["project_id"],
                    "resident_watch_change_task_created",
                    "%s -> %s" % (watch_id, task_id), now)
    return poll


def process_due_resident_watches(conn, fetcher=None, now=None, limit=50):
    """驱动批量入口：列出到期监听，逐条走 run_due（每条自带
    claim 门，批量内重复与并发驱动都不会重复轮询）。返回
    本轮实际产出的轮询记录列表（抢不到的条目不在内）。"""
    if now is None:
        now = time.time()
    due = due_watches(conn, now)[:limit]
    out = []
    for watch in due:
        poll = run_due_resident_watch(conn, watch["id"],
                                      fetcher=fetcher, now=now)
        if poll is not None:
            out.append(poll)
    return out