From 0722ca3fd8add6113a84adddeb26d95a166c51fc Mon Sep 17 00:00:00 2001 From: host Date: Sun, 1 Nov 2026 19:01:39 +0000 Subject: [PATCH] feat: resident watch tasks with pause/resume/end Long-running website monitoring: new cloud/resident.py layer on top of the automations claim/quota model (no task state-machine change). States running/paused/ended; pause preserves the content-hash cursor and stops polling, resume re-checks from the cursor, end is terminal (resume after end is 409, a new watch is required). Polling is a cheap fetch + hash comparison; only a real change creates a queued model task. Quota exhaustion pauses (quota_exceeded) instead of ending and keeps the cursor. Driver is the worker run_loop round boundary; server exposes create/list/get/update/pause/resume/end with audit. Console UI intentionally not wired this round (see verification summary for deviations). --- cloud/resident.py | 730 ++++++++++++++++++++++++++ cloud/server.py | 187 +++++++ cloud/tests/test_resident.py | 319 +++++++++++ cloud/tests/test_resident_server.py | 180 +++++++ cloud/worker.py | 31 ++ verification/resident-task-summary.md | 69 +++ 6 files changed, 1516 insertions(+) create mode 100644 cloud/resident.py create mode 100644 cloud/tests/test_resident.py create mode 100644 cloud/tests/test_resident_server.py create mode 100644 verification/resident-task-summary.md diff --git a/cloud/resident.py b/cloud/resident.py new file mode 100644 index 0000000..37684e9 --- /dev/null +++ b/cloud/resident.py @@ -0,0 +1,730 @@ +"""常驻监听任务（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 的两条复用函数 +（_audit_row、_automation_quota_exceeded），不 import +cloud 其它模块，避免循环 import。 +""" + +import hashlib +import ipaddress +import re +import time +import urllib.request +import uuid +from urllib.parse import urlsplit + +from cloud.automations import _audit_row as _auto_audit_row +from cloud.automations import _automation_quota_exceeded + +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+") + + +def row_get(row, name): + """行取值容错：与 automations.row_get 同款语义。""" + try: + return row[name] + except (KeyError, IndexError, TypeError): + return None + + +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)") + 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 秒。""" + 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 validate_target_url(url): + """目标 URL 校验：仅 http/https、无 userinfo、主机非空、 + 字面 localhost / 私有 / 回环 / 链路本地 IP 拒绝（与 + websearch._is_blocked_fetch_url 的字面 IP 口径同款， + DNS 解析后的私有地址不在 v1 覆盖内，见摘要已知偏差）。""" + 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() + if not host: + raise ValueError("target_url host required") + if host == "localhost": + raise ValueError("localhost 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 + 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)) + 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") + 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)) + 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() + + +def default_fetch(url, timeout): + """默认廉价抓取：urllib，超时可配，读满 1MB 即断， + 解码失败按 replace 处理。调用方可注入 fetcher 替换它 + （测试与未来选择器提取都走注入点，不改本函数签名）。""" + req = urllib.request.Request(url, headers={ + "User-Agent": "codex-app-server-resident-watch/1.0"}) + with urllib.request.urlopen(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 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() + watch = watch_to_dict(row, now) + 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) + conn.execute( + "UPDATE resident_watches SET consecutive_failures = ?," + " next_poll_at = ?, updated_at = ? WHERE id = ?", + (failures, nxt, now, watch_id)) + 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: + conn.execute( + "UPDATE resident_watches SET last_content_hash = ?," + " consecutive_failures = 0, next_poll_at = ?," + " updated_at = ? WHERE id = ?", + (observed, now + interval, now, watch_id)) + poll = _record_poll(conn, watch_id, watch["project_id"], + "baseline", observed, "first poll", None, now) + conn.commit() + return poll + if observed == old_hash: + conn.execute( + "UPDATE resident_watches SET consecutive_failures = 0," + " next_poll_at = ?, updated_at = ? WHERE id = ?", + (now + interval, now, watch_id)) + poll = _record_poll(conn, watch_id, watch["project_id"], + "unchanged", observed, "", None, now) + conn.commit() + return poll + # 真实变化：配额门只在这里过。 + gate = _automation_quota_exceeded(conn, watch["created_by"]) + if gate is not None and "quota_check_failed" not in gate: + conn.execute( + "UPDATE resident_watches SET status = ?," + " paused_reason = ?, next_poll_at = NULL, updated_at = ?" + " WHERE id = ?", + (PAUSED, "quota_exceeded", now, watch_id)) + 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) + conn.execute( + "UPDATE resident_watches SET consecutive_failures = ?," + " next_poll_at = ?, updated_at = ? WHERE id = ?", + (failures, nxt, now, watch_id)) + poll = _record_poll(conn, watch_id, watch["project_id"], + "changed_quota_check_failed", observed, gate, + None, now) + conn.commit() + return poll + task_id = _create_change_task(conn, watch, excerpt, now) + conn.execute( + "UPDATE resident_watches SET last_content_hash = ?," + " last_change_at = ?, last_change_text = ?," + " consecutive_failures = 0, next_poll_at = ?," + " updated_at = ? WHERE id = ?", + (observed, now, excerpt, now + interval, now, watch_id)) + 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..21f3614 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,137 @@ 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 + with _auth_store(self.server.db_path) as store: + store.audit(user_id, project_id, + "resident_watch_%s" % ("ended" if action == "end" + else action + "d" + if action == "pause" + else "resumed"), + 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..f488731 --- /dev/null +++ b/cloud/tests/test_resident.py @@ -0,0 +1,319 @@ +"""常驻监听任务单测（cloud/resident.py）：不跑真实模型、不联网， +抓取一律注入假 fetcher。覆盖创建校验、暂停保游标、恢复续跑、 +结束终态、防重复轮询、仅变化建任务、失败退避、配额耗尽转暂停、 +worker 驱动批量入口。""" +import os +import sqlite3 +import sys +import tempfile +import time +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + from cloud.resident import ( + claim_due_watch, + create_watch, + end_watch, + ensure_resident_tables, + get_watch, + list_polls, + pause_watch, + process_due_resident_watches, + resume_watch, + run_due_resident_watch, + update_watch, + ) + 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])) + from cloud.resident import ( + claim_due_watch, + create_watch, + end_watch, + ensure_resident_tables, + get_watch, + list_polls, + pause_watch, + process_due_resident_watches, + resume_watch, + run_due_resident_watch, + update_watch, + ) + 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) + + +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/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-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 实测与部署窗口。