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

- DNS 级 SSRF pin：解析后 pin IP 建连，防 DNS 重绑定
- 批量总预算与每日上限/抖动冷却/保留期
- PG schema 与迁移清单
- 双连接死锁隔离
- 改目标重置基线
- 显式代理配置与结构化失败原因
- /metrics resident 口径

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

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"})

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

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__":

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

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

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: