diff --git a/cloud/approvals.py b/cloud/approvals.py new file mode 100644 index 0000000..5396cab --- /dev/null +++ b/cloud/approvals.py @@ -0,0 +1,302 @@ +"""运行时审批队列(细粒度审批,P1 第 4 项)。 + +选型:只打通 elicitation 一类,其他审批类全部记为遗留。 + +选 elicitation 的理由: +- 它是 host.py 里唯一已经特判的 server_request 类,未来 hook + 的落点就是 _answer_server_request 的 elicitation 分支,加法 + 可接,不用先修一条错的通用回包。 +- 决定语义最干净:二值,批准 = accept(content 为空对象, + 与今天 approval=auto 时的行为完全一致),拒绝 = decline, + 不需要像 permissions 那样现拼权限 profile,也不需要像 + file_change 的 cancel 那样附带“中断整轮”的副作用。 +- 触发方是 MCP 服务器向用户要表单输入,正是“第三方工具向 + 用户要东西”的场景,用户在控制台点允许或拒绝最符合直觉。 + +关键语义(先定清楚,再写代码): +- 待审批期间原流程挂起等待,不先拒绝后补。队列里的 pending + 项本身不做任何放行或拦截动作,它只是等一个决定。 +- 等待有超时:默认 3600 秒,可在创建时改(服务端入口只接受 + 60 到 86400 秒)。超时默认是拒绝,绝不是放行:过期扫荡把 + pending 置为 expired,同时把 effective decision 记为 denied, + decided_by 记为 system-timeout,回执里看得见。 +- 批准放行、拒绝拦截:决定一旦写入就是终态,再次决定返回 + already_decided,原决定绝不被覆盖。回执字段:decided_by、 + decided_at、decision,批准与拒绝、过期都留痕。 +- 决定是 CAS:只有 pending 行能被改状态,并发下两个决定只 + 有一个能赢,输的那个拿到 already_decided 而不是假成功。 + +诚实边界(本片做到哪、没做到哪): +- 本片交付的是队列服务本身:建表迁移、创建、查询、批准、 + 拒绝、过期扫荡、回执、服务端端点、控制台 UI、单测。 +- 本片没有把实时 elicitation 接到队列上,host.py 完全没动。 + 原因不是协议不让挂起:JSON-RPC 请求本来就可以晚回包, + 挂起在协议上是成立的。真正缺的是通道:host.py 跑在 worker + 拉起的子进程里,手头没有云 DB 连接,也没有到控制面的 + 带权请求通道(worker.py 本片禁动,没法给它递 DB 路径或 + token)。在没有通道的情况下往 _answer_server_request 里 + 加“等决定”只会把回合无限挂死,所以不做假拦截。 +- 在通道落地之前,实时 elicitation 的行为与今天完全一致: + approval=auto 时 accept 空 content,其他一律 decline。 + 队列里的决定目前是记录加事后审计口径,供通道落地后的 + hook 消费,绝不要对外宣称“实时请求已被事前拦截”。 +- 通道落地时 host 侧要做的事只有两件,都已在本模块留好 + 映射函数:elicitation 到达时先建 pending 项再挂起等决定, + 决定到了按 elicitation_response 回包;超时按 expired 的 + denied 处理。通道本身(worker 递 DB 路径还是控制面 HTTP + 带权轮询,生产是 Postgres 所以 DB 直连只够 SQLite 开发 + 环境用)是合流阶段的单独设计,不在本片里猜一个。 + +遗留的审批类(明确不塞进本片): +- item/commandExecution/requestApproval:命令审批。现 host + 通用回包写的是 approve/deny,但新协议的决定词表是 + accept/decline/acceptForSession 等,通用回包本身就是错的, + 要先修词表再接队列;kind 里已预留 command,映射函数 + command_response 已按新协议写好,但不接实时流。 +- item/fileChange/requestApproval:文件改动审批。决定里有 + cancel(拒绝并立即中断整轮),不是纯二值,接队列要先定 + cancel 的轮次语义,暂不接。 +- item/permissions/requestApproval:权限审批。回包不是二值, + 而是一份 GrantedPermissionProfile,要现拼 profile,暂不接。 +- item/tool/requestUserInput:这根本不是审批,是向用户要 + 输入内容(问题列表),blocking 与否另有语义,应另起输入 + 队列,暂不接。 +- item/tool/call 与 account/chatgptAuthTokens/refresh:前者是 + 动态工具执行,后者是凭证刷新,都不是用户审批,排除。 + +本模块只依赖标准库,不 import cloud 其他模块,避免循环 +import;服务端与未来 worker/hook 共用。占位符用 ?,建表与 +列补齐写法与 peer_messages 同款(db 适配层负责翻译)。 +""" + +import json +import time +import uuid + +KINDS = ("elicitation", "command", "file_change", "permissions", + "user_input") +DECISIONS = ("approved", "denied") +DEFAULT_TIMEOUT_SECONDS = 3600 +MAX_SUMMARY_CHARS = 2000 +MAX_DETAIL_CHARS = 20000 + + +def row_get(row, name): + """行取值容错:sqlite3.Row 缺列报 IndexError、dict 缺键报 + KeyError、空行报 TypeError,一律回 None(与 peer_messages + 的同名函数同款,模块独立所以这里自带一份)。""" + try: + return row[name] + except (KeyError, IndexError, TypeError): + return None + + +def ensure_approvals_table(conn): + """幂等建审批队列表。调用方负责 commit(服务端入口在写 + 操作后统一 commit;本函数自己不 commit,避免半截事务)。""" + conn.execute( + "CREATE TABLE IF NOT EXISTS task_approvals(" + "id TEXT PRIMARY KEY, " + "task_id TEXT, " + "project_id TEXT, " + "kind TEXT, " + "summary TEXT, " + "detail TEXT, " + "status TEXT, " + "requested_by TEXT, " + "decided_by TEXT, " + "decided_at DOUBLE PRECISION, " + "decision TEXT, " + "created_at DOUBLE PRECISION, " + "updated_at DOUBLE PRECISION, " + "expires_at DOUBLE PRECISION)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_approvals_task" + " ON task_approvals(task_id, status)") + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_approvals_project" + " ON task_approvals(project_id, created_at)") + + +def _detail_to_str(detail): + if detail is None: + return None + if isinstance(detail, str): + text = detail + else: + try: + text = json.dumps(detail, ensure_ascii=False) + except (TypeError, ValueError): + text = str(detail) + if len(text) > MAX_DETAIL_CHARS: + raise ValueError("detail too long") + return text + + +def _detail_from_str(text): + if text is None: + return None + try: + return json.loads(text) + except (TypeError, ValueError): + return text + + +def approval_to_dict(row): + """行转 dict:回执字段原样带出,detail 尽量还原成对象。""" + if row is None: + return None + return { + "id": row_get(row, "id"), + "task_id": row_get(row, "task_id"), + "project_id": row_get(row, "project_id"), + "kind": row_get(row, "kind"), + "summary": row_get(row, "summary"), + "detail": _detail_from_str(row_get(row, "detail")), + "status": row_get(row, "status"), + "requested_by": row_get(row, "requested_by"), + "decided_by": row_get(row, "decided_by"), + "decided_at": row_get(row, "decided_at"), + "decision": row_get(row, "decision"), + "created_at": row_get(row, "created_at"), + "updated_at": row_get(row, "updated_at"), + "expires_at": row_get(row, "expires_at"), + } + + +def create_approval(conn, task_id, project_id, kind="elicitation", + summary="", detail=None, requested_by=None, + timeout_seconds=DEFAULT_TIMEOUT_SECONDS, now=None): + """创建一条待审批项,返回 dict。非法输入抛 ValueError, + 调用方据此回 400;成功时 commit。 + + timeout_seconds:None 用默认值;0 表示不过期(只允许 + 模块级调用,服务端入口另有限制);负数非法。""" + if kind not in KINDS: + raise ValueError("kind must be one of: " + ", ".join(KINDS)) + if not isinstance(summary, str) or not summary.strip(): + raise ValueError("summary required") + summary = summary.strip() + if len(summary) > MAX_SUMMARY_CHARS: + raise ValueError("summary too long") + if timeout_seconds is None: + timeout_seconds = DEFAULT_TIMEOUT_SECONDS + if not isinstance(timeout_seconds, (int, float)) \ + or isinstance(timeout_seconds, bool) \ + or timeout_seconds < 0: + raise ValueError("timeout_seconds invalid") + detail_str = _detail_to_str(detail) + now = time.time() if now is None else now + expires_at = None + if timeout_seconds: + expires_at = now + timeout_seconds + item_id = uuid.uuid4().hex + ensure_approvals_table(conn) + conn.execute( + "INSERT INTO task_approvals(id, task_id, project_id," + " kind, summary, detail, status, requested_by," + " decided_by, decided_at, decision," + " created_at, updated_at, expires_at)" + " VALUES(?, ?, ?, ?, ?, ?, 'pending', ?," + " NULL, NULL, NULL, ?, ?, ?)", + (item_id, task_id, project_id, kind, summary, detail_str, + requested_by, now, now, expires_at), + ) + conn.commit() + return approval_to_dict(conn.execute( + "SELECT * FROM task_approvals WHERE id = ?", + (item_id,)).fetchone()) + + +def get_approval(conn, approval_id): + """取一条,缺失回 None(不抛,让调用方定 404 还是 409)。""" + ensure_approvals_table(conn) + return approval_to_dict(conn.execute( + "SELECT * FROM task_approvals WHERE id = ?", + (approval_id,)).fetchone()) + + +def list_approvals(conn, task_id=None, project_id=None, status=None): + """列审批项:纯读,不扫荡过期(扫荡是写操作,调用方在 + 需要时先调 expire_overdue)。过滤全是可选的精确匹配。""" + ensure_approvals_table(conn) + query = "SELECT * FROM task_approvals WHERE 1 = 1" + args = [] + if task_id is not None: + query += " AND task_id = ?" + args.append(task_id) + if project_id is not None: + query += " AND project_id = ?" + args.append(project_id) + if status is not None: + query += " AND status = ?" + args.append(status) + query += " ORDER BY created_at, id" + rows = conn.execute(query, args).fetchall() + return [approval_to_dict(r) for r in rows] + + +def decide_approval(conn, approval_id, decision, decided_by): + """写决定。返回 (status, item):decided 表示本次写赢; + already_decided 表示该项已有终态(批准、拒绝或过期), + 原决定绝不被覆盖;not_found 表示无此项。 + + 决定是 CAS:UPDATE 只改 status='pending' 的行,rowcount + 为 0 说明并发下已被别人先决定,调用方按 already_decided + 处理,不要按成功回。成功与过期扫荡都 commit。""" + if decision not in DECISIONS: + raise ValueError("decision must be approved or denied") + ensure_approvals_table(conn) + item = get_approval(conn, approval_id) + if item is None: + return "not_found", None + if item["status"] != "pending": + return "already_decided", item + now = time.time() + cur = conn.execute( + "UPDATE task_approvals SET status = ?, decision = ?," + " decided_by = ?, decided_at = ?, updated_at = ?" + " WHERE id = ? AND status = 'pending'", + (decision, decision, decided_by, now, now, approval_id)) + conn.commit() + if cur.rowcount == 1: + return "decided", get_approval(conn, approval_id) + return "already_decided", get_approval(conn, approval_id) + + +def expire_overdue(conn, now=None): + """过期扫荡:把已到 expires_at 的 pending 项置为 expired, + 同时把 effective decision 记为 denied(超时默认拒绝,绝不 + 放行),decided_by 记为 system-timeout。返回过期条数。""" + ensure_approvals_table(conn) + now = time.time() if now is None else now + cur = conn.execute( + "UPDATE task_approvals SET status = 'expired'," + " decision = 'denied', decided_by = 'system-timeout'," + " decided_at = ?, updated_at = ?" + " WHERE status = 'pending' AND expires_at IS NOT NULL" + " AND expires_at <= ?", + (now, now, now)) + conn.commit() + return cur.rowcount + + +def elicitation_response(decision): + """未来 host hook 的回包映射(本片未接实时流,只定映射): + 批准 = accept 空 content,与今天 approval=auto 的行为一致; + 拒绝或过期 = decline。decision 取 approved/denied/expired + 之外的任何值都按拒绝处理,绝不默认放行。""" + if decision == "approved": + return {"action": "accept", "content": {}} + return {"action": "decline"} + + +def command_response(decision): + """未来命令审批 hook 的回包映射,按新协议词表写(accept + / decline),不是 host.py 现通用回包的 approve/deny——后者 + 与新协议不符,见模块 docstring 的遗留清单。非批准一律 + decline,绝不默认放行。""" + if decision == "approved": + return {"decision": "accept"} + return {"decision": "decline"} diff --git a/cloud/automations.py b/cloud/automations.py new file mode 100644 index 0000000..4c21cf1 --- /dev/null +++ b/cloud/automations.py @@ -0,0 +1,525 @@ +"""自动化与分诊收件箱(P1 第 2 项)。 + +与 host.py 现有 automations / run-due 的关系(先说清楚,避免 +两套调度打架): + +- host.py 的自动化存在 host state JSON 文件里(字段 name / + task / repo / interval / last_run),run-due 按 + now - last_run >= interval 判断到期,到期派发一个 host 任务 + 跑一轮,跑完才回写 last_run。daemon tick 与 crontab 都只 + 调这一个 run-due,它是 host 侧唯一的执行器。 +- 本模块是云端控制面的另一套登记口径:自动化定义存在控制面 + 数据库里(automations 表,带 project_id),不是 host state + 文件。两者存储不相交,同一个自动化绝不会同时出现在两边, + 所以本批不存在双重执行。 +- 本批不给云端自动化加执行器、不动 host.py 的 run-due 区: + host 的 run-due 跑的是 host 任务模型(codex 协议线程),云 + 端任务模型(POST /tasks 建、worker 认领)是另一条链路,把 + 云端定义硬塞进 host run-due 会造出第二个调度器。云端执行 + 器是后续一步,它必须走 claim_due_automation 先抢单(条件 + UPDATE,rowcount 为 1 才算抢到),抢到后再建云端任务、再 + 记 automation_runs 一行;抢不到的执行器绝不能建任务。这 + 个 claim 门就是防重复执行的唯一关口,比 host run-due 的 + “跑完才写 last_run”更严(中途被杀也不会在当间隔内重跑, + 代价是崩掉的那次要等到下个间隔,见 claim 文档)。 +- 分诊收件箱是只读视图:到期自动化、失败或中断的任务、待 + 处理的后续消息、最近的自动化产出记录。它不跑模型、不改 + 任务状态,只把待处理项聚到一处。 + +通知接入(本批结论:遗留,不新写通知代码): + +- server 没有接 webhooks.NotificationService(它只有可注入 + sender 的占位实现,server docstring 也写明未接成);真正 + 的终态派发是 worker 走 cloud/push.py dispatch_event(订阅 + 制,completed / failed / goal-unmet 等终态才发)。 +- 自动化本身没有终态事件可派:定义创建不是完成,取消不是 + 失败;硬在创建/取消时发通知是伪造成功。本批只记审计 + (automation_created / automation_cancelled)。等执行器 + 落地后,它建的云端任务自然走 worker 现有终态派发(任务 + id 记在 automation_runs.task_id 里,派发与任务绑定,不用 + 为自动化另起一套通知系统)。 + +本模块只依赖标准库,不 import cloud 其它模块,避免循环 +import;表迁移写法仿 peer_messages(CREATE TABLE IF NOT +EXISTS + PRAGMA 查列补齐,SQLite 与 Postgres 适配层同款)。 +""" + +import time +import uuid + +MIN_INTERVAL_SECONDS = 60 +MAX_INTERVAL_SECONDS = 30 * 86400 +MAX_NAME_LEN = 100 +MAX_PROMPT_LEN = 20000 +MAX_REPO_URL_LEN = 1000 +TRIAGE_LIMIT = 50 + +ACTIVE = "active" +CANCELLED = "cancelled" + + +def row_get(row, name): + """行取值容错:sqlite3.Row 缺列报 IndexError、dict 缺键报 + KeyError、空行报 TypeError,一律回 None(老库行缺列时 + 读自动化列表不能因此炸掉整页)。""" + try: + return row[name] + except (KeyError, IndexError, TypeError): + return None + + +AUTOMATION_COLUMNS = ( + "id", "project_id", "name", "prompt", "repo_url", + "interval_seconds", "status", "last_run_at", "last_claimed_at", + "created_by", "created_at", "cancelled_at") +RUN_COLUMNS = ( + "id", "automation_id", "project_id", "task_id", "status", + "detail", "created_at") + + +def _row_dict(row, columns): + """把行统一成可按名取值的形态:dict 与 sqlite3.Row 原样 + 用;裸 sqlite3 连接给的是 tuple(按 SELECT 列顺序), + 按建表列顺序 zip 成 dict——直接调模块的调用方不一定设 + row_factory,模块不能因此在取 row[id] 处炸掉。""" + if row is None or isinstance(row, dict): + return row + try: + row["id"] + return row + except TypeError: + 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_automations_tables(conn): + """幂等建自动化表与产出记录表,并补齐新增列。 + + automations:一条定义;status 只取 active / cancelled, + 取消是软删除(行保留、审计可查,名字可重用)。 + automation_runs:一次执行产出;task_id 为空表示执行器 + 抢单后尚未建任务(只应短暂存在,执行器建任务后补记)。 + 调用方在写成功后 commit(server 的 handler 与 peer 模式 + 同款:建表后由业务写操作一并 commit)。""" + conn.execute( + "CREATE TABLE IF NOT EXISTS automations(" + "id TEXT PRIMARY KEY, " + "project_id TEXT, " + "name TEXT, " + "prompt TEXT, " + "repo_url TEXT, " + "interval_seconds INTEGER, " + "status TEXT, " + "last_run_at DOUBLE PRECISION, " + "last_claimed_at DOUBLE PRECISION, " + "created_by TEXT, " + "created_at DOUBLE PRECISION, " + "cancelled_at DOUBLE PRECISION)" + ) + conn.execute( + "CREATE TABLE IF NOT EXISTS automation_runs(" + "id TEXT PRIMARY KEY, " + "automation_id TEXT, " + "project_id TEXT, " + "task_id TEXT, " + "status TEXT, " + "detail TEXT, " + "created_at DOUBLE PRECISION)" + ) + existing = _existing_columns(conn, "automations") + for col, ddl in ( + ("repo_url", "TEXT"), + ("last_claimed_at", "DOUBLE PRECISION"), + ("cancelled_at", "DOUBLE PRECISION")): + if col not in existing: + try: + conn.execute( + "ALTER TABLE automations ADD COLUMN %s %s" + % (col, ddl)) + except Exception: + pass + existing_runs = _existing_columns(conn, "automation_runs") + for col, ddl in ( + ("task_id", "TEXT"), + ("detail", "TEXT")): + if col not in existing_runs: + try: + conn.execute( + "ALTER TABLE automation_runs ADD COLUMN %s %s" + % (col, ddl)) + except Exception: + pass + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_automations_project" + " ON automations(project_id, status)") + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_automation_runs" + " ON automation_runs(automation_id, created_at)") + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_automation_runs_project" + " ON automation_runs(project_id, created_at)") + + +def next_run_at(auto, now=None): + """下次到期时刻:从未跑过时就是创建时间(建完即到期, + 执行器下一轮即可认领);跑过时为上次执行 + 间隔。""" + last = row_get(auto, "last_run_at") + interval = row_get(auto, "interval_seconds") or 0 + if last is None: + return row_get(auto, "created_at") + return last + interval + + +def is_due(auto, now=None): + """是否到期:仅 active 的定义可到期;now 缺省取当前时间。 + 到期边界用 >=,与 host run-due 的 now - last_run >= + interval 同款(含边界那一秒)。""" + if row_get(auto, "status") != ACTIVE: + return False + if now is None: + now = time.time() + nxt = next_run_at(auto) + if nxt is None: + return True + return now >= nxt + + +def automation_to_dict(row, now=None): + row = _row_dict(row, AUTOMATION_COLUMNS) + nxt = next_run_at(row) + return { + "id": row["id"], + "project_id": row["project_id"], + "name": row["name"], + "prompt": row["prompt"], + "repo_url": row_get(row, "repo_url"), + "interval_seconds": row["interval_seconds"], + "status": row["status"], + "last_run_at": row_get(row, "last_run_at"), + "last_claimed_at": row_get(row, "last_claimed_at"), + "created_by": row["created_by"], + "created_at": row["created_at"], + "cancelled_at": row_get(row, "cancelled_at"), + "next_run_at": nxt, + "due": is_due(row, now), + } + + +def _validate_fields(name, prompt, interval_seconds, repo_url): + 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") + 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 repo_url is not None: + if not isinstance(repo_url, str): + raise ValueError("repo_url must be a string") + repo_url = repo_url.strip() or None + if repo_url is not None and len(repo_url) > MAX_REPO_URL_LEN: + raise ValueError("repo_url too long") + return name, prompt, interval_seconds, repo_url + + +def create_automation(conn, project_id, name, prompt, created_by, + interval_seconds=86400, repo_url=None, now=None): + """创建自动化定义:同项目内 active 的名字唯一(取消过的 + 名字可重用,与环境名唯一规则同款语义)。成功时 commit + 并返回 dict;校验失败抛 ValueError(调用方转 400)。""" + name, prompt, interval_seconds, repo_url = _validate_fields( + name, prompt, interval_seconds, repo_url) + if now is None: + now = time.time() + ensure_automations_tables(conn) + existing = conn.execute( + "SELECT id FROM automations WHERE project_id = ?" + " AND name = ? AND status = ?", + (project_id, name, ACTIVE)).fetchone() + if existing is not None: + raise ValueError("automation name already exists 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)" + " VALUES(?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?, ?, NULL)", + (auto_id, project_id, name, prompt, repo_url, + interval_seconds, ACTIVE, created_by, now)) + conn.commit() + row = conn.execute( + "SELECT * FROM automations WHERE id = ?", (auto_id,)).fetchone() + return automation_to_dict(row, now) + + +def list_automations(conn, project_id, now=None): + """列出某项目全部自动化定义(含已取消,取消的 due 为 + False),按创建时间排序。调用方须先 ensure 表。""" + if now is None: + now = time.time() + ensure_automations_tables(conn) + rows = conn.execute( + "SELECT * FROM automations WHERE project_id = ?" + " ORDER BY created_at, id", + (project_id,)).fetchall() + return [automation_to_dict(r, now) for r in rows] + + +def get_automation(conn, automation_id): + """取单条定义;不存在回 None(调用方转 404)。""" + ensure_automations_tables(conn) + row = conn.execute( + "SELECT * FROM automations WHERE id = ?", (automation_id,) + ).fetchone() + if row is None: + return None + return automation_to_dict(row) + + +def cancel_automation(conn, automation_id, now=None): + """取消自动化(软删除):仅 active 的行能取消。 + + 返回 (ok, auto):ok 为 True 时 auto 是取消后的 dict; + 行不存在时 (False, None);已取消时 (False, 旧 dict), + 调用方据此区分 404 与 409。成功时 commit。""" + if now is None: + now = time.time() + ensure_automations_tables(conn) + row = conn.execute( + "SELECT * FROM automations WHERE id = ?", (automation_id,) + ).fetchone() + if row is None: + return False, None + row = _row_dict(row, AUTOMATION_COLUMNS) + if row["status"] != ACTIVE: + 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)) + conn.commit() + if cur.rowcount != 1: + return False, automation_to_dict(row, now) + row = conn.execute( + "SELECT * FROM automations WHERE id = ?", (automation_id,) + ).fetchone() + return True, automation_to_dict(row, now) + + +def due_automations(conn, project_id, now=None): + """列出某项目当前到期的 active 定义(纯读,不抢单)。 + SQL 层算到期:从未跑过(last_run_at 为空)或上次执行距 + 今已满间隔;与 is_due 的 Python 算法同款。""" + if now is None: + now = time.time() + ensure_automations_tables(conn) + rows = conn.execute( + "SELECT * FROM automations WHERE project_id = ?" + " AND status = ?" + " AND (last_run_at IS NULL" + " OR last_run_at + interval_seconds <= ?)" + " ORDER BY created_at, id", + (project_id, ACTIVE, now)).fetchall() + return [automation_to_dict(r, now) for r in rows] + + +def claim_due_automation(conn, automation_id, now=None): + """抢单:未来执行器的唯一入口。 + + 条件 UPDATE 只有一行能成功:同一自动化在同一间隔内, + 两个并发执行器只有一个拿到 rowcount 1,另一个拿到 0 + 就必须放弃建任务——这是防重复执行的硬门。抢单即把 + last_run_at 写成 now(不是等执行完才写,与 host run-due + 相反):好处是崩掉的执行不会在当间隔内被另一执行器重 + 跑;代价是抢单后崩掉的那次产出会缺失,直到下个间隔才 + 补跑。执行器抢单成功后必须调 record_automation_run 记 + 产出,否则分诊收件箱里这次执行不可见。成功时 commit。 + 返回 True 表示抢到,False 表示不存在、已取消或未到期。 + """ + if now is None: + now = time.time() + ensure_automations_tables(conn) + cur = conn.execute( + "UPDATE automations SET last_run_at = ?," + " last_claimed_at = ?" + " WHERE id = ? AND status = ?" + " AND (last_run_at IS NULL" + " OR last_run_at + interval_seconds <= ?)", + (now, now, automation_id, ACTIVE, now)) + conn.commit() + return cur.rowcount == 1 + + +def record_automation_run(conn, automation_id, project_id, task_id=None, + status="created", detail="", now=None): + """记一次自动化产出:执行器建任务后调它,task_id 为建好 + 的云端任务 id;status 取 created / completed / failed 等 + 执行器自定的终态词(本模块不校验任务状态机,只原样存)。 + 自动化不存在时抛 ValueError(执行器绝不能给不存在的 + 定义记产出)。成功时 commit 并返回 run dict。""" + if now is None: + now = time.time() + ensure_automations_tables(conn) + auto = conn.execute( + "SELECT id FROM automations WHERE id = ?", + (automation_id,)).fetchone() + if auto is None: + raise ValueError("automation not found") + run_id = uuid.uuid4().hex + conn.execute( + "INSERT INTO automation_runs(id, automation_id, project_id," + " task_id, status, detail, created_at)" + " VALUES(?, ?, ?, ?, ?, ?, ?)", + (run_id, automation_id, project_id, task_id, status, + detail or "", now)) + conn.commit() + return { + "id": run_id, + "automation_id": automation_id, + "project_id": project_id, + "task_id": task_id, + "status": status, + "detail": detail or "", + "created_at": now, + } + + +def list_automation_runs(conn, automation_id, limit=TRIAGE_LIMIT): + """列某自动化的最近产出记录,按时间倒序。""" + ensure_automations_tables(conn) + rows = conn.execute( + "SELECT * FROM automation_runs WHERE automation_id = ?" + " ORDER BY created_at DESC, id LIMIT ?", + (automation_id, limit)).fetchall() + out = [] + for raw in rows: + r = _row_dict(raw, RUN_COLUMNS) + out.append({ + "id": r["id"], + "automation_id": r["automation_id"], + "project_id": r["project_id"], + "task_id": row_get(r, "task_id"), + "status": r["status"], + "detail": row_get(r, "detail"), + "created_at": r["created_at"], + }) + return out + + +def triage_inbox(conn, project_id, now=None): + """分诊收件箱(只读,不写业务行、不改任务状态): + + - automations_due:当前到期的自动化定义。 + - failed_tasks:失败或中断的任务(最需要人工看的项), + 按更新时间倒序。 + - pending_followups:仍 pending 的后续消息(含任务间 + 消息排队的那部分,带 sender_task_id),按创建时间排序; + followups 表不存在时该段为空(不因此炸掉整页)。 + - recent_runs:最近的自动化产出记录,带对应任务的当前 + 状态(任务删了则 task_status 为 None,记录本身保留)。 + + 调用方(server handler)须先 ensure 自动化表与 followups + 表;本函数内 followups 缺表时只吞缺表错,其它错照常抛 + 出(调用方转 502,不伪造一页空收件箱)。""" + if now is None: + now = time.time() + ensure_automations_tables(conn) + due = due_automations(conn, project_id, now) + task_rows = conn.execute( + "SELECT id, status, prompt, created_at, updated_at," + " error_text FROM tasks WHERE project_id = ?" + " AND status IN ('failed', 'interrupted')" + " ORDER BY updated_at DESC, id LIMIT ?", + (project_id, TRIAGE_LIMIT)).fetchall() + failed = [] + for raw in task_rows: + r = _row_dict(raw, ("id", "status", "prompt", "created_at", + "updated_at", "error_text")) + failed.append({ + "id": r["id"], + "status": r["status"], + "prompt": (r["prompt"] or "")[:300], + "created_at": r["created_at"], + "updated_at": r["updated_at"], + "error_text": (row_get(r, "error_text") or "")[:300], + }) + try: + fu_rows = conn.execute( + "SELECT f.id, f.task_id, f.body, f.mode, f.status," + " f.sender_task_id, f.created_at FROM task_followups f" + " JOIN tasks t ON t.id = f.task_id" + " WHERE f.project_id = ? AND f.status = 'pending'" + " ORDER BY f.created_at, f.id LIMIT ?", + (project_id, TRIAGE_LIMIT)).fetchall() + pending = [] + for raw in fu_rows: + r = _row_dict(raw, ("id", "task_id", "body", "mode", + "status", "sender_task_id", + "created_at")) + pending.append({ + "id": r["id"], + "task_id": r["task_id"], + "body": (r["body"] or "")[:300], + "mode": r["mode"], + "status": r["status"], + "sender_task_id": row_get(r, "sender_task_id"), + "created_at": r["created_at"], + }) + except Exception as exc: + msg = str(exc) + if "no such table" in msg or "does not exist" in msg: + pending = [] + else: + raise + run_rows = conn.execute( + "SELECT r.id, r.automation_id, r.task_id, r.status," + " r.detail, r.created_at, t.status AS task_status" + " FROM automation_runs r" + " LEFT JOIN tasks t ON t.id = r.task_id" + " WHERE r.project_id = ?" + " ORDER BY r.created_at DESC, r.id LIMIT ?", + (project_id, TRIAGE_LIMIT)).fetchall() + runs = [] + for raw in run_rows: + r = _row_dict(raw, ("id", "automation_id", "task_id", "status", + "detail", "created_at", "task_status")) + runs.append({ + "id": r["id"], + "automation_id": r["automation_id"], + "task_id": row_get(r, "task_id"), + "status": r["status"], + "detail": row_get(r, "detail"), + "created_at": r["created_at"], + "task_status": row_get(r, "task_status"), + }) + return { + "project_id": project_id, + "automations_due": due, + "failed_tasks": failed, + "pending_followups": pending, + "recent_runs": runs, + "counts": { + "automations_due": len(due), + "failed_tasks": len(failed), + "pending_followups": len(pending), + "recent_runs": len(runs), + }, + } diff --git a/cloud/caps.py b/cloud/caps.py new file mode 100644 index 0000000..01364cd --- /dev/null +++ b/cloud/caps.py @@ -0,0 +1,76 @@ +"""Worker 并发上限:配置解析与认领判断(纯函数,无 DB、无 IO)。 + +配置来源与优先级(高到低): +1. 显式值:Worker 构造参数 capacity / CLI --worker-capacity。 +2. 环境变量 CLOUD_WORKER_CAPACITY(CLI 未给时生效)。 +3. 默认值 1:默认行为不得弱于现状,即旧串行行为。 +非法值(非整数、<= 0)一律归一化为 1,绝不出现 0 上限导致 +worker 永远认领不到任务,也绝不出现负数上限。 + +workers 表的 capacity 是注册表展示值,不是控制面下推: +本地 worker 只执行自己启动时的配置,表里被改大的值不会 +悄悄抬高本 worker 的上限(否则 stale 行或误写会放大并发)。 + +上限口径(单 worker 本地计数): +- running 数 = Worker._running 的条目数(_running_lock 下读)。 +- 启动中的任务也算:_claim 占位时就先在 _running 里建空条目, + 执行器起 proc / 容器后 _running_set 只往该条目里补字段, + 不新增条目;所以“已认领未起进程”与“已起进程”都占一个槽。 +- 同一任务重复占位不重复计数(dict 按 task_id 去重)。 +- 本模块只管单 worker 上限;多 worker 全局上限不在此列, + 见汇报里的诚实边界(项目级上限同样是遗留项)。 +""" + +import os + +DEFAULT_CAPACITY = 1 +ENV_VAR = "CLOUD_WORKER_CAPACITY" + + +def parse_capacity(value): + """把任意输入归一化为 >= 1 的整数上限;非法值返回 1。""" + try: + capacity = int(value) + except (TypeError, ValueError): + return DEFAULT_CAPACITY + if capacity < 1: + return DEFAULT_CAPACITY + return capacity + + +def resolve_capacity(explicit=None, env=None): + """按“显式 > 环境变量 > 默认 1”解析上限。 + + explicit 为 None 表示调用方未给(CLI 未传 --worker-capacity + 时 argparse 默认即 None);env 默认读进程环境,单测可注入 + 普通 dict。显式值即使非法也按非法处理(归一为 1),不回落 + 到环境变量:显式给错了就该按 1 跑,而不是悄悄用环境值。 + """ + if explicit is not None: + return parse_capacity(explicit) + source = env if env is not None else os.environ + try: + raw = source.get(ENV_VAR) + except Exception: + raw = None + if raw is None or (isinstance(raw, str) and not raw.strip()): + return DEFAULT_CAPACITY + return parse_capacity(raw) + + +def effective_capacity(capacity): + """运行期上限:构造后 self.capacity 可能来自旧调用方,统一归一。""" + return parse_capacity(capacity) + + +def running_count(running_snapshot): + """_running 快照的条目数:空条目(启动中)同样计数。""" + try: + return len(running_snapshot) + except TypeError: + return 0 + + +def can_claim(running_count_value, capacity): + """未满返回 True;running 数达到上限时不再认领。""" + return running_count_value < effective_capacity(capacity) diff --git a/cloud/console.html b/cloud/console.html index f2bdd7c..4674f2c 100644 --- a/cloud/console.html +++ b/cloud/console.html @@ -250,6 +250,24 @@