"""常驻监听任务（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 门，两个驱动
同时跑也不会在同一间隔内重复轮询。

S2 二选一（本批选批量总预算，不新增独立驱动进程，论证
见验证摘要）：轮询间隔只是下限、不是 SLA——模型任务一轮
最长 1800 秒仍会拖后轮询，任何对外口径都必须写“尽力而
为、延迟无界”。批量总预算（拟定默认 120 秒，见
resident_config）把单批最坏耗时 bound 在“预算 + 单条超时
上限 60 秒”，剩余到期监听保持 due、下批再跑；延迟本身由
resident_metrics 的 due_count / max_overdue_seconds 度量，
进 /metrics，不靠间隔数值自证送达。独立驱动进程的运维
成本（新 unit、独立 DB 连接、启停监管）是审核第 7 节留给
用户的拍板项，本批不擅自新增；claim 门已保证未来加驱动
时多驱动不重复轮询。

本模块只依赖标准库 + cloud.automations 的复用函数
（row_get、_audit_row、_automation_quota_exceeded）与
cloud.db 的完整性错误类型（仅用于把唯一索引冲突转成
ValueError，不 import 其它 cloud 模块），避免循环 import。
"""

import hashlib
import http.client
import ipaddress
import os
import re
import socket
import sqlite3
import ssl
import time
import uuid
from urllib.parse import urljoin
from urllib.parse import urlsplit

from cloud.automations import _audit_row as _auto_audit_row
from cloud.automations import _automation_quota_exceeded
from cloud.peer_messages import row_get
from cloud.db import PostgresIntegrityError as _PgIntegrityError

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
MAX_REDIRECTS = 10
RESIDENT_USER_AGENT = "codex-app-server-resident-watch/1.0"

# P0/P1 拟定默认值（全部是拟定值、供最终验收改，不是用户已
# 拍板；数值只在这里与 resident_config() 一处定义，环境变量
# 可覆盖，见下文各 CODEX_RESIDENT_* 变量）：
# - 每监听每日变化任务上限 24 个：3600 秒间隔的监听一天最多
#   24 次变化，本值即“每小时最多 1 个任务”的量级；60 秒间隔
#   的抖动页面一天最多也只建 24 个任务，超限的变化记
#   changed_capped 轮询、游标不推进（次日/下次仍可报，不丢）。
# - 变化抖动冷却 300 秒：上次建任务 300 秒内的新变化记
#   changed_cooldown、游标不推进，冷却过后复查仍可报。
# - polls 保留期 90 天：批量驱动每批开头 prune 一次，过期行
#   删除、未过期保留；当日上限计数只依赖当日轮询行，保留期
#   远大于 1 天，不影响上限计数。
# - 批量总预算 120 秒：见 S2 论证（模块头“驱动”一段）。
DEFAULT_MAX_DAILY_CHANGE_TASKS = 24
DEFAULT_CHANGE_COOLDOWN_SECONDS = 300
DEFAULT_POLL_RETENTION_DAYS = 90
DEFAULT_BATCH_BUDGET_SECONDS = 120
# 结构化失败原因口径（落 poll.failure_reason 列，detail 前缀
# 同样写 reason=<原因>，老读法只看 detail 也不丢信息）：
# dns / proxy / timeout / http_status / ssrf_blocked / tls /
# redirect / network / invalid_url / unknown。
FAILURE_REASONS = frozenset(
    {"dns", "proxy", "timeout", "http_status", "ssrf_blocked", "tls",
     "redirect", "network", "invalid_url", "unknown"})


def _env_int(name, default):
    """环境变量整数：解析失败或 <=0 时回默认值，绝不因此炸掉
    驱动（配置错误按默认跑，见摘要已知偏差）。"""
    try:
        value = int(os.environ.get(name) or default)
    except (TypeError, ValueError):
        return default
    return value if value > 0 else default


def resident_config():
    """常驻监听的可配置口径（每次调用时读环境变量，重启与
    改环境变量即生效，不需改代码；数值本身是拟定默认值）：
    - proxy：CODEX_RESIDENT_PROXY，未设/空/direct = 直连（默认
      直连，不跟随进程 http_proxy/https_proxy）；env = 显式跟随
      进程环境代理（旧 urllib 默认行为，须显式 opt-in）；其它值
      = 显式代理 URL（http/https），no_proxy 仍读进程环境。
    - max_daily_change_tasks：CODEX_RESIDENT_MAX_DAILY_CHANGE_TASKS。
    - change_cooldown_seconds：CODEX_RESIDENT_CHANGE_COOLDOWN_SECONDS。
    - poll_retention_days：CODEX_RESIDENT_POLL_RETENTION_DAYS。
    - batch_budget_seconds：CODEX_RESIDENT_BATCH_BUDGET_SECONDS。
    """
    raw = os.environ.get("CODEX_RESIDENT_PROXY")
    if raw is None or not raw.strip():
        proxy = {"mode": "direct", "url": None}
    elif raw.strip().lower() == "direct":
        proxy = {"mode": "direct", "url": None}
    elif raw.strip().lower() == "env":
        proxy = {"mode": "env", "url": None}
    else:
        proxy = {"mode": "url", "url": raw.strip()}
    return {
        "proxy": proxy,
        "max_daily_change_tasks": _env_int(
            "CODEX_RESIDENT_MAX_DAILY_CHANGE_TASKS",
            DEFAULT_MAX_DAILY_CHANGE_TASKS),
        "change_cooldown_seconds": _env_int(
            "CODEX_RESIDENT_CHANGE_COOLDOWN_SECONDS",
            DEFAULT_CHANGE_COOLDOWN_SECONDS),
        "poll_retention_days": _env_int(
            "CODEX_RESIDENT_POLL_RETENTION_DAYS",
            DEFAULT_POLL_RETENTION_DAYS),
        "batch_budget_seconds": _env_int(
            "CODEX_RESIDENT_BATCH_BUDGET_SECONDS",
            DEFAULT_BATCH_BUDGET_SECONDS),
    }


def resolve_proxy_config(raw=None):
    """解析代理配置：非法代理 URL 在这里就抛 FetchError
    （reason=proxy），调用方按结构化失败记轮询，不炸驱动。"""
    cfg = resident_config()
    if raw is not None:
        # 显式传入的 raw 只用于测试与未来调用方：与环境变量
        # 同款三态（direct/env/URL），不接受其它形态。
        if raw.strip().lower() in ("direct", "env"):
            return {"mode": raw.strip().lower(), "url": None}
        if raw.strip():
            cfg = dict(cfg)
            cfg["proxy"] = {"mode": "url", "url": raw.strip()}
        else:
            cfg["proxy"] = {"mode": "direct", "url": None}
    proxy = cfg["proxy"]
    if proxy["mode"] == "url":
        parts = urlsplit(proxy["url"])
        if parts.scheme not in ("http", "https") or not parts.hostname:
            raise FetchError(
                "proxy",
                "invalid resident proxy URL: %s" % proxy["url"])
    return proxy


def _host_in_no_proxy(host, no_proxy_value):
    """no_proxy 匹配：精确、后缀（.example.com 匹配子域）、*
    全匹配；主机名统一小写去尾点，与 validate 口径一致。"""
    host = (host or "").strip().lower().rstrip(".")
    for entry in (no_proxy_value or "").split(","):
        entry = entry.strip().lower().rstrip(".")
        if not entry:
            continue
        if entry == "*":
            return True
        if host == entry or host.endswith("." + entry):
            return True
    return False


def _proxy_for_target(host, scheme, proxy):
    """本跳的目标代理决策：返回 None = 直连，否则
    (proxy_host, proxy_port)。env 模式读进程环境
    http_proxy/https_proxy（大小写均认）；url 模式用配置的
    URL；两种模式都认进程 no_proxy/NO_PROXY。直连模式永远
    返回 None——旧代码跟随进程代理的行为至此结束，改动见
    摘要已知偏差（部署时若宿主出口必须走代理，须显式配
    CODEX_RESIDENT_PROXY=env 或代理 URL）。"""
    no_proxy = os.environ.get("no_proxy") or os.environ.get("NO_PROXY") or ""
    if _host_in_no_proxy(host, no_proxy):
        return None
    if proxy["mode"] == "direct":
        return None
    if proxy["mode"] == "env":
        if scheme == "https":
            raw = os.environ.get("https_proxy") \
                or os.environ.get("HTTPS_PROXY") or ""
        else:
            raw = os.environ.get("http_proxy") \
                or os.environ.get("HTTP_PROXY") or ""
        if not raw.strip():
            return None
        parts = urlsplit(raw.strip())
        if parts.scheme not in ("http", "https") or not parts.hostname:
            raise FetchError("proxy",
                             "invalid proxy URL in environment")
        return (parts.hostname,
                parts.port or (443 if parts.scheme == "https" else 80))
    parts = urlsplit(proxy["url"])
    return (parts.hostname,
            parts.port or (443 if parts.scheme == "https" else 80))

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

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


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", "failure_reason")


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 写法）。

    调用约定（G1）：本函数只在启动/迁移/API 路径调——
    server init_db、worker 首次批量、create/list/get 等 API
    函数；热路径（claim / run_due / due 列表 / 批量循环）绝
    不调它，批量内每条的 CREATE INDEX DDL 锁与 claim UPDATE
    的锁交互正是旧双 worker 死锁的疑似根因。热路径只在
    表确实缺失时（missing-table 错误）ensure 后重试一次。
    """
    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, "
        "failure_reason TEXT)")
    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
    existing_polls = _existing_columns(conn, "resident_polls")
    if "failure_reason" not in existing_polls:
        try:
            conn.execute(
                "ALTER TABLE resident_polls"
                " ADD COLUMN failure_reason TEXT")
        except Exception:
            pass
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_resident_watches_project"
        " ON resident_watches(project_id, status)")
    # 同项目未结束名唯一：先查后插挡不住并发，唯一索引才是
    # 真正的关口（server environments 的同款部分唯一索引，
    # WHERE status != 'archived' 的 resident 版）；已结束名
    # 不在索引内，可重用。已存在重复名的旧库建索引会失败，
    # 此时不炸掉整次建表调用，SELECT 查重仍挡正常路径。
    try:
        conn.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS"
            " idx_resident_watches_name"
            " ON resident_watches(project_id, name)"
            " WHERE status != 'ended'")
    except Exception:
        pass
    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 秒。
    调用方把它加在间隔之上（下次 = now + interval + backoff），
    失败后绝不比正常间隔更频繁地重试，行为见失败分支测试。"""
    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 _is_noncanonical_numeric_host(host):
    """非规范数字 IP 写法：单整数（2130706433）、十六进制
    （0x7f000001）、前导零八进制（0177.0.0.1）、含十六进制段
    的点分写法——urllib/socket 的 inet_aton 会把它们解析成
    数字 IP，ipaddress 却认不出来。规范点分十进制（4 段、
    每段 0-255、无前导零，0 本身除外）不在此列；普通域名
    （含字母的标签）也不在此列。"""
    if re.fullmatch(r"0[xX][0-9a-fA-F]+", host):
        return True
    if re.fullmatch(r"[0-9]+", host):
        return True
    if "." not in host:
        return False
    labels = host.split(".")
    if not all(re.fullmatch(r"0[xX][0-9a-fA-F]+|[0-9]+", lab)
               for lab in labels):
        return False
    if len(labels) != 4:
        return True
    for lab in labels:
        if lab.startswith("0x") or lab.startswith("0X"):
            return True
        if len(lab) > 1 and lab.startswith("0"):
            return True
        if int(lab) > 255:
            return True
    return False


