From 8d4771bce374c0cda47ad6aa3e80c2d64dbbbfc5 Mon Sep 17 00:00:00 2001 From: host Date: Fri, 6 Nov 2026 22:42:31 +0000 Subject: [PATCH] feat(chv): align Muse P8 background work (crons/swarm/presentation/per-task browser) - crons: pause/resume, exponential-backoff retry (cap 1h), retry claim single-winner reusing quota+task build, run_cron_tick cap 5/project - swarm: per-agent task id + isolated workspace (resolve_inside gate), tool register + invocation audit (no execution in swarm layer) - presentation: worth-presenting rules (needs_input/failed/new content only) + notification ledger dedup 1h per (task, reason) - browser_tasks: one BrowserTransport instance per browser task, single-use (reuse after destroy refused), shots dir per instance - automations: paused status + retry columns (max_retries, retry_backoff_seconds, consecutive_failures, next_retry_at) - gap table docs/chv-muse-p8-gap.md; self-defined values listed there (not Meta official): tick cap 5, swarm max 32/default 8, retry base 60s cap 3600s, ledger window 3600s, snapshot trunc 200/64KB - tests: 49 new in cloud/tests/test_chv_muse_p8.py (incl. negatives: cron unauthorized trigger, cross-workspace access, no-content silence, dedup); full cloud suite 1916 OK (skipped=8) --- cloud/automations.py | 144 +++++-- cloud/chv/browser_tasks.py | 211 ++++++++++ cloud/chv/crons.py | 281 +++++++++++++ cloud/chv/presentation.py | 172 ++++++++ cloud/chv/swarm.py | 383 +++++++++++++++++ cloud/tests/test_chv_muse_p8.py | 699 ++++++++++++++++++++++++++++++++ docs/chv-muse-p8-gap.md | 109 +++++ 7 files changed, 1975 insertions(+), 24 deletions(-) create mode 100644 cloud/chv/browser_tasks.py create mode 100644 cloud/chv/crons.py create mode 100644 cloud/chv/presentation.py create mode 100644 cloud/chv/swarm.py create mode 100644 cloud/tests/test_chv_muse_p8.py create mode 100644 docs/chv-muse-p8-gap.md diff --git a/cloud/automations.py b/cloud/automations.py index 3320da2..651068a 100644 --- a/cloud/automations.py +++ b/cloud/automations.py @@ -54,8 +54,19 @@ MAX_NAME_LEN = 100 MAX_PROMPT_LEN = 20000 MAX_REPO_URL_LEN = 1000 TRIAGE_LIMIT = 50 +# 单项目自动化定义上限（本项目自定，非 Meta 官方值：Meta 未公开 +# cron 的数量上限；超限建定义即 ValueError）。 +MAX_AUTOMATIONS_PER_PROJECT = 100 +# 失败重试默认值（本项目自定，非 Meta 官方值：Meta 未公开失败 +# 重试策略；退避 = 基数 × 2^(n-1)，上限见 crons 模块）。 +DEFAULT_MAX_RETRIES = 3 +DEFAULT_RETRY_BACKOFF_SECONDS = 60 +MAX_RETRIES_LIMIT = 10 +MIN_RETRY_BACKOFF_SECONDS = 10 +MAX_RETRY_BACKOFF_SECONDS = 3600 ACTIVE = "active" +PAUSED = "paused" CANCELLED = "cancelled" @@ -73,7 +84,9 @@ AUTOMATION_COLUMNS = ( "id", "project_id", "name", "prompt", "repo_url", "interval_seconds", "status", "last_run_at", "last_claimed_at", "created_by", "created_at", "cancelled_at", - "first_run_at", "run_once") + "first_run_at", "run_once", + "max_retries", "retry_backoff_seconds", + "consecutive_failures", "next_retry_at") RUN_COLUMNS = ( "id", "automation_id", "project_id", "task_id", "status", "detail", "created_at") @@ -104,8 +117,11 @@ def _existing_columns(conn, table): def ensure_automations_tables(conn): """幂等建自动化表与产出记录表，并补齐新增列。 - automations：一条定义；status 只取 active / cancelled， - 取消是软删除(行保留、审计可查，名字可重用)。 + automations：一条定义；status 只取 active / paused / + cancelled，取消是软删除(行保留、审计可查，名字可重用)； + paused 是停用(可恢复，不到期、不抢单，见 crons 模块)。 + max_retries / retry_backoff_seconds / consecutive_failures / + next_retry_at 是失败重试列（P8 crons，见该模块）。 automation_runs：一次执行产出；task_id 为空表示执行器 抢单后尚未建任务(只应短暂存在，执行器建任务后补记)。 调用方在写成功后 commit(server 的 handler 与 peer 模式 @@ -125,7 +141,11 @@ def ensure_automations_tables(conn): "created_at DOUBLE PRECISION, " "cancelled_at DOUBLE PRECISION, " "first_run_at DOUBLE PRECISION, " - "run_once INTEGER)" + "run_once INTEGER, " + "max_retries INTEGER, " + "retry_backoff_seconds INTEGER, " + "consecutive_failures INTEGER, " + "next_retry_at DOUBLE PRECISION)" ) conn.execute( "CREATE TABLE IF NOT EXISTS automation_runs(" @@ -143,7 +163,11 @@ def ensure_automations_tables(conn): ("last_claimed_at", "DOUBLE PRECISION"), ("cancelled_at", "DOUBLE PRECISION"), ("first_run_at", "DOUBLE PRECISION"), - ("run_once", "INTEGER")): + ("run_once", "INTEGER"), + ("max_retries", "INTEGER"), + ("retry_backoff_seconds", "INTEGER"), + ("consecutive_failures", "INTEGER"), + ("next_retry_at", "DOUBLE PRECISION")): if col not in existing: try: conn.execute( @@ -232,11 +256,30 @@ def automation_to_dict(row, now=None): "cancelled_at": row_get(row, "cancelled_at"), "first_run_at": row_get(row, "first_run_at"), "run_once": bool(row_get(row, "run_once")), + "max_retries": _retry_int(row, "max_retries", + DEFAULT_MAX_RETRIES), + "retry_backoff_seconds": _retry_int( + row, "retry_backoff_seconds", + DEFAULT_RETRY_BACKOFF_SECONDS), + "consecutive_failures": _retry_int( + row, "consecutive_failures", 0), + "next_retry_at": row_get(row, "next_retry_at"), "next_run_at": nxt, "due": is_due(row, now), } +def _retry_int(row, name, default): + """重试列取值：老库行缺列回 None 时用默认值；脏值（非 + int/负数）同样回默认值，绝不因此炸掉整页。""" + value = row_get(row, name) + if isinstance(value, bool) or not isinstance(value, int): + return default + if value < 0: + return default + return value + + def _validate_fields(name, prompt, interval_seconds, repo_url, first_run_at=None, run_once=False): if not name or not isinstance(name, str) or not name.strip(): @@ -276,18 +319,45 @@ def _validate_fields(name, prompt, interval_seconds, repo_url, first_run_at, run_once +def _validate_retry(max_retries, retry_backoff_seconds): + """重试参数校验（本项目自定范围）：max_retries 0–10， + retry_backoff_seconds 10–3600 秒；非法即 ValueError。""" + if isinstance(max_retries, bool) \ + or not isinstance(max_retries, int): + raise ValueError("max_retries must be an integer") + if not 0 <= max_retries <= MAX_RETRIES_LIMIT: + raise ValueError("max_retries must be between 0 and %d" + % MAX_RETRIES_LIMIT) + if isinstance(retry_backoff_seconds, bool) \ + or not isinstance(retry_backoff_seconds, int): + raise ValueError("retry_backoff_seconds must be an integer") + if not MIN_RETRY_BACKOFF_SECONDS <= retry_backoff_seconds \ + <= MAX_RETRY_BACKOFF_SECONDS: + raise ValueError("retry_backoff_seconds must be between %d" + " and %d" % (MIN_RETRY_BACKOFF_SECONDS, + MAX_RETRY_BACKOFF_SECONDS)) + return max_retries, retry_backoff_seconds + + def create_automation(conn, project_id, name, prompt, created_by, interval_seconds=86400, repo_url=None, now=None, - first_run_at=None, run_once=False): + first_run_at=None, run_once=False, + max_retries=DEFAULT_MAX_RETRIES, + retry_backoff_seconds=DEFAULT_RETRY_BACKOFF_SECONDS): """创建自动化定义：同项目内 active 的名字唯一(取消过的 名字可重用，与环境名唯一规则同款语义)。成功时 commit 并返回 dict；校验失败抛 ValueError(调用方转 400)。 first_run_at 为首次运行时刻（epoch 秒，过去则建完即 - 到期）；run_once 为 True 时跑过一次后不再到期。""" + 到期）；run_once 为 True 时跑过一次后不再到期。 + max_retries / retry_backoff_seconds 是失败重试策略 + （P8 crons，本项目自定）；单项目定义上限 + MAX_AUTOMATIONS_PER_PROJECT，超限即 ValueError。""" (name, prompt, interval_seconds, repo_url, first_run_at, run_once) = _validate_fields( name, prompt, interval_seconds, repo_url, first_run_at, run_once) + (max_retries, retry_backoff_seconds) = _validate_retry( + max_retries, retry_backoff_seconds) if now is None: now = time.time() ensure_automations_tables(conn) @@ -297,16 +367,30 @@ def create_automation(conn, project_id, name, prompt, created_by, (project_id, name, ACTIVE)).fetchone() if existing is not None: raise ValueError("automation name already exists in this project") + count_row = conn.execute( + "SELECT COUNT(*) FROM automations WHERE project_id = ?" + " AND status IN (?, ?)", + (project_id, ACTIVE, PAUSED)).fetchone() + try: + count = int(count_row[0]) + except (TypeError, ValueError, IndexError): + count = 0 + if count >= MAX_AUTOMATIONS_PER_PROJECT: + raise ValueError("too many automations in this project") auto_id = uuid.uuid4().hex conn.execute( "INSERT INTO automations(id, project_id, name, prompt," " repo_url, interval_seconds, status, last_run_at," " last_claimed_at, created_by, created_at, cancelled_at," - " first_run_at, run_once)" - " VALUES(?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?, ?, NULL, ?, ?)", + " first_run_at, run_once, max_retries," + " retry_backoff_seconds, consecutive_failures," + " next_retry_at)" + " VALUES(?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?, ?, NULL, ?, ?," + " ?, ?, 0, NULL)", (auto_id, project_id, name, prompt, repo_url, interval_seconds, ACTIVE, created_by, now, - first_run_at, 1 if run_once else 0)) + first_run_at, 1 if run_once else 0, + max_retries, retry_backoff_seconds)) conn.commit() row = conn.execute( "SELECT * FROM automations WHERE id = ?", (auto_id,)).fetchone() @@ -338,7 +422,7 @@ def get_automation(conn, automation_id): def cancel_automation(conn, automation_id, now=None): - """取消自动化(软删除)：仅 active 的行能取消。 + """取消自动化(软删除)：仅 active / paused 的行能取消。 返回 (ok, auto)：ok 为 True 时 auto 是取消后的 dict； 行不存在时 (False, None)；已取消时 (False, 旧 dict)， @@ -352,12 +436,12 @@ def cancel_automation(conn, automation_id, now=None): if row is None: return False, None row = _row_dict(row, AUTOMATION_COLUMNS) - if row["status"] != ACTIVE: + if row["status"] not in (ACTIVE, PAUSED): return False, automation_to_dict(row, now) cur = conn.execute( "UPDATE automations SET status = ?, cancelled_at = ?" - " WHERE id = ? AND status = ?", - (CANCELLED, now, automation_id, ACTIVE)) + " WHERE id = ? AND status IN (?, ?)", + (CANCELLED, now, automation_id, ACTIVE, PAUSED)) conn.commit() if cur.rowcount != 1: return False, automation_to_dict(row, now) @@ -370,16 +454,16 @@ def cancel_automation(conn, automation_id, now=None): def update_automation(conn, automation_id, fields, now=None): """更新自动化定义（automation_update）：fields 只认 name / prompt / interval_seconds / repo_url / first_run_at / - run_once 六个键，其它键一律忽略（不报错、不存入，避免 - 调用方拼错字段名后以为已改）；六个键全没给时抛 - ValueError（空更新不是成功）。校验与创建同款 + run_once / max_retries / retry_backoff_seconds 八个键，其它键 + 一律忽略（不报错、不存入，避免调用方拼错字段名后以为已 + 改）；八个键全没给时抛 ValueError（空更新不是成功）。校验与创建同款 （_validate_fields，名字仍按“同项目 active 名字唯一” 查重，改名撞名时 400）。 返回 (ok, auto)：行不存在时 (False, None)；已取消时 (False, 旧 dict)——取消是终态，改已取消的定义必须 重建，不能靠 update 复活（与 cancel 的软删除语义 - 一致）。成功时 commit。注意：更新不重置 last_run_at / + 一致）。paused 的行可改（改完仍是 paused，不自动恢复）。成功时 commit。注意：更新不重置 last_run_at / last_claimed_at——改间隔只影响下一次到期判断（next_run 由 last_run_at + 新间隔算），不立即触发一轮；想立即 跑一轮应走执行器的 claim 路径，不是改定义。""" @@ -392,7 +476,7 @@ def update_automation(conn, automation_id, fields, now=None): if row is None: return False, None row = _row_dict(row, AUTOMATION_COLUMNS) - if row["status"] != ACTIVE: + if row["status"] not in (ACTIVE, PAUSED): return False, automation_to_dict(row, now) if not isinstance(fields, dict): raise ValueError("fields must be an object") @@ -407,11 +491,16 @@ def update_automation(conn, automation_id, fields, now=None): run_once = fields.get("run_once", current["run_once"]) if not any(k in fields for k in ( "name", "prompt", "interval_seconds", "repo_url", - "first_run_at", "run_once")): + "first_run_at", "run_once", + "max_retries", "retry_backoff_seconds")): raise ValueError("no updatable field given") (name, prompt, interval, repo_url, first_run_at, run_once) = _validate_fields( name, prompt, interval, repo_url, first_run_at, run_once) + (max_retries, retry_backoff_seconds) = _validate_retry( + fields.get("max_retries", current["max_retries"]), + fields.get("retry_backoff_seconds", + current["retry_backoff_seconds"])) if name != current["name"]: clash = conn.execute( "SELECT id FROM automations WHERE project_id = ?" @@ -424,9 +513,11 @@ def update_automation(conn, automation_id, fields, now=None): conn.execute( "UPDATE automations SET name = ?, prompt = ?," " interval_seconds = ?, repo_url = ?, first_run_at = ?," - " run_once = ? WHERE id = ?", + " run_once = ?, max_retries = ?," + " retry_backoff_seconds = ? WHERE id = ?", (name, prompt, interval, repo_url, first_run_at, - 1 if run_once else 0, automation_id)) + 1 if run_once else 0, max_retries, retry_backoff_seconds, + automation_id)) conn.commit() row = conn.execute( "SELECT * FROM automations WHERE id = ?", (automation_id,) @@ -669,9 +760,13 @@ def _audit_row(conn, project_id, action, detail, now): pass -def run_due_automation(conn, automation_id, now=None): +def run_due_automation(conn, automation_id, now=None, claim_fn=None): """云端自动化执行器的唯一入口：抢单 -> 配额门 -> 建任务 -> 记产出。 + claim_fn 可注入抢单函数（默认 claim_due_automation）： + crons 模块的重试入口传自己的重试抢单，两条路共用配额门 + 与建任务逻辑，抢单语义各守各的（间隔抢单改 last_run_at， + 重试抢单只清 next_retry_at、不动间隔时钟）。 抢不到（不存在、已取消、未到期、已被别的执行器抢走）时 返回 None，不记产出、不建任务。抢到后先过配额门：创建人 超配额时不建任务，产出记 skipped_quota（不是失败，不进 @@ -685,7 +780,8 @@ def run_due_automation(conn, automation_id, now=None): if now is None: now = time.time() ensure_automations_tables(conn) - if not claim_due_automation(conn, automation_id, now): + claim = claim_fn if claim_fn is not None else claim_due_automation + if not claim(conn, automation_id, now): return None auto = conn.execute( "SELECT id, project_id, prompt, created_by FROM automations" diff --git a/cloud/chv/browser_tasks.py b/cloud/chv/browser_tasks.py new file mode 100644 index 0000000..55951f3 --- /dev/null +++ b/cloud/chv/browser_tasks.py @@ -0,0 +1,211 @@ +"""P8 browser_tasks：每浏览器任务一个实例——建/绑/毁生命周期。 + +对齐 Muse B4（官方安全文）：每个浏览器任务由专门浏览器子代理 +完成。P2 已把浏览器子代理独立成进程（chv-browser.py，常驻、 +close 只关页不杀进程）；本轨改为每任务一个实例：建实例、绑 +定一个浏览器任务、用完即毁，状态绝不跨任务复用。 + +transport 接口：真实 transport 走 P2 的 broker 协议 +（open/snapshot/screenshot/close/ping，经 /run/chv/browser.sock）； +客机侧“每任务起一个全新 chv-browser.py、close 后杀进程” +是差距表 §3/§6 的跟进项（P2 的 chv-broker.py 里改，本轨不 +复制 P2 的文件）。本层只管生命周期、单实例单任务与状态隔离， +transport 注入（测试用 fake）。 + +存储：state_dir/browser-tasks.json（原子写 + 损坏 quarantine， +orphans 同款口径）；P7 状态存储落地后迁移。 +""" + +from __future__ import annotations + +import json +import logging +import os +import time +import uuid + +from cloud.chv.atomic_file import atomic_write_text +from cloud.chv.compat215 import resolve_inside + +_LOGGER = logging.getLogger("cloud.cloud_hypervisor") + +_REGISTRY_FILENAME = "browser-tasks.json" + + +class BrowserTransport: + """浏览器实例 transport 接口：真实实现走 P2 的 broker 协议。 + + spawn 建实例、close 毁实例、ping 探活。子类须实现三者； + 基类方法一律 NotImplementedError（绝不静默假装成功）。""" + + def spawn(self, instance: dict) -> None: + """建一个浏览器实例（instance 含 instance_id/task_id）。""" + raise NotImplementedError + + def close(self, instance: dict) -> None: + """毁一个浏览器实例（进程退出、状态清理）。""" + raise NotImplementedError + + def ping(self, instance: dict) -> bool: + """实例是否还活着。""" + raise NotImplementedError + + +def _registry_path(state_dir: str) -> str: + """实例登记文件路径。""" + return os.path.join(state_dir, _REGISTRY_FILENAME) + + +def _quarantine(path: str) -> None: + """把损坏的登记文件挪到 .corrupt 旁（尽力而为）。""" + try: + os.replace(path, path + ".corrupt") + except OSError: + _LOGGER.warning("chv_browser_tasks_quarantine_failed " + "path=%s", path) + + +def _load_registry(state_dir: str) -> dict: + """读登记：无文件回空结构；损坏则 quarantine 后回空结构。""" + path = _registry_path(state_dir) + try: + with open(path, "r") as fh: + data = json.load(fh) + except FileNotFoundError: + return {"instances": {}} + except (OSError, ValueError): + _quarantine(path) + return {"instances": {}} + if not isinstance(data, dict) \ + or not isinstance(data.get("instances"), dict): + _quarantine(path) + return {"instances": {}} + return data + + +def _save_registry(state_dir: str, data: dict) -> None: + """原子落盘登记；写失败只记日志。""" + try: + os.makedirs(state_dir, exist_ok=True) + atomic_write_text(_registry_path(state_dir), json.dumps(data)) + except OSError: + _LOGGER.warning("chv_browser_tasks_save_failed " + "state_dir=%s", state_dir) + + +def _get_or_raise(data: dict, task_id: str) -> dict: + """取某任务的实例；不存在即 ValueError。""" + instance = data.get("instances", {}).get(task_id) + if not isinstance(instance, dict): + raise ValueError("browser instance not found") + return instance + + +def create_instance(state_dir: str, task_id: str, + transport: BrowserTransport, + now: float | None = None) -> dict: + """为一个浏览器任务建一个实例：单任务单实例（该任务已有 + 任何实例记录——active、destroyed 乃至 failed——即 + ValueError；一个任务一生只有一个实例，毁后不复建，审计 + 轨迹留在登记里）；先落盘 creating、再调 transport.spawn、 + 成功后置 active；spawn 抛错时记 failed 并原样抛出 + （不留半个实例）。返回实例 dict。""" + if now is None: + now = time.time() + if not task_id or not isinstance(task_id, str): + raise ValueError("task_id required") + data = _load_registry(state_dir) + existing = data["instances"].get(task_id) + if isinstance(existing, dict): + raise ValueError("browser instance already exists") + instance_id = uuid.uuid4().hex + instance = { + "instance_id": instance_id, + "task_id": task_id, + "status": "creating", + "shots_dir": "browser-tasks/%s/shots" % instance_id, + "created_at": now, + } + data["instances"][task_id] = instance + _save_registry(state_dir, data) + try: + transport.spawn(instance) + except Exception: + instance["status"] = "failed" + _save_registry(state_dir, data) + raise + instance["status"] = "active" + _save_registry(state_dir, data) + return dict(instance) + + +def get_instance(state_dir: str, task_id: str) -> dict | None: + """取某任务的实例；不存在回 None。""" + data = _load_registry(state_dir) + instance = data.get("instances", {}).get(task_id) + if not isinstance(instance, dict): + return None + return dict(instance) + + +def list_instances(state_dir: str) -> list[dict]: + """列出全部实例（按创建时间排序）。""" + data = _load_registry(state_dir) + out = [dict(v) for v in data["instances"].values() + if isinstance(v, dict)] + out.sort(key=lambda i: (i.get("created_at") or 0, + i.get("task_id") or "")) + return out + + +def destroy_instance(state_dir: str, task_id: str, + transport: BrowserTransport, + now: float | None = None) -> dict: + """毁一个实例：须是 active（不存在/已毁即 ValueError）； + 调 transport.close，close 抛错也记下并照常置 destroyed + （清理不能因一次 close 失败卡住），返回实例 dict。""" + if now is None: + now = time.time() + data = _load_registry(state_dir) + instance = _get_or_raise(data, task_id) + if instance.get("status") != "active": + raise ValueError("browser instance is not active") + close_error = "" + try: + transport.close(instance) + except Exception as exc: + close_error = str(exc)[:300] + instance["status"] = "destroyed" + instance["destroyed_at"] = now + instance["close_error"] = close_error + _save_registry(state_dir, data) + return dict(instance) + + +def ping_instance(state_dir: str, task_id: str, + transport: BrowserTransport) -> bool: + """实例是否还活着：非 active 一律 False；active 时问 + transport。""" + data = _load_registry(state_dir) + instance = data.get("instances", {}).get(task_id) + if not isinstance(instance, dict): + return False + if instance.get("status") != "active": + return False + try: + return bool(transport.ping(instance)) + except Exception: + return False + + +def instance_shots_dir(state_dir: str, task_id: str) -> str | None: + """取某任务实例的截图目录（state_dir 内、resolve_inside + 校验，越界回 None）；实例不存在回 None。""" + data = _load_registry(state_dir) + instance = data.get("instances", {}).get(task_id) + if not isinstance(instance, dict): + return None + rel = instance.get("shots_dir") + if not isinstance(rel, str) or not rel: + return None + return resolve_inside(state_dir, rel) diff --git a/cloud/chv/crons.py b/cloud/chv/crons.py new file mode 100644 index 0000000..70d37ce --- /dev/null +++ b/cloud/chv/crons.py @@ -0,0 +1,281 @@ +"""P8 crons：定时任务的启用/停用、失败重试、tick 驱动。 + +建在 cloud.automations 之上（不改 host.py run-due：它是 host +侧唯一的执行器，见 automations 模块 docstring）。本模块只依赖 +标准库 + cloud.automations，不 import 其它 cloud 模块。 + +语义（数值全部是本项目自定，见 docs/chv-muse-p8-gap.md §4； +Meta 未公开 cron 的调度粒度、数量上限与失败重试策略）： +- 停用（paused）：不到期、不抢单、可恢复；恢复时若同项目已有 + 同名 active 定义则恢复失败（调用方据 ok=False 区分）。 +- 失败重试：执行终态为 failed/interrupted 时记一次失败； + 失败次数 <= max_retries 时按下次重试时刻 next_retry_at = + now + min(backoff × 2^(n-1), 3600) 安排重试；超限后不再 + 重试，间隔时钟照常走（下个间隔仍会正常触发）。 + skipped_quota 不是失败，不进重试。 +- 抢单：间隔抢单与重试抢单各守各的门（claim_due_automation + 改 last_run_at；claim_retry_automation 只清 next_retry_at、 + 不动间隔时钟），同一执行同一时刻只一个执行器拿到。 +- tick：只列出本轮到期的定义（间隔到期 + 重试到期，单项目 + 单轮上限 5 个），不抢单；执行器对每条分别调 + run_due_automation / run_retry_automation（抢单在入口内）。 +""" + +from __future__ import annotations + +import sqlite3 +import time +from typing import TYPE_CHECKING + +from cloud.automations import ACTIVE +from cloud.automations import AUTOMATION_COLUMNS +from cloud.automations import DEFAULT_MAX_RETRIES +from cloud.automations import PAUSED +from cloud.automations import _audit_row +from cloud.automations import _row_dict +from cloud.automations import automation_to_dict +from cloud.automations import due_automations +from cloud.automations import ensure_automations_tables +from cloud.automations import get_automation +from cloud.automations import row_get +from cloud.automations import run_due_automation + +if TYPE_CHECKING: + from cloud.db import PgConnection + +# 单项目单轮 tick 最多认领的执行数（本项目自定）。 +MAX_CONCURRENT_CRON_RUNS_PER_PROJECT = 5 +# 重试退避上限（秒，本项目自定）。 +RETRY_BACKOFF_CAP_SECONDS = 3600 +# 计入失败重试的执行终态（skipped_quota 不是失败，不进重试）。 +RETRYABLE_STATUSES = ("failed", "interrupted") + + +def retry_delay_seconds(failures: int, backoff_seconds: int) -> int: + """第 failures 次失败后的重试等待秒数：基数 × 2^(n-1)， + 上限 3600 秒。failures < 1 时按 1 算。""" + n = failures if failures >= 1 else 1 + delay = backoff_seconds * (2 ** (n - 1)) + return min(delay, RETRY_BACKOFF_CAP_SECONDS) + + +def _fetch(conn: sqlite3.Connection | PgConnection, + automation_id: str) -> dict | None: + """取单条定义的原始行 dict；不存在回 None。""" + ensure_automations_tables(conn) + row = conn.execute( + "SELECT * FROM automations WHERE id = ?", + (automation_id,)).fetchone() + if row is None: + return None + return _row_dict(row, AUTOMATION_COLUMNS) + + +def pause_automation(conn: sqlite3.Connection | PgConnection, + automation_id: str, + now: float | None = None + ) -> tuple[bool, dict | None]: + """停用自动化：仅 active 的行能停用。 + + 返回 (ok, auto)：行不存在时 (False, None)；已停用/已取消 + 时 (False, 旧 dict)，调用方据此区分 404 与 409。成功时 + commit 并记审计。停用后不到期、不抢单，last_run_at 不动 + （恢复后按原间隔时钟继续，不立即补跑一轮）。""" + if now is None: + now = time.time() + row = _fetch(conn, automation_id) + if row is None: + return False, None + if row_get(row, "status") != ACTIVE: + return False, get_automation(conn, automation_id) + cur = conn.execute( + "UPDATE automations SET status = ?" + " WHERE id = ? AND status = ?", + (PAUSED, automation_id, ACTIVE)) + conn.commit() + if cur.rowcount != 1: + return False, get_automation(conn, automation_id) + _audit_row(conn, row_get(row, "project_id"), + "automation_paused", automation_id, now) + return True, get_automation(conn, automation_id) + + +def resume_automation(conn: sqlite3.Connection | PgConnection, + automation_id: str, + now: float | None = None + ) -> tuple[bool, dict | None]: + """恢复已停用的自动化：仅 paused 的行能恢复。 + + 同项目已有同名 active 定义时恢复失败（False, 旧 dict）： + 名字唯一规则只认 active，停用期间建了同名新定义是允许 + 的，恢复撞名时必须先改名或取消那一个。成功时 commit 并 + 记审计；last_run_at 不动（恢复后若间隔已过，下一轮 tick + 即到期，不立即触发）。""" + if now is None: + now = time.time() + row = _fetch(conn, automation_id) + if row is None: + return False, None + if row_get(row, "status") != PAUSED: + return False, get_automation(conn, automation_id) + clash = conn.execute( + "SELECT id FROM automations WHERE project_id = ?" + " AND name = ? AND status = ? AND id != ?", + (row_get(row, "project_id"), row_get(row, "name"), + ACTIVE, automation_id)).fetchone() + if clash is not None: + return False, get_automation(conn, automation_id) + cur = conn.execute( + "UPDATE automations SET status = ?" + " WHERE id = ? AND status = ?", + (ACTIVE, automation_id, PAUSED)) + conn.commit() + if cur.rowcount != 1: + return False, get_automation(conn, automation_id) + _audit_row(conn, row_get(row, "project_id"), + "automation_resumed", automation_id, now) + return True, get_automation(conn, automation_id) + + +def retry_due_automations(conn: sqlite3.Connection | PgConnection, + project_id: str, + now: float | None = None) -> list[dict]: + """列出某项目当前重试到期的定义（纯读，不抢单）： + active、next_retry_at 已到、失败次数 > 0 且未超 max_retries + （老库 NULL 列按默认值 3 算）。按创建时间排序。""" + if now is None: + now = time.time() + ensure_automations_tables(conn) + rows = conn.execute( + "SELECT * FROM automations WHERE project_id = ?" + " AND status = ? AND next_retry_at IS NOT NULL" + " AND next_retry_at <= ?" + " AND COALESCE(consecutive_failures, 0) > 0" + " AND COALESCE(consecutive_failures, 0)" + " <= COALESCE(max_retries, ?)" + " ORDER BY created_at, id", + (project_id, ACTIVE, now, DEFAULT_MAX_RETRIES)).fetchall() + return [automation_to_dict(r, now) for r in rows] + + +def claim_retry_automation(conn: sqlite3.Connection | PgConnection, + automation_id: str, + now: float | None = None) -> bool: + """重试抢单：重试执行的唯一入口（与间隔抢单分开）。 + + 条件 UPDATE 只有一行能成功；抢到即清 next_retry_at 并写 + last_claimed_at，**不动 last_run_at**（间隔时钟不受重试 + 影响）。返回 True 表示抢到，False 表示不存在、非 active、 + 未到重试时刻、失败次数已超限或已被别的执行器抢走。""" + if now is None: + now = time.time() + ensure_automations_tables(conn) + cur = conn.execute( + "UPDATE automations SET last_claimed_at = ?," + " next_retry_at = NULL" + " WHERE id = ? AND status = ?" + " AND next_retry_at IS NOT NULL AND next_retry_at <= ?" + " AND COALESCE(consecutive_failures, 0) > 0" + " AND COALESCE(consecutive_failures, 0)" + " <= COALESCE(max_retries, ?)", + (now, automation_id, ACTIVE, now, DEFAULT_MAX_RETRIES)) + conn.commit() + return cur.rowcount == 1 + + +def record_automation_outcome(conn: sqlite3.Connection | PgConnection, + automation_id: str, status: str, + now: float | None = None + ) -> dict | None: + """记一次自动化执行的终态，更新失败计数与下次重试时刻。 + + - failed/interrupted：失败次数 +1；未超 max_retries 时 + next_retry_at = now + 退避，超限后清 next_retry_at + （不再重试，间隔时钟照常走）；记审计。 + - completed：失败次数清零、next_retry_at 清空。 + - 其它终态（created/skipped_quota 等）：不改计数。 + 定义不存在时回 None。成功时 commit 并返回最新 dict。""" + if now is None: + now = time.time() + auto = get_automation(conn, automation_id) + if auto is None: + return None + if status in RETRYABLE_STATUSES: + return _record_failure(conn, auto, now) + if status == "completed": + return _record_success(conn, auto, now) + return auto + + +def _record_failure(conn: sqlite3.Connection | PgConnection, + auto: dict, now: float) -> dict | None: + """失败分支：失败次数 +1，未超限则安排下次重试时刻。""" + failures = int(auto["consecutive_failures"]) + 1 + max_retries = int(auto["max_retries"]) + if failures <= max_retries: + delay = retry_delay_seconds( + failures, int(auto["retry_backoff_seconds"])) + next_retry = now + delay + else: + next_retry = None + conn.execute( + "UPDATE automations SET consecutive_failures = ?," + " next_retry_at = ? WHERE id = ?", + (failures, next_retry, auto["id"])) + conn.commit() + _audit_row(conn, auto["project_id"], "automation_run_failed", + "%s: failures=%d next_retry_at=%s" + % (auto["id"], failures, next_retry), now) + return get_automation(conn, auto["id"]) + + +def _record_success(conn: sqlite3.Connection | PgConnection, + auto: dict, now: float) -> dict | None: + """成功分支：失败次数清零、下次重试时刻清空。""" + conn.execute( + "UPDATE automations SET consecutive_failures = 0," + " next_retry_at = NULL WHERE id = ?", + (auto["id"],)) + conn.commit() + return get_automation(conn, auto["id"]) + + +def run_retry_automation(conn: sqlite3.Connection | PgConnection, + automation_id: str, + now: float | None = None) -> dict | None: + """重试执行的唯一入口：重试抢单 -> 配额门 -> 建任务 -> 记产出。 + + 与 run_due_automation 共用配额门与建任务逻辑，只把抢单换 + 成重试抢单；抢不到时回 None。调用方（定时执行器）只调本 + 函数与 run_due_automation，绝不自己拼 claim + INSERT。""" + if now is None: + now = time.time() + return run_due_automation(conn, automation_id, now, + claim_fn=claim_retry_automation) + + +def run_cron_tick(conn: sqlite3.Connection | PgConnection, + project_id: str, + now: float | None = None) -> dict: + """一轮 tick 列出本项目当前到期的定义（纯读，不抢单）： + 间隔到期在前、重试到期在后，单轮总数上限 + MAX_CONCURRENT_CRON_RUNS_PER_PROJECT（本项目自定）。 + + 返回 {"project_id", "due", "retry_due", "capped"}；执行器 + 对 due 的每条调 run_due_automation、对 retry_due 的每条调 + run_retry_automation（抢单在入口内，两个执行器同时 tick + 也只一个能抢到）。""" + if now is None: + now = time.time() + due = due_automations(conn, project_id, now) + retry = retry_due_automations(conn, project_id, now) + cap = MAX_CONCURRENT_CRON_RUNS_PER_PROJECT + kept_due = due[:cap] + kept_retry = retry[:max(0, cap - len(kept_due))] + capped = len(due) + len(retry) > cap + return { + "project_id": project_id, + "due": kept_due, + "retry_due": kept_retry, + "capped": capped, + } diff --git a/cloud/chv/presentation.py b/cloud/chv/presentation.py new file mode 100644 index 0000000..d2fa8cb --- /dev/null +++ b/cloud/chv/presentation.py @@ -0,0 +1,172 @@ +"""P8 presentation：后台任务完成后“是否值得呈现”的评估，与 +通知去重节流账本。 + +对齐 Muse B2（官方设计文）：完成后台工作后评估是否值得呈现， +只在真正有新内容或需要输入时通知用户。 + +判定是确定性规则（不是模型判断）： +- needs_input：需要用户输入 -> 通知。 +- failed/interrupted：失败、需要决策 -> 通知。 +- completed 且有实质新内容（新产物、改动文件、或显式 notify + 标志）-> 通知；否则常规完成只记日志。 +- cancelled / 未知终态 -> 只记日志（未知终态 fail-closed， + 不通知）。 + +账本：state_dir/notification-ledger.json（原子写 + 损坏 +quarantine，orphans 同款口径）；同一任务同一原因在节流窗口 +（3600 秒，本项目自定）内只通知一次。P7 状态存储落地后迁移， +本模块 docstring 与差距表已标注；账本丢了只影响去重（可能 +重复通知一次），不影响任务执行本身。 +""" + +from __future__ import annotations + +import json +import logging +import os +import time + +from cloud.chv.atomic_file import atomic_write_text + +_LOGGER = logging.getLogger("cloud.cloud_hypervisor") + +# 同一任务同一原因的通知节流窗口（秒，本项目自定）。 +NOTIFY_THROTTLE_SECONDS = 3600 +_LEDGER_FILENAME = "notification-ledger.json" +_MAX_LEDGER_ENTRIES = 500 + + +def evaluate_presentation(result: dict) -> dict: + """评估一次后台任务终态是否值得呈现。 + + result 键：status（completed/failed/interrupted/cancelled/ + needs_input）、new_artifacts（list）、changed_files（list）、 + notify（bool，任务显式要求通知）。返回 {"present", "reason"}； + present 为 False 时调用方只记日志、不通知。""" + if not isinstance(result, dict): + return {"present": False, "reason": "unknown_status"} + status = result.get("status") + if status == "needs_input": + return {"present": True, "reason": "needs_input"} + if status in ("failed", "interrupted"): + return {"present": True, "reason": "failed_needs_decision"} + if status == "completed": + return _evaluate_completed(result) + if status == "cancelled": + return {"present": False, "reason": "cancelled"} + return {"present": False, "reason": "unknown_status"} + + +def _evaluate_completed(result: dict) -> dict: + """完成态分支：有实质新内容才通知。""" + artifacts = result.get("new_artifacts") + changed = result.get("changed_files") + has_new = bool(result.get("notify")) + if isinstance(artifacts, list) and artifacts: + has_new = True + if isinstance(changed, list) and changed: + has_new = True + if has_new: + return {"present": True, "reason": "substantive_new_content"} + return {"present": False, "reason": "routine_completed"} + + +def _ledger_path(state_dir: str) -> str: + """通知账本文件路径。""" + return os.path.join(state_dir, _LEDGER_FILENAME) + + +def _quarantine(path: str) -> None: + """把损坏的账本挪到 .corrupt 旁（尽力而为）。""" + try: + os.replace(path, path + ".corrupt") + except OSError: + _LOGGER.warning("chv_ledger_quarantine_failed path=%s", path) + + +def _load_ledger(state_dir: str) -> dict: + """读账本：无文件回空结构；损坏则 quarantine 后回空结构 + （去重信息丢了只会导致一次重复通知，不影响任务本身）。""" + path = _ledger_path(state_dir) + try: + with open(path, "r") as fh: + data = json.load(fh) + except FileNotFoundError: + return {"notified": {}} + except (OSError, ValueError): + _quarantine(path) + return {"notified": {}} + if not isinstance(data, dict) \ + or not isinstance(data.get("notified"), dict): + _quarantine(path) + return {"notified": {}} + return data + + +def _save_ledger(state_dir: str, data: dict) -> None: + """原子落盘账本（截断到上限）；写失败只记日志。""" + notified = data.get("notified", {}) + if len(notified) > _MAX_LEDGER_ENTRIES: + items = sorted(notified.items(), key=lambda kv: kv[1]) + data["notified"] = dict(items[-_MAX_LEDGER_ENTRIES:]) + try: + os.makedirs(state_dir, exist_ok=True) + atomic_write_text(_ledger_path(state_dir), json.dumps(data)) + except OSError: + _LOGGER.warning("chv_ledger_save_failed state_dir=%s", state_dir) + + +class NotificationLedger: + """通知去重节流账本（每任务每原因节流）。 + + should_notify(task_id, reason) 在窗口内未通知过时回 True + 并记下这次通知；窗口内已通知过回 False。账本先走原子 + JSON 文件，待迁移到 P7 状态存储（见模块 docstring）。 + """ + + def __init__(self, state_dir: str) -> None: + """建账本：state_dir 为空即 ValueError。""" + if not state_dir or not isinstance(state_dir, str): + raise ValueError("state_dir required") + self.state_dir = state_dir + + def should_notify(self, task_id: str, reason: str, + now: float | None = None) -> bool: + """该任务该原因现在能不能通知：窗口内没通知过就记下 + 并回 True，否则回 False。task_id/reason 为空即 + ValueError。""" + if now is None: + now = time.time() + if not task_id or not isinstance(task_id, str): + raise ValueError("task_id required") + if not reason or not isinstance(reason, str): + raise ValueError("reason required") + key = "%s:%s" % (task_id, reason) + data = _load_ledger(self.state_dir) + notified = data.get("notified", {}) + last = notified.get(key) + if isinstance(last, (int, float)) \ + and now - last < NOTIFY_THROTTLE_SECONDS: + return False + notified[key] = now + _save_ledger(self.state_dir, data) + return True + + +def present_or_log(result: dict, ledger: NotificationLedger, + task_id: str, + now: float | None = None) -> dict: + """评估 + 去重的一站式入口（worker/push 终态派发处调它）。 + + 返回 {"present", "reason", "notified"}：present 是评估结论， + notified 是本次是否真的发出通知（评估通过但窗口内已通知 + 过时 notified 为 False、只记日志）。""" + if now is None: + now = time.time() + decision = evaluate_presentation(result) + if not decision["present"]: + return {"present": False, "reason": decision["reason"], + "notified": False} + notified = ledger.should_notify(task_id, decision["reason"], now) + return {"present": True, "reason": decision["reason"], + "notified": notified} diff --git a/cloud/chv/swarm.py b/cloud/chv/swarm.py new file mode 100644 index 0000000..0732d90 --- /dev/null +++ b/cloud/chv/swarm.py @@ -0,0 +1,383 @@ +"""P8 swarm：子代理群——建群、派代理、隔离、工具注册与审计。 + +对齐 Muse B3（官方安全文）：启动成群的子代理、构建自己的工具； +VM 需要足够资源处理并发子代理与 crons。资源上限是本项目自定 +（见 docs/chv-muse-p8-gap.md §4），不是 Meta 官方值。 + +隔离模型：每代理一个独立任务 id + 独立工作区（父任务工作区下 +按代理 id 分目录，`resolve_inside` 门守着）；代理之间互不可见 +对方工作区（`check_agent_path` 否定测试钉住）。真正执行在代 +理自己的 VM/单元里（P1 常驻池 + P2 单元是底座），本层只做登 +记、派发与审计，不执行工具代码。 + +存储：state_dir/swarms.json（原子写 + 损坏 quarantine，orphans +同款口径）；P7 状态存储落地后迁移，本模块 docstring 与差距表 +已标注。 +""" + +from __future__ import annotations + +import json +import logging +import os +import time +import uuid +from collections.abc import Callable + +from cloud.chv.atomic_file import atomic_write_text +from cloud.chv.compat215 import resolve_inside + +_LOGGER = logging.getLogger("cloud.cloud_hypervisor") + +MAX_AGENTS_PER_SWARM = 8 +MAX_AGENTS_LIMIT = 32 +MAX_SWARMS_PER_PROJECT = 10 +MAX_TOOL_NAME_LEN = 100 +MAX_AGENT_PROMPT_LEN = 20000 +MAX_AUDIT_ENTRIES = 200 +MAX_INVOCATION_ENTRIES = 200 +AGENT_STATUSES = ("pending", "running", "done", "failed") +_REGISTRY_FILENAME = "swarms.json" + + +def _registry_path(state_dir: str) -> str: + """群登记文件路径。""" + return os.path.join(state_dir, _REGISTRY_FILENAME) + + +def _quarantine(path: str) -> None: + """把损坏的登记文件挪到 .corrupt 旁（尽力而为）。""" + try: + os.replace(path, path + ".corrupt") + except OSError: + _LOGGER.warning("chv_swarm_registry_quarantine_failed " + "path=%s", path) + + +def _empty_registry() -> dict: + """空登记结构。""" + return {"swarms": {}, "tools": {}, "invocations": [], + "audit": []} + + +def _load_registry(state_dir: str) -> dict: + """读登记：无文件回空结构；损坏则 quarantine 后回空结构 + （损坏丢状态是 orphans 同款 fail-safe 口径，见差距表）。""" + path = _registry_path(state_dir) + try: + with open(path, "r") as fh: + data = json.load(fh) + except FileNotFoundError: + return _empty_registry() + except (OSError, ValueError): + _quarantine(path) + return _empty_registry() + if not isinstance(data, dict): + _quarantine(path) + return _empty_registry() + data.setdefault("swarms", {}) + data.setdefault("tools", {}) + data.setdefault("invocations", []) + data.setdefault("audit", []) + return data + + +def _save_registry(state_dir: str, data: dict) -> None: + """原子落盘登记（审计与调用记录各截断到上限）；写失败 + 只记日志，不炸调用方。""" + data["audit"] = data.get("audit", [])[-MAX_AUDIT_ENTRIES:] + data["invocations"] = data.get("invocations", [])[ + -MAX_INVOCATION_ENTRIES:] + try: + os.makedirs(state_dir, exist_ok=True) + atomic_write_text(_registry_path(state_dir), + json.dumps(data)) + except OSError: + _LOGGER.warning("chv_swarm_registry_save_failed " + "state_dir=%s", state_dir) + + +def _audit(data: dict, action: str, detail: str, now: float) -> None: + """记一条审计（调用方负责落盘）。""" + data.setdefault("audit", []).append( + {"action": action, "detail": detail, "created_at": now}) + + +def _get_swarm_or_raise(data: dict, swarm_id: str) -> dict: + """取群，不存在即 ValueError（调用方转 404）。""" + swarm = data.get("swarms", {}).get(swarm_id) + if not isinstance(swarm, dict): + raise ValueError("swarm not found") + return swarm + + +def create_swarm(state_dir: str, project_id: str, + parent_task_id: str, max_agents: int = 8, + now: float | None = None) -> dict: + """建群：每项目群上限 10（本项目自定），每群代理上限 + 1–32（默认 8，本项目自定）。返回群 dict。""" + if now is None: + now = time.time() + if isinstance(max_agents, bool) \ + or not isinstance(max_agents, int): + raise ValueError("max_agents must be an integer") + if not 1 <= max_agents <= MAX_AGENTS_LIMIT: + raise ValueError("max_agents must be between 1 and %d" + % MAX_AGENTS_LIMIT) + if not project_id or not parent_task_id: + raise ValueError("project_id and parent_task_id required") + data = _load_registry(state_dir) + count = sum(1 for s in data["swarms"].values() + if isinstance(s, dict) + and s.get("project_id") == project_id) + if count >= MAX_SWARMS_PER_PROJECT: + raise ValueError("too many swarms in this project") + swarm_id = uuid.uuid4().hex + swarm = { + "id": swarm_id, + "project_id": project_id, + "parent_task_id": parent_task_id, + "max_agents": max_agents, + "status": "active", + "agents": {}, + "created_at": now, + } + data["swarms"][swarm_id] = swarm + _audit(data, "swarm_created", + "%s parent=%s" % (swarm_id, parent_task_id), now) + _save_registry(state_dir, data) + return dict(swarm) + + +def get_swarm(state_dir: str, swarm_id: str) -> dict | None: + """取群；不存在回 None。""" + data = _load_registry(state_dir) + swarm = data.get("swarms", {}).get(swarm_id) + if not isinstance(swarm, dict): + return None + return dict(swarm) + + +def spawn_agent(state_dir: str, swarm_id: str, prompt: str, + workspace_root: str, + now: float | None = None) -> dict: + """派一个子代理：群须 active、人数未满上限；代理得独立 + 任务 id + 独立工作区（父工作区下按代理 id 分目录， + resolve_inside 门守着，越界即 ValueError）。返回代理 dict。""" + if now is None: + now = time.time() + if not prompt or not isinstance(prompt, str) \ + or not prompt.strip(): + raise ValueError("prompt required") + if len(prompt) > MAX_AGENT_PROMPT_LEN: + raise ValueError("prompt too long") + if not workspace_root or not isinstance(workspace_root, str): + raise ValueError("workspace_root required") + data = _load_registry(state_dir) + swarm = _get_swarm_or_raise(data, swarm_id) + if swarm.get("status") != "active": + raise ValueError("swarm is not active") + agents = swarm.setdefault("agents", {}) + if len(agents) >= int(swarm["max_agents"]): + raise ValueError("swarm agent limit reached") + agent_id = uuid.uuid4().hex + workspace = resolve_inside(workspace_root, agent_id) + if workspace is None: + raise ValueError("workspace_root is not usable") + agent = { + "id": agent_id, + "task_id": uuid.uuid4().hex, + "prompt": prompt.strip(), + "workspace": workspace, + "status": "pending", + "created_at": now, + } + agents[agent_id] = agent + _audit(data, "swarm_agent_spawned", + "%s agent=%s" % (swarm_id, agent_id), now) + _save_registry(state_dir, data) + return dict(agent) + + +def list_agents(state_dir: str, swarm_id: str) -> list[dict]: + """列出某群的全部代理（按创建时间排序）。群不存在即 + ValueError。""" + data = _load_registry(state_dir) + swarm = _get_swarm_or_raise(data, swarm_id) + agents = [dict(a) for a in + swarm.get("agents", {}).values() + if isinstance(a, dict)] + agents.sort(key=lambda a: (a.get("created_at") or 0, + a.get("id") or "")) + return agents + + +def agent_workspace(state_dir: str, swarm_id: str, + agent_id: str) -> str | None: + """取某代理的工作区路径；群或代理不存在回 None。""" + data = _load_registry(state_dir) + swarm = data.get("swarms", {}).get(swarm_id) + if not isinstance(swarm, dict): + return None + agent = swarm.get("agents", {}).get(agent_id) + if not isinstance(agent, dict): + return None + workspace = agent.get("workspace") + return workspace if isinstance(workspace, str) else None + + +def check_agent_path(state_dir: str, swarm_id: str, agent_id: str, + path: str) -> bool: + """隔离门：path 是否在该代理自己的工作区内（realpath + + commonpath 校验，符号链接逃逸、.. 逃逸、跨盘符一律 False）。 + 群或代理不存在时同样 False（fail-closed）。""" + workspace = agent_workspace(state_dir, swarm_id, agent_id) + if not workspace or not isinstance(path, str) or not path: + return False + try: + rel = os.path.relpath(path, workspace) + except ValueError: + return False + return resolve_inside(workspace, rel) is not None + + +def _get_agent_or_raise(data: dict, swarm_id: str, + agent_id: str) -> tuple[dict, dict]: + """取（群，代理），任一不存在即 ValueError。""" + swarm = _get_swarm_or_raise(data, swarm_id) + agent = swarm.get("agents", {}).get(agent_id) + if not isinstance(agent, dict): + raise ValueError("agent not found") + return swarm, agent + + +def _refresh_swarm_status(swarm: dict) -> None: + """群终态推导：代理全终态后，全部 done 即 done、有 failed + 即 failed；否则保持 active。""" + agents = [a for a in swarm.get("agents", {}).values() + if isinstance(a, dict)] + if not agents: + return + if any(a.get("status") not in ("done", "failed") + for a in agents): + return + if any(a.get("status") == "failed" for a in agents): + swarm["status"] = "failed" + else: + swarm["status"] = "done" + + +def set_agent_status(state_dir: str, swarm_id: str, agent_id: str, + status: str, + now: float | None = None) -> dict: + """改代理状态（pending/running/done/failed）；非法状态或 + 代理不存在即 ValueError。改完推导群终态，返回代理 dict。""" + if now is None: + now = time.time() + if status not in AGENT_STATUSES: + raise ValueError("invalid agent status") + data = _load_registry(state_dir) + swarm, agent = _get_agent_or_raise(data, swarm_id, agent_id) + agent["status"] = status + _refresh_swarm_status(swarm) + _audit(data, "swarm_agent_status", + "%s agent=%s status=%s" % (swarm_id, agent_id, status), + now) + _save_registry(state_dir, data) + return dict(agent) + + +def register_tool(state_dir: str, swarm_id: str, agent_id: str, + name: str, description: str, + now: float | None = None) -> dict: + """子代理构建自己的工具：登记进群内工具表（名字在该代理 + 内唯一，1–100 字符），带审计。登记只登记，不执行。""" + if now is None: + now = time.time() + if not name or not isinstance(name, str) or not name.strip(): + raise ValueError("tool name required") + name = name.strip() + if len(name) > MAX_TOOL_NAME_LEN: + raise ValueError("tool name too long") + if not isinstance(description, str): + raise ValueError("description must be a string") + data = _load_registry(state_dir) + _get_agent_or_raise(data, swarm_id, agent_id) + tools = data.setdefault("tools", {}) + by_agent = tools.setdefault(swarm_id, {}).setdefault(agent_id, {}) + if name in by_agent: + raise ValueError("tool already registered for this agent") + tool = {"name": name, "description": description, + "agent_id": agent_id, "swarm_id": swarm_id, + "created_at": now} + by_agent[name] = tool + _audit(data, "swarm_tool_registered", + "%s agent=%s tool=%s" % (swarm_id, agent_id, name), now) + _save_registry(state_dir, data) + return dict(tool) + + +def record_tool_invocation(state_dir: str, swarm_id: str, + agent_id: str, name: str, + args_summary: str, + now: float | None = None) -> dict: + """记一次工具调用意图（登记 + 审计）：工具须是该代理自己 + 注册的，别代理的工具名在这里叫不动。 + + 注意：本函数**不执行工具代码**——执行在子代理自己的 VM/ + 单元里（执行器派发代理任务时发生）。这里只留下“谁、在 + 哪群、调了哪个工具、参数摘要”的审计记录。""" + if now is None: + now = time.time() + if not isinstance(args_summary, str): + raise ValueError("args_summary must be a string") + data = _load_registry(state_dir) + _get_agent_or_raise(data, swarm_id, agent_id) + by_agent = data.get("tools", {}).get(swarm_id, {}).get(agent_id, + {}) + if name not in by_agent: + raise ValueError("tool not registered for this agent") + record = { + "id": uuid.uuid4().hex, + "swarm_id": swarm_id, + "agent_id": agent_id, + "name": name, + "args_summary": args_summary[:500], + "created_at": now, + } + data.setdefault("invocations", []).append(record) + _audit(data, "swarm_tool_invoked", + "%s agent=%s tool=%s" % (swarm_id, agent_id, name), now) + _save_registry(state_dir, data) + return dict(record) + + +def dispatch_agent(state_dir: str, swarm_id: str, agent_id: str, + runner: Callable[[dict], dict], + now: float | None = None) -> dict: + """派发代理去执行：代理须是 pending；runner 是调用方注入 + 的执行函数（生产上接执行器 dispatch，测试里传 fake）， + 收到任务 dict {"id", "prompt", "swarm_id", "agent_id", + "workspace"}，返回其执行结果。 + + 派发前把代理置为 running；runner 抛错时代理状态不动 + （调用方按失败处理，不在这里把异常吞成 done）。""" + if now is None: + now = time.time() + data = _load_registry(state_dir) + swarm, agent = _get_agent_or_raise(data, swarm_id, agent_id) + if agent.get("status") != "pending": + raise ValueError("agent is not pending") + task = { + "id": agent["task_id"], + "prompt": agent["prompt"], + "swarm_id": swarm_id, + "agent_id": agent_id, + "workspace": agent["workspace"], + } + agent["status"] = "running" + _audit(data, "swarm_agent_dispatched", + "%s agent=%s task=%s" + % (swarm_id, agent_id, agent["task_id"]), now) + _save_registry(state_dir, data) + return runner(task) diff --git a/cloud/tests/test_chv_muse_p8.py b/cloud/tests/test_chv_muse_p8.py new file mode 100644 index 0000000..c50f869 --- /dev/null +++ b/cloud/tests/test_chv_muse_p8.py @@ -0,0 +1,699 @@ +"""对齐 Muse P8：后台工作（crons / 子代理群 / 回访通知 / +每任务浏览器子代理）单测。 + +不跑真实模型、不联网、不碰真机：临时 SQLite 文件 + 临时 +state_dir + 模块函数直调 + fake transport/runner。 +覆盖：cron 停用/恢复/重试退避/重试到期与抢单/tick 上限/ +定义上限；群建群/派代理/工作区隔离/工具注册与调用/派发； +值得呈现评估/通知账本去重节流；浏览器实例单任务单实例/ +用完即毁。 +""" + +import json +import os +import sqlite3 +import sys +import tempfile +import time +import unittest +from pathlib import Path + +try: + from cloud.automations import cancel_automation + from cloud.automations import create_automation + from cloud.automations import ensure_automations_tables + from cloud.automations import get_automation + from cloud.automations import update_automation + from cloud.chv.browser_tasks import BrowserTransport + from cloud.chv.browser_tasks import create_instance + from cloud.chv.browser_tasks import destroy_instance + from cloud.chv.browser_tasks import get_instance + from cloud.chv.browser_tasks import instance_shots_dir + from cloud.chv.browser_tasks import ping_instance + from cloud.chv.crons import claim_retry_automation + from cloud.chv.crons import pause_automation + from cloud.chv.crons import record_automation_outcome + from cloud.chv.crons import resume_automation + from cloud.chv.crons import retry_delay_seconds + from cloud.chv.crons import retry_due_automations + from cloud.chv.crons import run_cron_tick + from cloud.chv.crons import run_retry_automation + from cloud.chv.presentation import NotificationLedger + from cloud.chv.presentation import evaluate_presentation + from cloud.chv.presentation import present_or_log + from cloud.chv.swarm import check_agent_path + from cloud.chv.swarm import create_swarm + from cloud.chv.swarm import dispatch_agent + from cloud.chv.swarm import get_swarm + from cloud.chv.swarm import list_agents + from cloud.chv.swarm import record_tool_invocation + from cloud.chv.swarm import register_tool + from cloud.chv.swarm import set_agent_status + from cloud.chv.swarm import spawn_agent + from cloud.server import init_db +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.automations import cancel_automation + from cloud.automations import create_automation + from cloud.automations import ensure_automations_tables + from cloud.automations import get_automation + from cloud.automations import update_automation + from cloud.chv.browser_tasks import BrowserTransport + from cloud.chv.browser_tasks import create_instance + from cloud.chv.browser_tasks import destroy_instance + from cloud.chv.browser_tasks import get_instance + from cloud.chv.browser_tasks import instance_shots_dir + from cloud.chv.browser_tasks import ping_instance + from cloud.chv.crons import claim_retry_automation + from cloud.chv.crons import pause_automation + from cloud.chv.crons import record_automation_outcome + from cloud.chv.crons import resume_automation + from cloud.chv.crons import retry_delay_seconds + from cloud.chv.crons import retry_due_automations + from cloud.chv.crons import run_cron_tick + from cloud.chv.crons import run_retry_automation + from cloud.chv.presentation import NotificationLedger + from cloud.chv.presentation import evaluate_presentation + from cloud.chv.presentation import present_or_log + from cloud.chv.swarm import check_agent_path + from cloud.chv.swarm import create_swarm + from cloud.chv.swarm import dispatch_agent + from cloud.chv.swarm import get_swarm + from cloud.chv.swarm import list_agents + from cloud.chv.swarm import record_tool_invocation + from cloud.chv.swarm import register_tool + from cloud.chv.swarm import set_agent_status + from cloud.chv.swarm import spawn_agent + from cloud.server import init_db + +T0 = 1800000000.0 + + +class CronTestCase(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_automations_tables(self.conn) + + def tearDown(self): + self.conn.close() + self.tmp.cleanup() + + def _auto(self, name="a", **kwargs): + kwargs.setdefault("interval_seconds", 3600) + return create_automation(self.conn, "p1", name, "do thing", + "tester", now=T0, **kwargs) + + def test_pause_resume_roundtrip(self): + auto = self._auto() + ok, paused = pause_automation(self.conn, auto["id"], now=T0) + self.assertTrue(ok) + self.assertEqual(paused["status"], "paused") + self.assertFalse(paused["due"]) + ok, resumed = resume_automation(self.conn, auto["id"], now=T0) + self.assertTrue(ok) + self.assertEqual(resumed["status"], "active") + + def test_pause_twice_and_resume_active_fail(self): + auto = self._auto() + self.assertTrue(pause_automation(self.conn, auto["id"], + now=T0)[0]) + ok, _ = pause_automation(self.conn, auto["id"], now=T0) + self.assertFalse(ok) + ok, _ = resume_automation(self.conn, auto["id"], now=T0) + self.assertTrue(ok) + ok, _ = resume_automation(self.conn, auto["id"], now=T0) + self.assertFalse(ok) + + def test_pause_missing_returns_false_none(self): + ok, auto = pause_automation(self.conn, "nope", now=T0) + self.assertFalse(ok) + self.assertIsNone(auto) + + def test_cancel_paused_works(self): + auto = self._auto() + pause_automation(self.conn, auto["id"], now=T0) + ok, cancelled = cancel_automation(self.conn, auto["id"], + now=T0) + self.assertTrue(ok) + self.assertEqual(cancelled["status"], "cancelled") + + def test_resume_name_clash_fails(self): + first = self._auto("same") + pause_automation(self.conn, first["id"], now=T0) + self._auto("same") + ok, _ = resume_automation(self.conn, first["id"], now=T0) + self.assertFalse(ok) + self.assertEqual(get_automation(self.conn, first["id"]) + ["status"], "paused") + + def test_retry_delay_backoff_and_cap(self): + self.assertEqual(retry_delay_seconds(1, 60), 60) + self.assertEqual(retry_delay_seconds(2, 60), 120) + self.assertEqual(retry_delay_seconds(3, 60), 240) + self.assertEqual(retry_delay_seconds(0, 60), 60) + self.assertEqual(retry_delay_seconds(10, 60), 3600) + self.assertEqual(retry_delay_seconds(2, 3600), 3600) + + def test_record_failure_schedules_retry(self): + auto = self._auto() + out = record_automation_outcome(self.conn, auto["id"], + "failed", now=T0) + self.assertEqual(out["consecutive_failures"], 1) + self.assertEqual(out["next_retry_at"], T0 + 60) + out = record_automation_outcome(self.conn, auto["id"], + "failed", now=T0 + 60) + self.assertEqual(out["consecutive_failures"], 2) + self.assertEqual(out["next_retry_at"], T0 + 60 + 120) + + def test_record_success_clears_failures(self): + auto = self._auto() + record_automation_outcome(self.conn, auto["id"], "failed", + now=T0) + out = record_automation_outcome(self.conn, auto["id"], + "completed", now=T0 + 60) + self.assertEqual(out["consecutive_failures"], 0) + self.assertIsNone(out["next_retry_at"]) + + def test_skipped_quota_is_not_a_failure(self): + auto = self._auto() + record_automation_outcome(self.conn, auto["id"], "failed", + now=T0) + out = record_automation_outcome(self.conn, auto["id"], + "skipped_quota", now=T0 + 10) + self.assertEqual(out["consecutive_failures"], 1) + self.assertEqual(out["next_retry_at"], T0 + 60) + + def test_retry_exhausted_after_max_retries(self): + auto = self._auto(max_retries=1) + out = record_automation_outcome(self.conn, auto["id"], + "failed", now=T0) + self.assertEqual(out["next_retry_at"], T0 + 60) + out = record_automation_outcome(self.conn, auto["id"], + "failed", now=T0 + 60) + self.assertEqual(out["consecutive_failures"], 2) + self.assertIsNone(out["next_retry_at"]) + self.assertEqual(retry_due_automations(self.conn, "p1", + now=T0 + 61), []) + + def test_retry_due_boundary(self): + auto = self._auto() + record_automation_outcome(self.conn, auto["id"], "failed", + now=T0) + self.assertEqual(retry_due_automations(self.conn, "p1", + now=T0 + 59), []) + due = retry_due_automations(self.conn, "p1", now=T0 + 60) + self.assertEqual([d["id"] for d in due], [auto["id"]]) + + def test_claim_retry_single_winner_and_clock_untouched(self): + auto = self._auto() + record_automation_outcome(self.conn, auto["id"], "failed", + now=T0) + before = get_automation(self.conn, auto["id"]) + self.assertTrue(claim_retry_automation(self.conn, auto["id"], + now=T0 + 60)) + self.assertFalse(claim_retry_automation(self.conn, auto["id"], + now=T0 + 60)) + after = get_automation(self.conn, auto["id"]) + self.assertEqual(after["last_run_at"], + before["last_run_at"]) + self.assertIsNone(after["next_retry_at"]) + + def test_run_retry_automation_builds_task(self): + auto = self._auto() + record_automation_outcome(self.conn, auto["id"], "failed", + now=T0) + run = run_retry_automation(self.conn, auto["id"], now=T0 + 60) + self.assertIsNotNone(run) + self.assertEqual(run["status"], "created") + self.assertIsNotNone(run["task_id"]) + self.assertIsNone(run_retry_automation(self.conn, auto["id"], + now=T0 + 60)) + + def test_run_retry_before_due_returns_none(self): + auto = self._auto() + record_automation_outcome(self.conn, auto["id"], "failed", + now=T0) + self.assertIsNone(run_retry_automation(self.conn, auto["id"], + now=T0 + 59)) + + def test_tick_caps_at_five_interval_first(self): + for i in range(4): + self._auto("due%d" % i, first_run_at=T0 - 10) + retry_ids = [] + for i in range(2): + auto = self._auto("retry%d" % i, + first_run_at=T0 + 100000) + record_automation_outcome(self.conn, auto["id"], "failed", + now=T0) + retry_ids.append(auto["id"]) + tick = run_cron_tick(self.conn, "p1", now=T0 + 61) + self.assertEqual(len(tick["due"]), 4) + self.assertEqual(len(tick["retry_due"]), 1) + self.assertTrue(tick["capped"]) + # 两重试任务 next_retry_at 与 created_at 完全相同， + # 排序按 id 排，不按创建顺序；只断言属于其中之一。 + self.assertIn(tick["retry_due"][0]["id"], retry_ids) + + def test_tick_empty_when_nothing_due(self): + self._auto(first_run_at=T0 + 100000) + tick = run_cron_tick(self.conn, "p1", now=T0) + self.assertEqual(tick["due"], []) + self.assertEqual(tick["retry_due"], []) + self.assertFalse(tick["capped"]) + + def test_definition_cap_100(self): + for i in range(100): + self._auto("cap%d" % i) + with self.assertRaises(ValueError): + self._auto("cap100") + + def test_update_retry_params(self): + auto = self._auto() + ok, updated = update_automation( + self.conn, auto["id"], + {"max_retries": 5, "retry_backoff_seconds": 120}, + now=T0) + self.assertTrue(ok) + self.assertEqual(updated["max_retries"], 5) + self.assertEqual(updated["retry_backoff_seconds"], 120) + with self.assertRaises(ValueError): + update_automation(self.conn, auto["id"], + {"max_retries": 11}, now=T0) + with self.assertRaises(ValueError): + update_automation(self.conn, auto["id"], + {"retry_backoff_seconds": 5}, now=T0) + + def test_create_retry_params_validated(self): + with self.assertRaises(ValueError): + self._auto("bad1", max_retries=11) + with self.assertRaises(ValueError): + self._auto("bad2", retry_backoff_seconds=4000) + auto = self._auto("ok", max_retries=0, + retry_backoff_seconds=10) + self.assertEqual(auto["max_retries"], 0) + + +class SwarmTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.state_dir = os.path.join(self.tmp.name, "state") + os.makedirs(self.state_dir) + self.root = os.path.join(self.tmp.name, "workspaces") + os.makedirs(self.root) + + def tearDown(self): + self.tmp.cleanup() + + def _swarm(self, max_agents=8): + return create_swarm(self.state_dir, "p1", "parent-1", + max_agents=max_agents, now=T0) + + def test_create_swarm_defaults_and_caps(self): + swarm = self._swarm() + self.assertEqual(swarm["max_agents"], 8) + self.assertEqual(swarm["status"], "active") + with self.assertRaises(ValueError): + create_swarm(self.state_dir, "p1", "parent-1", + max_agents=0, now=T0) + with self.assertRaises(ValueError): + create_swarm(self.state_dir, "p1", "parent-1", + max_agents=33, now=T0) + + def test_project_swarm_cap_10(self): + for _ in range(10): + self._swarm() + with self.assertRaises(ValueError): + self._swarm() + + def test_spawn_agent_own_workspace(self): + swarm = self._swarm() + a1 = spawn_agent(self.state_dir, swarm["id"], "task one", + self.root, now=T0) + a2 = spawn_agent(self.state_dir, swarm["id"], "task two", + self.root, now=T0) + self.assertNotEqual(a1["task_id"], a2["task_id"]) + self.assertNotEqual(a1["workspace"], a2["workspace"]) + self.assertTrue(a1["workspace"].startswith(self.root)) + self.assertEqual(a1["status"], "pending") + + def test_spawn_beyond_max_agents_fails(self): + swarm = self._swarm(max_agents=1) + spawn_agent(self.state_dir, swarm["id"], "one", self.root, + now=T0) + with self.assertRaises(ValueError): + spawn_agent(self.state_dir, swarm["id"], "two", self.root, + now=T0) + + def test_check_agent_path_isolation(self): + swarm = self._swarm() + a1 = spawn_agent(self.state_dir, swarm["id"], "one", + self.root, now=T0) + a2 = spawn_agent(self.state_dir, swarm["id"], "two", + self.root, now=T0) + self.assertTrue(check_agent_path( + self.state_dir, swarm["id"], a1["id"], + os.path.join(a1["workspace"], "out.txt"))) + self.assertTrue(check_agent_path( + self.state_dir, swarm["id"], a1["id"], a1["workspace"])) + self.assertFalse(check_agent_path( + self.state_dir, swarm["id"], a1["id"], a2["workspace"])) + self.assertFalse(check_agent_path( + self.state_dir, swarm["id"], a1["id"], self.root)) + self.assertFalse(check_agent_path( + self.state_dir, swarm["id"], a1["id"], + os.path.join(a1["workspace"], "..", "escape.txt"))) + self.assertFalse(check_agent_path( + self.state_dir, swarm["id"], "nope", a1["workspace"])) + + def test_swarm_status_derived_from_agents(self): + swarm = self._swarm() + a1 = spawn_agent(self.state_dir, swarm["id"], "one", + self.root, now=T0) + a2 = spawn_agent(self.state_dir, swarm["id"], "two", + self.root, now=T0) + set_agent_status(self.state_dir, swarm["id"], a1["id"], + "done", now=T0) + self.assertEqual(get_swarm(self.state_dir, swarm["id"]) + ["status"], "active") + set_agent_status(self.state_dir, swarm["id"], a2["id"], + "failed", now=T0) + self.assertEqual(get_swarm(self.state_dir, swarm["id"]) + ["status"], "failed") + with self.assertRaises(ValueError): + spawn_agent(self.state_dir, swarm["id"], "three", + self.root, now=T0) + + def test_set_agent_status_invalid(self): + swarm = self._swarm() + agent = spawn_agent(self.state_dir, swarm["id"], "one", + self.root, now=T0) + with self.assertRaises(ValueError): + set_agent_status(self.state_dir, swarm["id"], agent["id"], + "nope", now=T0) + with self.assertRaises(ValueError): + set_agent_status(self.state_dir, swarm["id"], "nope", + "done", now=T0) + + def test_list_agents_sorted(self): + swarm = self._swarm() + spawn_agent(self.state_dir, swarm["id"], "one", self.root, + now=T0) + spawn_agent(self.state_dir, swarm["id"], "two", self.root, + now=T0 + 1) + agents = list_agents(self.state_dir, swarm["id"]) + self.assertEqual(len(agents), 2) + self.assertEqual(agents[0]["prompt"], "one") + with self.assertRaises(ValueError): + list_agents(self.state_dir, "nope") + + def test_register_tool_namespaced_per_agent(self): + swarm = self._swarm() + a1 = spawn_agent(self.state_dir, swarm["id"], "one", + self.root, now=T0) + a2 = spawn_agent(self.state_dir, swarm["id"], "two", + self.root, now=T0) + tool = register_tool(self.state_dir, swarm["id"], a1["id"], + "search", "web search", now=T0) + self.assertEqual(tool["name"], "search") + with self.assertRaises(ValueError): + register_tool(self.state_dir, swarm["id"], a1["id"], + "search", "again", now=T0) + register_tool(self.state_dir, swarm["id"], a2["id"], + "search", "same name other agent", now=T0) + with self.assertRaises(ValueError): + register_tool(self.state_dir, swarm["id"], a1["id"], + "x" * 101, "too long", now=T0) + + def test_record_tool_invocation_scoped(self): + swarm = self._swarm() + a1 = spawn_agent(self.state_dir, swarm["id"], "one", + self.root, now=T0) + a2 = spawn_agent(self.state_dir, swarm["id"], "two", + self.root, now=T0) + register_tool(self.state_dir, swarm["id"], a1["id"], + "search", "web search", now=T0) + record = record_tool_invocation( + self.state_dir, swarm["id"], a1["id"], "search", + "q=example", now=T0) + self.assertEqual(record["name"], "search") + with self.assertRaises(ValueError): + record_tool_invocation(self.state_dir, swarm["id"], + a2["id"], "search", "q=x", now=T0) + with self.assertRaises(ValueError): + record_tool_invocation(self.state_dir, swarm["id"], + a1["id"], "nosuch", "q=x", now=T0) + + def test_dispatch_agent_runs_runner_once(self): + swarm = self._swarm() + agent = spawn_agent(self.state_dir, swarm["id"], "do it", + self.root, now=T0) + seen = [] + + def runner(task): + seen.append(task) + return {"status": "ok", "task_id": task["id"]} + + result = dispatch_agent(self.state_dir, swarm["id"], + agent["id"], runner, now=T0) + self.assertEqual(result["status"], "ok") + self.assertEqual(seen[0]["id"], agent["task_id"]) + self.assertEqual(seen[0]["workspace"], agent["workspace"]) + with self.assertRaises(ValueError): + dispatch_agent(self.state_dir, swarm["id"], agent["id"], + runner, now=T0) + + +if __name__ == "__main__": + unittest.main() + + +class PresentationTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.state_dir = os.path.join(self.tmp.name, "state") + os.makedirs(self.state_dir) + self.ledger = NotificationLedger(self.state_dir) + + def tearDown(self): + self.tmp.cleanup() + + def test_needs_input_and_failure_present(self): + self.assertEqual( + evaluate_presentation({"status": "needs_input"}), + {"present": True, "reason": "needs_input"}) + self.assertEqual( + evaluate_presentation({"status": "failed"}), + {"present": True, "reason": "failed_needs_decision"}) + self.assertEqual( + evaluate_presentation({"status": "interrupted"}), + {"present": True, "reason": "failed_needs_decision"}) + + def test_completed_with_new_content_presents(self): + self.assertTrue(evaluate_presentation( + {"status": "completed", + "new_artifacts": ["report.pdf"]})["present"]) + self.assertTrue(evaluate_presentation( + {"status": "completed", + "changed_files": ["a.py"]})["present"]) + self.assertTrue(evaluate_presentation( + {"status": "completed", "notify": True})["present"]) + + def test_routine_completed_and_cancelled_log_only(self): + self.assertEqual( + evaluate_presentation({"status": "completed"}), + {"present": False, "reason": "routine_completed"}) + self.assertEqual( + evaluate_presentation({"status": "completed", + "new_artifacts": [], + "changed_files": []}), + {"present": False, "reason": "routine_completed"}) + self.assertEqual( + evaluate_presentation({"status": "cancelled"}), + {"present": False, "reason": "cancelled"}) + + def test_unknown_status_fail_closed(self): + self.assertEqual( + evaluate_presentation({"status": "weird"}), + {"present": False, "reason": "unknown_status"}) + self.assertEqual( + evaluate_presentation("not-a-dict"), + {"present": False, "reason": "unknown_status"}) + + def test_ledger_dedups_same_reason(self): + self.assertTrue(self.ledger.should_notify("t1", "failed", + now=T0)) + self.assertFalse(self.ledger.should_notify("t1", "failed", + now=T0 + 10)) + self.assertTrue(self.ledger.should_notify("t1", "needs_input", + now=T0 + 10)) + self.assertTrue(self.ledger.should_notify("t2", "failed", + now=T0 + 10)) + + def test_ledger_window_expires(self): + self.assertTrue(self.ledger.should_notify("t1", "failed", + now=T0)) + self.assertFalse(self.ledger.should_notify("t1", "failed", + now=T0 + 3599)) + self.assertTrue(self.ledger.should_notify("t1", "failed", + now=T0 + 3600)) + + def test_ledger_rejects_empty_keys(self): + with self.assertRaises(ValueError): + self.ledger.should_notify("", "failed", now=T0) + with self.assertRaises(ValueError): + self.ledger.should_notify("t1", "", now=T0) + + def test_ledger_corrupt_file_recovers(self): + self.assertTrue(self.ledger.should_notify("t1", "failed", + now=T0)) + with open(os.path.join(self.state_dir, + "notification-ledger.json"), + "w") as fh: + fh.write("{not json") + self.assertTrue(self.ledger.should_notify("t2", "failed", + now=T0)) + + def test_present_or_log_combined(self): + out = present_or_log({"status": "failed"}, self.ledger, + "t1", now=T0) + self.assertEqual(out, {"present": True, + "reason": "failed_needs_decision", + "notified": True}) + out = present_or_log({"status": "failed"}, self.ledger, + "t1", now=T0 + 5) + self.assertTrue(out["present"]) + self.assertFalse(out["notified"]) + out = present_or_log({"status": "completed"}, self.ledger, + "t2", now=T0) + self.assertEqual(out, {"present": False, + "reason": "routine_completed", + "notified": False}) + + +class FakeTransport(BrowserTransport): + """测试用 transport：记录 spawn/close，可注入失败。""" + + def __init__(self): + self.spawned = [] + self.closed = [] + self.fail_spawn = False + self.fail_close = False + + def spawn(self, instance): + if self.fail_spawn: + raise RuntimeError("spawn exploded") + self.spawned.append(instance["instance_id"]) + + def close(self, instance): + if self.fail_close: + raise RuntimeError("close exploded") + self.closed.append(instance["instance_id"]) + + def ping(self, instance): + return instance["instance_id"] in self.spawned \ + and instance["instance_id"] not in self.closed + + +class BrowserTasksTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.state_dir = os.path.join(self.tmp.name, "state") + os.makedirs(self.state_dir) + self.transport = FakeTransport() + + def tearDown(self): + self.tmp.cleanup() + + def test_create_and_get_instance(self): + inst = create_instance(self.state_dir, "task-1", + self.transport, now=T0) + self.assertEqual(inst["status"], "active") + self.assertEqual(inst["task_id"], "task-1") + self.assertEqual(self.transport.spawned, [inst["instance_id"]]) + got = get_instance(self.state_dir, "task-1") + self.assertEqual(got["instance_id"], inst["instance_id"]) + self.assertIsNone(get_instance(self.state_dir, "nope")) + + def test_two_tasks_get_distinct_instances(self): + a = create_instance(self.state_dir, "task-1", self.transport, + now=T0) + b = create_instance(self.state_dir, "task-2", self.transport, + now=T0) + self.assertNotEqual(a["instance_id"], b["instance_id"]) + self.assertNotEqual( + instance_shots_dir(self.state_dir, "task-1"), + instance_shots_dir(self.state_dir, "task-2")) + + def test_second_create_same_task_fails(self): + create_instance(self.state_dir, "task-1", self.transport, + now=T0) + with self.assertRaises(ValueError): + create_instance(self.state_dir, "task-1", self.transport, + now=T0) + + def test_destroy_marks_destroyed_and_rejects_reuse(self): + inst = create_instance(self.state_dir, "task-1", + self.transport, now=T0) + destroyed = destroy_instance(self.state_dir, "task-1", + self.transport, now=T0 + 5) + self.assertEqual(destroyed["status"], "destroyed") + self.assertEqual(destroyed["destroyed_at"], T0 + 5) + self.assertEqual(self.transport.closed, [inst["instance_id"]]) + with self.assertRaises(ValueError): + destroy_instance(self.state_dir, "task-1", self.transport, + now=T0 + 6) + with self.assertRaises(ValueError): + create_instance(self.state_dir, "task-1", self.transport, + now=T0 + 6) + self.assertFalse(ping_instance(self.state_dir, "task-1", + self.transport)) + + def test_destroy_missing_fails(self): + with self.assertRaises(ValueError): + destroy_instance(self.state_dir, "nope", self.transport, + now=T0) + + def test_spawn_failure_records_failed(self): + self.transport.fail_spawn = True + with self.assertRaises(RuntimeError): + create_instance(self.state_dir, "task-1", self.transport, + now=T0) + got = get_instance(self.state_dir, "task-1") + self.assertEqual(got["status"], "failed") + + def test_close_failure_still_destroys(self): + create_instance(self.state_dir, "task-1", self.transport, + now=T0) + self.transport.fail_close = True + destroyed = destroy_instance(self.state_dir, "task-1", + self.transport, now=T0) + self.assertEqual(destroyed["status"], "destroyed") + self.assertIn("close exploded", destroyed["close_error"]) + + def test_ping_active_and_missing(self): + self.assertFalse(ping_instance(self.state_dir, "task-1", + self.transport)) + create_instance(self.state_dir, "task-1", self.transport, + now=T0) + self.assertTrue(ping_instance(self.state_dir, "task-1", + self.transport)) + + def test_shots_dir_inside_state_dir(self): + create_instance(self.state_dir, "task-1", self.transport, + now=T0) + shots = instance_shots_dir(self.state_dir, "task-1") + self.assertTrue(shots.startswith(self.state_dir)) + self.assertIsNone(instance_shots_dir(self.state_dir, "nope")) + + def test_base_transport_raises(self): + base = BrowserTransport() + with self.assertRaises(NotImplementedError): + base.spawn({}) + with self.assertRaises(NotImplementedError): + base.close({}) + with self.assertRaises(NotImplementedError): + base.ping({}) diff --git a/docs/chv-muse-p8-gap.md b/docs/chv-muse-p8-gap.md new file mode 100644 index 0000000..d2997dc --- /dev/null +++ b/docs/chv-muse-p8-gap.md @@ -0,0 +1,109 @@ +# 对齐 Muse P8 差距表：后台工作（crons / 子代理群 / 回访通知 / 每任务浏览器子代理） + +> 先写表再动手。口径分层：= Meta 安全文（security.muse.ai 2026-09-08）与产品 +> 设计文（introducing.muse.ai）原文；= 媒体/评测转述；= +> 二进制拆解与用户 VM 观察。凡是 Meta 未公开的数值（cron 调度粒度、数量上限、失败重试 +> 策略），一律写清是**本项目自定**，不得写成 Meta 官方值。 + +## 1. Muse 官方要求（出处：研究报告 §2.5） + +| # | 要求 | 来源 | +|---|---|---| +| B1 | 关应用后继续工作；按计划与相关事件推进 | 设计文 | +| B2 | 完成后台工作后评估是否“值得呈现”，只在真正有新内容或需要输入时通知用户 | 设计文 | +| B3 | 启动成群的子代理、构建自己的工具、编辑自身；VM 需要足够存储、CPU 与内存来做真正的工作——如编译代理写的代码、开发自定义 Skills、处理并发子代理与 crons | 安全文 | +| B4 | 每个浏览器任务由专门浏览器子代理完成 | 安全文 | +| B5 | 定时任务（crons）与并发子代理并存 | 安全文（只提存在） | +| B6 | cron 的具体调度粒度、数量上限、失败重试策略 | Meta **未公开**——本项目自定，见 §4 | + +**不在本轨官方口径内的东西：** “目标工作器（objective workers）”一词只见于第三方 +VM 观察，未见于官方页面原文，本轨不把它当官方术语用；长期目标的分派与通知步骤 +用本轨的 swarm + presentation 实现，不挂这个名头。 + +## 2. 现状（基线 06c3e17，实测代码为准） + +- **crons，两套现成的、都不完整：** +- `host.py run-due`：host state JSON（name/task/repo/interval/last_run）， +`now - last_run >= interval` 到期即派发 host 任务，跑完才回写 last_run。 +无重试、无失败计数、无启用/停用（只有删定义）。 +- `cloud/automations.py`：控制面 automations 表（SQLite/Postgres）， +`claim_due_automation` 条件 UPDATE 抢单（同一间隔只一个执行器拿到）， +配额门、`automation_runs` 产出记录、分诊收件箱。无重试策略、无 paused +状态、无单项目上限。`MIN_INTERVAL_SECONDS=60`、`MAX_INTERVAL_SECONDS=30d` 已有。 +- **子代理群：没有。** 只有模型层的子代理名（`cloud/harness2.py` +`subagent_names`，prompt 树里的名字），没有执行层的并行派发、没有隔离、 +没有工具注册。 +- **后台回访通知：没有“值得呈现”评估。** `cloud/push.py dispatch_event` +按订阅派发终态（completed/failed/goal-unmet 等）；自动化模块 docstring +已写明“不新写通知代码”。无去重、无节流——同一终态可重复派发。 +- **每任务浏览器子代理：没有。** P2 的 `chv-browser.py` 是常驻进程， +`close` 只关当前页、进程继续跑、等下一次 open；broker 固定转发到 +`/run/chv/browser.sock`。状态跨任务复用。 + +## 3. 改什么（本轨四个新模块 + 一处扩展） + +| 模块 | 内容 | 归属 | +|---|---|---| +| `cloud/chv/crons.py`（新） | 在 `cloud/automations.py` 之上加：启用/停用（paused 状态）、失败重试（指数退避 + 上限）、单项目并发上限与定义上限、tick 驱动（到期 + 重试到期统一认领） | 本轨 | +| `cloud/chv/swarm.py`（新） | 子代理群：建群/派代理/生命周期；每代理独立任务 id + 独立工作区（`resolve_inside` 门，越界拒绝）；工具注册与审计（只登记与审计调用意图，执行在代理自己的 VM 里走执行器）；上限可配 | 本轨 | +| `cloud/chv/presentation.py`（新） | “值得呈现”评估（需要输入/失败需决策/有实质新内容才通知，常规完成只记日志）+ 通知账本（每任务每原因节流去重） | 本轨 | +| `cloud/chv/browser_tasks.py`（新） | 每浏览器任务一个实例：建/绑/毁生命周期、单实例单任务（复用即错、毁后复用即错）、每实例独立状态目录；transport 接口注入（测试用 fake） | 本轨 | +| `cloud/automations.py`（改） | 只加重试/停用所需的列与抢单入口（`max_retries`、`retry_backoff_seconds`、`consecutive_failures`、`next_retry_at`、`paused` 状态）；host.py run-due **不动**（host 侧唯一执行器，见该模块 docstring） | 本轨 | + +**P7 状态落点（更新 2026-11-07 06:39）：** P7 已完成并提交（分支 +`feat-chv-muse-p7`，提交 `29fa0c8`：每 VM 一个 SQLite 状态库 +AppStateStore，0600 WAL，tasks/goals/sessions/只追加 audit/meta）。 +本轨仍先用 `state_dir` 下的原子 JSON 文件（复用 +`cloud/chv/atomic_file.py`），模块 docstring 与本表明确标注“待迁移到 +P7 状态存储”；不重复造一套状态库。集成轨（round-12）把 +`swarms.json` / `notification-ledger.json` / `browser-tasks.json` +迁到 AppStateStore，接口（load/save/原子写）已按同一模式写好，迁移 +只是换后端。crons 与 swarm 的状态复用 `automations` 表与 swarm 登记 +文件（同款原子写），同样标注待迁移。 + +**客机侧跟进（本轨只写规约、不改 P2 的文件）：** 每任务浏览器实例的客机 +进程拉起/销毁在 P2 的 `chv-guest/chv-broker.py`（`browser` 操作固定转发到 +单例 socket）里改：每浏览器任务起一个全新的 `chv-browser.py`（每任务 +socket），`close` 后杀掉进程。P2 的文件不在本分支基线里，本轨不复制、不 +分叉，规约写进本表 §6，留给 round-12 集成时落在 P2 分支上。 + +## 4. 自定数值清单（全部是本项目自定，非 Meta 官方值） + +| 数值 | 本项目自定值 | 说明 | +|---|---|---| +| cron 最小间隔 | 60 秒（沿用现有 `MIN_INTERVAL_SECONDS`） | 已有，不新定 | +| cron 最大间隔 | 30 天（沿用现有 `MAX_INTERVAL_SECONDS`） | 已有，不新定 | +| 失败重试默认次数 | 3 次 | 超过即停，不再重试 | +| 重试退避 | 基数 60 秒 × 2^(n-1)，上限 3600 秒 | n 为第几次重试 | +| 单项目并发 cron 执行 | 5 | tick 单轮最多认领 5 个 | +| 单项目自动化定义上限 | 100 | 超过建定义即 ValueError | +| 每群子代理上限 | 8 | 建群时可配，1–32 | +| 每项目群上限 | 10 | 超过建群即 ValueError | +| 通知节流窗口 | 同一任务同一原因 3600 秒内只通知一次 | 账本持久化 | +| 工具名长度上限 | 100 字符 | 与自动化名上限同款 | + +## 5. 诚实边界 + +- B3 的“VM 需要足够资源”是容量要求，不是本轨能代码保证的：本轨只做上限与 +隔离，宿主容量仍由部署方保障；群上限 8 是防炸宿主的默认值，不是性能承诺。 +- swarm 层**不执行工具**：工具调用意图只登记与审计，真正执行在子代理自己的 +VM/单元里（P1 常驻池 + P2 单元是底座）。“子代理构建自己的工具”在本轨指 +注册进群内工具表、可被审计、可被别的代理复用调用名——不是沙箱外任意执行。 +- “实质新内容”的判定是确定性规则（新产物/新改动文件/显式 notify 标志）， +不是模型判断；模型觉得重要但规则判无新内容时，以规则为准、只记日志。 +- cron 重试只针对 failed/interrupted 的执行；超配额跳过（skipped_quota） +不是失败，不进重试（沿用 automations 模块现有口径）。 +- 通知账本先走原子 JSON 文件，P7 存储（`feat-chv-muse-p7` 提交 +`29fa0c8` 的 AppStateStore）已落地，集成轨再迁移；迁移前账本丢了只 +影响去重（可能重复通知一次），不影响任务执行本身。 + +## 6. 真机待验证清单（round-11 与 P4/P5/P6 之后，等通知） + +1. cron 端到端：定义在测试机上按间隔触发、失败重试 3 次后停、停用后不再触发。 +2. 群端到端：3 个子代理在各自 VM 里并行跑、互不可见对方工作区。 +3. 回访通知端到端：后台任务完成后有新内容才通知、无新内容不通知、同一原因 +1 小时内不重复通知。 +4. 每任务浏览器实例端到端：连续两个浏览器任务各起全新实例、第二个看不到 +第一个的截图与快照状态（需 §3 客机侧跟进先落地）。 +5. 账本迁移：P7 存储已落地（AppStateStore，`feat-chv-muse-p7` +提交 `29fa0c8`），集成轨把通知账本从 JSON 文件迁过去、去重语义不变。