diff --git a/cloud/tests/test_resident_website_loop.py b/cloud/tests/test_resident_website_loop.py
new file mode 100644
index 0000000..f9d25fe
--- /dev/null
+++ b/cloud/tests/test_resident_website_loop.py
@@ -0,0 +1,364 @@
+"""常驻监听网站闭环端到端测试。
+
+上一轮只验到注入假抓取的单测：真实抓取、真实 run_loop 轮次
+边界、经服务端 API 的全链路（创建 -> 基线 -> 变化建任务 ->
+暂停 -> 恢复 -> 失败退避 -> 结束）都没有跑过。本文件补这一环，
+不改任何生产默认值、不改生产校验口径：
+
+- 间隔仍是生产下限 60 秒；到期不是改间隔，而是给驱动传一个
+  更晚的 now（模拟时间流逝），与上一轮单测同款时间注入。
+- 目标 URL 仍走生产校验：主机名 website-loop.test 不是 IP 字
+  面、不是 localhost，所以校验通过；测试进程内只把这个主机
+  名的 DNS 解析指向本地测试 HTTP 服务（相当于 /etc/hosts 的
+  一条测试条目），抓取本身走真实 default_fetch（urllib、超时、
+  1MB 截断），不注入假 fetcher、不绕过生产代码。
+- 内容服务是线程内真实 http.server，可变内容、也可停服；
+  控制面是真实 ThreadingHTTPServer + 真实 SQLite 临时库；
+  变化建出的任务、轮询、审计都直接读库/接口核对。
+
+不联网、不跑真实模型、不碰生产。
+"""
+import http.client
+import http.server
+import json
+import os
+import socket
+import sqlite3
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.resident import _content_hash
+    from cloud.server import create_server
+    from cloud.worker import Worker
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.resident import _content_hash
+    from cloud.server import create_server
+    from cloud.worker import Worker
+
+HOST = "website-loop.test"
+
+
+class WebsiteLoopTestCase(unittest.TestCase):
+    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)
+        # 可变内容服务：state 字典跨重启保留，停服/重启可复现。
+        self.state = {"content": "version one"}
+        state = self.state
+
+        class Handler(http.server.BaseHTTPRequestHandler):
+            def do_GET(self):
+                body = state["content"].encode("utf-8")
+                self.send_response(200)
+                self.send_header("Content-Type", "text/plain")
+                self.send_header("Content-Length", str(len(body)))
+                self.end_headers()
+                self.wfile.write(body)
+
+            def log_message(self, *args):
+                pass
+
+        self._handler = Handler
+        self.content_server = http.server.ThreadingHTTPServer(
+            ("127.0.0.1", 0), Handler)
+        self.content_port = self.content_server.server_address[1]
+        self.content_thread = threading.Thread(
+            target=self.content_server.serve_forever, daemon=True)
+        self.content_thread.start()
+        # 测试专用 DNS：只把 website-loop.test 指到本地服务，
+        # 其它主机一律走原始解析；生产校验与抓取代码一字不改。
+        self._orig_getaddrinfo = socket.getaddrinfo
+
+        def patched(host, port, family=0, type=0, proto=0, flags=0):
+            if host == HOST:
+                return [(socket.AF_INET, socket.SOCK_STREAM, 6, "",
+                         ("127.0.0.1", port))]
+            return self._orig_getaddrinfo(
+                host, port, family, type, proto, flags)
+
+        socket.getaddrinfo = patched
+        # 测试环境自带 egress 代理：urllib 默认会把这个测试主机名
+        # 也送代理并被断开。只给本测试主机名加 no_proxy（测试
+        # 进程内生效），生产抓取代码与代理行为一字不改。
+        self._old_no_proxy = os.environ.get("no_proxy")
+        self._old_NO_PROXY = os.environ.get("NO_PROXY")
+        os.environ["no_proxy"] = (
+            (self._old_no_proxy + ",") if self._old_no_proxy
+            else "") + HOST
+        os.environ["NO_PROXY"] = (
+            (self._old_NO_PROXY + ",") if self._old_NO_PROXY
+            else "") + HOST
+        self.worker = Worker(
+            self.db_path, "/fake/host.py",
+            env={"XCM_API_KEY": "test-placeholder"},
+            runner=lambda cmd: SimpleNamespace(
+                returncode=0, stdout="{}", stderr=""))
+
+    def tearDown(self):
+        socket.getaddrinfo = self._orig_getaddrinfo
+        if self._old_no_proxy is None:
+            os.environ.pop("no_proxy", None)
+        else:
+            os.environ["no_proxy"] = self._old_no_proxy
+        if self._old_NO_PROXY is None:
+            os.environ.pop("NO_PROXY", None)
+        else:
+            os.environ["NO_PROXY"] = self._old_NO_PROXY
+        try:
+            self.content_server.shutdown()
+            self.content_server.server_close()
+        except Exception:
+            pass
+        self.content_thread.join(timeout=5)
+        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 _create_watch(self):
+        body = {"name": "site1",
+                "target_url": "http://%s:%d/data" % (
+                    HOST, self.content_port),
+                "prompt": "变化时汇报",
+                "interval_seconds": 60}
+        status, watch = self._req("POST", self._url(), body,
+                                  self.owner_key)
+        self.assertEqual(status, 201, watch)
+        return watch
+
+    def _get_watch(self, watch_id):
+        status, body = self._req(
+            "GET", self._url("/" + watch_id), None, self.owner_key)
+        self.assertEqual(status, 200, body)
+        return body
+
+    def _tasks(self):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            return conn.execute(
+                "SELECT id, prompt, status FROM tasks"
+                " ORDER BY created_at, id").fetchall()
+        finally:
+            conn.close()
+
+    def _audit(self):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            return [r[0] for r in conn.execute(
+                "SELECT action FROM audit_log WHERE project_id = ?",
+                (self.project["id"],)).fetchall()]
+        finally:
+            conn.close()
+
+    def test_full_website_loop_end_to_end(self):
+        # 1. 创建：经服务端 API，running、游标为空、下次轮询在位。
+        watch = self._create_watch()
+        wid = watch["id"]
+        self.assertEqual(watch["status"], "running")
+        self.assertEqual(watch["interval_seconds"], 60)
+        self.assertEqual(watch["fetch_timeout_seconds"], 10)
+        self.assertIsNone(watch["last_content_hash"])
+        self.assertIsNotNone(watch["next_poll_at"])
+        self.assertIn("resident_watch_created", self._audit())
+        t0 = time.time()
+
+        # 2. 基线：真实抓取 version one，只建游标，不建任务。
+        polls = self.worker.process_due_resident_watches(now=t0 + 120)
+        self.assertEqual(len(polls), 1, polls)
+        self.assertEqual(polls[0]["status"], "baseline")
+        self.assertEqual(
+            polls[0]["observed_hash"], _content_hash("version one"))
+        self.assertEqual(self._tasks(), [])
+        body = self._get_watch(wid)
+        self.assertEqual(body["last_content_hash"],
+                         _content_hash("version one"))
+        self.assertEqual(len(body["polls"]), 1)
+        self.assertEqual(body["polls"][0]["status"], "baseline")
+
+        # 3. 变化：内容改成 version two，到期轮询建出 queued 任务。
+        self.state["content"] = "version two"
+        polls = self.worker.process_due_resident_watches(now=t0 + 240)
+        self.assertEqual(len(polls), 1, polls)
+        self.assertEqual(polls[0]["status"], "changed")
+        self.assertIsNotNone(polls[0]["task_id"])
+        tasks = self._tasks()
+        self.assertEqual(len(tasks), 1)
+        self.assertEqual(tasks[0][0], polls[0]["task_id"])
+        self.assertEqual(tasks[0][2], "queued")
+        self.assertIn("变化时汇报", tasks[0][1])
+        self.assertIn("version two", tasks[0][1])
+        self.assertIn("resident_watch_change_task_created",
+                      self._audit())
+        body = self._get_watch(wid)
+        self.assertEqual(body["last_content_hash"],
+                         _content_hash("version two"))
+        self.assertIn("version two", body["last_change_text"])
+
+        # 4. 暂停：暂停期间再改内容，驱动不轮询、游标不动。
+        status, body = self._req(
+            "POST", self._url("/%s/pause" % wid), {}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["status"], "paused")
+        self.assertIsNone(body["next_poll_at"])
+        self.state["content"] = "version three"
+        polls = self.worker.process_due_resident_watches(now=t0 + 400)
+        self.assertEqual(polls, [])
+        body = self._get_watch(wid)
+        self.assertEqual(len(body["polls"]), 2)
+        self.assertEqual(body["last_content_hash"],
+                         _content_hash("version two"))
+
+        # 5. 恢复：用旧游标比对，报出暂停期间的变化并建任务，
+        #    不是从头建基线。
+        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)
+        self.assertEqual(len(polls), 1, polls)
+        self.assertEqual(polls[0]["status"], "changed")
+        tasks = self._tasks()
+        self.assertEqual(len(tasks), 2)
+        self.assertIn("version three", tasks[1][1])
+        body = self._get_watch(wid)
+        self.assertEqual(body["last_content_hash"],
+                         _content_hash("version three"))
+
+        # 6. 失败：停服后轮询记 failed、退避、不结束、游标不动；
+        #    恢复服务后内容未变，轮询记 unchanged、失败清零。
+        self.content_server.shutdown()
+        self.content_server.server_close()
+        self.content_thread.join(timeout=5)
+        polls = self.worker.process_due_resident_watches(now=t0 + 600)
+        self.assertEqual(len(polls), 1, polls)
+        self.assertEqual(polls[0]["status"], "failed")
+        body = self._get_watch(wid)
+        self.assertEqual(body["status"], "running")
+        self.assertEqual(body["consecutive_failures"], 1)
+        self.assertEqual(body["last_content_hash"],
+                         _content_hash("version three"))
+        self.assertGreater(body["next_poll_at"], t0 + 600 + 60)
+        self.content_server = http.server.ThreadingHTTPServer(
+            ("127.0.0.1", self.content_port), self._handler)
+        self.content_thread = threading.Thread(
+            target=self.content_server.serve_forever, daemon=True)
+        self.content_thread.start()
+        polls = self.worker.process_due_resident_watches(now=t0 + 800)
+        self.assertEqual(len(polls), 1, polls)
+        self.assertEqual(polls[0]["status"], "unchanged")
+        body = self._get_watch(wid)
+        self.assertEqual(body["consecutive_failures"], 0)
+        self.assertEqual(len(self._tasks()), 2)
+
+        # 7. 结束：终态，驱动不再轮询，改/停/恢复一律 409。
+        status, body = self._req(
+            "POST", self._url("/%s/end" % wid), {}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["status"], "ended")
+        self.assertIsNone(body["next_poll_at"])
+        self.assertEqual(body["last_content_hash"],
+                         _content_hash("version three"))
+        polls = self.worker.process_due_resident_watches(now=t0 + 1000)
+        self.assertEqual(polls, [])
+        for action in ("pause", "resume", "end"):
+            status, body = self._req(
+                "POST", self._url("/%s/%s" % (wid, action)), {},
+                self.owner_key)
+            self.assertEqual(status, 409, action)
+        status, _ = self._req(
+            "PATCH", self._url("/" + wid), {"interval_seconds": 120},
+            self.owner_key)
+        self.assertEqual(status, 409)
+        body = self._get_watch(wid)
+        self.assertEqual(body["status"], "ended")
+        self.assertEqual(len(body["polls"]), 5)
+
+    def test_run_loop_boundary_calls_resident_driver(self):
+        # 8. 驱动接线：run_loop 轮次边界确实调到常驻驱动，
+        #    不只是测试直接调 process_due。run_all 等无关边界
+        #    在本测试里打桩，只留 resident 驱动走真实抓取。
+        watch = self._create_watch()
+        wid = watch["id"]
+        worker = self.worker
+        calls = []
+        original = Worker.process_due_resident_watches
+
+        def recording():
+            # 实例属性上的函数不会自动绑定 self，必须闭包捕获
+            # worker，不能写成 def recording(self)。
+            calls.append(time.time())
+            return original(worker)
+
+        worker.process_due_resident_watches = recording
+        worker.run_all = lambda: []
+        worker.cleanup_stale_containers = lambda: None
+        worker._write_heartbeat = lambda: None
+        worker.register = lambda: None
+        worker._process_push_due = lambda: None
+        worker.process_auto_memories = lambda: None
+        worker._reap_kill_requests = lambda *a, **k: None
+        worker.auto_resume_interrupted = lambda: None
+        worker._heartbeat_running_tasks = lambda: None
+        worker._registry_heartbeat = lambda: None
+        worker._code_changed = lambda: False
+        worker._code_signature_valid = lambda: True
+        worker.run_loop(interval=0.05, max_seconds=0.4)
+        self.assertGreaterEqual(len(calls), 1)
+        body = self._get_watch(wid)
+        self.assertEqual(len(body["polls"]), 1)
+        self.assertEqual(body["polls"][0]["status"], "baseline")
+        self.assertEqual(body["last_content_hash"],
+                         _content_hash("version one"))
+
+
+if __name__ == "__main__":
+    unittest.main()