def validate_target_url(url):
    """目标 URL 字面校验：仅 http/https、无 userinfo、主机非空、
    字面 localhost（含尾点、*.localhost）/ 私有 / 回环 /
    链路本地 IP / 非规范数字 IP 写法拒绝（与
    websearch._is_blocked_fetch_url 的字面 IP 口径同款）。
    本函数只做字面层：创建监听时调它，域名解析到私网的
    情形不在创建时拦（DNS 会变，创建时放行不等于抓取时放
    行）；抓取时的 DNS 级门在 default_fetch 里逐跳执行
    （resolve_public_addresses + pin），那才是权威关口。
    私网抓取在抓取层直接失败、不产生摘录、不建任务。"""
    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().lower().rstrip(".")
    if not host:
        raise ValueError("target_url host required")
    if host == "localhost" or host.endswith(".localhost"):
        raise ValueError("localhost URLs are blocked")
    if _is_noncanonical_numeric_host(host):
        raise ValueError("non-canonical numeric IP 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
    try:
        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))
    except (sqlite3.IntegrityError, _PgIntegrityError) as exc:
        # 并发下 SELECT 查重与 INSERT 之间可能插进同名行，
        # 唯一索引是真正的关口：冲突转回与查重同款 ValueError，
        # 调用方（server）据此返回 400，不把数据库错误外泄。
        try:
            conn.rollback()
        except Exception:
            pass
        raise ValueError(
            "resident watch name already exists in this project") from exc
    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"],
            "failure_reason": row_get(r, "failure_reason"),
        })
    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 不变，下一次轮询
    完成后才按新间隔重算（想立即复查请暂停后恢复）。
    改目标 URL 则强制走一次 baseline：last_content_hash
    置 NULL，下一轮只建基线、不建任务——旧 hash 与新站
    内容比对必然报一次假变化的旧行为至此结束；改间隔、
    改提示、改名一律不重置游标（语义保留，见摘要）。"""
    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")
    target_changed = target_url != current["target_url"]
    try:
        if target_changed:
            conn.execute(
                "UPDATE resident_watches SET name = ?, prompt = ?,"
                " target_url = ?, interval_seconds = ?,"
                " fetch_timeout_seconds = ?, updated_at = ?,"
                " last_content_hash = NULL WHERE id = ?",
                (name, prompt, target_url, interval, timeout, now,
                 watch_id))
        else:
            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))
    except (sqlite3.IntegrityError, _PgIntegrityError) as exc:
        try:
            conn.rollback()
        except Exception:
            pass
        raise ValueError(
            "resident watch name already exists in this project") from exc
    conn.commit()
    row = conn.execute(
        "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
    return True, watch_to_dict(row, now)


def _is_missing_table_error(exc):
    """缺表错误判定（SQLite 与 PG 两种文案）：热路径只在这种
    错误上 ensure 后重试一次，其它错误一律不触发 DDL。"""
    msg = str(exc).lower()
    return "no such table" in msg or "does not exist" in msg


def due_watches(conn, now=None):
    """全部项目当前到期的 running 监听（纯读，不抢单），
    供驱动列出待处理项；执行入口是 claim + run_due。
    热路径不调 ensure（见 ensure_resident_tables 的约定）；
    表缺失时由批量入口的 fallback 统一处理。"""
    if now is None:
        now = time.time()
    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 _due_watches_with_fallback(conn, now):
    """批量入口的 due 列表：正常路径零 DDL；表缺失时
    ensure 一次后重试，仍失败则把原始错误抛给调用方。"""
    try:
        return due_watches(conn, now)
    except Exception as exc:
        if not _is_missing_table_error(exc):
            raise
        try:
            conn.rollback()
        except Exception:
            pass
        ensure_resident_tables(conn)
        return due_watches(conn, now)


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()
    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()


class FetchError(ValueError):
    """结构化抓取失败：reason 见 FAILURE_REASONS，status 仅
    http_status 时有值。ValueError 子类，故旧调用方按
    ValueError 捕获的行为不变；run_due 据此写 failure_reason
    与 reason= 前缀的 detail，不把原始内网响应写进任何记录。"""

    def __init__(self, reason, message, status=None):
        super().__init__(message)
        self.reason = reason if reason in FAILURE_REASONS else "unknown"
        self.status = status


class SSRFBlockedError(FetchError):
    """SSRF 拦截：字面或解析后地址命中内网/私有/回环/
    链路本地/保留/组播/未指定地址，或非规范数字写法。"""

    def __init__(self, message):
        super().__init__("ssrf_blocked", message)


class HttpStatusFetchError(FetchError):
    """非 200 响应：不把错误页当成“变化”处理，也不把它
    的正文写进变化任务（正文在 default_fetch 里直接丢弃）。"""

    def __init__(self, status, message=None):
        super().__init__(
            "http_status",
            message or ("HTTP status %s" % status), status=status)


def classify_fetch_error(exc):
    """异常 -> 结构化原因：FetchError 自带 reason；超时、
    TLS、连接类按类型分；其余一律 unknown（不谎称是网络
    错误，也不把异常文本当原因）。"""
    if isinstance(exc, FetchError):
        return exc.reason
    if isinstance(exc, (socket.timeout, TimeoutError)):
        return "timeout"
    if isinstance(exc, ssl.SSLError):
        return "tls"
    if isinstance(exc, (ConnectionError, OSError)):
        return "network"
    if isinstance(exc, ValueError):
        return "invalid_url"
    return "unknown"


def _default_resolver(host, port):
    """默认解析器：socket.getaddrinfo，去重保序。返回 IP
    字符串列表；测试可注入同签名的 resolver（见 default_fetch
    的 resolver 参数——生产路径一律不传，默认即本函数）。"""
    infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP)
    ips = []
    for info in infos:
        try:
            ip = info[4][0]
        except (IndexError, TypeError):
            continue
        if ip not in ips:
            ips.append(ip)
    return ips


def _address_is_blocked(addr):
    """单个解析地址是否命中拦截面：私有/回环/链路本地/
    保留/组播/未指定；IPv6 的 IPv4 映射地址同样按映射后
    的 IPv4 地址再判一次（::ffff:127.0.0.1 必须拦住）。"""
    if addr.is_private or addr.is_loopback or addr.is_link_local \
            or addr.is_reserved or addr.is_multicast \
            or addr.is_unspecified:
        return True
    mapped = getattr(addr, "ipv4_mapped", None)
    if mapped is not None and (
            mapped.is_private or mapped.is_loopback
            or mapped.is_link_local or mapped.is_reserved
            or mapped.is_multicast or mapped.is_unspecified):
        return True
    return False


def resolve_public_addresses(host, port, resolver=None):
    """DNS 级 SSRF 门：解析全部 A/AAAA，任一地址命中拦截面
    即整批拒绝（只放行“全部解析结果都是公网地址”的目标，
    不搞“取第一个公网地址蒙混过关”）；解析失败、空结果、
    非 IP 结果一律按 dns 失败处理，不放行、不重试建连。
    注意：本门只在直连模式下配合下文的 pin 才是完整封口；
    代理模式下建连由代理方二次解析，见 default_fetch 的
    残留风险说明。"""
    try:
        ips = (resolver or _default_resolver)(host, port)
    except FetchError:
        raise
    except Exception as exc:
        raise FetchError(
            "dns", "DNS resolution failed for %s: %s" % (host, exc))
    if not ips:
        raise FetchError("dns", "DNS resolution returned no address"
                         " for %s" % host)
    out = []
    for ipstr in ips:
        try:
            addr = ipaddress.ip_address(str(ipstr))
        except ValueError:
            raise FetchError(
                "dns", "DNS resolution returned non-IP %r for %s"
                % (ipstr, host))
        if _address_is_blocked(addr):
            raise SSRFBlockedError(
                "resolved address %s for %s is not public"
                % (ipstr, host))
        out.append(str(ipstr))
    return out


class _PinnedHTTPConnection(http.client.HTTPConnection):
    """直连时 pin 住解析结果的连接：建连目标是解析出的 IP，
    而不是再次按主机名解析（解析与建连之间的 TOCTOU 窗口
    至此关闭）；Host 头仍用原始主机名（http.client 按
    self.host 生成），服务端看到的 Host 与用户填的目标一致。"""

    def __init__(self, host, port=None, pinned_ips=(), timeout=None):
        super().__init__(host, port, timeout=timeout)
        self._pinned_ips = list(pinned_ips)

    def connect(self):
        last = None
        for ip in self._pinned_ips:
            try:
                self.sock = socket.create_connection(
                    (ip, self.port), self.timeout)
                return
            except OSError as exc:
                last = exc
        if last is not None:
            raise last
        raise OSError("no pinned address for %s" % self.host)


class _PinnedHTTPSConnection(_PinnedHTTPConnection,
                             http.client.HTTPSConnection):
    """直连 https 的 pin 连接：先按 IP 建裸 socket，再用
    默认校验上下文 wrap，server_hostname 必须是原始主机名
    （不是 IP），否则 SNI 与证书主机名校验都会错——这是
    “为 pin 而破坏 TLS 校验”的反模式，本实现明确不这么做：
    证书仍按原始主机名校验，IP 只决定 TCP 对端。"""

    def connect(self):
        _PinnedHTTPConnection.connect(self)
        context = ssl.create_default_context()
        self.sock = context.wrap_socket(self.sock,
                                        server_hostname=self.host)


def _fetch_single_hop(url, timeout, pinned_ips, proxy):
    """单跳抓取（不跟随重定向）：返回 (status, location, raw)。
    直连走 pin 连接；代理走 http.client 的代理连接（https 用
    set_tunnel，CPython 按 tunnel 主机名做 SNI，与直连同款
    证书主机名校验）。transport 异常由调用方按 _wrap 处理，
    本函数不把异常文本加工成原因。"""
    parts = urlsplit(url)
    host = (parts.hostname or "").strip().lower().rstrip(".")
    port = parts.port or (443 if parts.scheme == "https" else 80)
    path = parts.path or "/"
    if parts.query:
        path = path + "?" + parts.query
    headers = {"User-Agent": RESIDENT_USER_AGENT}
    if proxy is None:
        if parts.scheme == "https":
            conn = _PinnedHTTPSConnection(host, port, pinned_ips,
                                          timeout=timeout)
        else:
            conn = _PinnedHTTPConnection(host, port, pinned_ips,
                                         timeout=timeout)
        conn.request("GET", path, headers=headers)
    elif parts.scheme == "https":
        conn = http.client.HTTPSConnection(
            proxy[0], proxy[1], timeout=timeout,
            context=ssl.create_default_context())
        conn.set_tunnel(host, port)
        conn.request("GET", path, headers=headers)
    else:
        conn = http.client.HTTPConnection(proxy[0], proxy[1],
                                          timeout=timeout)
        conn.request("GET", url, headers=headers)
    try:
        resp = conn.getresponse()
        raw = resp.read(MAX_FETCH_BYTES + 1)[:MAX_FETCH_BYTES]
        return resp.status, resp.getheader("Location"), raw
    finally:
        try:
            conn.close()
        except Exception:
            pass


def _wrap_transport_error(exc, via_proxy):
    """transport 异常 -> FetchError：代理模式下的连接失败记
    proxy（代理本身不可达/拒绝都算代理侧失败，不算目标站
    失败）；直连按类型分超时/TLS/网络。"""
    if isinstance(exc, FetchError):
        return exc
    if via_proxy and isinstance(exc, (ConnectionError, OSError,
                                      socket.timeout, TimeoutError)):
        return FetchError("proxy", "proxy connection failed: %s" % exc)
    if isinstance(exc, (socket.timeout, TimeoutError)):
        return FetchError("timeout", "fetch timed out: %s" % exc)
    if isinstance(exc, ssl.SSLError):
        return FetchError("tls", "TLS failed: %s" % exc)
    if isinstance(exc, (ConnectionError, OSError)):
        return FetchError("network", "fetch failed: %s" % exc)
    return FetchError("unknown", "fetch failed: %s" % exc)


def default_fetch(url, timeout, resolver=None, proxy_config=None):
    """默认廉价抓取：超时可配，读满 1MB 即断，解码失败按
    replace 处理；只在 200 时返回正文，其它状态一律抛
    HttpStatusFetchError（错误页正文直接丢弃，绝不进入
    变化判定与任务摘录）。

    SSRF 封口（直连模式，完整）：
    1. 每跳先过 validate_target_url 的字面校验；
    2. 再解析全部 A/AAAA 并过 resolve_public_addresses；
    3. 建连只连解析出的 IP（pin），https 的 SNI 与证书主机
       名仍是原始主机名，TLS 校验不被 pin 破坏；
    4. 重定向每跳重复 1-3 步，相对跳转先 urljoin 再校验；
    5. 超过 MAX_REDIRECTS 跳抛 redirect 失败。
    残留风险（诚实写明，不假装已封口）：
    - 代理模式下：客户端的解析 gate 只是前置检查，真正建连
      的是代理方、由代理方二次解析；若代理方解析结果与客
      户端不一致（DNS 变化、代理方走不同解析器），pin 不生
      效，须靠代理自身的 egress allowlist 兜底。生产部署若
      用代理模式，必须同时给代理配内网地址拦截。
    - 解析与建连之间仍有理论 TOCTOU：直连模式下 pin 已把
      窗口关到“解析结果本身”这一层（建连不再二次解析），
      但 DNS 记录在解析后、建连前被改写不影响本次建连目
      标（目标就是解析时刻的 IP），残留的是“解析时刻的
      IP 在之后被重新分配给内网用途”的极端情形，不在本次
      封口承诺内。
    - resolver 参数是测试注入点（与 run_due 的 fetcher 注入
      同款）：生产路径一律不传；传了自定义 resolver 的调用
      方自行承担该解析器的可信责任。
    - 代理配置默认值是直连（见 resident_config）：旧代码跟
      随进程 http_proxy/https_proxy 的行为已改，部署时若宿
      主出口必须走代理，须显式配 CODEX_RESIDENT_PROXY=env
      或代理 URL，否则公网抓取会直连失败（记 proxy/timeout/
      network 失败轮询，不建任务）。
    调用方可注入 fetcher 替换本函数（测试与未来选择器提取
    都走注入点，不改本函数签名；注入的 fetcher 绕过本门，
    生产驱动不用注入）。"""
    if proxy_config is None:
        proxy = resolve_proxy_config()
    elif isinstance(proxy_config, dict):
        proxy = resolve_proxy_config() if proxy_config.get("mode") is None \
            else proxy_config
        if proxy.get("mode") == "url":
            resolve_proxy_config(proxy["url"])
    else:
        proxy = resolve_proxy_config(proxy_config)
    current = url
    for _hop in range(MAX_REDIRECTS + 1):
        try:
            current = validate_target_url(current)
        except ValueError as exc:
            if "blocked" in str(exc):
                raise SSRFBlockedError(str(exc))
            raise FetchError("invalid_url", str(exc))
        parts = urlsplit(current)
        host = (parts.hostname or "").strip().lower().rstrip(".")
        try:
            port = parts.port or (443 if parts.scheme == "https" else 80)
        except ValueError:
            raise FetchError("invalid_url",
                             "invalid port in URL: %s" % current)
        ips = resolve_public_addresses(host, port, resolver=resolver)
        target_proxy = _proxy_for_target(host, parts.scheme, proxy)
        try:
            status, location, raw = _fetch_single_hop(
                current, timeout, ips, target_proxy)
        except Exception as exc:
            raise _wrap_transport_error(exc, target_proxy is not None)
        if status in (301, 302, 303, 307, 308):
            if not location:
                raise FetchError(
                    "http_status", "redirect without Location")
            current = urljoin(current, location)
            continue
        if status != 200:
            raise HttpStatusFetchError(status)
        return raw.decode("utf-8", errors="replace")
    raise FetchError("redirect", "too many redirects")


def _record_poll(conn, watch_id, project_id, status, observed_hash,
                 detail, task_id, now, failure_reason=None):
    """记轮询行：failure_reason 是结构化失败原因（见
    FAILURE_REASONS），成功/变化类轮询为 None；detail 前缀
    同样写 reason=<原因>（失败类），老读法只看 detail 也不
    丢原因。失败轮询的 detail 只含异常文本，绝不含抓取正文
    （SSRF 拦截时根本没有正文可写，见 default_fetch）。"""
    poll_id = uuid.uuid4().hex
    conn.execute(
        "INSERT INTO resident_polls(id, watch_id, project_id, status,"
        " observed_hash, detail, task_id, created_at, failure_reason)"
        " VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)",
        (poll_id, watch_id, project_id, status, observed_hash,
         (detail or "")[:MAX_POLL_DETAIL_LEN], task_id, now,
         failure_reason))
    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,
        "failure_reason": failure_reason,
    }


def _utc_day_start(now):
    """now 所在 UTC 日的起点 epoch 秒：每日变化任务上限按
    UTC 日切分（不按本地时区，避免时区与夏令时切分歧义）。"""
    import datetime
    dt = datetime.datetime.fromtimestamp(now, tz=datetime.timezone.utc)
    start = dt.replace(hour=0, minute=0, second=0, microsecond=0)
    return start.timestamp()


def _changed_tasks_today(conn, watch_id, now):
    """本 UTC 日内已建的变化任务数：只数 status='changed'
    且 task_id 非空的轮询行（changed_capped / changed_cooldown /
    changed_quota_paused 都不占名额）。表缺失时按 0 计（热路
    径 fallback 之前已建表，正常路径不会走到这里）。"""
    try:
        row = conn.execute(
            "SELECT COUNT(*) FROM resident_polls WHERE watch_id = ?"
            " AND status = ? AND task_id IS NOT NULL AND created_at >= ?",
            (watch_id, "changed", _utc_day_start(now))).fetchone()
    except Exception:
        return 0
    try:
        return int(row[0])
    except (TypeError, ValueError, IndexError):
        return 0


def prune_old_polls(conn, now=None, retention_days=None):
    """删过期轮询行：created_at 早于 now - 保留期的行全部删，
    返回删除行数；保留期内（包括当日）的行一律不动。
    本函数自己 commit（批量驱动每批开头调它，失败由调用方
    兜底，不炸批量）。"""
    if now is None:
        now = time.time()
    if retention_days is None:
        retention_days = resident_config()["poll_retention_days"]
    cutoff = now - retention_days * 86400
    cur = conn.execute(
        "DELETE FROM resident_polls WHERE created_at < ?", (cutoff,))
    conn.commit()
    try:
        return int(cur.rowcount)
    except (TypeError, ValueError):
        return 0


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 _guarded_watch_update(conn, watch_id, sets, vals):
    """终局 UPDATE 带 status='running' 条件：抓取中途被用户
    暂停/结束时，条件不满足、rowcount 为 0，调用方据此回滚
    并放弃本轮（不建任务、不记轮询），而不是把 paused/ended
    行的 next_poll_at 改回未来值。返回是否成功。"""
    cur = conn.execute(
        "UPDATE resident_watches SET %s WHERE id = ? AND status = ?"
        % ", ".join(sets),
        tuple(vals) + (watch_id, RUNNING))
    return cur.rowcount == 1


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

    抢不到返回 None（不存在、已暂停/已结束、未到期、已被
    别的驱动抢走），不记轮询、不建任务。抓取失败记 failed
    （failure_reason 结构化，见 classify_fetch_error）并退避，
    游标不动。首次抓取记 baseline，只建游标。未变化记
    unchanged，不建任务。变化时依次过三道门：每日上限
    （changed_capped）、抖动冷却（changed_cooldown）、配额门
    （changed_quota_paused / changed_quota_check_failed）——
    前两道被拦时游标不推进（变化不丢，冷却/次日仍可报），
    配额耗尽转 paused（paused_reason='quota_exceeded'，游标
    不推进）；全部通过才建 queued 任务并推进游标。
    热路径不调 ensure；抢单时表缺失才 ensure 后重试一次。
    fetcher 为 None 时用 default_fetch（resolver 透传给它，
    生产路径 resolver 为 None 即默认 DNS 门）。"""
    if now is None:
        now = time.time()
    try:
        claimed = claim_due_watch(conn, watch_id, now)
    except Exception as exc:
        if not _is_missing_table_error(exc):
            raise
        try:
            conn.rollback()
        except Exception:
            pass
        ensure_resident_tables(conn)
        claimed = claim_due_watch(conn, watch_id, now)
    if not claimed:
        return None
    row = conn.execute(
        "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone()
    if row is None:
        return None
    watch = watch_to_dict(row, now)
    if watch["status"] != RUNNING:
        # 认领后、抓取前被用户暂停/结束：不抓取、不建任务、
        # 不改行（暂停/结束已把 next_poll_at 清掉，保持原样）。
        return None
    interval = watch["interval_seconds"]
    timeout = watch["fetch_timeout_seconds"]
    if fetcher is not None:
        fetch = fetcher
    else:
        def fetch(url, timeout, _resolver=resolver):
            return default_fetch(url, timeout, resolver=_resolver)
    try:
        text = fetch(watch["target_url"], timeout)
    except Exception as exc:
        reason = classify_fetch_error(exc)
        failures = watch["consecutive_failures"] + 1
        nxt = now + interval + _backoff_seconds(failures)
        if not _guarded_watch_update(
                conn, watch_id,
                ("consecutive_failures = ?", "next_poll_at = ?",
                 "updated_at = ?"),
                (failures, nxt, now)):
            conn.rollback()
            return None
        detail = "reason=%s; %s" % (reason, str(exc))
        poll = _record_poll(conn, watch_id, watch["project_id"],
                            "failed", None, detail,
                            None, now, failure_reason=reason)
        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:
        if not _guarded_watch_update(
                conn, watch_id,
                ("last_content_hash = ?", "consecutive_failures = 0",
                 "next_poll_at = ?", "updated_at = ?"),
                (observed, now + interval, now)):
            conn.rollback()
            return None
        poll = _record_poll(conn, watch_id, watch["project_id"],
                            "baseline", observed, "first poll", None, now)
        conn.commit()
        return poll
    if observed == old_hash:
        if not _guarded_watch_update(
                conn, watch_id,
                ("consecutive_failures = 0", "next_poll_at = ?",
                 "updated_at = ?"),
                (now + interval, now)):
            conn.rollback()
            return None
        poll = _record_poll(conn, watch_id, watch["project_id"],
                            "unchanged", observed, "", None, now)
        conn.commit()
        return poll
    # 真实变化，先过成本上限与抖动冷却（拟定值，见模块头）：
    # 两道门被拦时都只记轮询、不建任务、不推进游标（变化不
    # 丢：下次复查内容仍与旧游标不同，冷却过后/次日仍可报），
    # next_poll_at 按正常间隔重算（不是失败退避——这不是失败）。
    cfg = resident_config()
    if _changed_tasks_today(conn, watch_id, now) \
            >= cfg["max_daily_change_tasks"]:
        if not _guarded_watch_update(
                conn, watch_id,
                ("consecutive_failures = 0", "next_poll_at = ?",
                 "updated_at = ?"),
                (now + interval, now)):
            conn.rollback()
            return None
        poll = _record_poll(conn, watch_id, watch["project_id"],
                            "changed_capped", observed,
                            "daily change task cap reached", None, now)
        conn.commit()
        return poll
    last_change_at = watch["last_change_at"]
    if last_change_at is not None \
            and now - last_change_at < cfg["change_cooldown_seconds"]:
        if not _guarded_watch_update(
                conn, watch_id,
                ("consecutive_failures = 0", "next_poll_at = ?",
                 "updated_at = ?"),
                (now + interval, now)):
            conn.rollback()
            return None
        poll = _record_poll(conn, watch_id, watch["project_id"],
                            "changed_cooldown", observed,
                            "change cooldown active", None, now)
        conn.commit()
        return poll
    # 真实变化：配额门只在这里过。配额门契约是前缀
    # （quota_check_failed: / quota_exceeded: / None），不用
    # 子串判定，以免超限文本含该子串时误判。
    gate = _automation_quota_exceeded(conn, watch["created_by"])
    if gate is not None and not gate.startswith("quota_check_failed"):
        cur = conn.execute(
            "UPDATE resident_watches SET status = ?,"
            " paused_reason = ?, next_poll_at = NULL, updated_at = ?"
            " WHERE id = ? AND status = ?",
            (PAUSED, "quota_exceeded", now, watch_id, RUNNING))
        if cur.rowcount != 1:
            conn.rollback()
            return None
        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)
        if not _guarded_watch_update(
                conn, watch_id,
                ("consecutive_failures = ?", "next_poll_at = ?",
                 "updated_at = ?"),
                (failures, nxt, now)):
            conn.rollback()
            return None
        poll = _record_poll(conn, watch_id, watch["project_id"],
                            "changed_quota_check_failed", observed, gate,
                            None, now)
        conn.commit()
        return poll
    # 先受保护地推进游标，再建任务：抓取中途被暂停/结束时，
    # 条件不满足即回滚，本轮不建任务、不记轮询。
    if not _guarded_watch_update(
            conn, watch_id,
            ("last_content_hash = ?", "last_change_at = ?",
             "last_change_text = ?", "consecutive_failures = 0",
             "next_poll_at = ?", "updated_at = ?"),
            (observed, now, excerpt, now + interval, now)):
        conn.rollback()
        return None
    task_id = _create_change_task(conn, watch, excerpt, now)
    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 _is_retryable_lock_error(exc):
    """可重试的锁错误：PG 死锁/锁不可用（适配层 locked=True）
    与 SQLite 的 database is locked。其它错误重试无用。"""
    if getattr(exc, "locked", False):
        return True
    return "database is locked" in str(exc).lower()


def _run_one_isolated(conn, watch_id, fetcher, now, resolver, errors):
    """批量内单条隔离：锁错误回滚后重试一次（第一次死锁时
    claim 可能已提交——重试抢不到即干净返回 None；claim 未
    提交时重试可成功），仍失败则记入 errors 并继续下一条，
    绝不让一条死锁 abort 整批剩余监听。非锁异常同样隔离
    （抓取类失败已在 run_due 内记 failed 轮询，不会走到
    这里；走到这里的是 DB/建任务类异常）。"""
    for attempt in range(2):
        try:
            return run_due_resident_watch(conn, watch_id,
                                          fetcher=fetcher, now=now,
                                          resolver=resolver)
        except Exception as exc:
            try:
                conn.rollback()
            except Exception:
                pass
            if _is_retryable_lock_error(exc) and attempt == 0:
                continue
            if errors is not None:
                errors.append((watch_id, str(exc)[:200]))
            return None
    return None


def process_due_resident_watches(conn, fetcher=None, now=None, limit=50,
                                 budget_seconds=None,
                                 monotonic_fn=None, resolver=None,
                                 errors=None):
    """驱动批量入口：列出到期监听，逐条走 run_due（每条自带
    claim 门，批量内重复与并发驱动都不会重复轮询）。返回
    本轮实际产出的轮询记录列表（抢不到的条目不在内）。

    批量总预算（S2）：budget_seconds 缺省取拟定默认值
    （resident_config，120 秒）；每条开始前检查单调时钟，
     elapsed >= 预算即停——剩余到期监听保持 due 状态，下批
    再跑，不丢。预算不含“正在跑的最后一条”：最坏单批耗时
    = 预算 + 单条超时上限（60 秒），相对旧的 50×60=3000 秒
    已有界；单条超时本身仍由 fetch_timeout_seconds 卡。
    每批开头 prune 一次过期轮询（失败只回滚、不炸批量）。
    errors 为调用方传入的列表时，逐条隔离的异常与预算耗
    尽信号（watch_id=None 的条目）记入其中，供 worker 记
    日志；不传时静默丢弃（旧调用方行为不变）。"""
    if now is None:
        now = time.time()
    if budget_seconds is None:
        budget_seconds = resident_config()["batch_budget_seconds"]
    mono = monotonic_fn or time.monotonic
    started = mono()
    due = _due_watches_with_fallback(conn, now)[:limit]
    try:
        prune_old_polls(conn, now=now)
    except Exception:
        try:
            conn.rollback()
        except Exception:
            pass
    out = []
    for watch in due:
        if budget_seconds is not None \
                and mono() - started >= budget_seconds:
            if errors is not None:
                errors.append(
                    (None, "batch_budget_exceeded after %d polls"
                     % len(out)))
            break
        poll = _run_one_isolated(conn, watch["id"], fetcher, now,
                                 resolver, errors)
        if poll is not None:
            out.append(poll)
    return out


