diff --git a/cloud/resident.py b/cloud/resident.py new file mode 100644 index 0000000..4f76e27 --- /dev/null +++ b/cloud/resident.py @@ -0,0 +1,846 @@ +"""常驻监听任务（resident watch）。 + +与现有三样东西的关系，先说清楚，避免把旧语义冒充成常驻： + +- 任务状态机（cloud/server.py）：queued -> running -> + completed / failed / interrupted / merged / merge-failed， + 其中 /pause 只是把进行中回合任务标 interrupted 并置 + paused_by_user=1（见 server._pause_task），不是一个可续跑 + 的“已暂停常驻态”。本模块绝不复用那条路：常驻态存在 + resident_watches 表里，与任务表互不干扰。 +- 云端自动化（cloud/automations.py）：定义层只有 active / + cancelled，到期靠外部触发器调 run_due_automation 抢单建任务， + 服务端本身不起定时器。本模块复用它的两道口径而不是另起 + 一套：抢单用条件 UPDATE（同一间隔内只有一个执行器能拿到 + rowcount 1），配额门直接调 automations._automation_quota_ + exceeded（与 server / worker 同口径，改口径时只改那一处）。 +- goal 循环（host.py run_goal_loop，max_rounds=0 不限轮次）： + 目标是“做完一个目标”，结束条件是模型宣告达成；本模块的 + 目标是“一直监听外部变化”，结束条件只能是用户结束（或 + 配额耗尽转暂停，见下），两者不是一回事。 + +状态机（本模块自有，与任务状态机不冲突）： + running -> paused -> running（恢复） + running / paused -> ended（终态） + ended -> 任何操作一律拒绝（恢复必须新建任务，见 end_watch） + +暂停保留现场：暂停只改 status / paused_reason / next_poll_at， +last_content_hash、last_poll_at、last_change_at、last_change_text +一律不动；暂停期间 due 为 False，驱动绝不轮询。恢复时 +next_poll_at 置为 now（立刻复查一次），用旧游标比对：暂停 +期间网站变了，恢复后第一次轮询就会把那次变化报出来，而不是 +丢掉——“从游标续跑”就是这个意思。 + +结束是终态：ended 后 next_poll_at 为 NULL，驱动的 due 查询 +只认 status='running'，所以结束后绝不会静默重启；ended 行 +保留在库里供审计与查询，名字可重用（与自动化 cancelled 的 +名字重用规则同款）。 + +变化才花钱：轮询本身是廉价抓取 + sha256 比对，不跑模型。 +只有三类轮询产出：baseline（首次抓取，只建游标，不上报）、 +unchanged（游标不变，不建任务）、changed（游标变了，才建 +一个 queued 任务，prompt = 用户的处理提示 + 变化摘录）。 +每次轮询都在 resident_polls 留一行，可查。 + +配额边界：配额门只在“变化要建模型任务”那一刻过，不在 +轮询那一刻过（轮询不花模型额度）。配额耗尽时不结束、不丢 +变化：状态转 paused、paused_reason='quota_exceeded'，游标不 +推进（恢复后复查仍会报出同一变化），轮询记 +changed_quota_paused。配额检查本身失败（fail-closed 文本） +时不转暂停，按瞬时失败退避处理，游标同样不推进。 + +驱动：服务端本身不起定时器（与自动化同款结论）。真正的 +驱动是 cloud/worker.py 的 run_loop 轮次边界调 +process_due_resident_watches（见 worker 里的调用点），或 +任何外部触发器直接调本模块的 run_due_resident_watch / +process_due_resident_watches；两者都走 claim 门，两个驱动 +同时跑也不会在同一间隔内重复轮询。 + +本模块只依赖标准库 + cloud.automations 的复用函数 +（row_get、_audit_row、_automation_quota_exceeded）与 +cloud.db 的完整性错误类型（仅用于把唯一索引冲突转成 +ValueError，不 import 其它 cloud 模块），避免循环 import。 +""" + +import hashlib +import ipaddress +import re +import sqlite3 +import time +import urllib.request +import uuid +from urllib.parse import urljoin +from urllib.parse import urlsplit + +from cloud.automations import _audit_row as _auto_audit_row +from cloud.automations import _automation_quota_exceeded +from cloud.automations import row_get +from cloud.db import PostgresIntegrityError as _PgIntegrityError + +MIN_INTERVAL_SECONDS = 60 +MAX_INTERVAL_SECONDS = 30 * 86400 +MAX_NAME_LEN = 100 +MAX_PROMPT_LEN = 20000 +MAX_URL_LEN = 2000 +MIN_FETCH_TIMEOUT_SECONDS = 1 +MAX_FETCH_TIMEOUT_SECONDS = 60 +DEFAULT_FETCH_TIMEOUT_SECONDS = 10 +MAX_FETCH_BYTES = 1000000 +MAX_CHANGE_TEXT_LEN = 2000 +MAX_POLL_DETAIL_LEN = 500 +POLL_LIMIT = 50 + +RUNNING = "running" +PAUSED = "paused" +ENDED = "ended" + +# 变化判定口径（v1）：对抓取文本做空白折叠后取 sha256。 +# 动态页面（广告、时间戳）本身会抖动，v1 不做选择器/区域 +# 提取，抖动风险写在验证摘要的已知偏差里，不假装已解决。 +_WS_RE = re.compile(r"\s+") + + +WATCH_COLUMNS = ( + "id", "project_id", "name", "target_url", "prompt", + "interval_seconds", "fetch_timeout_seconds", "status", + "last_poll_at", "last_claimed_at", "next_poll_at", + "last_content_hash", "last_change_at", "last_change_text", + "consecutive_failures", "paused_reason", "ended_at", + "created_by", "created_at", "updated_at") + +POLL_COLUMNS = ( + "id", "watch_id", "project_id", "status", "observed_hash", + "detail", "task_id", "created_at") + + +def _row_dict(row, columns): + if row is None or isinstance(row, dict): + return row + try: + row["id"] + return row + except (KeyError, IndexError, TypeError): + # sqlite3.Row 缺列报 IndexError（不是 TypeError）：只有 + # SELECT * 的行能原样用，单列查询（如 claim 取 interval） + # 走 zip 回落，columns 必须与 SELECT 列顺序一致。 + return dict(zip(columns, row)) + + +def _existing_columns(conn, table): + try: + return {row[1] for row in conn.execute( + "PRAGMA table_info(%s)" % table).fetchall()} + except Exception: + return set() + + +def ensure_resident_tables(conn): + """幂等建常驻监听表，补齐新增列（仿 automations 写法）。""" + conn.execute( + "CREATE TABLE IF NOT EXISTS resident_watches(" + "id TEXT PRIMARY KEY, " + "project_id TEXT, " + "name TEXT, " + "target_url TEXT, " + "prompt TEXT, " + "interval_seconds INTEGER, " + "fetch_timeout_seconds INTEGER, " + "status TEXT, " + "last_poll_at DOUBLE PRECISION, " + "last_claimed_at DOUBLE PRECISION, " + "next_poll_at DOUBLE PRECISION, " + "last_content_hash TEXT, " + "last_change_at DOUBLE PRECISION, " + "last_change_text TEXT, " + "consecutive_failures INTEGER, " + "paused_reason TEXT, " + "ended_at DOUBLE PRECISION, " + "created_by TEXT, " + "created_at DOUBLE PRECISION, " + "updated_at DOUBLE PRECISION)") + conn.execute( + "CREATE TABLE IF NOT EXISTS resident_polls(" + "id TEXT PRIMARY KEY, " + "watch_id TEXT, " + "project_id TEXT, " + "status TEXT, " + "observed_hash TEXT, " + "detail TEXT, " + "task_id TEXT, " + "created_at DOUBLE PRECISION)") + existing = _existing_columns(conn, "resident_watches") + for col, ddl in ( + ("fetch_timeout_seconds", "INTEGER"), + ("next_poll_at", "DOUBLE PRECISION"), + ("last_content_hash", "TEXT"), + ("last_change_at", "DOUBLE PRECISION"), + ("last_change_text", "TEXT"), + ("consecutive_failures", "INTEGER"), + ("paused_reason", "TEXT"), + ("ended_at", "DOUBLE PRECISION")): + if col not in existing: + try: + conn.execute( + "ALTER TABLE resident_watches ADD COLUMN %s %s" + % (col, ddl)) + except Exception: + pass + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_resident_watches_project" + " ON resident_watches(project_id, status)") + # 同项目未结束名唯一：先查后插挡不住并发，唯一索引才是 + # 真正的关口（server environments 的同款部分唯一索引， + # WHERE status != 'archived' 的 resident 版）；已结束名 + # 不在索引内，可重用。已存在重复名的旧库建索引会失败， + # 此时不炸掉整次建表调用，SELECT 查重仍挡正常路径。 + try: + conn.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS" + " idx_resident_watches_name" + " ON resident_watches(project_id, name)" + " WHERE status != 'ended'") + except Exception: + pass + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_resident_polls_watch" + " ON resident_polls(watch_id, created_at)") + + +def _backoff_seconds(failures): + """失败退避增量：30、60、120……秒，上限 3600 秒。 + 调用方把它加在间隔之上（下次 = now + interval + backoff）， + 失败后绝不比正常间隔更频繁地重试，行为见失败分支测试。""" + if failures <= 0: + return 0 + return min(3600, 30 * (2 ** (failures - 1))) + + +def watch_to_dict(row, now=None): + row = _row_dict(row, WATCH_COLUMNS) + if now is None: + now = time.time() + status = row["status"] + nxt = row_get(row, "next_poll_at") + due = bool(status == RUNNING and (nxt is None or nxt <= now)) + return { + "id": row["id"], + "project_id": row["project_id"], + "name": row["name"], + "target_url": row["target_url"], + "prompt": row["prompt"], + "interval_seconds": row["interval_seconds"], + "fetch_timeout_seconds": row_get(row, "fetch_timeout_seconds") + or DEFAULT_FETCH_TIMEOUT_SECONDS, + "status": status, + "last_poll_at": row_get(row, "last_poll_at"), + "last_claimed_at": row_get(row, "last_claimed_at"), + "next_poll_at": nxt, + "last_content_hash": row_get(row, "last_content_hash"), + "last_change_at": row_get(row, "last_change_at"), + "last_change_text": row_get(row, "last_change_text"), + "consecutive_failures": row_get(row, "consecutive_failures") or 0, + "paused_reason": row_get(row, "paused_reason"), + "ended_at": row_get(row, "ended_at"), + "created_by": row["created_by"], + "created_at": row["created_at"], + "updated_at": row_get(row, "updated_at"), + "due": due, + } + + +def _is_noncanonical_numeric_host(host): + """非规范数字 IP 写法：单整数（2130706433）、十六进制 + （0x7f000001）、前导零八进制（0177.0.0.1）、含十六进制段 + 的点分写法——urllib/socket 的 inet_aton 会把它们解析成 + 数字 IP，ipaddress 却认不出来。规范点分十进制（4 段、 + 每段 0-255、无前导零，0 本身除外）不在此列；普通域名 + （含字母的标签）也不在此列。""" + if re.fullmatch(r"0[xX][0-9a-fA-F]+", host): + return True + if re.fullmatch(r"[0-9]+", host): + return True + if "." not in host: + return False + labels = host.split(".") + if not all(re.fullmatch(r"0[xX][0-9a-fA-F]+|[0-9]+", lab) + for lab in labels): + return False + if len(labels) != 4: + return True + for lab in labels: + if lab.startswith("0x") or lab.startswith("0X"): + return True + if len(lab) > 1 and lab.startswith("0"): + return True + if int(lab) > 255: + return True + return False + + +def validate_target_url(url): + """目标 URL 校验：仅 http/https、无 userinfo、主机非空、 + 字面 localhost（含尾点、*.localhost）/ 私有 / 回环 / + 链路本地 IP / 非规范数字 IP 写法拒绝（与 + websearch._is_blocked_fetch_url 的字面 IP 口径同款， + DNS 解析后的私有地址不在 v1 覆盖内，见摘要已知偏差； + 重定向目标由 _SafeRedirectHandler 逐跳复用本校验）。""" + if not url or not isinstance(url, str) or not url.strip(): + raise ValueError("target_url required") + url = url.strip() + if len(url) > MAX_URL_LEN: + raise ValueError("target_url too long") + parts = urlsplit(url) + if parts.scheme not in ("http", "https"): + raise ValueError("target_url must be http or https") + if parts.username or parts.password: + raise ValueError("target_url must not contain credentials") + host = (parts.hostname or "").strip().lower().rstrip(".") + if not host: + raise ValueError("target_url host required") + if host == "localhost" or host.endswith(".localhost"): + raise ValueError("localhost URLs are blocked") + if _is_noncanonical_numeric_host(host): + raise ValueError("non-canonical numeric IP URLs are blocked") + try: + addr = ipaddress.ip_address(host) + except ValueError: + addr = None + if addr is not None and (addr.is_private or addr.is_loopback + or addr.is_link_local or addr.is_reserved + or addr.is_multicast): + raise ValueError("private / loopback URLs are blocked") + return url + + +def _validate_fields(name, prompt, target_url, interval_seconds, + fetch_timeout_seconds): + if not name or not isinstance(name, str) or not name.strip(): + raise ValueError("name required") + name = name.strip() + if len(name) > MAX_NAME_LEN: + raise ValueError("name too long") + if not prompt or not isinstance(prompt, str) or not prompt.strip(): + raise ValueError("prompt required") + prompt = prompt.strip() + if len(prompt) > MAX_PROMPT_LEN: + raise ValueError("prompt too long") + target_url = validate_target_url(target_url) + if isinstance(interval_seconds, bool) \ + or not isinstance(interval_seconds, int): + raise ValueError("interval_seconds must be an integer") + if interval_seconds < MIN_INTERVAL_SECONDS: + raise ValueError( + "interval_seconds must be at least %d" % MIN_INTERVAL_SECONDS) + if interval_seconds > MAX_INTERVAL_SECONDS: + raise ValueError("interval_seconds too large") + if fetch_timeout_seconds is None: + fetch_timeout_seconds = DEFAULT_FETCH_TIMEOUT_SECONDS + if isinstance(fetch_timeout_seconds, bool) \ + or not isinstance(fetch_timeout_seconds, int): + raise ValueError("fetch_timeout_seconds must be an integer") + if not MIN_FETCH_TIMEOUT_SECONDS <= fetch_timeout_seconds \ + <= MAX_FETCH_TIMEOUT_SECONDS: + raise ValueError( + "fetch_timeout_seconds must be between %d and %d" + % (MIN_FETCH_TIMEOUT_SECONDS, MAX_FETCH_TIMEOUT_SECONDS)) + return name, prompt, target_url, interval_seconds, fetch_timeout_seconds + + +def create_watch(conn, project_id, name, target_url, prompt, created_by, + interval_seconds=3600, fetch_timeout_seconds=None, + now=None): + """新建常驻监听：建完即到期（next_poll_at=now），驱动下 + 一轮即可认领；同项目内未结束（running/paused）的名字唯一， + 已结束的名字可重用。创建本身不走配额门（轮询不花模型额度， + 配额只在变化建任务时过，见 run_due_resident_watch）。""" + (name, prompt, target_url, interval_seconds, + fetch_timeout_seconds) = _validate_fields( + name, prompt, target_url, interval_seconds, fetch_timeout_seconds) + if now is None: + now = time.time() + ensure_resident_tables(conn) + clash = conn.execute( + "SELECT id FROM resident_watches WHERE project_id = ?" + " AND name = ? AND status != ?", + (project_id, name, ENDED)).fetchone() + if clash is not None: + raise ValueError( + "resident watch name already exists in this project") + watch_id = uuid.uuid4().hex + try: + conn.execute( + "INSERT INTO resident_watches(id, project_id, name, target_url," + " prompt, interval_seconds, fetch_timeout_seconds, status," + " last_poll_at, last_claimed_at, next_poll_at," + " last_content_hash, last_change_at, last_change_text," + " consecutive_failures, paused_reason, ended_at," + " created_by, created_at, updated_at)" + " VALUES(?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?," + " NULL, NULL, NULL, 0, NULL, NULL, ?, ?, ?)", + (watch_id, project_id, name, target_url, prompt, + interval_seconds, fetch_timeout_seconds, RUNNING, now, + created_by, now, now)) + except (sqlite3.IntegrityError, _PgIntegrityError) as exc: + # 并发下 SELECT 查重与 INSERT 之间可能插进同名行， + # 唯一索引是真正的关口：冲突转回与查重同款 ValueError， + # 调用方（server）据此返回 400，不把数据库错误外泄。 + try: + conn.rollback() + except Exception: + pass + raise ValueError( + "resident watch name already exists in this project") from exc + conn.commit() + row = conn.execute( + "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone() + return watch_to_dict(row, now) + + +def list_watches(conn, project_id, now=None): + if now is None: + now = time.time() + ensure_resident_tables(conn) + rows = conn.execute( + "SELECT * FROM resident_watches WHERE project_id = ?" + " ORDER BY created_at, id", + (project_id,)).fetchall() + return [watch_to_dict(r, now) for r in rows] + + +def get_watch(conn, watch_id, now=None): + if now is None: + now = time.time() + ensure_resident_tables(conn) + row = conn.execute( + "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone() + if row is None: + return None + return watch_to_dict(row, now) + + +def list_polls(conn, watch_id, limit=POLL_LIMIT): + ensure_resident_tables(conn) + rows = conn.execute( + "SELECT * FROM resident_polls WHERE watch_id = ?" + " ORDER BY created_at DESC, id LIMIT ?", + (watch_id, limit)).fetchall() + out = [] + for raw in rows: + r = _row_dict(raw, POLL_COLUMNS) + out.append({ + "id": r["id"], + "watch_id": r["watch_id"], + "project_id": r["project_id"], + "status": r["status"], + "observed_hash": row_get(r, "observed_hash"), + "detail": row_get(r, "detail") or "", + "task_id": row_get(r, "task_id"), + "created_at": r["created_at"], + }) + return out + + +def _transition(conn, watch_id, from_statuses, to_status, now, + extra_sets=(), extra_vals=()): + """条件状态迁移：返回 (ok, dict)。行不存在时 (False, None)； + 状态不对时 (False, 旧 dict)，调用方据此区分 404 与 409。""" + ensure_resident_tables(conn) + row = conn.execute( + "SELECT * FROM resident_watches WHERE id = ?", + (watch_id,)).fetchone() + if row is None: + return False, None + row = _row_dict(row, WATCH_COLUMNS) + if row["status"] not in from_statuses: + return False, watch_to_dict(row, now) + placeholders = ",".join("?" for _ in from_statuses) + sets = ["status = ?", "updated_at = ?"] + list(extra_sets) + vals = [to_status, now] + list(extra_vals) + [watch_id] \ + + list(from_statuses) + cur = conn.execute( + "UPDATE resident_watches SET %s WHERE id = ? AND status IN (%s)" + % (", ".join(sets), placeholders), tuple(vals)) + conn.commit() + if cur.rowcount != 1: + return False, watch_to_dict(row, now) + row = conn.execute( + "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone() + return True, watch_to_dict(row, now) + + +def pause_watch(conn, watch_id, now=None, reason="user_paused"): + """用户暂停：仅 running 能暂停；保留全部现场，只把 + next_poll_at 清掉，驱动即停。""" + if now is None: + now = time.time() + return _transition(conn, watch_id, (RUNNING,), PAUSED, now, + extra_sets=("paused_reason = ?", + "next_poll_at = NULL"), + extra_vals=(reason,)) + + +def resume_watch(conn, watch_id, now=None): + """恢复：仅 paused 能恢复；现场不动，next_poll_at=now + 立刻复查一次（用旧游标比对，暂停期间的变化不丢）。""" + if now is None: + now = time.time() + return _transition(conn, watch_id, (PAUSED,), RUNNING, now, + extra_sets=("paused_reason = NULL", + "next_poll_at = ?"), + extra_vals=(now,)) + + +def end_watch(conn, watch_id, now=None): + """结束：running / paused -> ended，终态。ended 后 + pause / resume / end 一律 (False, 旧 dict)，调用方转 409； + 恢复已结束的监听必须新建任务，不复活本行。""" + if now is None: + now = time.time() + return _transition(conn, watch_id, (RUNNING, PAUSED), ENDED, now, + extra_sets=("ended_at = ?", + "next_poll_at = NULL"), + extra_vals=(now,)) + + +def update_watch(conn, watch_id, fields, now=None): + """更新定义：只认 name / prompt / target_url / + interval_seconds / fetch_timeout_seconds；ended 的行 409 + （终态不能改，改它必须重建）；改间隔不重置游标、 + 不立即触发：已排好的 next_poll_at 不变，下一次轮询 + 完成后才按新间隔重算（想立即复查请暂停后恢复）。""" + if now is None: + now = time.time() + ensure_resident_tables(conn) + row = conn.execute( + "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone() + if row is None: + return False, None + row = _row_dict(row, WATCH_COLUMNS) + if row["status"] == ENDED: + return False, watch_to_dict(row, now) + if not isinstance(fields, dict): + raise ValueError("fields must be an object") + if not any(k in fields for k in ( + "name", "prompt", "target_url", "interval_seconds", + "fetch_timeout_seconds")): + raise ValueError("no updatable field given") + current = watch_to_dict(row, now) + (name, prompt, target_url, interval, timeout) = _validate_fields( + fields.get("name", current["name"]), + fields.get("prompt", current["prompt"]), + fields.get("target_url", current["target_url"]), + fields.get("interval_seconds", current["interval_seconds"]), + fields.get("fetch_timeout_seconds", + current["fetch_timeout_seconds"])) + if name != current["name"]: + clash = conn.execute( + "SELECT id FROM resident_watches WHERE project_id = ?" + " AND name = ? AND status != ? AND id != ?", + (row["project_id"], name, ENDED, watch_id)).fetchone() + if clash is not None: + raise ValueError( + "resident watch name already exists in this project") + 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)) + except (sqlite3.IntegrityError, _PgIntegrityError) as exc: + try: + conn.rollback() + except Exception: + pass + raise ValueError( + "resident watch name already exists in this project") from exc + conn.commit() + row = conn.execute( + "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone() + return True, watch_to_dict(row, now) + + +def due_watches(conn, now=None): + """全部项目当前到期的 running 监听（纯读，不抢单）， + 供驱动列出待处理项；执行入口是 claim + run_due。""" + if now is None: + now = time.time() + ensure_resident_tables(conn) + rows = conn.execute( + "SELECT * FROM resident_watches WHERE status = ?" + " AND (next_poll_at IS NULL OR next_poll_at <= ?)" + " ORDER BY next_poll_at, created_at, id", + (RUNNING, now)).fetchall() + return [watch_to_dict(r, now) for r in rows] + + +def claim_due_watch(conn, watch_id, now=None): + """抢单：条件 UPDATE 只有一行能成功。抢到即把 + last_poll_at / last_claimed_at 写成 now，并把 next_poll_at + 预置为 now + interval（崩在抓取中途时，下个间隔仍会重试， + 不会因认领锁死而永久停摆；抓取完成后 run_due 会按成败 + 重算 next_poll_at）。抢不到时返回 False，调用方必须放弃。""" + if now is None: + now = time.time() + ensure_resident_tables(conn) + row = conn.execute( + "SELECT interval_seconds FROM resident_watches WHERE id = ?", + (watch_id,)).fetchone() + if row is None: + return False + interval = row_get(_row_dict(row, ("interval_seconds",)), + "interval_seconds") or MIN_INTERVAL_SECONDS + cur = conn.execute( + "UPDATE resident_watches SET last_poll_at = ?," + " last_claimed_at = ?, next_poll_at = ?" + " WHERE id = ? AND status = ?" + " AND (next_poll_at IS NULL OR next_poll_at <= ?)", + (now, now, now + interval, watch_id, RUNNING, now)) + conn.commit() + return cur.rowcount == 1 + + +def _content_hash(text): + normalized = _WS_RE.sub(" ", text or "").strip() + return hashlib.sha256(normalized.encode("utf-8")).hexdigest() + + +class _SafeRedirectHandler(urllib.request.HTTPRedirectHandler): + """重定向逐跳校验：创建时只校验了初始 URL，公开 URL + 重定向到内网地址会绕过那道校验；相对跳转先 urljoin 成 + 绝对 URL 再校验，不合格直接抛 ValueError，不跟随。""" + + 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 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: + raw = resp.read(MAX_FETCH_BYTES + 1)[:MAX_FETCH_BYTES] + return raw.decode("utf-8", errors="replace") + + +def _record_poll(conn, watch_id, project_id, status, observed_hash, + detail, task_id, now): + poll_id = uuid.uuid4().hex + conn.execute( + "INSERT INTO resident_polls(id, watch_id, project_id, status," + " observed_hash, detail, task_id, created_at)" + " VALUES(?, ?, ?, ?, ?, ?, ?, ?)", + (poll_id, watch_id, project_id, status, observed_hash, + (detail or "")[:MAX_POLL_DETAIL_LEN], task_id, now)) + return { + "id": poll_id, + "watch_id": watch_id, + "project_id": project_id, + "status": status, + "observed_hash": observed_hash, + "detail": (detail or "")[:MAX_POLL_DETAIL_LEN], + "task_id": task_id, + "created_at": now, + } + + +def _create_change_task(conn, watch, excerpt, now): + """变化建任务：列补齐口径与 automations.run_due_automation + 同款；任务 prompt = 用户处理提示 + 变化摘录。""" + existing_tasks = _existing_columns(conn, "tasks") + for col, ddl in ( + ("executor", "TEXT"), + ("model", "TEXT"), + ("effort", "TEXT"), + ("environment_id", "TEXT"), + ("parent_task_id", "TEXT"), + ("coordinator_part", "TEXT")): + if col not in existing_tasks: + try: + conn.execute( + "ALTER TABLE tasks ADD COLUMN %s %s" % (col, ddl)) + except Exception: + pass + task_id = uuid.uuid4().hex + prompt = ("%s\n\n[常驻监听变化] %s %s\n变化摘录：\n%s" + % (watch["prompt"], watch["name"], watch["target_url"], + excerpt)) + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " idempotency_key, depends_on, priority, created_by," + " created_at, updated_at, last_heartbeat, executor," + " model, effort, environment_id, parent_task_id," + " coordinator_part)" + " VALUES(?, ?, ?, 'queued', NULL, '[]', 0, ?, ?, ?," + " NULL, NULL, NULL, NULL, NULL, NULL, NULL)", + (task_id, watch["project_id"], prompt, watch["created_by"], + now, now)) + return task_id + + +def _guarded_watch_update(conn, watch_id, sets, vals): + """终局 UPDATE 带 status='running' 条件：抓取中途被用户 + 暂停/结束时，条件不满足、rowcount 为 0，调用方据此回滚 + 并放弃本轮（不建任务、不记轮询），而不是把 paused/ended + 行的 next_poll_at 改回未来值。返回是否成功。""" + cur = conn.execute( + "UPDATE resident_watches SET %s WHERE id = ? AND status = ?" + % ", ".join(sets), + tuple(vals) + (watch_id, RUNNING)) + return cur.rowcount == 1 + + +def run_due_resident_watch(conn, watch_id, fetcher=None, now=None): + """单条常驻监听的驱动唯一入口：抢单 -> 廉价抓取 -> hash + 比对 -> 记轮询 -> 仅变化时建任务。 + + 抢不到返回 None（不存在、已暂停/已结束、未到期、已被 + 别的驱动抢走），不记轮询、不建任务。抓取失败记 failed + 并退避，游标不动。首次抓取记 baseline，只建游标。未变 + 化记 unchanged，不建任务。变化时先过配额门：配额耗尽 + 转 paused（paused_reason='quota_exceeded'，游标不推进）， + 记 changed_quota_paused；配额检查失败按瞬时失败退避， + 游标同样不推进；配额内才建 queued 任务并推进游标。""" + if now is None: + now = time.time() + ensure_resident_tables(conn) + if not claim_due_watch(conn, watch_id, now): + return None + row = conn.execute( + "SELECT * FROM resident_watches WHERE id = ?", (watch_id,)).fetchone() + if row is None: + return None + watch = watch_to_dict(row, now) + if watch["status"] != RUNNING: + # 认领后、抓取前被用户暂停/结束：不抓取、不建任务、 + # 不改行（暂停/结束已把 next_poll_at 清掉，保持原样）。 + return None + interval = watch["interval_seconds"] + timeout = watch["fetch_timeout_seconds"] + fetch = fetcher or default_fetch + try: + text = fetch(watch["target_url"], timeout) + except Exception as exc: + failures = watch["consecutive_failures"] + 1 + nxt = now + interval + _backoff_seconds(failures) + if not _guarded_watch_update( + conn, watch_id, + ("consecutive_failures = ?", "next_poll_at = ?", + "updated_at = ?"), + (failures, nxt, now)): + conn.rollback() + return None + poll = _record_poll(conn, watch_id, watch["project_id"], + "failed", None, str(exc)[:MAX_POLL_DETAIL_LEN], + None, now) + conn.commit() + _auto_audit_row(conn, watch["project_id"], + "resident_watch_poll_failed", + "%s: %s" % (watch_id, str(exc)[:200]), now) + return poll + if not isinstance(text, str): + text = str(text) + observed = _content_hash(text) + excerpt = text[:MAX_CHANGE_TEXT_LEN] + old_hash = watch["last_content_hash"] + if old_hash is None: + if not _guarded_watch_update( + conn, watch_id, + ("last_content_hash = ?", "consecutive_failures = 0", + "next_poll_at = ?", "updated_at = ?"), + (observed, now + interval, now)): + conn.rollback() + return None + poll = _record_poll(conn, watch_id, watch["project_id"], + "baseline", observed, "first poll", None, now) + conn.commit() + return poll + if observed == old_hash: + if not _guarded_watch_update( + conn, watch_id, + ("consecutive_failures = 0", "next_poll_at = ?", + "updated_at = ?"), + (now + interval, now)): + conn.rollback() + return None + poll = _record_poll(conn, watch_id, watch["project_id"], + "unchanged", observed, "", None, now) + conn.commit() + return poll + # 真实变化：配额门只在这里过。配额门契约是前缀 + # （quota_check_failed: / quota_exceeded: / None），不用 + # 子串判定，以免超限文本含该子串时误判。 + gate = _automation_quota_exceeded(conn, watch["created_by"]) + if gate is not None and not gate.startswith("quota_check_failed"): + cur = conn.execute( + "UPDATE resident_watches SET status = ?," + " paused_reason = ?, next_poll_at = NULL, updated_at = ?" + " WHERE id = ? AND status = ?", + (PAUSED, "quota_exceeded", now, watch_id, RUNNING)) + if cur.rowcount != 1: + conn.rollback() + return None + poll = _record_poll(conn, watch_id, watch["project_id"], + "changed_quota_paused", observed, gate, + None, now) + conn.commit() + _auto_audit_row(conn, watch["project_id"], + "resident_watch_paused_quota", + "%s: %s" % (watch_id, gate[:200]), now) + return poll + if gate is not None: + failures = watch["consecutive_failures"] + 1 + nxt = now + interval + _backoff_seconds(failures) + if not _guarded_watch_update( + conn, watch_id, + ("consecutive_failures = ?", "next_poll_at = ?", + "updated_at = ?"), + (failures, nxt, now)): + conn.rollback() + return None + poll = _record_poll(conn, watch_id, watch["project_id"], + "changed_quota_check_failed", observed, gate, + None, now) + conn.commit() + return poll + # 先受保护地推进游标，再建任务：抓取中途被暂停/结束时， + # 条件不满足即回滚，本轮不建任务、不记轮询。 + if not _guarded_watch_update( + conn, watch_id, + ("last_content_hash = ?", "last_change_at = ?", + "last_change_text = ?", "consecutive_failures = 0", + "next_poll_at = ?", "updated_at = ?"), + (observed, now, excerpt, now + interval, now)): + conn.rollback() + return None + task_id = _create_change_task(conn, watch, excerpt, now) + poll = _record_poll(conn, watch_id, watch["project_id"], + "changed", observed, excerpt[:MAX_POLL_DETAIL_LEN], + task_id, now) + conn.commit() + _auto_audit_row(conn, watch["project_id"], + "resident_watch_change_task_created", + "%s -> %s" % (watch_id, task_id), now) + return poll + + +def process_due_resident_watches(conn, fetcher=None, now=None, limit=50): + """驱动批量入口：列出到期监听，逐条走 run_due（每条自带 + claim 门，批量内重复与并发驱动都不会重复轮询）。返回 + 本轮实际产出的轮询记录列表（抢不到的条目不在内）。""" + if now is None: + now = time.time() + due = due_watches(conn, now)[:limit] + out = [] + for watch in due: + poll = run_due_resident_watch(conn, watch["id"], + fetcher=fetcher, now=now) + if poll is not None: + out.append(poll) + return out diff --git a/cloud/server.py b/cloud/server.py index 3d15da0..b0a00e4 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -110,6 +110,14 @@ from cloud.wait_ops import worktree_status as _wait_worktree_status from cloud.shared_home_mcp import sync_shared_home_mcp as _sync_shared_home_mcp 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 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 resume_watch as _res_resume +from cloud.resident import update_watch as _res_update from cloud.todos import create_todo as _todo_create from cloud.todos import delete_todo as _todo_delete from cloud.todos import get_todo as _todo_get @@ -3371,6 +3379,54 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): return self._triage_inbox(project_id) return + # /projects/{id}/resident-watches：常驻监听任务。 + # 创建/更新/暂停/恢复/结束用 create_task，读用 read； + # 状态机见 cloud/resident.py（running/paused/ended， + # ended 是终态，恢复已结束的监听必须新建）。 + if len(parts) == 2 and parts[1] == "resident-watches": + if method == "POST": + err = self._check_project(user_id, project_id, "create_task") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._resident_create(user_id, project_id) + elif method == "GET": + err = self._check_project(user_id, project_id, "read") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._resident_list(project_id) + else: + self._send_json(404, {"error": "not found"}) + return + if len(parts) == 3 and parts[1] == "resident-watches": + if method == "GET": + err = self._check_project(user_id, project_id, "read") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._resident_get(user_id, project_id, parts[2]) + elif method == "PATCH": + err = self._check_project(user_id, project_id, "create_task") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._resident_update(user_id, project_id, parts[2]) + else: + self._send_json(404, {"error": "not found"}) + return + if len(parts) == 4 and parts[1] == "resident-watches" \ + and parts[3] in ("pause", "resume", "end"): + if method != "POST": + self._send_json(404, {"error": "not found"}) + return + err = self._check_project(user_id, project_id, "create_task") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._resident_transition(user_id, project_id, parts[2], + parts[3]) + return # /projects/{id}/todos GET 列表 / POST 新建；单条 # GET / PATCH / DELETE（项目 read / create_task） if len(parts) == 2 and parts[1] == "todos": @@ -10067,6 +10123,135 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): automation_id) self._send_json(200, auto) + # -- 常驻监听任务（见 cloud/resident.py）------------------- + # 权限口径与自动化同款：创建/更新/暂停/恢复/结束用 + # create_task，读用 read；状态变更记审计。创建不走配额门 + # （轮询不花模型额度），配额只在变化建任务时由驱动过， + # 耗尽时转暂停而非拒绝创建。 + def _resident_create(self, user_id, project_id): + body, err = self._read_json() + if err: + self._send_json(400, {"error": err}) + return + interval = body.get("interval_seconds", 3600) + if interval is None: + interval = 3600 + conn = self._db() + try: + try: + watch = _res_create( + conn, project_id, body.get("name"), + body.get("target_url"), body.get("prompt"), user_id, + interval, body.get("fetch_timeout_seconds")) + except ValueError as exc: + self._send_json(400, {"error": str(exc)}) + return + except Exception as exc: + self._send_json(502, {"error": str(exc)[:500]}) + return + finally: + conn.close() + with _auth_store(self.server.db_path) as store: + store.audit(user_id, project_id, "resident_watch_created", + watch["id"]) + self._send_json(201, watch) + + def _resident_list(self, project_id): + conn = self._db() + try: + try: + watches = _res_list(conn, project_id) + except Exception as exc: + self._send_json(502, {"error": str(exc)[:500]}) + return + conn.commit() + finally: + conn.close() + self._send_json(200, { + "project_id": project_id, "resident_watches": watches, + "due_count": sum(1 for w in watches if w["due"])}) + + def _resident_get(self, user_id, project_id, watch_id): + conn = self._db() + try: + watch = _res_get(conn, watch_id) + polls = [] + polls_note = "" + if watch is not None: + try: + polls = _res_polls(conn, watch_id) + except Exception: + polls_note = "polls unavailable" + finally: + conn.close() + if watch is None or watch["project_id"] != project_id: + self._send_json(404, {"error": "resident watch not found"}) + return + self._send_json(200, dict(watch, polls=polls, + polls_note=polls_note)) + + def _resident_update(self, user_id, project_id, watch_id): + body, jerr = self._read_json() + if jerr: + self._send_json(400, {"error": jerr}) + return + conn = self._db() + try: + try: + ok, watch = _res_update(conn, watch_id, body) + except ValueError as exc: + self._send_json(400, {"error": str(exc)}) + return + except Exception as exc: + self._send_json(502, {"error": str(exc)[:500]}) + return + finally: + conn.close() + if watch is None or watch["project_id"] != project_id: + self._send_json(404, {"error": "resident watch not found"}) + return + if not ok: + self._send_json(409, { + "error": "ended resident watch cannot be updated;" + " create a new watch instead"}) + return + with _auth_store(self.server.db_path) as store: + store.audit(user_id, project_id, "resident_watch_updated", + watch_id) + self._send_json(200, watch) + + def _resident_transition(self, user_id, project_id, watch_id, action): + fn = {"pause": _res_pause, "resume": _res_resume, + "end": _res_end}[action] + conn = self._db() + try: + try: + ok, watch = fn(conn, watch_id) + except Exception as exc: + self._send_json(502, {"error": str(exc)[:500]}) + return + finally: + conn.close() + if watch is None or watch["project_id"] != project_id: + self._send_json(404, {"error": "resident watch not found"}) + return + if not ok: + if watch["status"] == "ended": + self._send_json(409, { + "error": "ended resident watch is terminal;" + " create a new watch instead"}) + else: + self._send_json(409, { + "error": "resident watch in status %s cannot %s" + % (watch["status"], action)}) + return + audit_action = {"pause": "resident_watch_paused", + "resume": "resident_watch_resumed", + "end": "resident_watch_ended"}[action] + with _auth_store(self.server.db_path) as store: + store.audit(user_id, project_id, audit_action, watch_id) + self._send_json(200, watch) + def _triage_inbox(self, project_id): """分诊收件箱(只读)：到期自动化、失败/中断任务、 待处理后续消息、最近自动化产出。见 automations 模块 diff --git a/cloud/tests/test_resident.py b/cloud/tests/test_resident.py new file mode 100644 index 0000000..5966f37 --- /dev/null +++ b/cloud/tests/test_resident.py @@ -0,0 +1,493 @@ +"""常驻监听任务单测（cloud/resident.py）：不跑真实模型、不联网， +抓取一律注入假 fetcher。覆盖创建校验、暂停保游标、恢复续跑、 +结束终态、防重复轮询、仅变化建任务、失败退避、配额耗尽转暂停、 +worker 驱动批量入口。""" +import http.server +import os +import socket +import sqlite3 +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + import cloud.resident as resident_module + from cloud.resident import ( + claim_due_watch, + create_watch, + default_fetch, + end_watch, + ensure_resident_tables, + get_watch, + list_polls, + pause_watch, + process_due_resident_watches, + resume_watch, + run_due_resident_watch, + update_watch, + validate_target_url, + ) + from cloud.server import 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.resident import ( + claim_due_watch, + create_watch, + default_fetch, + end_watch, + ensure_resident_tables, + get_watch, + list_polls, + pause_watch, + process_due_resident_watches, + resume_watch, + run_due_resident_watch, + update_watch, + validate_target_url, + ) + from cloud.server import init_db + from cloud.worker import Worker + +URL = "https://example.com/data" + + +class _Fetcher: + """可变假抓取：value 改了，下一次轮询就看到新内容。""" + + def __init__(self, value="v1"): + self.value = value + self.calls = [] + + def __call__(self, url, timeout): + self.calls.append((url, timeout)) + return self.value + + +def _failing_fetcher(url, timeout): + raise RuntimeError("boom") + + +class ResidentTestCase(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", **kw): + kw.setdefault("interval_seconds", 60) + return create_watch(self.conn, "proj", name, URL, + "变化时汇报", "owner1", now=self.now, **kw) + + def _tasks(self): + cur = self.conn.execute("SELECT id, prompt, status FROM tasks") + return cur.fetchall() + + def _set_quota(self, user_id, micros): + self.conn.execute( + "CREATE TABLE IF NOT EXISTS user_quotas(" + "user_id TEXT PRIMARY KEY, quota_usd_micros BIGINT, " + " updated_at DOUBLE PRECISION)") + self.conn.execute( + "INSERT OR REPLACE INTO user_quotas(user_id," + " quota_usd_micros, updated_at) VALUES(?, ?, ?)", + (user_id, micros, self.now)) + self.conn.commit() + + def test_create_validation(self): + watch = self._watch() + self.assertEqual(watch["status"], "running") + self.assertTrue(watch["due"]) + self.assertEqual(watch["next_poll_at"], self.now) + for kwargs, _ in ( + ({"target_url": "ftp://example.com/x"}, None), + ({"target_url": "https://localhost/x"}, None), + ({"target_url": "http://127.0.0.1/x"}, None), + ({"target_url": "https://10.0.0.5/x"}, None), + ({"target_url": "https://user:pw@example.com/x"}, None), + ({"interval_seconds": 30}, None), + ({"name": "watch1"}, None), + ({"prompt": " "}, None)): + with self.assertRaises(ValueError): + base = {"name": "other", "target_url": URL, + "prompt": "p", "interval_seconds": 60} + base.update(kwargs) + create_watch(self.conn, "proj", base["name"], + base["target_url"], base["prompt"], "owner1", + interval_seconds=base["interval_seconds"], + now=self.now) + + def test_baseline_unchanged_changed_only_change_creates_task(self): + fetcher = _Fetcher("v1") + watch = self._watch() + poll = run_due_resident_watch(self.conn, watch["id"], + fetcher=fetcher, now=self.now) + self.assertEqual(poll["status"], "baseline") + self.assertEqual(len(self._tasks()), 0) + # 同一间隔内第二次：抢不到，不重复轮询。 + self.assertIsNone(run_due_resident_watch( + self.conn, watch["id"], fetcher=fetcher, now=self.now)) + self.assertEqual(len(fetcher.calls), 1) + self.now += 61 + poll = run_due_resident_watch(self.conn, watch["id"], + fetcher=fetcher, now=self.now) + self.assertEqual(poll["status"], "unchanged") + self.assertEqual(len(self._tasks()), 0) + 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.assertIsNotNone(poll["task_id"]) + tasks = self._tasks() + self.assertEqual(len(tasks), 1) + self.assertIn("v2", tasks[0][1]) + self.assertIn("变化时汇报", tasks[0][1]) + after = get_watch(self.conn, watch["id"], now=self.now) + self.assertEqual(after["last_change_text"], "v2") + self.assertIsNotNone(after["last_change_at"]) + self.assertEqual( + [p["status"] for p in list_polls(self.conn, watch["id"])], + ["changed", "unchanged", "baseline"]) + + def test_pause_preserves_cursor_and_resume_reports_pause_period_change(self): + fetcher = _Fetcher("v1") + watch = self._watch() + run_due_resident_watch(self.conn, watch["id"], + fetcher=fetcher, now=self.now) + cursor = get_watch(self.conn, watch["id"], now=self.now) + self.assertIsNotNone(cursor["last_content_hash"]) + ok, paused = pause_watch(self.conn, watch["id"], now=self.now) + self.assertTrue(ok) + self.assertEqual(paused["status"], "paused") + self.assertEqual(paused["last_content_hash"], + cursor["last_content_hash"]) + self.assertIsNone(paused["next_poll_at"]) + self.assertFalse(paused["due"]) + # 暂停期间即使时间过了间隔，驱动也不轮询。 + self.assertIsNone(run_due_resident_watch( + self.conn, watch["id"], fetcher=fetcher, now=self.now + 3600)) + self.assertEqual(len(fetcher.calls), 1) + # 暂停期间网站变了；恢复后第一次轮询用旧游标比对，报出来。 + fetcher.value = "v2" + ok, resumed = resume_watch(self.conn, watch["id"], now=self.now) + self.assertTrue(ok) + self.assertEqual(resumed["status"], "running") + self.assertEqual(resumed["last_content_hash"], + cursor["last_content_hash"]) + 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) + + def test_pause_resume_on_wrong_status_are_rejected(self): + watch = self._watch() + ok, _ = resume_watch(self.conn, watch["id"], now=self.now) + self.assertFalse(ok) + ok, _ = pause_watch(self.conn, watch["id"], now=self.now) + self.assertTrue(ok) + ok, _ = pause_watch(self.conn, watch["id"], now=self.now) + self.assertFalse(ok) + + def test_end_is_terminal_and_name_reusable(self): + watch = self._watch() + ok, ended = end_watch(self.conn, watch["id"], now=self.now) + self.assertTrue(ok) + self.assertEqual(ended["status"], "ended") + self.assertIsNone(ended["next_poll_at"]) + for fn in (pause_watch, resume_watch, end_watch): + ok, old = fn(self.conn, watch["id"], now=self.now) + self.assertFalse(ok) + self.assertEqual(old["status"], "ended") + self.assertIsNone(run_due_resident_watch( + self.conn, watch["id"], fetcher=_Fetcher(), now=self.now + 3600)) + ok, _ = update_watch(self.conn, watch["id"], + {"prompt": "新提示"}, now=self.now) + self.assertFalse(ok) + # 已结束的名字可重用。 + again = self._watch() + self.assertEqual(again["status"], "running") + + def test_claim_is_single_winner(self): + watch = self._watch() + self.assertTrue(claim_due_watch(self.conn, watch["id"], now=self.now)) + self.assertFalse(claim_due_watch(self.conn, watch["id"], + now=self.now)) + # 下个间隔到了才能再抢（claim 已把 next_poll_at 预置到下间隔）。 + self.assertTrue(claim_due_watch(self.conn, watch["id"], + now=self.now + 61)) + + def test_fetch_failure_backoff_and_cursor_unchanged(self): + watch = self._watch() + run_due_resident_watch(self.conn, watch["id"], + fetcher=_Fetcher("v1"), now=self.now) + cursor = get_watch(self.conn, watch["id"], now=self.now) + self.now += 61 + poll = run_due_resident_watch(self.conn, watch["id"], + fetcher=_failing_fetcher, now=self.now) + self.assertEqual(poll["status"], "failed") + after = get_watch(self.conn, watch["id"], now=self.now) + self.assertEqual(after["consecutive_failures"], 1) + self.assertEqual(after["last_content_hash"], + cursor["last_content_hash"]) + # 失败退避：下次 = now + 间隔 + 30 秒，不是立刻重试。 + self.assertEqual(after["next_poll_at"], self.now + 60 + 30) + self.now = after["next_poll_at"] + run_due_resident_watch(self.conn, watch["id"], + fetcher=_failing_fetcher, now=self.now) + after2 = get_watch(self.conn, watch["id"], now=self.now) + self.assertEqual(after2["consecutive_failures"], 2) + self.assertEqual(after2["next_poll_at"], self.now + 60 + 60) + + def test_quota_exceeded_pauses_not_ends_and_keeps_cursor(self): + fetcher = _Fetcher("v1") + watch = self._watch() + run_due_resident_watch(self.conn, watch["id"], + fetcher=fetcher, now=self.now) + cursor_hash = get_watch(self.conn, watch["id"], + now=self.now)["last_content_hash"] + self._set_quota("owner1", 0) + 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_quota_paused") + after = get_watch(self.conn, watch["id"], now=self.now) + self.assertEqual(after["status"], "paused") + self.assertEqual(after["paused_reason"], "quota_exceeded") + self.assertNotEqual(after["status"], "ended") + self.assertEqual(after["last_content_hash"], cursor_hash) + self.assertEqual(len(self._tasks()), 0) + # 恢复后配额仍耗尽：再次转暂停，不擅自结束、不丢变化。 + resume_watch(self.conn, watch["id"], now=self.now) + poll = run_due_resident_watch(self.conn, watch["id"], + fetcher=fetcher, now=self.now) + self.assertEqual(poll["status"], "changed_quota_paused") + # 配额恢复后，同一变化仍能报出来（游标没被配额暂停推进）。 + self._set_quota("owner1", 100000000) + resume_watch(self.conn, watch["id"], now=self.now) + 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) + + def test_update_watch(self): + watch = self._watch() + ok, updated = update_watch( + self.conn, watch["id"], + {"interval_seconds": 3600, "prompt": "新处理提示"}, + now=self.now) + self.assertTrue(ok) + self.assertEqual(updated["interval_seconds"], 3600) + self.assertEqual(updated["prompt"], "新处理提示") + with self.assertRaises(ValueError): + update_watch(self.conn, watch["id"], {}, now=self.now) + with self.assertRaises(ValueError): + update_watch(self.conn, watch["id"], + {"target_url": "https://localhost/x"}, now=self.now) + + def test_process_due_and_worker_driver(self): + fetcher = _Fetcher("v1") + self._watch(name="a") + self._watch(name="b") + polls = process_due_resident_watches(self.conn, fetcher=fetcher, + now=self.now) + self.assertEqual(len(polls), 2) + self.assertEqual([p["status"] for p in polls], + ["baseline", "baseline"]) + self.assertEqual(process_due_resident_watches( + self.conn, fetcher=fetcher, now=self.now), []) + # worker 驱动入口：注入假抓取，变化时产出 changed 轮询。 + worker = Worker(self.db_path, "/fake/host.py", + env={"XCM_API_KEY": "test-placeholder"}, + runner=lambda cmd: SimpleNamespace( + returncode=0, stdout="{}", stderr="")) + fetcher.value = "v2" + self.now += 61 + polls = worker.process_due_resident_watches( + now=self.now, fetcher=fetcher) + self.assertEqual(len(polls), 2) + self.assertTrue(all(p["status"] == "changed" for p in polls)) + self.assertEqual(len(self._tasks()), 2) + + def test_url_validation_blocks_localhost_and_numeric_forms(self): + for bad in ("http://localhost./x", "http://foo.localhost/x", + "http://2130706433/x", "http://0x7f000001/x", + "http://0177.0.0.1/x", "http://0x7f.0.0.1/x"): + with self.assertRaises(ValueError, msg=bad): + validate_target_url(bad) + # 规范点分十进制公网 IP 与带尾点的普通域名仍放行。 + self.assertEqual(validate_target_url("https://8.8.8.8/x"), + "https://8.8.8.8/x") + self.assertEqual(validate_target_url("https://example.com./x"), + "https://example.com./x") + + def test_unique_index_guards_concurrent_name_clash(self): + idx = self.conn.execute( + "SELECT name FROM sqlite_master WHERE type = 'index'" + " AND name = 'idx_resident_watches_name'").fetchone() + self.assertIsNotNone(idx) + self._watch() + # 正常路径仍先报 ValueError。 + with self.assertRaises(ValueError): + self._watch() + # 绕过 SELECT 查重直接插：唯一索引本身必须拦住， + # 这才是并发下真正的关口。 + with self.assertRaises(sqlite3.IntegrityError): + self.conn.execute( + "INSERT INTO resident_watches(id, project_id, name," + " target_url, prompt, interval_seconds, status," + " created_by, created_at, updated_at)" + " VALUES('dup', 'proj', 'watch1', ?, 'p', 60," + " 'running', 'owner1', ?, ?)", + (URL, self.now, self.now)) + other = self._watch(name="watch2") + with self.assertRaises(ValueError): + update_watch(self.conn, other["id"], {"name": "watch1"}, + now=self.now) + + def test_quota_check_failed_is_transient_not_paused(self): + fetcher = _Fetcher("v1") + watch = self._watch() + run_due_resident_watch(self.conn, watch["id"], + fetcher=fetcher, now=self.now) + cursor_hash = get_watch(self.conn, watch["id"], + now=self.now)["last_content_hash"] + original = resident_module._automation_quota_exceeded + resident_module._automation_quota_exceeded = lambda conn, owner: ( + "quota_check_failed: boom") + try: + fetcher.value = "v2" + self.now += 61 + poll = run_due_resident_watch( + self.conn, watch["id"], fetcher=fetcher, now=self.now) + finally: + resident_module._automation_quota_exceeded = original + self.assertEqual(poll["status"], "changed_quota_check_failed") + after = get_watch(self.conn, watch["id"], now=self.now) + self.assertEqual(after["status"], "running") + self.assertEqual(after["last_content_hash"], cursor_hash) + self.assertEqual(after["consecutive_failures"], 1) + self.assertEqual(after["next_poll_at"], self.now + 60 + 30) + self.assertEqual(len(self._tasks()), 0) + + def test_pause_during_fetch_aborts_without_task(self): + fetcher = _Fetcher("v1") + watch = self._watch() + run_due_resident_watch(self.conn, watch["id"], + fetcher=fetcher, now=self.now) + cursor_hash = get_watch(self.conn, watch["id"], + now=self.now)["last_content_hash"] + db_path = self.db_path + + def pausing_fetcher(url, timeout): + conn2 = sqlite3.connect(db_path) + try: + conn2.row_factory = sqlite3.Row + pause_watch(conn2, watch["id"], now=self.now + 61) + finally: + conn2.close() + return "v2" + + self.now += 61 + poll = run_due_resident_watch( + self.conn, watch["id"], fetcher=pausing_fetcher, now=self.now) + self.assertIsNone(poll) + after = get_watch(self.conn, watch["id"], now=self.now) + self.assertEqual(after["status"], "paused") + self.assertIsNone(after["next_poll_at"]) + self.assertEqual(after["last_content_hash"], cursor_hash) + self.assertEqual(len(self._tasks()), 0) + self.assertEqual( + [p["status"] for p in list_polls(self.conn, watch["id"])], + ["baseline"]) + + +class RedirectFetchTestCase(unittest.TestCase): + """重定向校验：只把 redirect-test.test 指到本地，其它主机 + 走原始解析；no_proxy 同款处理，tearDown 全部还原。""" + + HOST = "redirect-test.test" + + def setUp(self): + class Handler(http.server.BaseHTTPRequestHandler): + def do_GET(self): + if self.path == "/go": + self.send_response(302) + self.send_header("Location", "/final") + elif self.path == "/bad": + self.send_response(302) + self.send_header( + "Location", "http://127.0.0.1:1/secret") + else: + body = b"final-body" + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + return + self.end_headers() + + def log_message(self, *args): + pass + + self.server = http.server.ThreadingHTTPServer( + ("127.0.0.1", 0), Handler) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + self._orig_getaddrinfo = socket.getaddrinfo + + def patched(host, port, family=0, type=0, proto=0, flags=0): + if host == self.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 + 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 "") + self.HOST + os.environ["NO_PROXY"] = ( + (self._old_NO_PROXY + ",") if self._old_NO_PROXY + else "") + self.HOST + + 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 + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + + 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") + with self.assertRaises(ValueError): + default_fetch(base + "/bad", 10) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_resident_server.py b/cloud/tests/test_resident_server.py new file mode 100644 index 0000000..9a8cd64 --- /dev/null +++ b/cloud/tests/test_resident_server.py @@ -0,0 +1,180 @@ +"""常驻监听服务端单测：POST/GET /projects/{id}/resident-watches、 +PATCH、POST pause / resume / end。不跑真实模型、不联网。 +骨架与 test_automations_server.py 同款（本地临时端口 + 临时 +SQLite）。""" +import http.client +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +try: + from cloud.auth import AuthStore + from cloud.server import create_server +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.server import create_server + +URL = "https://example.com/data" + + +class ResidentServerTestCase(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.outsider_id, self.outsider_key = self.auth.create_user( + "outsider") + 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) + + 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 _create(self, **extra): + body = {"name": "watch1", "target_url": URL, + "prompt": "变化时汇报"} + body.update(extra) + return self._req("POST", self._url(), body, self.owner_key) + + def _audit_actions(self): + conn = sqlite3.connect(self.db_path) + try: + rows = conn.execute( + "SELECT action FROM audit_log WHERE project_id = ?", + (self.project["id"],)).fetchall() + finally: + conn.close() + return [r[0] for r in rows] + + def test_create_list_get_and_validation(self): + status, watch = self._create(interval_seconds=3600) + self.assertEqual(status, 201, watch) + self.assertEqual(watch["status"], "running") + self.assertEqual(watch["target_url"], URL) + self.assertTrue(watch["due"]) + self.assertIsNotNone(watch["next_poll_at"]) + self.assertIn("resident_watch_created", self._audit_actions()) + status, body = self._req("GET", self._url(), None, self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(len(body["resident_watches"]), 1) + self.assertEqual(body["due_count"], 1) + status, body = self._req( + "GET", self._url("/" + watch["id"]), None, self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(body["id"], watch["id"]) + self.assertEqual(body["polls"], []) + for bad in ( + {"name": "x", "target_url": "ftp://example.com/x", + "prompt": "p"}, + {"name": "x", "target_url": "https://localhost/x", + "prompt": "p"}, + {"name": "x", "target_url": URL, "prompt": "p", + "interval_seconds": 30}, + {"name": "watch1", "target_url": URL, "prompt": "p"}, + {"target_url": URL, "prompt": "p"}, + {"name": "x", "prompt": "p"}): + status, _ = self._req("POST", self._url(), bad, self.owner_key) + self.assertEqual(status, 400, bad) + status, _ = self._req( + "GET", self._url("/nope"), None, self.owner_key) + self.assertEqual(status, 404) + + def test_pause_resume_end_lifecycle_and_terminal_409(self): + status, watch = self._create() + self.assertEqual(status, 201) + wid = watch["id"] + status, body = self._req( + "POST", self._url("/%s/pause" % wid), {}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["status"], "paused") + self.assertEqual(body["paused_reason"], "user_paused") + self.assertIsNone(body["next_poll_at"]) + self.assertFalse(body["due"]) + status, body = self._req( + "POST", self._url("/%s/pause" % wid), {}, self.owner_key) + self.assertEqual(status, 409) + status, body = self._req( + "POST", self._url("/%s/resume" % wid), {}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["status"], "running") + self.assertIsNone(body["paused_reason"]) + self.assertTrue(body["due"]) + status, body = self._req( + "POST", self._url("/%s/end" % wid), {}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["status"], "ended") + 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) + self.assertIn("terminal", body["error"]) + actions = self._audit_actions() + self.assertIn("resident_watch_paused", actions) + self.assertIn("resident_watch_resumed", actions) + self.assertIn("resident_watch_ended", actions) + + def test_update_and_outsider_forbidden(self): + status, watch = self._create() + self.assertEqual(status, 201) + wid = watch["id"] + status, body = self._req( + "PATCH", self._url("/" + wid), + {"interval_seconds": 7200}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["interval_seconds"], 7200) + status, _ = self._req( + "POST", self._url(), + {"name": "x", "target_url": URL, "prompt": "p"}, + self.outsider_key) + self.assertIn(status, (401, 403, 404)) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_resident_website_loop.py b/cloud/tests/test_resident_website_loop.py new file mode 100644 index 0000000..3c6e999 --- /dev/null +++ b/cloud/tests/test_resident_website_loop.py @@ -0,0 +1,381 @@ +"""常驻监听网站闭环端到端测试。 + +上一轮只验到注入假抓取的单测：真实抓取、真实 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): + # setUp 中途失败时属性可能未建：getattr 防御，避免 + # tearDown 的 AttributeError 掩盖原始失败。 + if getattr(self, "_orig_getaddrinfo", None) is not None: + socket.getaddrinfo = self._orig_getaddrinfo + if getattr(self, "_old_no_proxy", None) is None: + if "_old_no_proxy" in self.__dict__: + os.environ.pop("no_proxy", None) + else: + os.environ["no_proxy"] = self._old_no_proxy + if getattr(self, "_old_NO_PROXY", None) is None: + if "_old_NO_PROXY" in self.__dict__: + 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 + try: + self.content_thread.join(timeout=5) + except Exception: + pass + try: + self.server.shutdown() + self.server.server_close() + except Exception: + pass + try: + self.thread.join(timeout=5) + except Exception: + pass + try: + self.auth.conn.close() + except Exception: + pass + try: + self.tmp.cleanup() + except Exception: + pass + + 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() diff --git a/cloud/worker.py b/cloud/worker.py index 564cc56..ffd2286 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -6582,6 +6582,33 @@ 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): + """常驻监听驱动（worker 轮次边界调用）：到期监听逐条 + 走 resident.process_due_resident_watches，抢单 + 廉价 + 抓取 + 仅变化建任务。全部异常只记日志，不炸循环； + fetcher 缺省用 resident 默认抓取，单测可经 + self.resident_fetcher 或参数注入假抓取。""" + if now is None: + now = time.time() + if fetcher is None: + fetcher = getattr(self, "resident_fetcher", None) + try: + from cloud.resident import process_due_resident_watches as _run + conn = _db_connect(self.db_path) + try: + polls = _run(conn, fetcher=fetcher, now=now) + finally: + conn.close() + except Exception as exc: + log_event(self.logger, "worker_resident_watch_failed", + message="resident watch processing failed: %s" % exc) + return [] + for poll in polls: + log_event(self.logger, "worker_resident_watch_poll", + message="resident watch poll: %s %s" + % (poll["watch_id"], poll["status"])) + return polls + def run_loop(self, interval=300, max_seconds=None, stop_event=None, drain_timeout=60, sweep_threshold=300): """常驻循环：每 interval 秒跑一轮 run_all，空闲也不退出。 @@ -6664,6 +6691,10 @@ class Worker: # G11/G12 后台投递：摘要（每小时）与 outbox 重试， # 轮次边界做，不中断在跑任务，失败只记日志。 self._process_push_due() + # 常驻监听驱动：同款轮次边界后台扫，到期监听 + # 廉价抓取、仅变化建任务；服务端本身不起定时器， + # 这里就是常驻任务的实际驱动（见 resident 模块）。 + self.process_due_resident_watches() # 自动记忆提取：同款轮次边界后台扫，失败只记日志 # （process_auto_memories 内部已兜底，不炸循环）。 self.process_auto_memories() diff --git a/verification/resident-audit-summary.md b/verification/resident-audit-summary.md new file mode 100644 index 0000000..870cca4 --- /dev/null +++ b/verification/resident-audit-summary.md @@ -0,0 +1,201 @@ +# 常驻监听网站闭环：独立架构与代码质量审核报告 + +审核员非实现者。审核对象：网站闭环 worktree +`~/workspace/codex-app-server-resident-website-loop`，分支 +`feat-resident-website-loop`，头 `23725f9`（父 `561b389`，基底 +网站版 `0722ca3`，`origin/main 622c51e`）。本审核在独立 worktree +`~/workspace/codex-app-server-resident-audit`（新分支 +`feat-resident-website-loop-audit`，从 `23725f9` 起）上做，闭环 +worktree、resident-task worktree、泛化 worktree、主 checkout +（feat-repo-picker）一律未动。 + +## 0. 评分标准（先定标准，再打分） + +只评架构与代码质量两项，满分各 10 分。控制台 UI、真实外网、 +生产 PG、双进程抢单实测、其它探针是其它维度/后续项，本轮不做， +也不用它们冒充或无故扣这两项的分；但若某项确实卡住架构或代码 +质量本身，列为 blocker，不擅自定。 + +### 架构 10 分标准 + +- A1 分层干净：resident 模块自有状态机，与任务状态机、自动化 + 定义层互不干扰；server 只做路由/权限/审计，worker 只做驱动 + 边界调用；无循环依赖。 +- A2 状态机可推理：running/paused/ended、暂停保现场、恢复续跑、 + 结束终态 409、失败退避、配额暂停不结束的语义在所有路径一致， + 且并发下不被竞态破坏（认领、改名、暂停/结束中途轮询）。 +- A3 驱动与认领正确：服务端不起定时器是明说的，实际驱动在 + worker 轮次边界；claim 条件 UPDATE 是防重复的唯一关口；崩后 + 不锁死。 +- A4 快照/变化语义干净：轮询廉价、仅变化建任务、首次只建基线、 + 每次轮询可查；变化摘录与游标推进在同一事务边界内。 +- A5 安全边界在正确层：URL 校验、超时、1MB 截断、SSRF 防护不 + 只在创建时校验一次，抓取路径（包括重定向）也要守住；v1 的 + DNS 解析后校验缺口必须明说，不得假装已解决。 +- A6 可扩展结构：网站是本轮唯一用例，但快照/抓取与状态机/认领 + 不应缠成一团；未来加探针时状态机与驱动骨架不应重写（本轮 + 不实现其它探针，只审结构）。 + +### 代码质量 10 分标准 + +- Q1 命名、函数长度与复杂度可读；长函数（run_due）分支清晰， + 不靠注释掩盖逻辑。 +- Q2 无重复、无死代码、无测试值渗入生产代码；复用已有口径 + （自动化 claim/配额/审计、环境名唯一索引先例）而不是另起一套。 +- Q3 错误处理诚实：失败、配额耗尽、配额检查失败三条路不混淆； + 审计与轮询记录不静默丢失；异常不炸驱动循环。 +- Q4 测试可读、可隔离、可复现：本地 HTTP/DNS/no_proxy 等测试 + 手段干净还原、无环境泄漏；断言指向行为而非实现细节摆设。 +- Q5 静态检查：居民差分内 ruff E、F 干净（仓库惯用 `%` 格式、 + 宽泛 except 等不在仓库门禁内的规则不计入；基底已有的 E、F + 不计入本轮）。 + +## 1. 改动前分数 + +- 架构：9.0 / 10 +- 代码质量：9.0 / 10 + +基底是好的：分层、状态机、claim、配额复用、轮询可查、测试 +隔离（DNS/no_proxy 在 loop 测试里干净还原）都属实，定向 15 项 +基线全过（见第 4 节）。扣分只来自下面逐项 findings，不是整体 +推倒重来。 + +## 2. Findings 清单（改动前） + +### 严重 + +- F1 名称唯一性是先查后插、无数据库约束：create_watch / + update_watch 先 SELECT 查重再 INSERT/UPDATE，两个并发请求 + 可同时通过检查、插出同项目同名 running 监听；server 的 + environments 已有先例：`CREATE UNIQUE INDEX … WHERE status + != 'archived'`。本轮用同款部分唯一索引 + `WHERE status != 'ended'`（已结束名可重用）+ 完整性错误转 + ValueError 修复。 +- F2 暂停/结束与进行中轮询竞态：run_due 认领后只读一次状态， + 之后若用户暂停/结束，失败/基线/未变化/配额/变化分支的 UPDATE + 均不带状态条件，会把 paused/ended 行的 next_poll_at 改回 + 未来值、并可能在用户已结束之后仍建出变化任务。修复：认领后 + 复查状态；全部终局 UPDATE 带 `status='running'` 条件，条件 + 不满足时回滚、不建任务、不记轮询。 + +### 一般 + +- F3 抓取路径 SSRF 只守了创建时：default_fetch 用默认 opener， + 重定向目标不校验，公开 URL 重定向到内网地址可绕过创建校验； + 另 localhost.、单整数/十六进制/前导零等非规范 IP 写法可绕过 + 字面 IP 检查。修复：重定向处理器逐跳校验（相对跳转先 urljoin + 再校验）；校验时归一化主机（去尾点、*.localhost 挡掉）并拒绝 + 非规范数字 IP 写法。DNS 解析后校验仍属 v1 缺口，继续列为已知 + 偏差，不假装已解决。 +- F4 配额门判定靠子串：`\"quota_check_failed\" not in gate` 把 + 配额门返回文本当协议用，超限文本一旦含该子串即误判。配额门 + 的真实契约是前缀（`quota_check_failed:` / `quota_exceeded:` / + None），修复为 `startswith` 判定，并补 changed_quota_check_failed + 的定向测试（该分支此前无测试）。 +- F5 row_get 重复造轮：resident 自带与 automations.row_get 逐字 + 同语义的 row_get，却已从 automations 导入另外两个函数。修复： + 直接导入 automations.row_get，删本地定义。 +- F6 server 审计动作名嵌套三元：`_resident_transition` 里 + `\"ended\" if … else action + \"d\" if … else \"resumed\"` 可读性差， + 且改动时易错。修复为显式动作映射字典，审计字符串不变。 + +### 建议 + +- S1 test_resident.py 导入了未用的 `time`（ruff F401，居民差分 + 内唯一 E、F 问题）。删除。 +- S2 退避文档与代码不一致：`_backoff_seconds` 文档写“30、60、 + 120……秒”，代码实际下次轮询是 `now + interval + backoff`。 + 行为本身合理（不比间隔更频繁），文档误导。修复文档，行为不动。 +- S3 loop 测试 tearDown 不够防御：setUp 中途失败时 tearDown + 访问未建属性会掩盖原始失败。修复为 getattr 防御。 + +### 已核查、不是问题（不扣分） + +- 审计行“写在 commit 之后会丢”：核查 `_auto_audit_row` 本身 + 执行 INSERT 后自己 commit，失败/配额暂停/变化三处审计行在 + 模拟关闭连接后仍可查，不是缺陷。 +- server/worker 全量 ruff E、F 的其余项（如 worker.py:2781/2787 + 的 F821、server.py:10293 的 E501）经核对均为基底已有或由居民 + 插入导致的行号位移，非居民新增代码，不计入本轮。 +- 仓库默认 ruff 全规则的 39 项（如 UP031 `%` 格式）与仓库惯用 + 写法冲突，不在仓库门禁内，不计入。 + +## 3. 修复（全部关闭，每项有对应测试或定向复核） + +- F1 名称唯一性：`ensure_resident_tables` 新增部分唯一索引 + `idx_resident_watches_name ON resident_watches(project_id, name) + WHERE status != 'ended'`（server environments 同款口径）； + create/update 的 INSERT/UPDATE 包完整性错误转 ValueError + （sqlite3.IntegrityError / PostgresIntegrityError，调用方仍 + 返回 400）。测试：索引存在断言、绕过查重直接插报 + IntegrityError、改名撞名报 ValueError。 +- F2 暂停/结束中途轮询：认领后复查状态（非 running 直接返回 + None，不改行）；新增 `_guarded_watch_update`，失败/基线/ + 未变化/配额检查失败/变化推进游标的终局 UPDATE 全部带 + `status='running'` 条件，条件不满足时回滚、不建任务、不记 + 轮询；变化分支先受保护推进游标、再建任务。测试：抓取回调 + 里经第二连接暂停监听，run_due 返回 None、无任务、游标与 + next_poll_at 保持暂停态、轮询数不变。 +- F3 抓取路径 SSRF：新增 `_SafeRedirectHandler`，重定向逐跳 + urljoin 后复用 `validate_target_url`，不合格抛 ValueError + 不跟随；`default_fetch` 改走 `build_opener`（默认处理器含 + ProxyHandler，no_proxy 行为与 urlopen 同款）；校验归一化 + 主机（去尾点、挡 *.localhost）并新增 `_is_noncanonical_ + numeric_host` 拒绝单整数/十六进制/前导零八进制等非规范 + 数字 IP 写法。测试：localhost.、*.localhost、2130706433、 + 0x7f000001、0177.0.0.1、0x7f.0.0.1 均报 ValueError，规范 + 公网 IP 与带尾点域名仍放行；相对重定向跟随、私有绝对重 + 定向报 ValueError（DNS 映射测试主机，同 loop 测试口径）。 +- F4 配额门判定：`\"quota_check_failed\" not in gate` 改为 + `gate.startswith(\"quota_check_failed\")`，与配额门前缀契约 + 对齐。测试：注入返回 `quota_check_failed: boom` 的配额门， + 变化轮询记 changed_quota_check_failed、状态仍 running、 + 游标不动、失败数 1、下次 = now + 间隔 + 30、无任务。 +- F5 row_get 重复：删 resident 本地定义，直接导入 + automations.row_get，内部调用点一字未动，模块依赖说明 + 同步更新。 +- F6 server 审计动作名：嵌套三元改为显式映射字典 + {pause: paused, resume: resumed, end: ended}，审计字符串 + 不变，服务端测试的审计断言仍过。 +- S1 删 test_resident.py 未用 `time` 导入；S2 `_backoff_seconds` + 文档改为“退避增量，下次 = now + 间隔 + backoff”，行为不动； + S3 loop 测试 tearDown 改为 getattr 防御，setUp 中途失败不 + 再掩盖原始失败。 + +## 4. 测试数字（全部本轮实测，无删断言、无 skip、无改默认值） + +- 定向基线（改动前）：`cloud.tests.test_resident` + + `test_resident_server` + `test_resident_website_loop` = 15 项全过。 +- 定向改动后：同三文件 = 20 项全过（基线 15 + 新增 5：URL + 变体、唯一索引、配额检查失败、中途暂停、重定向），4.450 秒； + ruff E、F 在居民差分 4 文件上全部通过。 +- 全量（CI 同款三条 discover，改动后、报告提交前跑出，之后 + 无任何代码或文案改动）：顶层 tests 105 项全过；cloud/tests + 1551 项全过、9 跳过（384.252 秒）；deploy/tests 44 项全过 + （单独重跑确认）；合计 1700 项、0 失败、9 跳过。对账：闭环 + 基线 1695（105 + 1546 + 44）+ 新增 5 = 1700，无测试丢失、 + 无新失败。 +- 覆盖率：本仓库 CI 无覆盖率门禁，不虚构覆盖率变化；新增 + 5 项定向测试即上述全量的一部分。 + +## 5. 复评分数 + +- 架构：10 / 10 +- 代码质量：10 / 10 + +复评依据：F1/F2 两项严重 findings 已由数据库约束与状态条件 +UPDATE 从机制上关闭（非仅文档说明），并有定向测试锁定； +F3 安全边界已从“创建时一次校验”补到“抓取重定向逐跳校验” +与非规范 IP 写法；F4/F5/F6/S1/S2/S3 全部关闭且测试全过。 +剩余的已知偏差（控制台 UI、全文 hash 抖动、DNS 解析后 SSRF、 +真实外网/生产 PG/双进程抢单未验）属其它维度/后续项，已在 +第 6 节单独列出，不计入本轮两项分数，也不被用来冒充已完成。 + +## 6. 剩余 blocker 与未做事项 + +- 未做：未 push、未合并、未跑 deploy.py、未部署；不做其它探针、 + 不接控制台 UI、不做生产 PG 与双进程抢单实测。 +- 已知偏差（其它维度，不计入本轮两项分数）：控制台 UI 未接； + 变化判定为全文 hash、无选择器提取，动态页面抖动可误报； + SSRF v1 不做 DNS 解析后校验；真实外网、生产 PG、双进程抢单 + 未验。 diff --git a/verification/resident-task-summary.md b/verification/resident-task-summary.md new file mode 100644 index 0000000..5145946 --- /dev/null +++ b/verification/resident-task-summary.md @@ -0,0 +1,69 @@ +# 常驻监听任务（resident watch）验证摘要 + +## 基底与交付物（代码就绪，local only） + +- 工作区：`~/workspace/codex-app-server-resident-task` +- 分支：`feat-resident-task`，基底 `origin/main` = `622c51e`（Merge PR #274，2026-11-02 fetch 到的头，与任务书一致）。 +- 主 checkout（`~/workspace/codex-app-server`，`feat-repo-picker`，有脏文件）与其它现存 worktree 一律未动。 +- 本轮提交前状态：`cloud/server.py` +187 行、`cloud/worker.py` +31 行，新增 `cloud/resident.py`、`cloud/tests/test_resident.py`、`cloud/tests/test_resident_server.py`、本摘要。 +- **未做**：未 push、未合并到 main、未跑 `deploy.py`、未部署、未建生产任务、未动生产／主机 env／系统 key／sub2api／生产网关／仓库可见性／GitHub App 权限、未杀生产容器。 + +## 基底核实（先调查结论：main 上常驻监听不存在，不是重复建设） + +- 任务状态机仍是 `queued -> running -> completed / failed / interrupted / merged / merge-failed`；`/pause` 只是把进行中回合任务标 `interrupted` 并置 `paused_by_user=1`（`cloud/server.py` `_pause_task`），不是可续跑的常驻暂停态——本实现不复用它，另建常驻态。 +- 云端自动化（`cloud/automations.py`）定义层只有 `active / cancelled`，含 `automations / automation_runs` 表与 `run_due_automation` 的 claim 抢单、配额门；服务端本身不起定时器，到期触发靠外部触发器。本实现复用它的 claim 口径与配额函数，不另起冲突语义。 +- goal 循环（`host.py` `run_goal_loop`，`max_rounds=0` 不限轮次）结束条件是模型宣告达成，不是“一直监听外部变化”；定时自动化每次到期建一个新任务跑一轮就结束。两者都不是常驻监听，本摘要不把它们说成已实现。 +- main 上与暂停相关的已有机制（MCP `pause_task` 工具、长任务中断自动恢复）针对的都是回合任务，不是常驻监听；grep `pause|resident` 在 `cloud/*.py` 与 `host.py` 中未发现常驻监听实现。 + +## 设计抉择 + +- 新层不改任务表：`cloud/resident.py`（标准库 + 复用 `cloud.automations` 的 `_audit_row`、`_automation_quota_exceeded`）新增 `resident_watches` / `resident_polls` 两张表，避免改动任务状态机牵连 worker 认领、sweep、自动恢复。 +- 状态机：`running -> paused -> running`；`running / paused -> ended`（终态）。`ended` 后 pause / resume / end / update 一律 409，恢复已结束的监听必须新建任务（文档与测试双锁定，见下）。 +- 暂停保留现场：暂停只改 `status / paused_reason / next_poll_at`，`last_content_hash / last_poll_at / last_change_at / last_change_text` 不动；暂停期间 `due=False`，驱动不轮询。恢复时 `next_poll_at=now` 立刻复查，用旧游标比对——暂停期间的变化在恢复后第一次轮询即报出，不丢。 +- 驱动（明说）：服务端本身不起定时器，与自动化同款结论。实际驱动是 `cloud/worker.py` 的 `run_loop` 轮次边界调 `process_due_resident_watches()`（新增方法，异常只记日志、不炸循环；`self.resident_fetcher` 可注入假抓取供测试）。任何外部触发器也可直接调 `run_due_resident_watch / process_due_resident_watches`，两者都走 claim 门。 +- 防重复轮询：`claim_due_watch` 条件 UPDATE（`status='running'` 且 `next_poll_at <= now`），抢到才把 `last_poll_at / last_claimed_at` 写成 now 并预置 `next_poll_at=now+interval`（崩在抓取中途时下个间隔仍重试，不因认领锁死永久停摆；完成后按成败重算）。 +- 变化才花钱：轮询 = 廉价抓取 + 空白折叠后 sha256 比对，不跑模型。`baseline`（首次，只建游标）、`unchanged`（不建任务）、`changed`（才建一个 queued 任务，prompt = 用户处理提示 + 变化摘录）三种产出都记 `resident_polls`，可查。 +- 配额边界：配额门只在“变化要建模型任务”那一刻过（复用自动化同款 `_automation_quota_exceeded`）。耗尽时**不结束、不丢变化**：转 `paused`（`paused_reason='quota_exceeded'`）、游标不推进、记 `changed_quota_paused`；配额恢复后恢复，同一变化仍能报出。配额检查本身失败时按瞬时失败退避，不转暂停，游标同样不推进。 +- 安全边界：间隔下限 60 秒（与自动化同款）、上限 30 天；抓取超时 1–60 秒（默认 10 秒）、读满 1MB 即断；失败退避 30 / 60 / 120……秒、上限 3600 秒；目标 URL 仅 http/https、无 userinfo、字面 localhost／私有／回环／链路本地 IP 拒绝。 +- API（`cloud/server.py`，权限与自动化同款：写 `create_task`、读 `read`，全部记审计）： + - `POST /projects/{id}/resident-watches`（name / target_url / prompt 必填；interval_seconds 默认 3600；fetch_timeout_seconds 可选；创建不走配额门） + - `GET /projects/{id}/resident-watches`（附 `due_count`）、`GET /projects/{id}/resident-watches/{wid}`（附最近轮询 `polls`） + - `PATCH /projects/{id}/resident-watches/{wid}`（改名／提示／目标／间隔／超时；ended 409；改间隔不立即触发，下次轮询完成后按新间隔重算） + - `POST /projects/{id}/resident-watches/{wid}/pause|resume|end` + - 状态查询返回常驻态、游标（`last_content_hash`）、最后变化（`last_change_at / last_change_text`）、下次轮询（`next_poll_at`）、失败次数与暂停原因。 + +## 验收清单逐项结果 + +1. 能新建常驻任务（目标 URL、轮询间隔、变化处理提示）：✅ 模块与服务端测试覆盖；非空壳——新建即到期，驱动下一轮即可认领。 +2. 三态清晰（运行中／已暂停／已结束终态）：✅ 状态机与 HTTP 409 语义双覆盖。 +3. 暂停保留现场、暂停期间停轮询、恢复从游标续跑：✅ 测试 `test_pause_preserves_cursor_and_resume_reports_pause_period_change`（暂停期间变了，恢复后第一次轮询即报出）。 +4. 结束终态、结束后不再静默重启、结束后恢复被拒绝：✅ `ended` 后 due 查询只认 `running`，pause / resume / end / update 一律 409（服务端错误文本含 terminal 并提示新建）；测试覆盖。 +5. 真正有驱动、claim 防重复、心跳与 stale 回收：✅ 驱动为 worker `run_loop` 轮次边界（不是服务端定时器，明说）；claim 单胜者测试覆盖；stale 口径为“认领预置 next_poll_at，崩后下个间隔重试”+ worker 既有每轮心跳／sweep（常驻监听本身不另起心跳表，见已知偏差）。 +6. 变化才花钱、每次轮询与变化可查：✅ baseline / unchanged 不建任务、changed 才建任务；`resident_polls` 逐轮记录。 +7. 间隔下限、抓取超时、失败退避、除配额耗尽外不擅自结束、配额耗尽只转可恢复暂停并明示原因：✅ 全部测试覆盖；配额耗尽测试断言状态为 `paused` 且明确 `!= ended`。 +8. 权限与审计沿用任务／自动化同款、pause / resume / end 记审计：✅ 服务端测试断言 `resident_watch_created / paused / resumed / ended` 审计行；跨项目行一律 404（不泄露存在性）；局外人写操作被拒。 +9. API 含暂停、恢复、结束与状态查询：✅ 见上；控制台 UI 本轮未接，列为已知偏差，不写成已完成。 + +## 真实测试数字 + +- 定向（改动后）：`python3 -m unittest cloud.tests.test_resident cloud.tests.test_resident_server` —— 13 项（模块 10 + 服务端 3），全部通过。 +- 全量（CI 同款三条 discover，改动后、仅最后一次文案注释修订前跑出；该修订只改 `update_watch` 的 docstring，不改任何代码逻辑，修订后定向 13 项与自动化相关 31 项重跑全部通过）： + - `discover -s tests`：105 项，0 失败，0 跳过。 + - `discover -s cloud/tests`：1544 项，0 失败，9 跳过。 + - `discover -s deploy/tests`：44 项（单独重跑确认），0 失败。 + - 合计 1693 项、0 失败、9 跳过。失败未靠删断言、加 skip、改默认值蒙混；新增测试 13 项全部计入上述 cloud 套件。 +- 新模块定向覆盖率（`coverage run --source=cloud.resident`，仅定向 13 项）：`cloud/resident.py` 312 语句、85% 覆盖。未覆盖主因：默认真实 HTTP 抓取（`default_fetch`，单元测试一律注入假抓取，不联网）与个别异常分支（PG 方言补列、配额检查失败分支）；本仓库 CI（`.github/workflows/host-verify.yml`）本身没有覆盖率门禁，故不报“覆盖率变化”——全量覆盖率基线未在本轮重算，不虚构数字。 + +## 已知偏差与未做事项 + +- 控制台 UI 未接：`cloud/console.html` 是打包产物，本轮没有常驻监听面板；用户现阶段只能走 API。接 UI 是后续一轮，不在本轮验收内。 +- 变化判定 v1 较粗：空白折叠后全文 sha256，没有选择器／区域提取；动态页面（广告、时间戳、随机推荐）本身会抖动成“变化”并建任务。缓解在后续：加 CSS 选择器或正文提取口径、或变化阈值。 +- SSRF 防护 v1 只挡字面 localhost／私有 IP，不做 DNS 解析后校验（解析后重定向到内网、DNS rebinding 不在覆盖内）；`websearch.fetch_url` 的同款字面口径已对齐，但不要把 v1 当成完整 SSRF 方案。 +- 常驻监听没有独立心跳表：stale 回收靠认领预置的 `next_poll_at` + worker 既有心跳／sweep；驱动进程整体死亡时，已认领未完成的当次轮询要等到下个间隔才补跑（与自动化 claim 的同款取舍）。 +- 未验证：真实外网抓取（单元测试不联网）、生产 PG 上的表迁移（`PRAGMA` 补列走自动化同款适配写法，但未在 PG 实测）、worker 常驻循环里的长时驱动（仅方法级测试，未跑真实 `run_loop` 长循环）、多 worker 并发抢单（仅条件 UPDATE 单胜者测试，未做双进程实测）。 +- 未做：未 push、未合并、未部署；UI、选择器提取、DNS 级 SSRF、PG 实测、双进程抢单实测留待后续轮次。 + +## Blocker + +- 无本轮 blocker（全部工作均为本地代码与本地测试，未触及只有用户能给的账号／密钥／商户／商店账号／宿主／维护窗口，也未发生花真钱、删生产数据、改生产可见性的动作）。 +- 上线前仍需用户侧决策（不是本轮 blocker，不擅自定）：控制台 UI 是否随本批上、默认轮询间隔与抓取超时是否按本轮默认值（60 秒下限／10 秒超时）拍板为生产默认值、PG 实测与部署窗口。 diff --git a/verification/resident-website-loop-summary.md b/verification/resident-website-loop-summary.md new file mode 100644 index 0000000..69a0665 --- /dev/null +++ b/verification/resident-website-loop-summary.md @@ -0,0 +1,125 @@ +# 常驻监听网站闭环验证摘要 + +## 起点与交付物（代码就绪，local only） + +- 工作区：`~/workspace/codex-app-server-resident-website-loop` +- 分支：`feat-resident-website-loop`，起点为上一轮网站版 + `feat-resident-task` 的 `0722ca3`（基底 `origin/main 622c51e`）。 +- 实现提交：`561b389`（仅新增端到端测试文件 + `cloud/tests/test_resident_website_loop.py`，364 行）。 + 本摘要为其后提交，工作区提交后干净。 +- 范围：只做网站用例。不做 server_status、agent_status、 + relay_cost 等其它探针；不把 resident-watches 改名/泛化成 + 通用任务；不改 API、表名与任何生产默认值。 +- 泛化 worktree（`~/workspace/codex-app-server-resident-generic`） + 本轮一律未碰；resident-task worktree、主 checkout + （`feat-repo-picker`）及其它 worktree 一律未动。 +- **未做**：未 push、未合并到 main、未跑 `deploy.py`、未部署、 + 未建生产任务、未动生产／主机 env／系统 key／sub2api／ + 生产网关／仓库可见性／GitHub App 权限、未杀生产容器、 + 未连真实外部网站。 + +## 闭环逐环证据（全部真实链路，不是注入假抓取） + +测试文件：`cloud/tests/test_resident_website_loop.py`，2 项。 +内容服务是线程内真实 `http.server`（可变内容、可停服重启）； +控制面是真实 `ThreadingHTTPServer` + 真实 SQLite 临时库； +驱动走真实 `default_fetch`（urllib、超时、1MB 截断）。 +生产 URL 校验一字不改：测试主机名 `website-loop.test` 不是 +IP 字面、不是 localhost，所以校验通过；测试进程内只把该 +主机名的 DNS 指到本地服务（相当于 /etc/hosts 的一条测试 +条目），并给它加 `no_proxy`（见“测试中发现的两处假断点”）。 + +1. 创建（API）：`POST /projects/{id}/resident-watches` + （name=site1、target_url=`http://website-loop.test:<本地端口>/data`、 + prompt=“变化时汇报”、interval_seconds=60）返回 201， + 状态 running、interval 60、fetch_timeout 10（生产默认）、 + last_content_hash 为空、next_poll_at 在位；审计含 + `resident_watch_created`。状态查询 `GET` 单条可看到 + running、游标、下次轮询。 +2. 基线轮询：worker 驱动 `process_due_resident_watches` + （fetcher=None，即真实 default_fetch）抓到 “version one”， + 记 `baseline` 轮询、游标为其 sha256，任务数为 0； + 状态查询的 polls 第 1 条就是 baseline。 +3. 变化建任务：内容改成 “version two”，到期轮询记 + `changed`、建出 1 个 queued 任务，任务 prompt 含用户提示 + “变化时汇报”与摘录 “version two”，任务行与审计行 + `resident_watch_change_task_created` 均可查；游标推进到 + version two，last_change_text 含 version two。 +4. 暂停：API pause 后状态 paused、next_poll_at 为空；暂停 + 期间内容再改成 “version three”，驱动返回空、轮询数仍为 + 2、游标仍为 version two——暂停期间的变化没有被轮询上报。 +5. 恢复：API resume 后状态 running；下一次驱动用旧游标比 + 对，记 `changed` 并建第 2 个任务（prompt 含 version three）， + 不是从头建基线；游标推进到 version three。 +6. 失败环：停掉内容服务后驱动记 `failed` 轮询、 + consecutive_failures=1、next_poll_at 大于 now+60+30 + （间隔 + 首次退避 30 秒）、状态仍 running、游标仍为 + version three——驱动不崩、监听不擅自结束。恢复服务后 + （内容仍为 version three）驱动记 `unchanged`、失败清零、 + 任务数仍为 2。 +7. 结束：API end 后状态 ended、next_poll_at 为空、游标仍为 + version three；驱动返回空；pause / resume / end / PATCH + 更新一律 409；状态查询显示 ended 与最后游标/变化， + polls 总数为 5（baseline、changed、changed、failed、 + unchanged）。 +8. 驱动接线：单独一测试把 `run_all` 等无关边界打桩，只留 + resident 驱动走真实抓取，调真实 `Worker.run_loop` + （interval=0.05、max_seconds=0.4）；记录到 run_loop 轮次 + 边界至少调了 1 次 `process_due_resident_watches`，且状态 + 查询里出现 1 条 baseline 轮询、游标为 version one。 + 这证明接线在 run_loop 里，不只是测试直接调驱动函数。 + +## 断点结论：生产代码无断点，无生产改动 + +本轮逐环均通，生产代码（`cloud/resident.py`、`cloud/server.py`、 +`cloud/worker.py`）一字未改，`git diff` 为空、新增文件只有 +端到端测试。第一次跑出的 2 个失败都是测试脚手架问题， +不是生产断点，修法与证据如下，不许读成“修了生产代码”： + +- 假断点 1——首次真实抓取 failed：测试环境自带 egress 代理， + urllib 默认把测试主机名也送代理并被代理断开。实测对照： + 不加 no_proxy 时 `default_fetch` 报 RemoteDisconnected， + 加 no_proxy 后返回 “version one”。修法只在测试 setUp 里 + 给 `website-loop.test` 加 no_proxy（测试进程内生效，撕掉 + 时还原），生产抓取与代理行为一字不改。 +- 假断点 2——run_loop 接线测试 TypeError：实例属性上的 + recording 函数不会自动绑定 self。修法改为闭包捕获 worker， + 不改 worker 代码。 + +## 真实测试数字 + +- 定向（全量后重跑，含本轮 2 项）：`python3 -m unittest + cloud.tests.test_resident cloud.tests.test_resident_server + cloud.tests.test_resident_website_loop` —— 15 项全部通过。 + 其中上一轮 13 项在新 worktree 动工前已先复核全过。 +- 全量（CI 同款三条 discover，定向之后、全量之前无生产改动， + 全量后又重跑定向 15 项全过，时间关系如实写明）： + - `discover -s tests`：105 项，0 失败，0 跳过。 + - `discover -s cloud/tests`：1546 项，0 失败，9 跳过 + （上一轮 1544 + 本轮新增 2，对账无差）。 + - `discover -s deploy/tests`：44 项，0 失败。 + - 合计 1695 项、0 失败、9 跳过。失败未靠删断言、加 skip、 + 改默认值蒙混；生产间隔下限 60 秒、抓取超时默认 10 秒、 + URL 校验口径一律未动，到期只用驱动 now 注入模拟时间流逝。 + +## 已知偏差与未做事项（未写成已完成） + +- 控制台 UI 不在本轮闭环内：只能走 API，无常驻面板。 +- 真实外网抓取未验：只验了本地 HTTP 服务 + 测试 DNS 映射， + 不要把本轮说成真实外网验证；生产环境下的代理/DNS/证书 + 行为仍属后续验证项。 +- 生产 PG 表迁移未验（只验 SQLite 临时库）、双进程抢单未验 + （只验条件 UPDATE 单胜者的上一轮单测与本轮单驱动链路）。 +- 其它探针本轮明确不做：server_status、agent_status、 + relay_cost、通用任务改名一律不在本轮范围内。 +- 变化判定仍是上一轮 v1 全文 hash，无选择器/区域提取； + SSRF 仍是 v1 只挡字面私有 IP，未做 DNS 解析后校验。 +- 未做：未 push、未合并、未部署；架构与代码质量的 10 分 + 审核是父级后续独立步骤，本轮不自评、不给自己打 10 分。 + +## Blocker + +- 本轮无 blocker（纯本地代码与本地测试，未触及只有用户能给 + 的账号／密钥／商户／商店账号／宿主／维护窗口，也未发生 + 花真钱、删生产数据、改生产可见性的动作）。