From b391fe932087696b0552ae6e9a1e71299d1bf5ec Mon Sep 17 00:00:00 2001 From: host Date: Mon, 2 Nov 2026 03:30:57 +0000 Subject: [PATCH] resident P0+P1: DNS-level SSRF pin, batch budget, daily cap/cooldown/retention, PG schema/migration, dual-connection deadlock isolation, retarget baseline reset, proxy and structured failures, metrics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - DNS 级 SSRF pin：解析后 pin IP 建连，防 DNS 重绑定 - 批量总预算与每日上限/抖动冷却/保留期 - PG schema 与迁移清单 - 双连接死锁隔离 - 改目标重置基线 - 显式代理配置与结构化失败原因 - /metrics resident 口径 --- cloud/db.py | 43 ++ cloud/migrations.py | 5 + cloud/resident.py | 851 ++++++++++++++++++++-- cloud/server.py | 37 +- cloud/tests/test_resident.py | 45 +- cloud/tests/test_resident_p0_fixes.py | 684 +++++++++++++++++ cloud/tests/test_resident_website_loop.py | 47 +- cloud/worker.py | 32 +- verification/resident-p0-fixes-summary.md | 236 ++++++ 9 files changed, 1906 insertions(+), 74 deletions(-) create mode 100644 cloud/tests/test_resident_p0_fixes.py create mode 100644 verification/resident-p0-fixes-summary.md diff --git a/cloud/db.py b/cloud/db.py index 31cc418..af24827 100644 --- a/cloud/db.py +++ b/cloud/db.py @@ -1044,6 +1044,49 @@ POSTGRES_SCHEMA = ( "status TEXT, " "requested_at DOUBLE PRECISION, " "done_at DOUBLE PRECISION)", + # 常驻监听（P0 S3：此前只由 resident.ensure_resident_tables + # 运行时懒建，POSTGRES_SCHEMA 与迁移清单里没有；本批正式 + # 写进 schema，与 ensure 的 CREATE 语句列一一对应，含 + # P1 新增的 resident_polls.failure_reason 列；TEXT 主键/ + # 复合部分唯一索引，无身份列，不需要序列拨动）。 + "CREATE TABLE IF NOT EXISTS resident_watches(" + "id TEXT PRIMARY KEY, " + "project_id TEXT, " + "name TEXT, " + "target_url TEXT, " + "prompt TEXT, " + "interval_seconds BIGINT, " + "fetch_timeout_seconds BIGINT, " + "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 BIGINT, " + "paused_reason TEXT, " + "ended_at DOUBLE PRECISION, " + "created_by TEXT, " + "created_at DOUBLE PRECISION, " + "updated_at DOUBLE PRECISION)", + "CREATE INDEX IF NOT EXISTS idx_resident_watches_project" + " ON resident_watches(project_id, status)", + "CREATE UNIQUE INDEX IF NOT EXISTS idx_resident_watches_name" + " ON resident_watches(project_id, name)" + " WHERE status != 'ended'", + "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)", + "CREATE INDEX IF NOT EXISTS idx_resident_polls_watch" + " ON resident_polls(watch_id, created_at)", ) diff --git a/cloud/migrations.py b/cloud/migrations.py index 952cd8d..81a593a 100644 --- a/cloud/migrations.py +++ b/cloud/migrations.py @@ -82,6 +82,11 @@ MIGRATION_TABLES = ( # 老库没有时迁移记 0，不中断其余表）。 "task_checkpoints", "task_steps", + # 常驻监听（P0 S3：与 POSTGRES_SCHEMA 同款两表，TEXT 主键， + # 无身份列；sqlite 侧由 resident.ensure 建出，老库没有时 + # 记 0，不中断其余表；failure_reason 列两端 schema 一致）。 + "resident_watches", + "resident_polls", ) _SKIP_TABLES = frozenset({"kb_fts", "sqlite_sequence"}) diff --git a/cloud/resident.py b/cloud/resident.py index 4f76e27..1d649dc 100644 --- a/cloud/resident.py +++ b/cloud/resident.py @@ -56,6 +56,18 @@ process_due_resident_watches（见 worker 里的调用点），或 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 的完整性错误类型（仅用于把唯一索引冲突转成 @@ -63,11 +75,14 @@ ValueError，不 import 其它 cloud 模块），避免循环 import。 """ import hashlib +import http.client import ipaddress +import os import re +import socket import sqlite3 +import ssl import time -import urllib.request import uuid from urllib.parse import urljoin from urllib.parse import urlsplit @@ -89,6 +104,153 @@ 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" @@ -110,7 +272,7 @@ WATCH_COLUMNS = ( POLL_COLUMNS = ( "id", "watch_id", "project_id", "status", "observed_hash", - "detail", "task_id", "created_at") + "detail", "task_id", "created_at", "failure_reason") def _row_dict(row, columns): @@ -135,7 +297,15 @@ def _existing_columns(conn, table): def ensure_resident_tables(conn): - """幂等建常驻监听表，补齐新增列（仿 automations 写法）。""" + """幂等建常驻监听表，补齐新增列（仿 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, " @@ -167,7 +337,8 @@ def ensure_resident_tables(conn): "observed_hash TEXT, " "detail TEXT, " "task_id TEXT, " - "created_at DOUBLE PRECISION)") + "created_at DOUBLE PRECISION, " + "failure_reason TEXT)") existing = _existing_columns(conn, "resident_watches") for col, ddl in ( ("fetch_timeout_seconds", "INTEGER"), @@ -185,6 +356,14 @@ def ensure_resident_tables(conn): % (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)") @@ -278,12 +457,15 @@ def _is_noncanonical_numeric_host(host): def validate_target_url(url): - """目标 URL 校验：仅 http/https、无 userinfo、主机非空、 + """目标 URL 字面校验：仅 http/https、无 userinfo、主机非空、 字面 localhost（含尾点、*.localhost）/ 私有 / 回环 / 链路本地 IP / 非规范数字 IP 写法拒绝（与 - websearch._is_blocked_fetch_url 的字面 IP 口径同款， - DNS 解析后的私有地址不在 v1 覆盖内，见摘要已知偏差； - 重定向目标由 _SafeRedirectHandler 逐跳复用本校验）。""" + 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() @@ -436,6 +618,7 @@ def list_polls(conn, watch_id, limit=POLL_LIMIT): "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 @@ -507,7 +690,11 @@ def update_watch(conn, watch_id, fields, now=None): 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) @@ -540,12 +727,23 @@ def update_watch(conn, watch_id, fields, now=None): if clash is not None: raise ValueError( "resident watch name already exists in this project") + target_changed = target_url != current["target_url"] try: - 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)) + 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() @@ -559,12 +757,20 @@ def update_watch(conn, watch_id, fields, now=None): 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。""" + 供驱动列出待处理项；执行入口是 claim + run_due。 + 热路径不调 ensure（见 ensure_resident_tables 的约定）； + 表缺失时由批量入口的 fallback 统一处理。""" 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 <= ?)" @@ -573,6 +779,22 @@ def due_watches(conn, now=None): 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 @@ -581,7 +803,6 @@ def claim_due_watch(conn, watch_id, now=None): 重算 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() @@ -604,40 +825,313 @@ def _content_hash(text): return hashlib.sha256(normalized.encode("utf-8")).hexdigest() -class _SafeRedirectHandler(urllib.request.HTTPRedirectHandler): - """重定向逐跳校验：创建时只校验了初始 URL，公开 URL - 重定向到内网地址会绕过那道校验；相对跳转先 urljoin 成 - 绝对 URL 再校验，不合格直接抛 ValueError，不跟随。""" +class FetchError(ValueError): + """结构化抓取失败：reason 见 FAILURE_REASONS，status 仅 + http_status 时有值。ValueError 子类，故旧调用方按 + ValueError 捕获的行为不变；run_due 据此写 failure_reason + 与 reason= 前缀的 detail，不把原始内网响应写进任何记录。""" - def redirect_request(self, req, fp, code, msg, headers, newurl): - validate_target_url(urljoin(req.full_url, newurl)) - return super().redirect_request(req, fp, code, msg, headers, - newurl) + def __init__(self, reason, message, status=None): + super().__init__(message) + self.reason = reason if reason in FAILURE_REASONS else "unknown" + self.status = status -def default_fetch(url, timeout): - """默认廉价抓取：urllib，超时可配，读满 1MB 即断， - 解码失败按 replace 处理。调用方可注入 fetcher 替换它 - （测试与未来选择器提取都走注入点，不改本函数签名）。 - build_opener 带默认处理器（含 ProxyHandler，no_proxy - 行为与 urlopen 同款），只把重定向处理器换成逐跳校验版。""" - req = urllib.request.Request(url, headers={ - "User-Agent": "codex-app-server-resident-watch/1.0"}) - opener = urllib.request.build_opener(_SafeRedirectHandler()) - with opener.open(req, timeout=timeout) as resp: +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 raw.decode("utf-8", errors="replace") + 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): + 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)" - " VALUES(?, ?, ?, ?, ?, ?, ?, ?)", + " 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)) + (detail or "")[:MAX_POLL_DETAIL_LEN], task_id, now, + failure_reason)) return { "id": poll_id, "watch_id": watch_id, @@ -647,9 +1141,56 @@ def _record_poll(conn, watch_id, project_id, status, 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 = 用户处理提示 + 变化摘录。""" @@ -696,21 +1237,38 @@ def _guarded_watch_update(conn, watch_id, sets, vals): return cur.rowcount == 1 -def run_due_resident_watch(conn, watch_id, fetcher=None, now=None): +def run_due_resident_watch(conn, watch_id, fetcher=None, now=None, + resolver=None): """单条常驻监听的驱动唯一入口：抢单 -> 廉价抓取 -> hash 比对 -> 记轮询 -> 仅变化时建任务。 抢不到返回 None（不存在、已暂停/已结束、未到期、已被 别的驱动抢走），不记轮询、不建任务。抓取失败记 failed - 并退避，游标不动。首次抓取记 baseline，只建游标。未变 - 化记 unchanged，不建任务。变化时先过配额门：配额耗尽 - 转 paused（paused_reason='quota_exceeded'，游标不推进）， - 记 changed_quota_paused；配额检查失败按瞬时失败退避， - 游标同样不推进；配额内才建 queued 任务并推进游标。""" + （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() - ensure_resident_tables(conn) - if not claim_due_watch(conn, watch_id, now): + 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() @@ -723,10 +1281,15 @@ def run_due_resident_watch(conn, watch_id, fetcher=None, now=None): return None interval = watch["interval_seconds"] timeout = watch["fetch_timeout_seconds"] - fetch = fetcher or default_fetch + 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( @@ -736,9 +1299,10 @@ def run_due_resident_watch(conn, watch_id, fetcher=None, now=None): (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, str(exc)[:MAX_POLL_DETAIL_LEN], - None, now) + "failed", None, detail, + None, now, failure_reason=reason) conn.commit() _auto_audit_row(conn, watch["project_id"], "resident_watch_poll_failed", @@ -773,6 +1337,40 @@ def run_due_resident_watch(conn, watch_id, fetcher=None, now=None): "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），不用 # 子串判定，以免超限文本含该子串时误判。 @@ -830,17 +1428,158 @@ def run_due_resident_watch(conn, watch_id, fetcher=None, now=None): return poll -def process_due_resident_watches(conn, fetcher=None, now=None, limit=50): +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() - due = due_watches(conn, now)[:limit] + 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: - poll = run_due_resident_watch(conn, watch["id"], - fetcher=fetcher, now=now) + 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()} diff --git a/cloud/server.py b/cloud/server.py index b0a00e4..c669e18 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -112,10 +112,12 @@ from cloud.automations import ensure_automations_tables as _auto_ensure from cloud.automations import list_automations as _auto_list from cloud.resident import create_watch as _res_create from cloud.resident import end_watch as _res_end +from cloud.resident import ensure_resident_tables as _res_ensure from cloud.resident import get_watch as _res_get from cloud.resident import list_polls as _res_polls from cloud.resident import list_watches as _res_list from cloud.resident import pause_watch as _res_pause +from cloud.resident import resident_metrics as _res_metrics from cloud.resident import resume_watch as _res_resume from cloud.resident import update_watch as _res_update from cloud.todos import create_todo as _todo_create @@ -1739,6 +1741,10 @@ def init_db(db_path): _ensure_task_checkpoints_table(conn) _ensure_task_steps_table(conn) ensure_kill_requests_table(conn) + # 常驻监听表（P0 S3 的 SQLite 侧）：启动时建一次， + # 热路径（claim/run_due/批量）不再逐条 ensure， + # 详见 cloud/resident.py ensure 的调用约定。 + _res_ensure(conn) # worker 每轮 poll 都按 status 取任务（见 load_tasks 的 # WHERE status IN (...)），无索引时是全表扫描；任务表随 # 时间变大后索引是必须的。IF NOT EXISTS 保证老库幂等。 @@ -8901,6 +8907,13 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): else: unpriced_tokens += row["total_tokens"] or 0 all_tokens = actual_tokens + estimated_tokens + # 常驻监听口径（P1 G5）：只统计调用者是成员的项目， + # 与任务计数同款成员过滤；表缺失时 resident_metrics + # 回 available=False，不炸 /metrics。 + try: + resident_stats = _res_metrics(conn, member_ids, now) + except Exception: # noqa: BLE001 - 观测端点不因 resident 失败 + resident_stats = {"available": False} finally: conn.close() try: @@ -8928,6 +8941,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): "by_model": sorted(by_model.values(), key=lambda e: -e["total_tokens"]), }, + "resident": resident_stats, "queue_depth": { "queued": by_status.get("queued", 0), "running": by_status.get("running", 0), @@ -10147,7 +10161,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(400, {"error": str(exc)}) return except Exception as exc: - self._send_json(502, {"error": str(exc)[:500]}) + # P2：502 不再透传异常文本（可能含表名/列名等 + # DB 细节），统一固定错误码，详情只记服务端日志。 + _server_log.warning("resident create failed: %s", exc) + self._send_json(502, {"error": "resident_operation_failed"}) return finally: conn.close() @@ -10162,7 +10179,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): try: watches = _res_list(conn, project_id) except Exception as exc: - self._send_json(502, {"error": str(exc)[:500]}) + _server_log.warning("resident list failed: %s", exc) + self._send_json(502, {"error": "resident_operation_failed"}) return conn.commit() finally: @@ -10180,7 +10198,13 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): if watch is not None: try: polls = _res_polls(conn, watch_id) - except Exception: + except Exception as exc: + # P2：polls 异常不再静默吞成空串，至少记 + # 服务端日志；响应仍保持 polls=[] + note + # 的兼容形态（排障信号在日志里，不在响应里丢）。 + _server_log.warning( + "resident polls unavailable for %s: %s", + watch_id, exc) polls_note = "polls unavailable" finally: conn.close() @@ -10203,7 +10227,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(400, {"error": str(exc)}) return except Exception as exc: - self._send_json(502, {"error": str(exc)[:500]}) + _server_log.warning("resident update failed: %s", exc) + self._send_json(502, {"error": "resident_operation_failed"}) return finally: conn.close() @@ -10228,7 +10253,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): try: ok, watch = fn(conn, watch_id) except Exception as exc: - self._send_json(502, {"error": str(exc)[:500]}) + _server_log.warning( + "resident %s failed: %s", action, exc) + self._send_json(502, {"error": "resident_operation_failed"}) return finally: conn.close() diff --git a/cloud/tests/test_resident.py b/cloud/tests/test_resident.py index 5966f37..c193875 100644 --- a/cloud/tests/test_resident.py +++ b/cloud/tests/test_resident.py @@ -16,6 +16,7 @@ from types import SimpleNamespace try: import cloud.resident as resident_module from cloud.resident import ( + SSRFBlockedError, claim_due_watch, create_watch, default_fetch, @@ -36,6 +37,7 @@ except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.p sys.path.insert(0, str(Path(__file__).resolve().parents[2])) import cloud.resident as resident_module from cloud.resident import ( + SSRFBlockedError, claim_due_watch, create_watch, default_fetch, @@ -416,8 +418,13 @@ class ResidentTestCase(unittest.TestCase): class RedirectFetchTestCase(unittest.TestCase): - """重定向校验：只把 redirect-test.test 指到本地，其它主机 - 走原始解析；no_proxy 同款处理，tearDown 全部还原。""" + """重定向与 DNS 门校验：默认解析器下 redirect-test.test + 经 getaddrinfo 补丁指到本地（127.0.0.1），DNS 门必须拦 + 住它（私网解析结果）；相对跳转的跟随逻辑走注入的测试 + resolver（返回公网测试 IP，门正常放行）+ socket 层重定向 + （测试进程内把该公网 IP:端口指回本地服务——门与 pin + 逻辑全程真实跑，只有传输层终点是测试替身）；no_proxy + 同款处理，tearDown 全部还原。""" HOST = "redirect-test.test" @@ -459,6 +466,18 @@ class RedirectFetchTestCase(unittest.TestCase): host, port, family, type, proto, flags) socket.getaddrinfo = patched + # socket 层测试替身：只重定向公网测试 IP 到本地服务端口， + # 其它地址一律走原始 create_connection。 + self._orig_create_connection = socket.create_connection + test_port = self.port + + def patched_create_connection(address, *args, **kwargs): + if address[0] == "93.184.216.34" \ + and address[1] == test_port: + address = ("127.0.0.1", test_port) + return self._orig_create_connection(address, *args, **kwargs) + + socket.create_connection = patched_create_connection self._old_no_proxy = os.environ.get("no_proxy") self._old_NO_PROXY = os.environ.get("NO_PROXY") os.environ["no_proxy"] = ( @@ -469,6 +488,7 @@ class RedirectFetchTestCase(unittest.TestCase): else "") + self.HOST def tearDown(self): + socket.create_connection = self._orig_create_connection socket.getaddrinfo = self._orig_getaddrinfo if self._old_no_proxy is None: os.environ.pop("no_proxy", None) @@ -484,9 +504,26 @@ class RedirectFetchTestCase(unittest.TestCase): def test_relative_redirect_followed_and_private_blocked(self): base = "http://%s:%d" % (self.HOST, self.port) - self.assertEqual(default_fetch(base + "/go", 10), "final-body") + # 注入 resolver 的跟随路径：解析门放行公网测试 IP， + # socket 替身把建连指回本地，相对跳转 /go -> /final 走通。 + def test_resolver(host, port): + return ["93.184.216.34"] + + self.assertEqual( + default_fetch(base + "/go", 10, resolver=test_resolver), + "final-body") + # 默认 DNS 门：测试主机名解析到 127.0.0.1，直接拦住。 + with self.assertRaises(SSRFBlockedError): + default_fetch(base + "/go", 10) + # 重定向跳本身：跳转目标是字面私网 IP，逐跳校验拦住。 with self.assertRaises(ValueError): - default_fetch(base + "/bad", 10) + default_fetch(base + "/bad", 10, resolver=test_resolver) + # 混合解析结果：任一地址私网即整批拒绝，不取公网那个蒙混。 + def mixed(host, port): + return ["93.184.216.34", "127.0.0.1"] + + with self.assertRaises(ValueError): + default_fetch(base + "/go", 10, resolver=mixed) if __name__ == "__main__": diff --git a/cloud/tests/test_resident_p0_fixes.py b/cloud/tests/test_resident_p0_fixes.py new file mode 100644 index 0000000..27c6e76 --- /dev/null +++ b/cloud/tests/test_resident_p0_fixes.py @@ -0,0 +1,684 @@ +"""常驻监听 P0+P1 修复批单测（cloud/resident.py 等）。 + +覆盖审核报告的 P0（S1 DNS 级 SSRF + 摘录口径、S2 批量总预算、 +G2 每日上限/抖动冷却/保留期）与 P1（S3 PG schema 与迁移清单、 +G1 ensure 移出热路径 + 逐条隔离 + 死锁重试、G3 改目标重置基线、 +G4 显式代理配置 + 结构化失败原因、G5 /metrics 口径）及 P2 +建议项（502 固定错误码、_resident_get polls 异常记日志不静默）。 + +不跑真实模型、不连真实外网：真实抓取只对本地测试 HTTP +服务，DNS 门用 getaddrinfo 补丁或注入 resolver 验证；本地 +PG 验证不在本文件（见 verification/resident-p0-fixes-summary.md +里的 /tmp 一次性库记录），本文件只做 SQLite 与覆盖断言。 +""" +import http.client +import http.server +import json +import os +import socket +import sqlite3 +import ssl +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +try: + import cloud.resident as resident_module + from cloud.auth import AuthStore + from cloud.db import PostgresOperationalError + from cloud.migrations import MIGRATION_TABLES, migration_coverage + from cloud.resident import ( + FetchError, + HttpStatusFetchError, + SSRFBlockedError, + _PinnedHTTPSConnection, + classify_fetch_error, + create_watch, + default_fetch, + ensure_resident_tables, + get_watch, + list_polls, + prune_old_polls, + process_due_resident_watches, + resident_config, + resident_metrics, + resolve_public_addresses, + run_due_resident_watch, + update_watch, + ) + from cloud.server import create_server, init_db + from cloud.worker import Worker +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + import cloud.resident as resident_module + from cloud.auth import AuthStore + from cloud.db import PostgresOperationalError + from cloud.migrations import MIGRATION_TABLES, migration_coverage + from cloud.resident import ( + FetchError, + HttpStatusFetchError, + SSRFBlockedError, + _PinnedHTTPSConnection, + classify_fetch_error, + create_watch, + default_fetch, + ensure_resident_tables, + get_watch, + list_polls, + prune_old_polls, + process_due_resident_watches, + resident_config, + resident_metrics, + resolve_public_addresses, + run_due_resident_watch, + update_watch, + ) + from cloud.server import create_server, init_db + from cloud.worker import Worker + +URL = "https://example.com/data" +URL2 = "https://example.com/other" + + +class _Fetcher: + def __init__(self, value="v1"): + self.value = value + self.calls = [] + + def __call__(self, url, timeout): + self.calls.append((url, timeout)) + return self.value + + +class ResidentP0TestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.conn = sqlite3.connect(self.db_path) + self.conn.row_factory = sqlite3.Row + ensure_resident_tables(self.conn) + self.now = 1700000000.0 + + def tearDown(self): + self.conn.close() + self.tmp.cleanup() + + def _watch(self, name="watch1", url=URL, **kw): + kw.setdefault("interval_seconds", 60) + return create_watch(self.conn, "proj", name, url, + "变化时汇报", "owner1", now=self.now, **kw) + + def _tasks(self): + return self.conn.execute( + "SELECT id, prompt, status FROM tasks").fetchall() + + def _polls(self, watch_id): + return list_polls(self.conn, watch_id) + + # -- S1：DNS 级 SSRF ------------------------------------------- + def test_resolve_public_rejects_private_and_mixed(self): + # 单一私网解析结果即拦。 + with self.assertRaises(SSRFBlockedError): + resolve_public_addresses( + "evil.test", 80, + resolver=lambda h, p: ["127.0.0.1"]) + with self.assertRaises(SSRFBlockedError): + resolve_public_addresses( + "evil.test", 80, resolver=lambda h, p: ["::1"]) + with self.assertRaises(SSRFBlockedError): + resolve_public_addresses( + "evil.test", 80, + resolver=lambda h, p: ["169.254.169.254"]) + with self.assertRaises(SSRFBlockedError): + resolve_public_addresses( + "evil.test", 80, + resolver=lambda h, p: ["::ffff:127.0.0.1"]) + # 混合结果：任一私网即整批拒绝，不取公网那个蒙混。 + with self.assertRaises(SSRFBlockedError): + resolve_public_addresses( + "evil.test", 80, + resolver=lambda h, p: ["93.184.216.34", "10.0.0.5"]) + # 全公网才放行。 + self.assertEqual( + resolve_public_addresses( + "ok.test", 80, resolver=lambda h, p: ["93.184.216.34"]), + ["93.184.216.34"]) + # 解析失败按 dns，不放行。 + with self.assertRaises(FetchError) as ctx: + resolve_public_addresses( + "nope.test", 80, + resolver=lambda h, p: (_ for _ in ()).throw( + socket.gaierror("no such host"))) + self.assertEqual(ctx.exception.reason, "dns") + + def test_ssrf_blocked_poll_has_no_excerpt_and_no_task(self): + fetcher = _Fetcher("v1") + watch = self._watch() + run_due_resident_watch(self.conn, watch["id"], + fetcher=fetcher, now=self.now) + self.now += 61 + + def ssrf_fetcher(url, timeout): + raise SSRFBlockedError("resolved address is not public") + + poll = run_due_resident_watch(self.conn, watch["id"], + fetcher=ssrf_fetcher, now=self.now) + self.assertEqual(poll["status"], "failed") + self.assertEqual(poll["failure_reason"], "ssrf_blocked") + self.assertTrue(poll["detail"].startswith("reason=ssrf_blocked")) + self.assertEqual(len(self._tasks()), 0) + after = get_watch(self.conn, watch["id"], now=self.now) + self.assertIsNone(after["last_change_text"]) + # 失败轮询的 detail 里绝不出现“内网响应正文”类内容。 + self.assertNotIn("secret", poll["detail"]) + + def test_pinned_https_uses_hostname_for_sni_not_ip(self): + seen = {} + + def fake_create_connection(address, timeout=None): + seen["address"] = address + return object() + + class FakeContext: + def wrap_socket(self, sock, server_hostname=None): + seen["server_hostname"] = server_hostname + return object() + + with mock.patch.object(socket, "create_connection", + fake_create_connection): + with mock.patch.object(ssl, "create_default_context", + lambda: FakeContext()): + conn = _PinnedHTTPSConnection( + "example.com", 443, ["93.184.216.34"], timeout=5) + conn.connect() + self.assertEqual(seen["address"], ("93.184.216.34", 443)) + self.assertEqual(seen["server_hostname"], "example.com") + + # -- S2：批量总预算 -------------------------------------------- + def test_batch_budget_stops_batch_and_keeps_rest_due(self): + class FakeClock: + def __init__(self): + self.now = 0.0 + + def __call__(self): + return self.now + + clock = FakeClock() + + def slow_fetcher(url, timeout): + clock.now += 50.0 + return "v1" + + for name in ("a", "b", "c", "d", "e"): + self._watch(name=name) + errors = [] + polls = process_due_resident_watches( + self.conn, fetcher=slow_fetcher, now=self.now, + budget_seconds=100, monotonic_fn=clock, errors=errors) + # 0 秒、50 秒处各跑一条，第 3 条开始前已满 100 秒即停。 + self.assertEqual(len(polls), 2) + self.assertTrue(any(w is None and "budget" in m + for w, m in errors)) + # 剩余 3 条仍是 due 状态，下批可跑，不丢。 + clock2 = FakeClock() + polls2 = process_due_resident_watches( + self.conn, fetcher=slow_fetcher, now=self.now, + budget_seconds=10000, monotonic_fn=clock2) + self.assertEqual(len(polls2), 3) + + # -- G1：逐条隔离 + 死锁重试 + 热路径零 DDL --------------------- + def test_batch_isolates_item_errors_and_retries_deadlock(self): + watch_a = self._watch(name="a") + watch_b = self._watch(name="b") + calls = [] + + def flaky(url, timeout): + return "v1" + + original = resident_module.run_due_resident_watch + + def patched(conn, watch_id, fetcher=None, now=None, + resolver=None): + calls.append(watch_id) + if watch_id == watch_a["id"]: + raise PostgresOperationalError("deadlock detected", + locked=True) + return original(conn, watch_id, fetcher=fetcher, now=now, + resolver=resolver) + + errors = [] + with mock.patch.object(resident_module, + "run_due_resident_watch", patched): + polls = process_due_resident_watches( + self.conn, fetcher=flaky, now=self.now, errors=errors) + self.assertEqual(len(polls), 1) + self.assertEqual(polls[0]["watch_id"], watch_b["id"]) + self.assertEqual(len(errors), 1) + self.assertEqual(errors[0][0], watch_a["id"]) + # 死锁重试成功路径：第一次锁、第二次成功，errors 为空。 + attempts = {"n": 0} + + def retry_once(conn, watch_id, fetcher=None, now=None, + resolver=None): + if watch_id == watch_a["id"] and attempts["n"] == 0: + attempts["n"] += 1 + raise PostgresOperationalError("deadlock detected", + locked=True) + return original(conn, watch_id, fetcher=fetcher, now=now, + resolver=resolver) + + errors2 = [] + with mock.patch.object(resident_module, + "run_due_resident_watch", retry_once): + polls2 = process_due_resident_watches( + self.conn, fetcher=flaky, now=self.now + 120, + errors=errors2) + self.assertEqual(errors2, []) + self.assertTrue(any(p["watch_id"] == watch_a["id"] + for p in polls2)) + + def test_hot_path_batch_executes_no_ddl(self): + class CountingConn: + def __init__(self, real): + self._real = real + self.statements = [] + + def execute(self, sql, params=()): + self.statements.append(sql) + return self._real.execute(sql, params) + + def commit(self): + return self._real.commit() + + def rollback(self): + return self._real.rollback() + + self._watch(name="a") + self._watch(name="b") + counting = CountingConn(self.conn) + polls = process_due_resident_watches( + counting, fetcher=_Fetcher("v1"), now=self.now) + self.assertEqual(len(polls), 2) + ddl = [s for s in counting.statements + if s.strip().upper().startswith("CREATE")] + self.assertEqual(ddl, []) + + # -- G2：每日上限 / 抖动冷却 / 保留期 ---------------------------- + def test_daily_cap_and_cooldown_suppress_without_losing_change(self): + env = { + "CODEX_RESIDENT_MAX_DAILY_CHANGE_TASKS": "2", + "CODEX_RESIDENT_CHANGE_COOLDOWN_SECONDS": "300", + } + with mock.patch.dict(os.environ, env): + fetcher = _Fetcher("v1") + watch = self._watch() + run_due_resident_watch(self.conn, watch["id"], + fetcher=fetcher, now=self.now) + # 第 1 次变化：建任务。 + fetcher.value = "v2" + self.now += 61 + poll = run_due_resident_watch( + self.conn, watch["id"], fetcher=fetcher, now=self.now) + self.assertEqual(poll["status"], "changed") + self.assertEqual(len(self._tasks()), 1) + # 冷却期内再变：记 changed_cooldown，不建任务，游标不动。 + fetcher.value = "v3" + self.now += 61 + poll = run_due_resident_watch( + self.conn, watch["id"], fetcher=fetcher, now=self.now) + self.assertEqual(poll["status"], "changed_cooldown") + self.assertEqual(len(self._tasks()), 1) + after = get_watch(self.conn, watch["id"], now=self.now) + self.assertEqual(after["last_change_text"], "v2") + # 冷却过后：第 2 次变化建任务，达到上限 2。 + self.now += 300 + poll = run_due_resident_watch( + self.conn, watch["id"], fetcher=fetcher, now=self.now) + self.assertEqual(poll["status"], "changed") + self.assertEqual(len(self._tasks()), 2) + # 再变：上限已满，记 changed_capped，不建任务，游标不动。 + fetcher.value = "v4" + self.now += 361 + poll = run_due_resident_watch( + self.conn, watch["id"], fetcher=fetcher, now=self.now) + self.assertEqual(poll["status"], "changed_capped") + self.assertEqual(len(self._tasks()), 2) + after = get_watch(self.conn, watch["id"], now=self.now) + self.assertEqual(after["last_change_text"], "v3") + + def test_prune_old_polls_keeps_recent(self): + watch = self._watch() + old = self.now - 100 * 86400 + self.conn.execute( + "INSERT INTO resident_polls(id, watch_id, project_id," + " status, observed_hash, detail, task_id, created_at)" + " VALUES('old1', ?, 'proj', 'unchanged', 'h', '', NULL, ?)", + (watch["id"], old)) + self.conn.execute( + "INSERT INTO resident_polls(id, watch_id, project_id," + " status, observed_hash, detail, task_id, created_at)" + " VALUES('new1', ?, 'proj', 'unchanged', 'h', '', NULL, ?)", + (watch["id"], self.now)) + self.conn.commit() + deleted = prune_old_polls(self.conn, now=self.now, + retention_days=90) + self.assertEqual(deleted, 1) + remaining = [p["id"] for p in self._polls(watch["id"])] + self.assertEqual(remaining, ["new1"]) + + # -- G3：改目标重置基线 ------------------------------------------ + def test_update_target_resets_baseline_but_interval_does_not(self): + fetcher = _Fetcher("v1") + watch = self._watch() + run_due_resident_watch(self.conn, watch["id"], + fetcher=fetcher, now=self.now) + before = get_watch(self.conn, watch["id"], now=self.now) + self.assertIsNotNone(before["last_content_hash"]) + # 只改间隔：游标保留。 + ok, updated = update_watch( + self.conn, watch["id"], {"interval_seconds": 3600}, + now=self.now) + self.assertTrue(ok) + self.assertIsNotNone(updated["last_content_hash"]) + # 改目标：游标清掉，下一轮只建基线、不建任务。 + ok, updated = update_watch( + self.conn, watch["id"], {"target_url": URL2}, now=self.now) + self.assertTrue(ok) + self.assertIsNone(updated["last_content_hash"]) + + def url_fetcher(url, timeout): + return "new-site-content" if url == URL2 else "v1" + + self.now += 61 + poll = run_due_resident_watch( + self.conn, watch["id"], fetcher=url_fetcher, now=self.now) + self.assertEqual(poll["status"], "baseline") + self.assertEqual(len(self._tasks()), 0) + + # -- G4：代理配置与结构化失败 ------------------------------------- + def test_proxy_config_defaults_direct_and_env_opt_in(self): + with mock.patch.dict(os.environ, {}, clear=False): + os.environ.pop("CODEX_RESIDENT_PROXY", None) + os.environ.pop("http_proxy", None) + os.environ.pop("https_proxy", None) + cfg = resident_config() + self.assertEqual(cfg["proxy"]["mode"], "direct") + with mock.patch.dict(os.environ, + {"CODEX_RESIDENT_PROXY": "env", + "https_proxy": "http://proxy.test:8080"}): + cfg = resident_config() + self.assertEqual(cfg["proxy"]["mode"], "env") + with mock.patch.dict( + os.environ, + {"CODEX_RESIDENT_PROXY": "http://proxy.test:8080"}): + cfg = resident_config() + self.assertEqual(cfg["proxy"]["mode"], "url") + self.assertEqual(cfg["proxy"]["url"], + "http://proxy.test:8080") + with mock.patch.dict(os.environ, + {"CODEX_RESIDENT_PROXY": "ftp://x/"}): + with self.assertRaises(FetchError) as ctx: + resident_module.resolve_proxy_config() + self.assertEqual(ctx.exception.reason, "proxy") + # 默认值本身是拟定值（见摘要拟定清单），这里只断言 + # 当前实现值，不把它写成用户已拍板。 + self.assertEqual(resident_config()["max_daily_change_tasks"], 24) + self.assertEqual( + resident_config()["change_cooldown_seconds"], 300) + self.assertEqual(resident_config()["poll_retention_days"], 90) + self.assertEqual(resident_config()["batch_budget_seconds"], 120) + + def test_classify_fetch_errors(self): + self.assertEqual(classify_fetch_error( + SSRFBlockedError("blocked")), "ssrf_blocked") + self.assertEqual(classify_fetch_error( + HttpStatusFetchError(404)), "http_status") + self.assertEqual(classify_fetch_error(socket.timeout("t")), + "timeout") + self.assertEqual(classify_fetch_error( + ConnectionRefusedError("refused")), "network") + self.assertEqual(classify_fetch_error(RuntimeError("boom")), + "unknown") + + def test_http_status_fetch_raises_structured_error(self): + class Handler(http.server.BaseHTTPRequestHandler): + def do_GET(self): + body = b"not here" + self.send_response(404) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): + pass + + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), + Handler) + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever, + daemon=True) + thread.start() + try: + # 单跳层真实 404（本地服务，绕过 DNS 门直接调单跳）。 + status, _loc, _raw = resident_module._fetch_single_hop( + "http://127.0.0.1:%d/missing" % port, 10, + ["127.0.0.1"], None) + self.assertEqual(status, 404) + # default_fetch 的状态分支：解析门放行公网 IP 后， + # 单跳返回 404 即抛结构化 http_status 错误（正文丢弃）。 + with mock.patch.object( + resident_module, "_fetch_single_hop", + return_value=(404, None, b"not here")): + with self.assertRaises(HttpStatusFetchError) as ctx: + default_fetch( + "http://status.test/missing", 10, + resolver=lambda h, p: ["93.184.216.34"]) + self.assertEqual(ctx.exception.reason, "http_status") + self.assertEqual(ctx.exception.status, 404) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + # -- S3：迁移覆盖（无 PG 也可断言）-------------------------------- + def test_migration_coverage_includes_resident_tables(self): + coverage = migration_coverage() + self.assertEqual(coverage["missing_from_migration"], ()) + self.assertIn("resident_watches", MIGRATION_TABLES) + self.assertIn("resident_polls", MIGRATION_TABLES) + self.assertIn("resident_watches", coverage["schema_tables"]) + self.assertIn("resident_polls", coverage["schema_tables"]) + + # -- G5：metrics 口径 --------------------------------------------- + def test_resident_metrics_counts(self): + fetcher = _Fetcher("v1") + watch = self._watch() + run_due_resident_watch(self.conn, watch["id"], + fetcher=fetcher, now=self.now) + fetcher.value = "v2" + self.now += 61 + run_due_resident_watch(self.conn, watch["id"], + fetcher=fetcher, now=self.now) + stats = resident_metrics(self.conn, {"proj"}, now=self.now) + self.assertTrue(stats["available"]) + self.assertEqual(stats["watches_by_status"], {"running": 1}) + self.assertEqual(stats["polls_last_24h"], 2) + self.assertEqual(stats["changed_last_24h"], 1) + self.assertEqual(stats["change_tasks_last_24h"], 1) + self.assertEqual(stats["failed_last_24h"], 0) + # 到期积压口径：把 next_poll_at 推到过去，due_count 见涨。 + later = self.now + 3600 + stats2 = resident_metrics(self.conn, {"proj"}, now=later) + self.assertEqual(stats2["due_count"], 1) + self.assertGreater(stats2["max_overdue_seconds"], 0) + # 非成员项目不计入。 + stats3 = resident_metrics(self.conn, {"other"}, now=self.now) + self.assertEqual(stats3["watches_by_status"], {}) + + +class ResidentP0ServerTestCase(unittest.TestCase): + """服务端侧：/metrics resident 段、502 固定错误码、polls + 异常记日志不静默（响应形态仍为 polls=[] + note）。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = self.auth.create_user("owner") + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + status, self.project = self._req( + "POST", "/projects", {"name": "demo"}, self.owner_key) + self.assertEqual(status, 201, self.project) + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + + def _req(self, method, path, body=None, key=None): + conn = http.client.HTTPConnection("127.0.0.1", self.port, + timeout=10) + hdrs = {} + payload = None + if body is not None: + payload = json.dumps(body).encode("utf-8") + hdrs["Content-Type"] = "application/json" + if key is not None: + hdrs["X-API-Key"] = key + try: + conn.request(method, path, body=payload, headers=hdrs) + resp = conn.getresponse() + data = resp.read() + finally: + conn.close() + try: + parsed = json.loads(data.decode("utf-8")) if data else None + except ValueError: + parsed = None + return resp.status, parsed + + def _url(self, suffix=""): + return "/projects/%s/resident-watches%s" % ( + self.project["id"], suffix) + + def test_metrics_contains_resident_section(self): + status, watch = self._req( + "POST", self._url(), + {"name": "w1", "target_url": URL, "prompt": "p"}, + self.owner_key) + self.assertEqual(status, 201, watch) + status, body = self._req("GET", "/metrics", None, self.owner_key) + self.assertEqual(status, 200, body) + resident = body["resident"] + self.assertTrue(resident["available"]) + self.assertEqual(resident["watches_by_status"], {"running": 1}) + self.assertEqual(resident["due_count"], 1) + self.assertIn("max_daily_change_tasks", resident["config"]) + + def test_resident_502_uses_fixed_error_code(self): + import cloud.server as server_module + original = server_module._res_create + + def boom(*args, **kwargs): + raise RuntimeError("db table resident_watches secret detail") + + server_module._res_create = boom + try: + status, body = self._req( + "POST", self._url(), + {"name": "w1", "target_url": URL, "prompt": "p"}, + self.owner_key) + finally: + server_module._res_create = original + self.assertEqual(status, 502) + self.assertEqual(body["error"], "resident_operation_failed") + self.assertNotIn("secret", json.dumps(body)) + + def test_resident_get_polls_failure_keeps_note(self): + status, watch = self._req( + "POST", self._url(), + {"name": "w1", "target_url": URL, "prompt": "p"}, + self.owner_key) + self.assertEqual(status, 201, watch) + import cloud.server as server_module + original = server_module._res_polls + + def boom(*args, **kwargs): + raise RuntimeError("poll query exploded") + + server_module._res_polls = boom + try: + status, body = self._req( + "GET", self._url("/" + watch["id"]), None, + self.owner_key) + finally: + server_module._res_polls = original + self.assertEqual(status, 200, body) + self.assertEqual(body["polls"], []) + self.assertEqual(body["polls_note"], "polls unavailable") + + +class ResidentP0WorkerTestCase(unittest.TestCase): + """worker 驱动：首次批量 ensure 一次、resolver 透传。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.conn = sqlite3.connect(self.db_path) + self.conn.row_factory = sqlite3.Row + ensure_resident_tables(self.conn) + self.now = 1700000000.0 + + def tearDown(self): + self.conn.close() + self.tmp.cleanup() + + def test_worker_driver_passes_resolver_and_ensures_once(self): + create_watch(self.conn, "proj", "w1", URL, "p", + "owner1", interval_seconds=60, now=self.now) + worker = Worker(self.db_path, "/fake/host.py", + env={"XCM_API_KEY": "test-placeholder"}, + runner=lambda cmd: SimpleNamespace( + returncode=0, stdout="{}", stderr="")) + seen = {} + fetcher_calls = [] + + def recording_fetch(url, timeout): + fetcher_calls.append(url) + return "v1" + + def recording_resolver(host, port): + seen["host"] = host + return ["127.0.0.1"] + + # worker 经 resident_fetcher/resident_resolver 注入点取用。 + worker.resident_fetcher = recording_fetch + worker.resident_resolver = recording_resolver + polls = worker.process_due_resident_watches(now=self.now) + self.assertEqual(len(polls), 1) + self.assertEqual(polls[0]["status"], "baseline") + # fetcher 注入时 resolver 不经过 default_fetch，不必被调用； + # 这里只断言 worker 首批 ensure 标志已立、次批不再重复建表 + # （标志位本身即“每 worker 一次”的证据）。 + self.assertTrue( + getattr(worker, "_resident_tables_ensured", False)) + self.assertEqual(fetcher_calls, [URL]) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_resident_website_loop.py b/cloud/tests/test_resident_website_loop.py index 3c6e999..d51748b 100644 --- a/cloud/tests/test_resident_website_loop.py +++ b/cloud/tests/test_resident_website_loop.py @@ -8,10 +8,15 @@ - 间隔仍是生产下限 60 秒；到期不是改间隔，而是给驱动传一个 更晚的 now（模拟时间流逝），与上一轮单测同款时间注入。 - 目标 URL 仍走生产校验：主机名 website-loop.test 不是 IP 字 - 面、不是 localhost，所以校验通过；测试进程内只把这个主机 - 名的 DNS 解析指向本地测试 HTTP 服务（相当于 /etc/hosts 的 - 一条测试条目），抓取本身走真实 default_fetch（urllib、超时、 - 1MB 截断），不注入假 fetcher、不绕过生产代码。 + 面、不是 localhost，所以校验通过；抓取本身走真实 + default_fetch（DNS 门、pin 连接、超时、1MB 截断），不注入 + 假 fetcher。注意：生产默认 DNS 门会把解析到 127.0.0.1 的 + 目标拦住，所以本文件用两段测试接缝：worker 的 + resident_resolver 把该主机名解析成公网测试 IP（门正常放 + 行），测试进程内再在 socket 层把该 IP:内容端口重定向回 + 本地服务（传输层终点是测试替身，门与 pin 全程真实跑）； + DNS 门本身由 test_resident_p0_fixes.py 的默认解析器用例 + 覆盖，不在本文件重复证明。 - 内容服务是线程内真实 http.server，可变内容、也可停服； 控制面是真实 ThreadingHTTPServer + 真实 SQLite 临时库； 变化建出的任务、轮询、审计都直接读库/接口核对。 @@ -112,12 +117,39 @@ class WebsiteLoopTestCase(unittest.TestCase): env={"XCM_API_KEY": "test-placeholder"}, runner=lambda cmd: SimpleNamespace( returncode=0, stdout="{}", stderr="")) + # 两段测试接缝（见文件头说明）：resolver 返回公网测试 + # IP，socket 层把该 IP:内容端口重定向回本地服务。 + try: + from cloud.resident import _default_resolver + except ImportError: + _default_resolver = None + + def _loop_resolver(host, port, _default=_default_resolver): + if host == HOST: + return ["93.184.216.34"] + if _default is not None: + return _default(host, port) + return [socket.getaddrinfo(host, port)[0][4][0]] + + self.worker.resident_resolver = _loop_resolver + self._orig_create_connection = socket.create_connection + + def _patched_create_connection(address, *args, **kwargs): + if address[0] == "93.184.216.34" \ + and address[1] == self.content_port: + address = ("127.0.0.1", self.content_port) + return self._orig_create_connection(address, *args, + **kwargs) + + socket.create_connection = _patched_create_connection def tearDown(self): # setUp 中途失败时属性可能未建：getattr 防御，避免 # tearDown 的 AttributeError 掩盖原始失败。 if getattr(self, "_orig_getaddrinfo", None) is not None: socket.getaddrinfo = self._orig_getaddrinfo + if getattr(self, "_orig_create_connection", None) is not None: + socket.create_connection = self._orig_create_connection if getattr(self, "_old_no_proxy", None) is None: if "_old_no_proxy" in self.__dict__: os.environ.pop("no_proxy", None) @@ -275,12 +307,15 @@ class WebsiteLoopTestCase(unittest.TestCase): _content_hash("version two")) # 5. 恢复：用旧游标比对，报出暂停期间的变化并建任务， - # 不是从头建基线。 + # 不是从头建基线。轮询时刻取 t0+540：距上次建任务 + # （t0+240）已过 300 秒抖动冷却，否则新冷却语义会 + # 正确地记 changed_cooldown 而不是 changed（冷却期 + # 内的变化不丢，冷却过后复查仍可报，见 P0 批单测）。 status, body = self._req( "POST", self._url("/%s/resume" % wid), {}, self.owner_key) self.assertEqual(status, 200, body) self.assertEqual(body["status"], "running") - polls = self.worker.process_due_resident_watches(now=t0 + 400) + polls = self.worker.process_due_resident_watches(now=t0 + 540) self.assertEqual(len(polls), 1, polls) self.assertEqual(polls[0]["status"], "changed") tasks = self._tasks() diff --git a/cloud/worker.py b/cloud/worker.py index ffd2286..59b79e5 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -6582,21 +6582,47 @@ class Worker: log_event(self.logger, "worker_push_due_failed", message="push due processing failed: %s" % exc) - def process_due_resident_watches(self, now=None, fetcher=None): + def process_due_resident_watches(self, now=None, fetcher=None, + resolver=None): """常驻监听驱动（worker 轮次边界调用）：到期监听逐条 走 resident.process_due_resident_watches，抢单 + 廉价 抓取 + 仅变化建任务。全部异常只记日志，不炸循环； fetcher 缺省用 resident 默认抓取，单测可经 - self.resident_fetcher 或参数注入假抓取。""" + self.resident_fetcher 或参数注入假抓取；resolver 同款 + 注入点（self.resident_resolver），生产路径为 None 即 + 默认 DNS 门，测试用它把测试主机名指到本地服务。 + 建表只在 worker 生命周期内做一次（首批）：热路径 + 不再逐条 ensure，见 resident.ensure 的调用约定； + 批量内逐条隔离的异常与预算耗尽信号记 + worker_resident_watch_failed 日志，不炸整批。""" if now is None: now = time.time() if fetcher is None: fetcher = getattr(self, "resident_fetcher", None) + if resolver is None: + resolver = getattr(self, "resident_resolver", None) try: + from cloud.resident import ensure_resident_tables as _ensure from cloud.resident import process_due_resident_watches as _run conn = _db_connect(self.db_path) try: - polls = _run(conn, fetcher=fetcher, now=now) + if not getattr(self, "_resident_tables_ensured", False): + try: + _ensure(conn) + conn.commit() + self._resident_tables_ensured = True + except Exception as exc: + log_event( + self.logger, "worker_resident_watch_failed", + message="resident ensure failed: %s" % exc) + errors = [] + polls = _run(conn, fetcher=fetcher, now=now, + resolver=resolver, errors=errors) + for watch_id, msg in errors: + log_event( + self.logger, "worker_resident_watch_failed", + message="resident watch item failed: %s: %s" + % (watch_id, msg)) finally: conn.close() except Exception as exc: diff --git a/verification/resident-p0-fixes-summary.md b/verification/resident-p0-fixes-summary.md new file mode 100644 index 0000000..448386b --- /dev/null +++ b/verification/resident-p0-fixes-summary.md @@ -0,0 +1,236 @@ +# 常驻监听 P0+P1 修复批验证摘要（本地实现与测试，未 push、未开 PR、未合并、未部署） + +- 起点：`feat-resident-p0-fixes`，从 `origin/main 60997b1`（PR #275 合并头）起； + 本批只改本 worktree，绝不动生产审核 worktree、闭环 worktree、主 checkout。 +- 审核依据：生产审核报告 + `~/workspace/codex-app-server-resident-prod-audit/verification/resident-prod-audit-summary.md` + 第 4、5、7 节（动手前已重读）；范围只限该报告的 P0 与 P1。 +- 不做事项：控制台 UI、选择器/区域提取、其它探针 + （server_status/agent_status/relay_cost）、磁盘清理、真实外网抓取 + （真实抓取只用本地测试 HTTP 服务验证）、生产 PG、生产双 worker、 + 真实外网只读演练（仍是需用户拍板的后续窗口，见 blocker）。 +- 红线：以下所有数值凡标“拟定默认值”者，均为本批实现工拟定的 + 默认值、供最终验收改，不是用户已拍板；摘要、代码注释与后续 + PR 描述素材里一律按此口径写，不得写成用户已定。 + +## 1. 拟定默认值清单（逐项标拟定，环境变量可覆盖） + +| 口径 | 拟定默认值 | 环境变量 | 说明 | +| --- | --- | --- | --- | +| 每监听每日变化任务上限 | 24 个 | `CODEX_RESIDENT_MAX_DAILY_CHANGE_TASKS` | 按 UTC 日切分；只数 `status='changed'` 且 `task_id` 非空的轮询行；超限记 `changed_capped`，游标不推进 | +| 变化抖动冷却 | 300 秒 | `CODEX_RESIDENT_CHANGE_COOLDOWN_SECONDS` | 上次建任务 300 秒内的新变化记 `changed_cooldown`，游标不推进，冷却过后复查仍可报 | +| resident_polls 保留期 | 90 天 | `CODEX_RESIDENT_POLL_RETENTION_DAYS` | 批量驱动每批开头 prune 一次；过期删、未过期保留；当日上限计数不受影响（保留期远大于 1 天） | +| 批量总预算 | 120 秒 | `CODEX_RESIDENT_BATCH_BUDGET_SECONDS` | 单批最坏 = 预算 + 单条超时上限 60 秒；超预算的剩余监听保持 due，下批再跑 | +| 代理模式 | 直连（默认） | `CODEX_RESIDENT_PROXY` | 未设/空/`direct` = 直连，不跟随进程代理；`env` = 显式跟随进程环境代理；其它值 = 显式代理 URL；两种代理模式都认进程 `no_proxy`/`NO_PROXY` | +| 重定向上限 | 10 跳 | （固定值） | 超过抛 `redirect` 结构化失败 | + +其它沿用旧值不变：间隔下限 60 秒、抓取超时 1–60 秒（默认 10 秒）、 +1MB 截断、变化摘录 2000 字、poll detail 500 字、批量单批上限 50 条。 + +## 2. P0 修复与测试证据 + +### S1. DNS 级 SSRF + 摘录口径（cloud/resident.py） + +- 抓取前解析全部 A/AAAA（`resolve_public_addresses`）：任一地址命中 + 私有/回环/链路本地/保留/组播/未指定即整批拒绝；IPv6 的 IPv4 映射 + 地址（`::ffff:127.0.0.1`）按映射后 IPv4 再判；解析失败/空结果/ + 非 IP 结果按 `dns` 失败，不放行。 +- 直连建连 pin 住解析结果（`_PinnedHTTPConnection` / + `_PinnedHTTPSConnection`）：TCP 对端就是解析时刻的 IP，不再二次 + 按主机名解析；https 的 SNI 与证书主机名仍是原始主机名（单测断言 + `wrap_socket(server_hostname='example.com')` 且建连地址为 pin 的 IP）， + TLS 校验不被 pin 破坏。 +- 重定向每跳重复字面校验 + 解析门 + pin（相对跳转先 urljoin）； + 非 200 一律抛 `HttpStatusFetchError`，错误页正文直接丢弃。 +- 摘录口径（拟定）：摘录只来自已通过校验的公开抓取结果；私网 + 抓取在抓取层直接失败（`ssrf_blocked`）、不产生摘录、不建任务、 + 轮询 detail 只含 `reason=ssrf_blocked; …` 异常文本，绝不含抓取正文。 + 创建监听时仍只做字面校验（DNS 会变，创建时放行不等于抓取时放行）， + 权威关口在抓取层——见代码注释。 +- 残留风险（诚实写明，不假装已封口）： + - 代理模式下客户端解析门只是前置检查，真正建连的是代理方、由 + 代理方二次解析；pin 在代理模式下不生效，须靠代理自身的 egress + allowlist 兜底。生产部署若用代理模式，必须同时给代理配内网拦截。 + - 直连 pin 已把解析-建连 TOCTOU 关到“解析结果本身”这一层，但 + “解析时刻的 IP 在之后被重新分配给内网用途”的极端情形不在 + 本次封口承诺内。 + - 旧 urllib 默认跟随进程代理的行为已改：默认直连。部署时若宿主 + 出口必须走代理，须显式配 `CODEX_RESIDENT_PROXY=env` 或代理 URL， + 否则公网抓取会直连失败（记失败轮询，不建任务）。 +- 测试：`test_resolve_public_rejects_private_and_mixed`（私网/混合/ + 映射地址/解析失败）、`test_ssrf_blocked_poll_has_no_excerpt_and_no_task`、 + `test_pinned_https_uses_hostname_for_sni_not_ip`、 + 旧重定向测试改写（默认解析器 + getaddrinfo 补丁必拦；注入 resolver + 返回公网测试 IP + socket 层重定向到本地服务时相对跳转走通）。 + +### S2. 驱动与延迟：选批量总预算，不新增独立驱动进程 + +- 论证：独立驱动进程的运维成本（新 unit、独立 DB 连接、启停监管） + 是审核第 7 节留给用户拍板的项，本批不擅自新增；批量总预算是纯 + 代码改动、本地可测，且 claim 门已保证未来加驱动时多驱动不重复 + 轮询。预算把旧最坏 `50×60=3000` 秒 bound 到“预算 120 秒 + 单条 + 超时上限 60 秒”，剩余到期监听保持 due、下批再跑，不丢。 +- 间隔只是下限、不是 SLA：`resident_metrics` 的 `due_count` / + `max_overdue_seconds` 直接度量“到期未跑”积压，进 `/metrics`； + 任何对外口径仍须写“尽力而为、延迟无界”（模型任务一轮最长 + 1800 秒仍会拖后轮询，预算只 bound 单批，不承诺送达时限）。 +- 测试：`test_batch_budget_stops_batch_and_keeps_rest_due` + （单调时钟注入：50 秒/条、预算 100 秒时跑 2 条即停，剩余 3 条仍 due）。 + +### G2. 成本上限与保留期 + +- 每日上限、抖动冷却、保留期见第 1 节清单；三者均持久化 + （上限数当日轮询行、冷却用既有 `last_change_at`、保留期靠 + `prune_old_polls` 的 DELETE），重启不丢计数。 +- 页面每轮都变时：冷却期内只记 `changed_cooldown`、上限满后只记 + `changed_capped`，游标不推进（变化不丢，冷却过后/次日仍可报）， + 任务数有界。 +- 测试：`test_daily_cap_and_cooldown_suppress_without_losing_change` + （上限 2、冷却 300 秒的加速版：建任务→冷却抑制→冷却后建任务→ + 上限抑制，游标始终指向最后一次真实建任务的内容）、 + `test_prune_old_polls_keeps_recent`（100 天前行删、当前行保留）。 + +## 3. P1 修复与测试证据 + +### S3. resident 表进 PG schema 与迁移清单 + +- `cloud/db.py` 的 `POSTGRES_SCHEMA` 新增 `resident_watches` / + `resident_polls`（含 `failure_reason` 列）与三条索引（含部分唯一索引 + `WHERE status != 'ended'`）；`cloud/migrations.py` 的 + `MIGRATION_TABLES` 同步新增两表（TEXT 主键，无身份列，无需序列拨动）。 +- 测试：`test_migration_coverage_includes_resident_tables` + （`migration_coverage()["missing_from_migration"] == ()`）； + 本地 PG 实测见第 5 节。 + +### G1. ensure 移出热路径 + 批量逐条隔离 + 死锁重试一次 + +- 热路径（`due_watches` / `claim_due_watch` / `run_due` / + 批量循环）不再调 `ensure_resident_tables`；建表只在 server + `init_db`（SQLite 分支新增 `_res_ensure`，PG 分支走 schema）、 + worker 生命周期首次批量、API 函数（create/list/get/update/ + transition）做。热路径只在表确实缺失时 ensure 后重试一次。 +- 批量入口逐条隔离：锁错误（PG `locked=True`、SQLite + `database is locked`）回滚后重试一次，仍失败记入 `errors` + 列表并继续下一条；非锁异常同样隔离；预算耗尽信号以 + `watch_id=None` 的条目记入 `errors`，供 worker 记日志。 +- 测试：`test_batch_isolates_item_errors_and_retries_deadlock` + （死锁始终失败时只影响本条、其余照常；第一次死锁第二次成功时 + errors 为空）、`test_hot_path_batch_executes_no_ddl` + （计数连接断言批量内零 CREATE 语句）。 + +### G3. 改目标 URL 强制走一次 baseline + +- `update_watch` 改 `target_url` 时 `last_content_hash` 置 NULL， + 下一轮只建基线、不建任务；改间隔/提示/名字不重置游标， + 已排好的 `next_poll_at` 不变（语义保留，见代码注释）。 +- 测试：`test_update_target_resets_baseline_but_interval_does_not`。 + +### G4. 显式代理配置 + 结构化失败原因 + +- 代理口径见第 1 节清单与 `resident_config()` 注释；默认直连 + 与环境变量的关系写在代码与本摘要里，不只靠隐式进程环境。 +- 失败原因进 `resident_polls.failure_reason` 列（ensure 与 PG + schema 同步补列），detail 前缀同样写 `reason=<原因>`； + 原因集：`dns / proxy / timeout / http_status / ssrf_blocked / + tls / redirect / network / invalid_url / unknown`。 +- 测试：`test_proxy_config_defaults_direct_and_env_opt_in`、 + `test_classify_fetch_errors`、`test_http_status_fetch_raises_structured_error` + （单跳层真实 404 + default_fetch 状态分支）。 + +### G5. resident 指标进 /metrics + +- `GET /metrics` 新增 `resident` 段（成员项目过滤，与任务计数 + 同款）：`watches_by_status`、`due_count`、`max_overdue_seconds`、 + 近 24h `polls_last_24h / failed_last_24h / changed_last_24h / + change_tasks_last_24h`、当前拟定 `config`；表缺失时 + `{"available": False}`，不炸 `/metrics`。 +- 测试：`test_resident_metrics_counts`（含到期积压与非成员过滤）、 + `test_metrics_contains_resident_section`（HTTP 层）。 + +### P2 建议项（顺手修，不改语义） + +- resident 四个 handler 的 502 统一为固定错误码 + `resident_operation_failed`，详情只记服务端日志（`_server_log`）， + 不再透传可能含表名/列名的异常文本；测试 + `test_resident_502_uses_fixed_error_code`（响应不含注入的细节串）。 +- `_resident_get` 的 polls 异常记服务端日志（`_server_log.warning`）， + 响应仍保持 `polls=[] + polls_note="polls unavailable"` 的兼容形态； + 测试 `test_resident_get_polls_failure_keeps_note`。 +- 未做（列为后续，不硬修）：改间隔立即生效的语义（仍为“暂停后 + 恢复才立即复查”，已写进注释与本摘要）；选择器/区域提取； + 控制台 UI。 + +## 4. 测试结果 + +- 定向（最后一次纯文案修订后重跑，见时间关系说明）： + `python3 -m unittest cloud.tests.test_resident + cloud.tests.test_resident_server cloud.tests.test_resident_website_loop + cloud.tests.test_resident_p0_fixes` —— 38 项全过（旧 20 项 + 新 18 项）。 +- ruff E,F（被碰文件）：与改动前基线一致的 12 项，全是旧代码 + （含 `cloud/worker.py` 的 F821、`cloud/server.py` 旧中文注释超长行等）， + 本批新增 0 项（新增的 3 项在自查中已修掉）。 +- CI 同款三条 discover 全量（均在本批代码定稿后跑出）： + - `python3 -m unittest discover -s cloud/tests`：1569 项，OK（跳过 9 项），393 秒； + - `python3 -m unittest discover -s deploy/tests`：44 项，OK，12 秒 + （尾部三行中文传输日志为某测试退出时打印，不影响结果）； + - `python3 -m unittest discover -s tests`：105 项，OK，12 秒。 +- 时间关系：定向 38 项先于全量跑出；全量三条跑完后，本摘要 + 填入全量数字（纯文案修订，不碰代码与测试），随后按任务要求 + 重跑定向——重跑结果仍为 38 项全过，与全量结论一致；摘要定稿 + 在定向重跑之后，不再有代码或测试改动。 + +## 5. 本地 PG 16 验证单独记录（一次性库，测完已删库删脚本） + +- 方式：`initdb --auth=trust` 一次性库（Unix socket，psycopg2 + 适配层），脚本放 `/tmp`（不进 worktree），测完停库、删库、 + 删脚本；结论只代表本地 PG，不代表生产 PG。 +- 结果（全部通过）： + - `init_postgres_schema` 后 `resident_watches` / `resident_polls` 齐全； + - 索引在位：`idx_resident_watches_project`、 + `idx_resident_watches_name`（部分唯一， + `WHERE (status <> 'ended'::text)`）、`idx_resident_polls_watch`； + - 二次 `init_postgres_schema` + `ensure_resident_tables` 幂等； + - `migration_coverage` 无缺失；SQLite→PG 迁移 resident 两表各 1 行； + - 基线→变化→queued 任务、已结束名重用全部通过； + - 双连接并发：胜者 1 条 baseline 轮询、败者 0 条且 `errors` 为空、 + 全程无死锁异常、无整批 abort、poll 总数 1（G1 修复后败者干净返回； + 旧死锁形态未再出现，但“双连接”不等于“生产双 worker”，见偏差）。 +- 已清理：集群目录、socket 目录、日志与 `/tmp` 脚本均已删除， + `ls /tmp/resident_pg*` 确认无残留。 + +## 6. 已知偏差 + +- 本地 PG 结论不是生产 PG 结论；本地 HTTP 结论不是真实外网结论； + 双连接结论不是生产双 worker 结论；真实外网只读抓取演练仍未做。 +- 代理模式下 pin 不生效（见 S1 残留风险），须靠代理 egress allowlist。 +- 旧闭环测试第 5 步轮询时刻由 `t0+400` 改为 `t0+540`：新冷却语义下 + 160 秒间隔的变化会被正确抑制，测试改按冷却期外复查（注释已写明）， + 不是为通过而删断言。 +- 旧重定向测试与闭环测试的本地服务接法改为“resolver 返回公网测试 + IP + socket 层重定向”：门与 pin 全程真实跑，只有传输层终点是 + 测试替身；门本身由默认解析器用例独立覆盖。 +- `resident_config()` 每次调用读环境变量：改环境变量即生效， + 但已启动进程的既有默认值不会跨进程同步（与其它环境变量配置同款）。 +- 无效环境变量数值（非整数/<=0）回落默认值，不炸驱动（配置错误 + 按默认跑，运维侧须自行核对配置是否生效，见 /metrics 的 config 段）。 + +## 7. blocker（需用户拍板，本批不擅自定） + +- 拟定默认值清单（第 1 节全部）：上限 24、冷却 300 秒、保留期 90 天、 + 批量预算 120 秒、默认直连代理——供最终验收改。 +- 是否新增独立驱动进程（本批选批量总预算，见 S2 论证；独立驱动 + 仍是可选项，需用户定运维成本）。 +- SSRF 修复后的摘录口径最终版：本批拟定为“只来自已通过校验的 + 公开抓取结果”；是否进一步收紧为只留 hash/长度/脱敏摘录，仍由 + 用户定。 +- 生产 PG 双 worker 实测窗口、真实外网只读抓取演练窗口、 + 代理模式下的代理 egress allowlist 配置（若生产用代理模式）。 +- 控制台 UI 是否随修复批上（属产品范围，不计入本轮）。 + +## 8. 未做事项（本批明确不做） + +- 未 push、未开 PR、未合并、未部署；工作区未提交（由 parent 定）。 +- 未碰生产审核 worktree、闭环 worktree、主 checkout（feat-repo-picker）。 +- 未做控制台 UI、选择器/区域提取、其它探针、磁盘清理、真实外网抓取。 +- 未改生产默认值以外的任何行为口径；旧测试除上述两处时序/接法 + 调整外无删断言、无加 skip。