def resident_metrics(conn, project_ids, now=None):
    """resident 可观测口径（供 /metrics 用，只读不建表）：
    返回 dict，表缺失时 {"available": False}（老库未跑过
    init_db/ensure 时不炸 /metrics）。字段：
    - watches_by_status：限定项目内的监听状态计数；
    - due_count：running 且已到期（next_poll_at <= now）的条数；
    - max_overdue_seconds：到期监听的最大逾期秒数（无到期为 0）；
    - polls_last_24h / failed_last_24h / changed_last_24h /
      change_tasks_last_24h：近 24h 轮询、失败、变化、变化任务数；
    - config：当前拟定配置值（见 resident_config）。
    延迟口径说明：due_count 与 max_overdue_seconds 直接度量
    “到期未跑”的积压——驱动被模型派发拖住时，这两项会如
    实变大，不存在“间隔写 60 秒就等于 60 秒送达”的谎言。"""
    if now is None:
        now = time.time()
    ids = list(project_ids or [])
    if not ids:
        return {"available": True, "watches_by_status": {},
                "due_count": 0, "max_overdue_seconds": 0,
                "polls_last_24h": 0, "failed_last_24h": 0,
                "changed_last_24h": 0, "change_tasks_last_24h": 0,
                "config": resident_config()}
    placeholders = ",".join("?" for _ in ids)
    try:
        watch_rows = conn.execute(
            "SELECT status, next_poll_at FROM resident_watches"
            " WHERE project_id IN (%s)" % placeholders,
            tuple(ids)).fetchall()
    except Exception:
        return {"available": False}
    by_status = {}
    due_count = 0
    max_overdue = 0
    for row in watch_rows:
        try:
            status = row["status"]
            nxt = row["next_poll_at"]
        except (KeyError, IndexError, TypeError):
            status, nxt = row[0], row[1]
        by_status[status] = by_status.get(status, 0) + 1
        if status == RUNNING and nxt is not None and nxt <= now:
            due_count += 1
            max_overdue = max(max_overdue, now - nxt)
    try:
        poll_rows = conn.execute(
            "SELECT status, task_id FROM resident_polls"
            " WHERE project_id IN (%s) AND created_at >= ?"
            % placeholders,
            tuple(ids) + (now - 86400,)).fetchall()
    except Exception:
        poll_rows = []
    polls_24h = len(poll_rows)
    failed_24h = 0
    changed_24h = 0
    tasks_24h = 0
    for row in poll_rows:
        try:
            status = row["status"]
            task_id = row["task_id"]
        except (KeyError, IndexError, TypeError):
            status, task_id = row[0], row[1]
        if status == "failed":
            failed_24h += 1
        if status == "changed":
            changed_24h += 1
            if task_id:
                tasks_24h += 1
    return {"available": True, "watches_by_status": by_status,
            "due_count": due_count,
            "max_overdue_seconds": max_overdue,
            "polls_last_24h": polls_24h, "failed_last_24h": failed_24h,
            "changed_last_24h": changed_24h,
            "change_tasks_last_24h": tasks_24h,
            "config": resident_config()}