Merge pull request #89 from openqj/feat-p1

P1 补齐:自动评审队列、自动化+分诊、并发 caps、运行时审批队列
This commit is contained in:
openqj authored and GitHub committed 2026-10-27 16:00:53 +08:00
commit e786e1f85d
18 files changed
+3474 -27

No files matched your search

+302
View File
@@ -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"}
+525
View File
@@ -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),
},
}
+76
View File
@@ -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)
+186 -1
View File
@@ -250,6 +250,24 @@
<pre id="detailBody" style="display:none"></pre>
<h3>子代理链路</h3>
<div id="agentBox" class="muted">点任务行查看详情后,这里显示子代理链路</div>
<h3>评审队列</h3>
<div id="reviewForm">
<select id="reviewTargetType">
<option value="uncommittedChanges">未提交改动(工作区)</option>
<option value="baseBranch">与分支的差异</option>
<option value="commit">指定提交</option>
<option value="custom">自定义评审指令</option>
</select>
<input id="reviewBranch" placeholder="与分支的差异用:分支名(如 main)" style="width:100%">
<input id="reviewSha" placeholder="指定提交用:提交 sha" style="width:100%">
<textarea id="reviewInstructions" placeholder="自定义评审指令用:评审指令正文"></textarea>
<div class="bar">
<button id="btnCreateReview" class="primary" onclick="postReview()">发起评审</button>
<button onclick="if (currentTask) loadReviews(currentTask.id)">刷新队列</button>
<span id="reviewMsg" class="muted"></span>
</div>
</div>
<div id="reviewList" class="muted">点任务行查看详情后,点“刷新队列”查看评审记录</div>
<h3>实时链路</h3>
<div id="traceBox" class="muted">点任务行查看详情后,这里显示过程链路</div>
<div id="costBox"></div>
@@ -263,6 +281,10 @@
</div>
</div>
<div id="followupList" class="muted"></div>
<h3>运行时审批队列</h3>
<div class="bar"><button id="btnRefreshApprovals" onclick="if(currentTask)loadApprovals(currentTask.id)">刷新审批队列</button><span id="approvalMsg" class="muted"></span></div>
<div class="muted">待审批项需人工点允许或拒绝;超时默认拒绝。本队列的决定会留回执;实时请求的事前拦截待 host 接入后生效,在此之前实时行为不变。</div>
<div id="approvalList" class="muted"></div>
<h3>给其他任务发消息</h3>
<div id="peerMessageForm">
<input id="peerTarget" placeholder="目标任务 id(与本任务同项目;对方运行中排队执行、已终态则续跑复活,绝不打断当前轮)" style="width:100%">
@@ -274,6 +296,22 @@
</div>
</div>
<div id="peerMessageList" class="muted"></div>
<h3>自动化</h3>
<div id="automationForm">
<input id="autoName" placeholder="自动化名称(同项目内启用中的名称唯一)" style="width:100%">
<textarea id="autoPrompt" placeholder="定时任务内容(创建只登记定义与到期状态;执行由后续执行器经 claim 抢单后建任务,不另起调度器)"></textarea>
<div class="bar">
<input id="autoInterval" placeholder="间隔秒数(默认 86400,最小 60)" style="width:40%">
<input id="autoRepo" placeholder="仓库(可选)" style="width:40%">
<button id="btnCreateAutomation" class="primary">创建自动化</button>
<button id="btnLoadAutomations">刷新自动化</button>
<span id="autoMsg" class="muted"></span>
</div>
</div>
<div id="automationList" class="muted"></div>
<h3>分诊收件箱</h3>
<div class="bar"><button id="btnLoadTriage">刷新分诊</button><span id="triageMsg" class="muted"></span></div>
<div id="triageBox" class="muted"></div>
</section>
<script>
@@ -399,14 +437,46 @@ async function loadProjects() {
} catch (e) { $("#projectMsg").textContent = "加载失败:" + e.message; }
}
async function loadApprovals(taskId) {
var box = $("#approvalList");
try {
var d = await api("/tasks/" + encodeURIComponent(taskId) + "/approvals", "GET");
var list = d.approvals || [];
if (!list.length) { box.innerHTML = "(暂无审批项)"; return; }
box.innerHTML = list.map(function (a) {
var receipt = "";
if (a.status === "approved" || a.status === "denied" || a.status === "expired") {
receipt = "|回执:" + esc(a.decision || a.status) + " by " + esc(String(a.decided_by || "").slice(0, 8)) + " " + fmt(a.decided_at);
}
var btns = "";
if (a.status === "pending") {
btns = " <button onclick=\"decideApproval('" + esc(a.id) + "', 'approved')\">允许</button>" +
" <button onclick=\"decideApproval('" + esc(a.id) + "', 'denied')\">拒绝</button>";
}
return "[" + esc(a.kind) + "|" + esc(a.status) + "] " + esc(a.summary) + receipt + btns;
}).join("<br>");
} catch (e) { box.innerHTML = "<span class='err'>审批队列加载失败:" + esc(e.message) + "</span>"; }
}
async function decideApproval(approvalId, decision) {
if (!currentTask) return;
var verb = decision === "approved" ? "approve" : "deny";
try {
await api("/tasks/" + encodeURIComponent(currentTask.id) + "/approvals/" + encodeURIComponent(approvalId) + "/" + verb, "POST", {});
$("#approvalMsg").innerHTML = "<span class='ok'>已" + (decision === "approved" ? "允许" : "拒绝") + "</span>";
loadApprovals(currentTask.id);
} catch (e) { $("#approvalMsg").innerHTML = "<span class='err'>决定失败:" + esc(e.message) + "</span>"; }
}
async function loadWorkers() {
try {
var d = await api("/workers", "GET");
var list = d.workers || d || [];
if (!list.length) { $("#workers").textContent = "无 worker 注册"; return; }
$("#workers").innerHTML = list.map(function (w) {
var run = (w.running === undefined || w.running === null) ? "?" : w.running;
var cap = (w.capacity === undefined || w.capacity === null) ? "?" : w.capacity;
var live = (w.alive === false) ? " / 离线(占用为上次上报值)" : "";
return esc(w.worker_id || w.id || "?") + " / " + esc(w.executor || "") +
" / capacity " + esc(w.capacity || "") +
" / 占用 " + esc(run) + "/" + esc(cap) + live +
" / last_seen " + esc(w.last_seen || w.last_seen_seconds_ago || "");
}).join("<br>");
} catch (e) { $("#workers").textContent = "加载失败:" + e.message; }
@@ -639,6 +709,52 @@ async function startTrace(taskId, status) {
if (done) { loadTasks(); showDetail(taskId); }
}
async function loadReviews(taskId) {
var box = $("#reviewList");
try {
var d = await api("/tasks/" + encodeURIComponent(taskId) + "/reviews", "GET");
var list = d.reviews || [];
if (!list.length) { box.innerHTML = "(该任务暂无评审记录)"; return; }
box.innerHTML = list.map(function (r) {
var target = esc(r.target_type);
if (r.branch) target += ":" + esc(r.branch);
if (r.sha) target += ":" + esc(String(r.sha).slice(0, 12));
if (r.instructions) target += ":" + esc(String(r.instructions).slice(0, 80));
var html = "<div style='margin:6px 0;padding:8px 10px;border:1px solid var(--border);border-radius:var(--radius);background:#fff'>";
html += "<b>评审</b> " + esc(String(r.id).slice(0, 8)) + " | " + target +
" | <span class='st-" + esc(r.status) + "'>" + esc(r.status) + "</span>" +
(r.created_at ? " | " + fmt(r.created_at) : "") + "<br>";
html += r.findings ? esc(String(r.findings).slice(0, 1000)) : "<span class='muted'>(暂无 findings:评审尚未执行或尚未写回)</span>";
html += "<br><button onclick=\"submitReviewFindings('" + esc(r.id) + "')\">写 findings</button>";
html += "</div>";
return html;
}).join("");
} catch (e) { box.innerHTML = "<span class='err'>评审队列加载失败:" + esc(e.message) + "</span>"; }
}
async function postReview() {
if (!currentTask) return;
var body = { target_type: $("#reviewTargetType").value };
var branch = $("#reviewBranch").value.trim();
var sha = $("#reviewSha").value.trim();
var instructions = $("#reviewInstructions").value.trim();
if (branch) body.branch = branch;
if (sha) body.sha = sha;
if (instructions) body.instructions = instructions;
try {
await api("/tasks/" + encodeURIComponent(currentTask.id) + "/reviews", "POST", body);
$("#reviewMsg").innerHTML = "<span class='ok'>评审已登记(requested)</span>";
loadReviews(currentTask.id);
} catch (e) { $("#reviewMsg").innerHTML = "<span class='err'>发起评审失败:" + esc(e.message) + "</span>"; }
}
async function submitReviewFindings(reviewId) {
if (!currentTask) return;
var text = prompt("输入该条评审的 findings(将写回记录并标为 completed):");
if (text === null) return;
try {
await api("/tasks/" + encodeURIComponent(currentTask.id) + "/reviews/" + encodeURIComponent(reviewId) + "/findings", "POST", { findings: text });
loadReviews(currentTask.id);
} catch (e) { $("#reviewMsg").innerHTML = "<span class='err'>写 findings 失败:" + esc(e.message) + "</span>"; }
}
async function showDetail(taskId) {
stopTrace();
try {
@@ -696,6 +812,75 @@ async function doAction(kind) {
} catch (e) { $("#actionMsg").innerHTML = "<span class='err'>" + kind + " 失败:" + esc(e.message) + "</span>"; }
}
async function loadAutomations() {
var pid = $("#projectSelect").value;
var box = $("#automationList");
if (!pid) { box.innerHTML = "(先选项目)"; return; }
try {
var d = await api("/projects/" + encodeURIComponent(pid) + "/automations", "GET");
var list = d.automations || [];
if (!list.length) { box.innerHTML = "(暂无自动化)"; return; }
box.innerHTML = list.map(function (a) {
var html = "[" + esc(a.status) + (a.due ? "|到期" : "") + "] " + esc(a.name) +
"|每 " + esc(a.interval_seconds) + " 秒|下次 " + esc(fmt(a.next_run_at));
if (a.status === "active") html += " <button data-auto-cancel='" + esc(a.id) + "'>取消</button>";
return html;
}).join("<br>");
box.querySelectorAll("[data-auto-cancel]").forEach(function (btn) {
btn.onclick = function () { cancelAutomation(btn.getAttribute("data-auto-cancel")); };
});
} catch (e) { box.innerHTML = "<span class='err'>自动化加载失败:" + esc(e.message) + "</span>"; }
}
async function createAutomation() {
var pid = $("#projectSelect").value;
if (!pid) { $("#autoMsg").textContent = "先选项目"; return; }
var name = $("#autoName").value.trim();
var prompt = $("#autoPrompt").value.trim();
if (!name) { $("#autoMsg").textContent = "自动化名称必填"; return; }
if (!prompt) { $("#autoMsg").textContent = "定时任务内容必填"; return; }
var body = { name: name, prompt: prompt };
var iv = $("#autoInterval").value.trim();
if (iv) body.interval_seconds = parseInt(iv, 10);
var repo = $("#autoRepo").value.trim();
if (repo) body.repo_url = repo;
try {
await api("/projects/" + encodeURIComponent(pid) + "/automations", "POST", body);
$("#autoMsg").innerHTML = "<span class='ok'>自动化已创建(登记定义,到期后由执行器认领)</span>";
$("#autoName").value = ""; $("#autoPrompt").value = ""; $("#autoInterval").value = ""; $("#autoRepo").value = "";
loadAutomations();
} catch (e) { $("#autoMsg").innerHTML = "<span class='err'>创建失败:" + esc(e.message) + "</span>"; }
}
async function cancelAutomation(autoId) {
var pid = $("#projectSelect").value;
if (!pid) return;
try {
await api("/projects/" + encodeURIComponent(pid) + "/automations/" + encodeURIComponent(autoId) + "/cancel", "POST", {});
$("#autoMsg").innerHTML = "<span class='ok'>自动化已取消</span>";
loadAutomations();
} catch (e) { $("#autoMsg").innerHTML = "<span class='err'>取消失败:" + esc(e.message) + "</span>"; }
}
async function loadTriage() {
var pid = $("#projectSelect").value;
var box = $("#triageBox");
if (!pid) { box.innerHTML = "(先选项目)"; return; }
try {
var d = await api("/projects/" + encodeURIComponent(pid) + "/triage", "GET");
var html = "";
html += "<b>到期自动化(" + (d.counts.automations_due || 0) + ")</b><br>";
html += (d.automations_due || []).length ? (d.automations_due || []).map(function (a) { return esc(a.name) + "|每 " + esc(a.interval_seconds) + " 秒"; }).join("<br>") : "(无)";
html += "<br><b>失败/中断任务(" + (d.counts.failed_tasks || 0) + ")</b><br>";
html += (d.failed_tasks || []).length ? (d.failed_tasks || []).map(function (t) { return esc(String(t.id).slice(0, 8)) + "|" + esc(t.status) + "|" + esc(String(t.prompt || "").slice(0, 80)); }).join("<br>") : "(无)";
html += "<br><b>待处理后续消息(" + (d.counts.pending_followups || 0) + ")</b><br>";
html += (d.pending_followups || []).length ? (d.pending_followups || []).map(function (f) { return "任务 " + esc(String(f.task_id).slice(0, 8)) + "|" + esc(String(f.body || "").slice(0, 80)); }).join("<br>") : "(无)";
html += "<br><b>最近自动化产出(" + (d.counts.recent_runs || 0) + ")</b><br>";
html += (d.recent_runs || []).length ? (d.recent_runs || []).map(function (r) { return esc(String(r.automation_id).slice(0, 8)) + "|" + esc(r.status) + (r.task_id ? "|任务 " + esc(String(r.task_id).slice(0, 8)) : "") + (r.task_status ? "|" + esc(r.task_status) : ""); }).join("<br>") : "(无)";
box.innerHTML = html;
} catch (e) { box.innerHTML = "<span class='err'>分诊加载失败:" + esc(e.message) + "</span>"; }
}
$("#btnCreateAutomation").onclick = createAutomation;
$("#btnLoadAutomations").onclick = loadAutomations;
$("#btnLoadTriage").onclick = loadTriage;
$("#projectSelect").addEventListener("change", function () { loadAutomations(); loadTriage(); });
async function loadFollowups(taskId) {
var box = $("#followupList");
try {
+189
View File
@@ -0,0 +1,189 @@
"""自动代码评审:评审队列记录。
评审队列是服务端的一张登记表,不跑模型:调用方对某个任务
发起评审时记一条 requested 记录,评审执行方(目前是
host.py code-review,走协议 review/start)跑完后把 findings
写回同一条记录,控制台评审区据此展示队列与结果。
与 peer_messages 同款分工:本模块只含纯表操作与协议目标
构造,不 import cloud 其它模块,避免循环 import;服务端
的表迁移委托到 ensure_reviews_table,失败不伪造成功。
协议锚点(protocol/codex_app_server_protocol.v2.schemas.json,
逐个对过枚举,不编造):
- 方法名 review/start,参数 ReviewStartParams:threadId 与
target 必填,delivery 可空(inline 为默认,detached 已
废弃,故本模块不产生 detached)。
- ReviewTarget 四种:uncommittedChanges(工作区未提交改动)、
baseBranch(须带 branch)、commit(须带 sha,title 可空)、
custom(须带 instructions)。
状态机:requested(已登记,未执行)-> running(执行中,
目前仅 host CLI 执行时可由调用方先置位,服务端不自动
流转)-> completed / failed(写 findings 时落终态)。
服务端不替执行方编造 findings:findings 为空时记录仍是
requested,控制台如实显示“暂无 findings”。
"""
import time
import uuid
REVIEW_TARGET_TYPES = (
"uncommittedChanges", "baseBranch", "commit", "custom")
REVIEW_STATUSES = ("requested", "running", "completed", "failed")
_REVIEW_COLUMNS = (
"id", "task_id", "project_id", "target_type", "branch", "sha",
"instructions", "status", "findings", "created_by",
"created_at", "updated_at")
def build_review_target(target_type, branch=None, sha=None,
instructions=None):
"""按协议构造 ReviewTarget dict;组合非法时 ValueError,
调用方转 400,绝不把非法目标发给 app-server。"""
if target_type == "uncommittedChanges":
return {"type": "uncommittedChanges"}
if target_type == "baseBranch":
if not branch or not isinstance(branch, str) or not branch.strip():
raise ValueError("baseBranch target requires branch")
return {"type": "baseBranch", "branch": branch.strip()}
if target_type == "commit":
if not sha or not isinstance(sha, str) or not sha.strip():
raise ValueError("commit target requires sha")
return {"type": "commit", "sha": sha.strip()}
if target_type == "custom":
if not instructions or not isinstance(instructions, str) \
or not instructions.strip():
raise ValueError("custom target requires instructions")
return {"type": "custom", "instructions": instructions.strip()}
raise ValueError("unknown target_type: %r" % (target_type,))
def ensure_reviews_table(conn):
"""幂等建评审队列表。调用方(服务端接口)在每次读写前
调它,老库第一次跑到这里时自动补表,不丢数据。"""
conn.execute(
"CREATE TABLE IF NOT EXISTS task_reviews("
"id TEXT PRIMARY KEY, "
"task_id TEXT, "
"project_id TEXT, "
"target_type TEXT, "
"branch TEXT, "
"sha TEXT, "
"instructions TEXT, "
"status TEXT, "
"findings TEXT, "
"created_by TEXT, "
"created_at DOUBLE PRECISION, "
"updated_at DOUBLE PRECISION)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_reviews_task"
" ON task_reviews(task_id, created_at)")
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_reviews_project"
" ON task_reviews(project_id, created_at)")
def _row_to_dict(row):
"""行转 dict:兼容 dict 行、sqlite3.Row 与裸元组(按
_REVIEW_COLUMNS 的 SELECT 顺序)。"""
if row is None:
return None
if isinstance(row, dict):
return {c: row.get(c) for c in _REVIEW_COLUMNS}
try:
keys = row.keys()
except AttributeError:
keys = None
if keys is not None:
d = dict(row)
return {c: d.get(c) for c in _REVIEW_COLUMNS}
return dict(zip(_REVIEW_COLUMNS, row))
def _select_cols():
return ", ".join(_REVIEW_COLUMNS)
def create_review(conn, task_id, project_id, target_type, created_by,
branch=None, sha=None, instructions=None):
"""登记一条评审,返回记录 dict。先校验目标组合(非法
时 ValueError,调用方转 400,表里不留脏行),成功才
commit。"""
build_review_target(target_type, branch, sha, instructions)
ensure_reviews_table(conn)
now = time.time()
review_id = uuid.uuid4().hex
conn.execute(
"INSERT INTO task_reviews(id, task_id, project_id,"
" target_type, branch, sha, instructions, status,"
" findings, created_by, created_at, updated_at)"
" VALUES(?, ?, ?, ?, ?, ?, ?, 'requested', NULL, ?, ?, ?)",
(review_id, task_id, project_id, target_type,
branch.strip() if isinstance(branch, str) else None,
sha.strip() if isinstance(sha, str) else None,
instructions.strip() if isinstance(instructions, str) else None,
created_by, now, now),
)
conn.commit()
return get_review(conn, review_id)
def list_reviews(conn, task_id):
"""某任务的评审队列,按创建时间升序(先登记的先审)。"""
ensure_reviews_table(conn)
rows = conn.execute(
"SELECT %s FROM task_reviews WHERE task_id = ?"
" ORDER BY created_at, id" % _select_cols(),
(task_id,)).fetchall()
return [_row_to_dict(r) for r in rows]
def list_project_reviews(conn, project_id):
"""某项目的评审队列(供未来项目级队列端点复用;本批
服务端只暴露任务级队列,函数先备好不接线)。"""
ensure_reviews_table(conn)
rows = conn.execute(
"SELECT %s FROM task_reviews WHERE project_id = ?"
" ORDER BY created_at, id" % _select_cols(),
(project_id,)).fetchall()
return [_row_to_dict(r) for r in rows]
def get_review(conn, review_id):
"""取单条评审记录,不存在时返回 None(调用方转 404)。"""
ensure_reviews_table(conn)
row = conn.execute(
"SELECT %s FROM task_reviews WHERE id = ?" % _select_cols(),
(review_id,)).fetchone()
return _row_to_dict(row)
def update_review(conn, review_id, status=None, findings=None):
"""更新评审状态与 findings,返回更新后记录,不存在时
返回 None。status 非法时 ValueError(调用方转 400);
findings 为字符串时照存(空串也存,区别于未写回的
NULL);不传 status 时 findings 写回默认落 completed,
即“有结果即完成”,调用方可显式传 failed 覆盖。"""
if status is not None and status not in REVIEW_STATUSES:
raise ValueError("unknown review status: %r" % (status,))
if findings is not None and not isinstance(findings, str):
raise ValueError("findings must be a string")
ensure_reviews_table(conn)
record = get_review(conn, review_id)
if record is None:
return None
new_status = status
if new_status is None:
new_status = "completed" if findings is not None \
else record["status"]
new_findings = findings if findings is not None \
else record["findings"]
conn.execute(
"UPDATE task_reviews SET status = ?, findings = ?,"
" updated_at = ? WHERE id = ?",
(new_status, new_findings, time.time(), review_id))
conn.commit()
return get_review(conn, review_id)
+435 -3
View File
@@ -66,6 +66,17 @@ from cloud.peer_messages import deliver_peer_message as _deliver_peer_message
from cloud.peer_messages import ensure_followups_table as _peer_ensure_followups
from cloud.peer_messages import peer_message_prefix as _peer_prefix
from cloud.peer_messages import row_get as _row_get
from cloud.automations import cancel_automation as _auto_cancel
from cloud.automations import create_automation as _auto_create
from cloud.automations import ensure_automations_tables as _auto_ensure
from cloud.automations import list_automations as _auto_list
from cloud.automations import triage_inbox as _auto_triage
from cloud.approvals import create_approval as _appr_create
from cloud.approvals import decide_approval as _appr_decide
from cloud.approvals import ensure_approvals_table as _appr_ensure
from cloud.approvals import expire_overdue as _appr_expire
from cloud.approvals import get_approval as _appr_get
from cloud.approvals import list_approvals as _appr_list
from cloud.kb import KBStore
from cloud.memory import MemoryStore
from cloud.pr import PrSkipped, PrStepError
@@ -1480,8 +1491,27 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
self._create_peer_message(user_id, rest[: -len("/messages")])
elif rest.endswith("/messages") and method == "GET":
self._list_peer_messages(user_id, rest[: -len("/messages")])
elif rest.endswith("/reviews") and method == "POST":
self._create_review(user_id, rest[: -len("/reviews")])
elif rest.endswith("/reviews") and method == "GET":
self._list_reviews(user_id, rest[: -len("/reviews")])
elif "/reviews/" in rest and rest.endswith("/findings") \
and method == "POST":
_task_part, _rev_part = rest.split("/reviews/", 1)
self._update_review_findings(
user_id, _task_part, _rev_part.split("/")[0])
elif rest.endswith("/fork") and method == "POST":
self._fork_task(user_id, rest[: -len("/fork")])
elif rest.endswith("/approvals") and method == "GET":
self._list_approvals(user_id, rest[: -len("/approvals")])
elif rest.endswith("/approvals") and method == "POST":
self._create_approval(user_id, rest[: -len("/approvals")])
elif rest.endswith("/approve") and method == "POST":
self._decide_approval(
user_id, rest[: -len("/approve")], "approved")
elif rest.endswith("/deny") and method == "POST":
self._decide_approval(
user_id, rest[: -len("/deny")], "denied")
elif rest.endswith("/continue") and method == "POST":
self._continue_task(user_id, rest[: -len("/continue")])
else:
@@ -1974,6 +2004,46 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
subscription_id=parts[3])
self._send_json(200, {"deleted": bool(deleted)})
return
# /projects/{id}/automations
if len(parts) == 2 and parts[1] == "automations":
if method == "POST":
err = self._check_project(user_id, project_id, "create_task")
if err:
self._send_json(err[0], {"error": err[1]})
return
self._automation_create(user_id, project_id)
elif method == "GET":
err = self._check_project(user_id, project_id, "read")
if err:
self._send_json(err[0], {"error": err[1]})
return
self._automation_list(project_id)
else:
self._send_json(404, {"error": "not found"})
return
# /projects/{id}/automations/{aid}/cancel
if len(parts) == 4 and parts[1] == "automations" \
and parts[3] == "cancel":
if method != "POST":
self._send_json(404, {"error": "not found"})
return
err = self._check_project(user_id, project_id, "create_task")
if err:
self._send_json(err[0], {"error": err[1]})
return
self._automation_cancel(user_id, project_id, parts[2])
return
# /projects/{id}/triage
if len(parts) == 2 and parts[1] == "triage":
if method != "GET":
self._send_json(404, {"error": "not found"})
return
err = self._check_project(user_id, project_id, "read")
if err:
self._send_json(err[0], {"error": err[1]})
return
self._triage_inbox(project_id)
return
self._send_json(404, {"error": "not found"})
def _memory_create(self, user_id, project_id):
@@ -2998,6 +3068,127 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
self._send_json(200, {
"messages": [_followup_to_dict(r) for r in rows]})
def _create_review(self, user_id, task_id):
"""发起评审:对某任务登记一条评审队列记录(status
requested),不跑模型、不伪造 findings——执行方跑完
后经 /findings 写回。权限:项目 create_task;目标组合
非法时 400,表里不留脏行;记审计 review_requested。"""
from cloud.review import build_review_target
from cloud.review import create_review as _review_create
body, err = self._read_json()
if err:
self._send_json(400, {"error": err})
return
row = self._fetch_task(task_id)
if row is None:
self._send_json(404, {"error": "task not found"})
return
task = _task_to_dict(row)
err = self._check_project(user_id, task["project_id"],
"create_task")
if err:
self._send_json(err[0], {"error": err[1]})
return
target_type = body.get("target_type", "uncommittedChanges")
branch = body.get("branch")
sha = body.get("sha")
instructions = body.get("instructions")
for name, value in (("branch", branch), ("sha", sha),
("instructions", instructions)):
if value is not None and not isinstance(value, str):
self._send_json(400, {"error": "%s must be a string" % name})
return
try:
build_review_target(target_type, branch, sha, instructions)
except ValueError as exc:
self._send_json(400, {"error": str(exc)})
return
conn = self._db()
try:
record = _review_create(
conn, task_id, task["project_id"], target_type,
user_id, branch=branch, sha=sha,
instructions=instructions)
finally:
conn.close()
with _auth_store(self.server.db_path) as store:
store.audit(user_id, task["project_id"],
"review_requested",
"%s -> %s" % (task_id, record["id"]))
self._send_json(201, record)
def _list_reviews(self, user_id, task_id):
"""评审队列查看:列出某任务的全部评审记录(含 findings,
未写回的 findings 为 null,如实展示不编造)。权限:
项目 read。"""
from cloud.review import list_reviews as _review_list
row = self._fetch_task(task_id)
if row is None:
self._send_json(404, {"error": "task not found"})
return
task = _task_to_dict(row)
err = self._check_project(user_id, task["project_id"], "read")
if err:
self._send_json(err[0], {"error": err[1]})
return
conn = self._db()
try:
reviews = _review_list(conn, task_id)
finally:
conn.close()
self._send_json(200, {"reviews": reviews})
def _update_review_findings(self, user_id, task_id, review_id):
"""评审结果写回:执行方把 findings 记到指定记录上。
不传 status 时 findings 写回默认落 completed;显式传
failed 可标失败。权限:项目 create_task;记录不属于
该任务时 404;记审计 review_findings_updated。"""
from cloud.review import get_review as _review_get
from cloud.review import update_review as _review_update
body, err = self._read_json()
if err:
self._send_json(400, {"error": err})
return
row = self._fetch_task(task_id)
if row is None:
self._send_json(404, {"error": "task not found"})
return
task = _task_to_dict(row)
err = self._check_project(user_id, task["project_id"],
"create_task")
if err:
self._send_json(err[0], {"error": err[1]})
return
findings = body.get("findings")
status = body.get("status")
if findings is None and status is None:
self._send_json(400, {
"error": "findings or status required"})
return
if findings is not None and not isinstance(findings, str):
self._send_json(400, {"error": "findings must be a string"})
return
conn = self._db()
try:
record = _review_get(conn, review_id)
if record is None or record["task_id"] != task_id:
self._send_json(404, {"error": "review not found"})
return
try:
updated = _review_update(conn, review_id,
status=status,
findings=findings)
except ValueError as exc:
self._send_json(400, {"error": str(exc)})
return
finally:
conn.close()
with _auth_store(self.server.db_path) as store:
store.audit(user_id, task["project_id"],
"review_findings_updated",
"%s -> %s" % (task_id, review_id))
self._send_json(200, updated)
def _fork_task(self, user_id, task_id):
"""会话分支:从已结束任务的 thread 分叉出新任务。
@@ -3180,6 +3371,136 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
store.audit(user_id, task["project_id"], "merge_task", task_id)
self._send_json(200, task)
# -- 运行时审批队列 ------------------------------------------------
def _list_approvals(self, user_id, task_id):
"""列出某任务的审批项(含已决定的,回执一并带出)。
权限:项目 read。列之前先扫荡过期:已到期的 pending
在这里变成 expired(effective decision 为 denied),
列表里看到的就是最终态,不会把已过期项继续显示为
待审批。下游异常不拦:抛给 _dispatch_safe 回 500,
绝不回一条伪造的空列表或成功。"""
row = self._fetch_task(task_id)
if row is None:
self._send_json(404, {"error": "task not found"})
return
task = _task_to_dict(row)
with _auth_store(self.server.db_path) as store:
if not store.check_permission(user_id, task["project_id"],
"read"):
self._send_json(403, {"error": "forbidden"})
return
conn = self._db()
try:
_appr_ensure(conn)
_appr_expire(conn)
items = _appr_list(conn, task_id=task_id)
finally:
conn.close()
self._send_json(200, {"approvals": items})
def _create_approval(self, user_id, task_id):
"""给某任务建一条待审批项(系统入口:未来 host hook 与
手动补录都走这里,实时 elicitation 本片未接,见
approvals 模块 docstring 的诚实边界)。
权限:项目 create_task(建审批项是写动作,与后续消息
同级)。kind 缺省 elicitation;summary 必填;
timeout_seconds 只接受 60 到 86400 秒整数,缺省 3600,
超出范围回 400。成功记审计 approval_created。"""
row = self._fetch_task(task_id)
if row is None:
self._send_json(404, {"error": "task not found"})
return
task = _task_to_dict(row)
with _auth_store(self.server.db_path) as store:
if not store.check_permission(user_id, task["project_id"],
"create_task"):
self._send_json(403, {"error": "forbidden"})
return
body, err = self._read_json()
if err:
self._send_json(400, {"error": err})
return
kind = body.get("kind", "elicitation")
summary = body.get("summary")
detail = body.get("detail")
if detail is not None and not isinstance(
detail, (dict, list, str)):
self._send_json(400, {"error": "detail must be an object"})
return
timeout = body.get("timeout_seconds", 3600)
if not isinstance(timeout, int) or isinstance(timeout, bool) \
or timeout < 60 or timeout > 86400:
self._send_json(
400, {"error": "timeout_seconds must be 60..86400"})
return
conn = self._db()
try:
try:
item = _appr_create(
conn, task_id, task["project_id"], kind=kind,
summary=summary, detail=detail,
requested_by=body.get("requested_by"),
timeout_seconds=timeout)
except ValueError as exc:
self._send_json(400, {"error": str(exc)})
return
finally:
conn.close()
with _auth_store(self.server.db_path) as store:
store.audit(user_id, task["project_id"],
"approval_created",
"%s %s" % (item["id"], item["kind"]))
self._send_json(201, {"approval": item})
def _decide_approval(self, user_id, prefix, decision):
"""批准或拒绝一条审批项:prefix 形如
<task_id>/approvals/<approval_id>(路由从 /approve
或 /deny 后缀剥出来的剩余部分)。
权限:项目 create_task(决定是写动作)。审批项不属于
该任务时回 404,不跨任务决定。已终态(批准、拒绝或
过期)时回 409 并带回执,原决定绝不被覆盖。成功记
审计 approval_decided,回执里有 decided_by 与
decided_at。"""
parts = prefix.split("/approvals/")
if len(parts) != 2 or not parts[0] or not parts[1]:
self._send_json(404, {"error": "not found"})
return
task_id, approval_id = parts
row = self._fetch_task(task_id)
if row is None:
self._send_json(404, {"error": "task not found"})
return
task = _task_to_dict(row)
with _auth_store(self.server.db_path) as store:
if not store.check_permission(user_id, task["project_id"],
"create_task"):
self._send_json(403, {"error": "forbidden"})
return
conn = self._db()
try:
_appr_ensure(conn)
_appr_expire(conn)
existing = _appr_get(conn, approval_id)
if existing is None or existing["task_id"] != task_id:
self._send_json(404, {"error": "approval not found"})
return
status, item = _appr_decide(
conn, approval_id, decision, user_id)
finally:
conn.close()
if status == "already_decided":
self._send_json(409, {"error": "already decided",
"approval": item})
return
with _auth_store(self.server.db_path) as store:
store.audit(user_id, task["project_id"],
"approval_decided",
"%s %s by %s" % (approval_id, decision, user_id))
self._send_json(200, {"approval": item})
def _continue_task(self, user_id, task_id):
"""旧任务接回来继续会话:终端任务(completed/failed/
interrupted/merge-failed)带一条新消息 + 全部选项重排队,
@@ -3566,8 +3887,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
def _workers_heartbeat(self, user_id):
"""worker 心跳:body 含 worker_id(必填,缺了 400)与可选
executor/capacity。心跳本身就是注册:worker_id 不存在时
自动注册,绝不 404。"""
executor/capacity/running。心跳本身就是注册:worker_id
不存在时自动注册,绝不 404。running 是 worker 上报的
当前占用槽数(启动中占位也算);未给时保留旧值。"""
body, err = self._read_json()
if err:
self._send_json(400, {"error": err})
@@ -3584,10 +3906,20 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
except (TypeError, ValueError):
self._send_json(400, {"error": "capacity must be an integer"})
return
running = body.get("running")
if running is not None:
try:
running = int(running)
except (TypeError, ValueError):
self._send_json(400, {"error": "running must be an integer"})
return
if running < 0:
self._send_json(400, {"error": "running must be >= 0"})
return
try:
worker = _registry_heartbeat(
self.server.db_path, worker_id,
executor=executor, capacity=capacity)
executor=executor, capacity=capacity, running=running)
except ValueError as exc:
self._send_json(400, {"error": str(exc)})
return
@@ -4534,6 +4866,106 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
self._send_json(200, result)
# -- webhooks(签名认证,不走 X-API-Key)---------------------------
# -- 自动化与分诊收件箱(P1 第 2 项)---------------------------
# 权限口径与其它项目级写操作同款:创建/取消用 create_task
# (不是 manage_project,因为自动化是项目成员的日常编排,
# 不是改项目归属/仓库这类 owner 专属操作;评审/审批等写
# 操作同样用 create_task);读用 read。状态变更记审计。
# 失败口径:校验失败 400;模块/数据库层意外失败时回 502
# 透传错误原文(截断 500 字),绝不回 201 假装建成功——
# 注意 _dispatch_safe 的兜底是 500,这里显式 502 优先于它。
def _automation_create(self, user_id, project_id):
"""创建自动化定义:只登记定义与到期状态,不跑模型、
不另起调度器(执行器后续走 claim 门,见 automations
模块文档)。body:name、prompt 必填;interval_seconds
可选(默认 86400,最小 60);repo_url 可选。"""
body, err = self._read_json()
if err:
self._send_json(400, {"error": err})
return
interval = body.get("interval_seconds", 86400)
if interval is None:
interval = 86400
conn = self._db()
try:
try:
auto = _auto_create(
conn, project_id, body.get("name"),
body.get("prompt"), user_id, interval,
body.get("repo_url"))
except ValueError as exc:
self._send_json(400, {"error": str(exc)})
return
except Exception as exc:
self._send_json(502, {"error": str(exc)[:500]})
return
finally:
conn.close()
with _auth_store(self.server.db_path) as store:
store.audit(user_id, project_id, "automation_created",
auto["id"])
self._send_json(201, auto)
def _automation_list(self, project_id):
"""列出某项目全部自动化定义(含已取消),附到期计数。"""
conn = self._db()
try:
try:
autos = _auto_list(conn, project_id)
except Exception as exc:
self._send_json(502, {"error": str(exc)[:500]})
return
conn.commit()
finally:
conn.close()
self._send_json(200, {
"project_id": project_id,
"automations": autos,
"due_count": sum(1 for a in autos if a["due"]),
})
def _automation_cancel(self, user_id, project_id, automation_id):
"""取消自动化(软删除):行不存在或属于别的项目时一
律 404(不泄露跨项目存在性);已取消时 409。"""
conn = self._db()
try:
try:
ok, auto = _auto_cancel(conn, automation_id)
except Exception as exc:
self._send_json(502, {"error": str(exc)[:500]})
return
finally:
conn.close()
if auto is None or auto["project_id"] != project_id:
self._send_json(404, {"error": "automation not found"})
return
if not ok:
self._send_json(409, {
"error": "automation already cancelled"})
return
with _auth_store(self.server.db_path) as store:
store.audit(user_id, project_id, "automation_cancelled",
automation_id)
self._send_json(200, auto)
def _triage_inbox(self, project_id):
"""分诊收件箱(只读):到期自动化、失败/中断任务、
待处理后续消息、最近自动化产出。见 automations 模块
triage_inbox 的字段说明。"""
conn = self._db()
try:
try:
_auto_ensure(conn)
_ensure_followups_table(conn)
inbox = _auto_triage(conn, project_id)
except Exception as exc:
self._send_json(502, {"error": str(exc)[:500]})
return
conn.commit()
finally:
conn.close()
self._send_json(200, inbox)
def _handle_github_webhook(self):
"""GitHub webhook:验签 -> delivery 去重 -> 解析 -> 执行建任务。
+157
View File
@@ -0,0 +1,157 @@
"""运行时审批队列模块单测:cloud/approvals.py。
不跑真实模型、不联网:内存 SQLite。覆盖批准放行、拒绝拦截、
回执留痕、二次决定不覆盖、超时默认拒绝、非法输入、回包映射。
"""
import sqlite3
import sys
import unittest
from pathlib import Path
try:
from cloud.approvals import (
command_response,
create_approval,
decide_approval,
elicitation_response,
ensure_approvals_table,
expire_overdue,
get_approval,
list_approvals,
)
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.approvals import (
command_response,
create_approval,
decide_approval,
elicitation_response,
ensure_approvals_table,
expire_overdue,
get_approval,
list_approvals,
)
class ApprovalsModuleTestCase(unittest.TestCase):
def setUp(self):
self.conn = sqlite3.connect(":memory:")
self.conn.row_factory = sqlite3.Row
ensure_approvals_table(self.conn)
self.conn.commit()
def tearDown(self):
self.conn.close()
def _mk(self, **kw):
kw.setdefault("task_id", "task-1")
kw.setdefault("project_id", "proj-1")
kw.setdefault("summary", "MCP 服务请求表单输入")
return create_approval(self.conn, **kw)
def test_create_defaults_to_pending_with_timeout(self):
item = self._mk()
self.assertEqual(item["status"], "pending")
self.assertEqual(item["kind"], "elicitation")
self.assertIsNone(item["decided_by"])
self.assertIsNone(item["decision"])
self.assertIsNotNone(item["expires_at"])
self.assertGreater(item["expires_at"], item["created_at"])
pending = list_approvals(self.conn, task_id="task-1",
status="pending")
self.assertEqual(len(pending), 1)
def test_approve_releases_with_receipt(self):
item = self._mk()
status, decided = decide_approval(
self.conn, item["id"], "approved", "user-9")
self.assertEqual(status, "decided")
self.assertEqual(decided["status"], "approved")
self.assertEqual(decided["decision"], "approved")
self.assertEqual(decided["decided_by"], "user-9")
self.assertIsNotNone(decided["decided_at"])
self.assertEqual(elicitation_response(decided["decision"]),
{"action": "accept", "content": {}})
self.assertEqual(command_response(decided["decision"]),
{"decision": "accept"})
def test_deny_blocks_with_receipt(self):
item = self._mk()
status, decided = decide_approval(
self.conn, item["id"], "denied", "user-9")
self.assertEqual(status, "decided")
self.assertEqual(decided["status"], "denied")
self.assertEqual(decided["decision"], "denied")
self.assertEqual(decided["decided_by"], "user-9")
self.assertEqual(elicitation_response(decided["decision"]),
{"action": "decline"})
self.assertEqual(command_response(decided["decision"]),
{"decision": "decline"})
def test_second_decision_never_overwrites_first(self):
item = self._mk()
decide_approval(self.conn, item["id"], "approved", "user-9")
status, kept = decide_approval(
self.conn, item["id"], "denied", "user-7")
self.assertEqual(status, "already_decided")
self.assertEqual(kept["status"], "approved")
self.assertEqual(kept["decided_by"], "user-9")
def test_expire_overdue_defaults_to_deny_not_allow(self):
item = self._mk(now=1000.0, timeout_seconds=60)
self.assertEqual(item["expires_at"], 1060.0)
count = expire_overdue(self.conn, now=1061.0)
self.assertEqual(count, 1)
done = get_approval(self.conn, item["id"])
self.assertEqual(done["status"], "expired")
self.assertEqual(done["decision"], "denied")
self.assertEqual(done["decided_by"], "system-timeout")
self.assertEqual(elicitation_response(done["decision"]),
{"action": "decline"})
status, kept = decide_approval(
self.conn, item["id"], "approved", "user-9")
self.assertEqual(status, "already_decided")
self.assertEqual(kept["status"], "expired")
def test_not_yet_due_is_not_expired(self):
self._mk(now=1000.0, timeout_seconds=3600)
self.assertEqual(expire_overdue(self.conn, now=2000.0), 0)
self.assertEqual(len(list_approvals(self.conn, status="pending")),
1)
def test_decide_missing_is_not_found(self):
status, item = decide_approval(self.conn, "nope", "approved",
"user-9")
self.assertEqual(status, "not_found")
self.assertIsNone(item)
def test_invalid_inputs_raise_value_error(self):
with self.assertRaises(ValueError):
self._mk(kind="bogus")
with self.assertRaises(ValueError):
self._mk(summary=" ")
with self.assertRaises(ValueError):
self._mk(summary="x" * 2001)
with self.assertRaises(ValueError):
self._mk(timeout_seconds=-1)
with self.assertRaises(ValueError):
decide_approval(self.conn, "x", "maybe", "user-9")
def test_ensure_is_idempotent_and_keeps_rows(self):
item = self._mk()
ensure_approvals_table(self.conn)
self.conn.commit()
self.assertEqual(get_approval(self.conn, item["id"])["id"],
item["id"])
def test_response_mapping_never_defaults_to_allow(self):
for weird in ("expired", "bogus", None, ""):
self.assertEqual(elicitation_response(weird),
{"action": "decline"})
self.assertEqual(command_response(weird),
{"decision": "decline"})
if __name__ == "__main__":
unittest.main()
+222
View File
@@ -0,0 +1,222 @@
"""运行时审批队列服务端单测:审批端点。
不跑真实模型、不联网:本地临时端口 + 临时 SQLite,与
test_peer_messages_server.py 同款骨架。覆盖创建、列表、批准、
拒绝、回执、二次决定 409、过期扫荡、权限、校验、跨任务隔离。
"""
import http.client
import json
import os
import sqlite3
import sys
import tempfile
import threading
import time
import unittest
from pathlib import Path
try:
from cloud.approvals import ensure_approvals_table
from cloud.auth import AuthStore
from cloud.server import create_server
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.approvals import ensure_approvals_table
from cloud.auth import AuthStore
from cloud.server import create_server
class ApprovalsServerTestCase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db_path = os.path.join(self.tmp.name, "control.db")
self.auth = AuthStore(self.db_path)
self.owner_id, self.owner_key = self.auth.create_user("owner")
self.outsider_id, self.outsider_key = self.auth.create_user(
"outsider")
self.server = create_server(self.db_path, "127.0.0.1", 0)
self.port = self.server.server_address[1]
self.thread = threading.Thread(
target=self.server.serve_forever, daemon=True)
self.thread.start()
status, self.project = self._req(
"POST", "/projects", {"name": "demo"}, self.owner_key)
self.assertEqual(status, 201)
def tearDown(self):
self.server.shutdown()
self.server.server_close()
self.thread.join(timeout=5)
try:
self.auth.conn.close()
except Exception:
pass
self.tmp.cleanup()
def _req(self, method, path, body=None, key=None):
conn = http.client.HTTPConnection(
"127.0.0.1", self.port, timeout=10)
hdrs = {}
payload = None
if body is not None:
payload = json.dumps(body).encode("utf-8")
hdrs["Content-Type"] = "application/json"
if key is not None:
hdrs["X-API-Key"] = key
try:
conn.request(method, path, body=payload, headers=hdrs)
resp = conn.getresponse()
data = resp.read()
finally:
conn.close()
try:
parsed = json.loads(data.decode("utf-8")) if data else None
except ValueError:
parsed = None
return resp.status, parsed
def _create_task(self, project_id):
status, task = self._req(
"POST", "/tasks",
{"project_id": project_id, "prompt": "做事"}, self.owner_key)
self.assertEqual(status, 201, task)
return task
def _create_approval(self, task_id, **extra):
body = {"summary": "MCP 服务请求表单输入"}
body.update(extra)
return self._req(
"POST", "/tasks/%s/approvals" % task_id, body, self.owner_key)
def _insert_expired_row(self, task_id, project_id):
conn = sqlite3.connect(self.db_path)
try:
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(?, ?, ?, 'elicitation', '过期项', NULL,"
" 'pending', NULL, NULL, NULL, NULL, ?, ?, ?)",
("expired-1", task_id, project_id,
time.time() - 7200, time.time() - 7200,
time.time() - 3600))
conn.commit()
finally:
conn.close()
def test_create_list_approve_receipt(self):
task = self._create_task(self.project["id"])
status, body = self._create_approval(task["id"])
self.assertEqual(status, 201, body)
item = body["approval"]
self.assertEqual(item["status"], "pending")
self.assertEqual(item["kind"], "elicitation")
status, body = self._req(
"GET", "/tasks/%s/approvals" % task["id"],
None, self.owner_key)
self.assertEqual(status, 200)
self.assertEqual(len(body["approvals"]), 1)
status, body = self._req(
"POST", "/tasks/%s/approvals/%s/approve" % (
task["id"], item["id"]), {}, self.owner_key)
self.assertEqual(status, 200, body)
decided = body["approval"]
self.assertEqual(decided["status"], "approved")
self.assertEqual(decided["decision"], "approved")
self.assertEqual(decided["decided_by"], self.owner_id)
self.assertIsNotNone(decided["decided_at"])
status, body = self._req(
"GET", "/tasks/%s/approvals" % task["id"],
None, self.owner_key)
self.assertEqual(body["approvals"][0]["status"], "approved")
def test_deny_and_second_decision_conflicts(self):
task = self._create_task(self.project["id"])
_, body = self._create_approval(task["id"])
item = body["approval"]
status, body = self._req(
"POST", "/tasks/%s/approvals/%s/deny" % (
task["id"], item["id"]), {}, self.owner_key)
self.assertEqual(status, 200)
self.assertEqual(body["approval"]["status"], "denied")
self.assertEqual(body["approval"]["decided_by"], self.owner_id)
status, body = self._req(
"POST", "/tasks/%s/approvals/%s/approve" % (
task["id"], item["id"]), {}, self.owner_key)
self.assertEqual(status, 409, body)
self.assertEqual(body["approval"]["status"], "denied")
def test_list_sweeps_overdue_to_expired_deny(self):
task = self._create_task(self.project["id"])
self._insert_expired_row(task["id"], self.project["id"])
status, body = self._req(
"GET", "/tasks/%s/approvals" % task["id"],
None, self.owner_key)
self.assertEqual(status, 200)
self.assertEqual(len(body["approvals"]), 1)
swept = body["approvals"][0]
self.assertEqual(swept["status"], "expired")
self.assertEqual(swept["decision"], "denied")
self.assertEqual(swept["decided_by"], "system-timeout")
def test_validation_and_not_found(self):
task = self._create_task(self.project["id"])
base = "/tasks/%s/approvals" % task["id"]
status, _ = self._req("POST", base, {}, self.owner_key)
self.assertEqual(status, 400)
status, _ = self._req(
"POST", base, {"summary": "x", "kind": "bogus"},
self.owner_key)
self.assertEqual(status, 400)
status, _ = self._req(
"POST", base, {"summary": "x", "timeout_seconds": 10},
self.owner_key)
self.assertEqual(status, 400)
status, _ = self._req(
"POST", base, {"summary": "x", "timeout_seconds": 90000},
self.owner_key)
self.assertEqual(status, 400)
status, _ = self._req(
"GET", "/tasks/nope/approvals", None, self.owner_key)
self.assertEqual(status, 404)
status, _ = self._req(
"POST", base, {"summary": "x"}, None)
self.assertEqual(status, 401)
def test_forbidden_for_outsider(self):
task = self._create_task(self.project["id"])
status, body = self._req(
"POST", "/tasks/%s/approvals" % task["id"],
{"summary": "x"}, self.outsider_key)
self.assertEqual(status, 403)
status, _ = self._req(
"GET", "/tasks/%s/approvals" % task["id"],
None, self.outsider_key)
self.assertEqual(status, 403)
_, body = self._create_approval(task["id"])
item = body["approval"]
status, _ = self._req(
"POST", "/tasks/%s/approvals/%s/approve" % (
task["id"], item["id"]), {}, self.outsider_key)
self.assertEqual(status, 403)
def test_cross_task_approval_id_is_not_found(self):
task_a = self._create_task(self.project["id"])
task_b = self._create_task(self.project["id"])
_, body = self._create_approval(task_a["id"])
item = body["approval"]
status, _ = self._req(
"POST", "/tasks/%s/approvals/%s/approve" % (
task_b["id"], item["id"]), {}, self.owner_key)
self.assertEqual(status, 404)
status, body = self._req(
"GET", "/tasks/%s/approvals" % task_b["id"],
None, self.owner_key)
self.assertEqual(body["approvals"], [])
if __name__ == "__main__":
unittest.main()
+232
View File
@@ -0,0 +1,232 @@
"""自动化与分诊收件箱模块单测(cloud/automations.py)。
不跑真实模型、不联网:临时 SQLite 文件 + 模块函数直调。
覆盖:建表幂等、创建校验与同名规则、到期边界、claim 抢单
防重、取消语义、产出记录、分诊四段聚合。
"""
import os
import sqlite3
import sys
import tempfile
import time
import unittest
from pathlib import Path
try:
from cloud.automations import (
cancel_automation,
claim_due_automation,
create_automation,
due_automations,
ensure_automations_tables,
get_automation,
is_due,
list_automation_runs,
list_automations,
record_automation_run,
triage_inbox,
)
from cloud.peer_messages import ensure_followups_table
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.automations import (
cancel_automation,
claim_due_automation,
create_automation,
due_automations,
ensure_automations_tables,
get_automation,
is_due,
list_automation_runs,
list_automations,
record_automation_run,
triage_inbox,
)
from cloud.peer_messages import ensure_followups_table
class AutomationsTestCase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db_path = os.path.join(self.tmp.name, "control.db")
self.conn = sqlite3.connect(self.db_path)
self.conn.row_factory = sqlite3.Row
self.conn.execute(
"CREATE TABLE tasks(id TEXT PRIMARY KEY, project_id TEXT,"
" prompt TEXT, status TEXT, created_at DOUBLE PRECISION,"
" updated_at DOUBLE PRECISION, error_text TEXT)")
ensure_automations_tables(self.conn)
ensure_followups_table(self.conn)
self.conn.commit()
self.now = 1700000000.0
self.project = "proj-1"
def tearDown(self):
self.conn.close()
self.tmp.cleanup()
def _make(self, name="daily", interval=3600, **kw):
kw.setdefault("project_id", self.project)
return create_automation(
self.conn, kw.pop("project_id"), name, "做每日检查",
"user-1", interval, now=self.now, **kw)
def test_ensure_is_idempotent(self):
ensure_automations_tables(self.conn)
ensure_automations_tables(self.conn)
self.conn.commit()
rows = self.conn.execute(
"SELECT name FROM sqlite_master WHERE type = 'table'"
" AND name IN ('automations', 'automation_runs')").fetchall()
self.assertEqual(len(rows), 2)
def test_create_fields_and_due_immediately(self):
auto = self._make()
self.assertEqual(auto["status"], "active")
self.assertEqual(auto["interval_seconds"], 3600)
self.assertIsNone(auto["last_run_at"])
self.assertEqual(auto["next_run_at"], self.now)
self.assertTrue(auto["due"])
self.assertTrue(is_due(auto, self.now))
got = get_automation(self.conn, auto["id"])
self.assertEqual(got["name"], "daily")
self.assertIsNone(get_automation(self.conn, "missing"))
def test_create_validation(self):
for name, prompt, interval in (
("", "x", 3600),
(" ", "x", 3600),
("n", "", 3600),
("n", "x", 59),
("n", "x", True),
("n", "x", 3600.5),
("x" * 101, "x", 3600),
("n", "x" * 20001, 3600)):
with self.assertRaises(ValueError, msg=(name, interval)):
create_automation(
self.conn, self.project, name, prompt, "user-1",
interval, now=self.now)
def test_duplicate_active_name_rejected_reusable_after_cancel(self):
self._make(name="dup")
with self.assertRaises(ValueError):
self._make(name="dup")
autos = list_automations(self.conn, self.project, self.now)
self.assertEqual(len(autos), 1)
ok, cancelled = cancel_automation(
self.conn, autos[0]["id"], self.now)
self.assertTrue(ok)
self.assertEqual(cancelled["status"], "cancelled")
again = self._make(name="dup")
self.assertEqual(again["status"], "active")
def test_due_boundary_matches_host_rundue_rule(self):
auto = self._make(interval=3600)
self.conn.execute(
"UPDATE automations SET last_run_at = ? WHERE id = ?",
(self.now - 3599, auto["id"]))
self.conn.commit()
self.assertEqual(due_automations(
self.conn, self.project, self.now), [])
self.conn.execute(
"UPDATE automations SET last_run_at = ? WHERE id = ?",
(self.now - 3600, auto["id"]))
self.conn.commit()
due = due_automations(self.conn, self.project, self.now)
self.assertEqual([d["id"] for d in due], [auto["id"]])
def test_claim_is_the_duplicate_gate(self):
auto = self._make(interval=3600)
self.assertTrue(claim_due_automation(
self.conn, auto["id"], self.now))
# 同一间隔内第二次抢单必须失败:这就是防重复执行。
self.assertFalse(claim_due_automation(
self.conn, auto["id"], self.now))
self.assertFalse(claim_due_automation(
self.conn, auto["id"], self.now + 3599))
# 下个间隔到了才能再抢。
self.assertTrue(claim_due_automation(
self.conn, auto["id"], self.now + 3600))
self.assertFalse(claim_due_automation(
self.conn, "missing", self.now))
ok, _ = cancel_automation(self.conn, auto["id"], self.now)
self.assertTrue(ok)
self.assertFalse(claim_due_automation(
self.conn, auto["id"], self.now + 7200))
def test_cancel_semantics(self):
ok, auto = cancel_automation(self.conn, "missing", self.now)
self.assertFalse(ok)
self.assertIsNone(auto)
made = self._make()
ok, auto = cancel_automation(self.conn, made["id"], self.now)
self.assertTrue(ok)
self.assertEqual(auto["status"], "cancelled")
self.assertFalse(auto["due"])
ok, auto = cancel_automation(self.conn, made["id"], self.now)
self.assertFalse(ok)
self.assertEqual(auto["status"], "cancelled")
def test_record_run_and_list_runs(self):
auto = self._make()
run = record_automation_run(
self.conn, auto["id"], self.project, task_id="task-1",
status="created", detail="claimed", now=self.now)
self.assertEqual(run["task_id"], "task-1")
self.assertEqual(run["status"], "created")
runs = list_automation_runs(self.conn, auto["id"])
self.assertEqual(len(runs), 1)
with self.assertRaises(ValueError):
record_automation_run(
self.conn, "missing", self.project, now=self.now)
def test_triage_inbox_four_sections(self):
auto = self._make()
other = self._make(name="weekly", interval=86400)
cancel_automation(self.conn, other["id"], self.now)
self.conn.execute(
"INSERT INTO tasks(id, project_id, prompt, status,"
" created_at, updated_at, error_text)"
" VALUES('t-fail', ?, '坏任务', 'failed', ?, ?, 'boom')",
(self.project, self.now, self.now))
self.conn.execute(
"INSERT INTO tasks(id, project_id, prompt, status,"
" created_at, updated_at, error_text)"
" VALUES('t-run', ?, '跑中任务', 'running', ?, ?, NULL)",
(self.project, self.now, self.now))
self.conn.execute(
"INSERT INTO task_followups(id, task_id, project_id, body,"
" mode, status, created_by, sender_task_id, created_at,"
" updated_at) VALUES('f-1', 't-run', ?, '请看一下',"
" 'queued', 'pending', 'user-1', NULL, ?, ?)",
(self.project, self.now, self.now))
self.conn.execute(
"INSERT INTO tasks(id, project_id, prompt, status,"
" created_at, updated_at, error_text)"
" VALUES('t-done', ?, '已完成', 'completed', ?, ?, NULL)",
(self.project, self.now, self.now))
record_automation_run(
self.conn, auto["id"], self.project, task_id="t-done",
status="created", detail="", now=self.now)
self.conn.commit()
inbox = triage_inbox(self.conn, self.project, self.now)
self.assertEqual(
[d["id"] for d in inbox["automations_due"]], [auto["id"]])
self.assertEqual(
[t["id"] for t in inbox["failed_tasks"]], ["t-fail"])
self.assertEqual(
[f["id"] for f in inbox["pending_followups"]], ["f-1"])
self.assertEqual(len(inbox["recent_runs"]), 1)
self.assertEqual(inbox["recent_runs"][0]["task_status"],
"completed")
self.assertEqual(inbox["counts"], {
"automations_due": 1,
"failed_tasks": 1,
"pending_followups": 1,
"recent_runs": 1,
})
if __name__ == "__main__":
unittest.main()
+238
View File
@@ -0,0 +1,238 @@
"""自动化与分诊收件箱服务端单测:POST/GET /projects/{id}/
automations、POST /projects/{id}/automations/{aid}/cancel、
GET /projects/{id}/triage。
不跑真实模型、不联网:本地临时端口 + 临时 SQLite,与
test_peer_messages_server.py 同款骨架。覆盖鉴权、权限、
校验、取消语义(404/409)、分诊聚合与审计。
"""
import http.client
import json
import os
import sqlite3
import sys
import tempfile
import threading
import unittest
from pathlib import Path
try:
from cloud.auth import AuthStore
from cloud.automations import record_automation_run
from cloud.server import create_server
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.auth import AuthStore
from cloud.automations import record_automation_run
from cloud.server import create_server
class AutomationServerTestCase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db_path = os.path.join(self.tmp.name, "control.db")
self.auth = AuthStore(self.db_path)
self.owner_id, self.owner_key = self.auth.create_user("owner")
self.outsider_id, self.outsider_key = self.auth.create_user(
"outsider")
self.server = create_server(self.db_path, "127.0.0.1", 0)
self.port = self.server.server_address[1]
self.thread = threading.Thread(
target=self.server.serve_forever, daemon=True)
self.thread.start()
status, self.project = self._req(
"POST", "/projects", {"name": "demo"}, self.owner_key)
self.assertEqual(status, 201)
status, self.project2 = self._req(
"POST", "/projects", {"name": "other"}, self.owner_key)
self.assertEqual(status, 201)
def tearDown(self):
self.server.shutdown()
self.server.server_close()
self.thread.join(timeout=5)
try:
self.auth.conn.close()
except Exception:
pass
self.tmp.cleanup()
def _req(self, method, path, body=None, key=None):
conn = http.client.HTTPConnection(
"127.0.0.1", self.port, timeout=10)
hdrs = {}
payload = None
if body is not None:
payload = json.dumps(body).encode("utf-8")
hdrs["Content-Type"] = "application/json"
if key is not None:
hdrs["X-API-Key"] = key
try:
conn.request(method, path, body=payload, headers=hdrs)
resp = conn.getresponse()
data = resp.read()
finally:
conn.close()
try:
parsed = json.loads(data.decode("utf-8")) if data else None
except ValueError:
parsed = None
return resp.status, parsed
def _autos_url(self, project_id=None):
return "/projects/%s/automations" % (
project_id or self.project["id"])
def _create_auto(self, project_id=None, **extra):
body = {"name": "daily", "prompt": "做每日检查"}
body.update(extra)
return self._req(
"POST", self._autos_url(project_id), body, self.owner_key)
def _audit_actions(self, project_id):
conn = sqlite3.connect(self.db_path)
try:
rows = conn.execute(
"SELECT action FROM audit_log WHERE project_id = ?",
(project_id,)).fetchall()
finally:
conn.close()
return [r[0] for r in rows]
def _create_task(self, project_id):
status, task = self._req(
"POST", "/tasks",
{"project_id": project_id, "prompt": "做事"}, self.owner_key)
self.assertEqual(status, 201, task)
return task
def _set_status(self, task_id, status):
conn = sqlite3.connect(self.db_path)
try:
conn.execute("UPDATE tasks SET status = ? WHERE id = ?",
(status, task_id))
conn.commit()
finally:
conn.close()
def test_create_and_list(self):
status, auto = self._create_auto(interval_seconds=3600)
self.assertEqual(status, 201, auto)
self.assertEqual(auto["status"], "active")
self.assertEqual(auto["interval_seconds"], 3600)
self.assertTrue(auto["due"])
self.assertIn("automation_created",
self._audit_actions(self.project["id"]))
status, body = self._req(
"GET", self._autos_url(), None, self.owner_key)
self.assertEqual(status, 200)
self.assertEqual(len(body["automations"]), 1)
self.assertEqual(body["due_count"], 1)
self.assertEqual(body["project_id"], self.project["id"])
def test_create_defaults_and_validation(self):
status, auto = self._create_auto()
self.assertEqual(status, 201, auto)
self.assertEqual(auto["interval_seconds"], 86400)
for body in (
{"prompt": "x"},
{"name": "n"},
{"name": "n", "prompt": "x", "interval_seconds": 30},
{"name": "n", "prompt": "x",
"interval_seconds": "3600"}):
status, _ = self._req(
"POST", self._autos_url(), body, self.owner_key)
self.assertEqual(status, 400, body)
status, _ = self._create_auto()
self.assertEqual(status, 400)
def test_auth_and_permission(self):
status, _ = self._req(
"POST", self._autos_url(),
{"name": "n", "prompt": "x"}, None)
self.assertEqual(status, 401)
status, _ = self._req(
"GET", self._autos_url(), None, None)
self.assertEqual(status, 401)
status, _ = self._req(
"POST", self._autos_url(),
{"name": "n", "prompt": "x"}, self.outsider_key)
self.assertEqual(status, 403)
status, _ = self._req(
"GET", self._autos_url(), None, self.outsider_key)
self.assertEqual(status, 403)
status, _ = self._req(
"POST", "/projects/nope/automations",
{"name": "n", "prompt": "x"}, self.owner_key)
self.assertEqual(status, 404)
def test_cancel_semantics(self):
_, auto = self._create_auto()
url = "%s/%s/cancel" % (self._autos_url(), auto["id"])
status, body = self._req("POST", url, {}, self.owner_key)
self.assertEqual(status, 200, body)
self.assertEqual(body["status"], "cancelled")
self.assertFalse(body["due"])
self.assertIn("automation_cancelled",
self._audit_actions(self.project["id"]))
status, _ = self._req("POST", url, {}, self.owner_key)
self.assertEqual(status, 409)
status, _ = self._req(
"POST", "%s/missing/cancel" % self._autos_url(), {},
self.owner_key)
self.assertEqual(status, 404)
# 别的项目的自动化 id 在本项目下取消也一律 404。
_, other = self._create_auto(project_id=self.project2["id"])
status, _ = self._req(
"POST", "%s/%s/cancel" % (self._autos_url(), other["id"]),
{}, self.owner_key)
self.assertEqual(status, 404)
status, _ = self._req(
"POST", url, {}, self.outsider_key)
self.assertEqual(status, 403)
def test_triage_inbox(self):
_, auto = self._create_auto()
task = self._create_task(self.project["id"])
self._set_status(task["id"], "failed")
runner = self._create_task(self.project["id"])
self._set_status(runner["id"], "running")
status, _ = self._req(
"POST", "/tasks/%s/followups" % runner["id"],
{"body": "请看一下", "mode": "queued"}, self.owner_key)
self.assertEqual(status, 201)
conn = sqlite3.connect(self.db_path)
try:
record_automation_run(
conn, auto["id"], self.project["id"],
task_id=task["id"], status="created", detail="")
finally:
conn.close()
status, body = self._req(
"GET", "/projects/%s/triage" % self.project["id"], None,
self.owner_key)
self.assertEqual(status, 200, body)
self.assertEqual(
[d["id"] for d in body["automations_due"]], [auto["id"]])
self.assertEqual(
[t["id"] for t in body["failed_tasks"]], [task["id"]])
self.assertEqual(len(body["pending_followups"]), 1)
self.assertEqual(
body["pending_followups"][0]["task_id"], runner["id"])
self.assertEqual(len(body["recent_runs"]), 1)
self.assertEqual(body["recent_runs"][0]["task_status"],
"failed")
self.assertEqual(body["counts"]["automations_due"], 1)
status, _ = self._req(
"GET", "/projects/%s/triage" % self.project["id"], None,
self.outsider_key)
self.assertEqual(status, 403)
status, _ = self._req(
"GET", "/projects/%s/triage" % self.project["id"], None,
None)
self.assertEqual(status, 401)
if __name__ == "__main__":
unittest.main()
+223
View File
@@ -0,0 +1,223 @@
"""并发上限单测(caps)。
覆盖:配置解析优先级与默认值、认领门的未满 / 已满 / 释放后
恢复、启动中任务计入占用、CAS 抢输后占位不泄漏、缺 key 门
提前返回后占位释放、双线程同时认领只赢一个。
只用注入的假环境与本地 SQLite,不跑真实模型、不联网。
"""
import os
import sqlite3
import sys
import tempfile
import threading
import time
import unittest
from pathlib import Path
try:
from cloud.caps import (
can_claim,
effective_capacity,
resolve_capacity,
running_count,
)
from cloud.server import init_db
from cloud.worker import Worker
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.caps import (
can_claim,
effective_capacity,
resolve_capacity,
running_count,
)
from cloud.server import init_db
from cloud.worker import Worker
def _insert_task(db_path, task_id, project_id="p1"):
now = time.time()
conn = sqlite3.connect(db_path)
try:
conn.execute(
"INSERT INTO tasks(id, project_id, prompt, status,"
" idempotency_key, depends_on, priority, created_by,"
" created_at, updated_at, last_heartbeat)"
" VALUES(?, ?, ?, 'queued', NULL, '[]', 0, 'tester', ?, ?, NULL)",
(task_id, project_id, "do work", now, now),
)
conn.commit()
finally:
conn.close()
def _status(db_path, task_id):
conn = sqlite3.connect(db_path)
try:
row = conn.execute(
"SELECT status FROM tasks WHERE id = ?",
(task_id,)).fetchone()
finally:
conn.close()
return row[0] if row else None
class CapsResolveTestCase(unittest.TestCase):
def test_explicit_beats_env(self):
env = {"CLOUD_WORKER_CAPACITY": "3"}
self.assertEqual(resolve_capacity(2, env=env), 2)
def test_env_used_when_no_explicit(self):
env = {"CLOUD_WORKER_CAPACITY": "3"}
self.assertEqual(resolve_capacity(None, env=env), 3)
def test_default_is_1(self):
self.assertEqual(resolve_capacity(None, env={}), 1)
self.assertEqual(resolve_capacity(), 1)
def test_invalid_values_fall_back_to_1(self):
self.assertEqual(resolve_capacity(0, env={}), 1)
self.assertEqual(resolve_capacity(-2, env={}), 1)
self.assertEqual(resolve_capacity("abc", env={}), 1)
self.assertEqual(resolve_capacity(None, env={
"CLOUD_WORKER_CAPACITY": "abc"}), 1)
self.assertEqual(resolve_capacity(None, env={
"CLOUD_WORKER_CAPACITY": "0"}), 1)
self.assertEqual(resolve_capacity(None, env={
"CLOUD_WORKER_CAPACITY": " "}), 1)
def test_effective_capacity(self):
self.assertEqual(effective_capacity(2), 2)
self.assertEqual(effective_capacity(0), 1)
self.assertEqual(effective_capacity(None), 1)
def test_can_claim_boundaries(self):
self.assertTrue(can_claim(0, 1))
self.assertFalse(can_claim(1, 1))
self.assertTrue(can_claim(1, 2))
self.assertFalse(can_claim(2, 2))
def test_running_count_counts_starting_entries(self):
# 空条目 = 已认领未起进程的启动中任务,同样占一个槽。
self.assertEqual(running_count({"a": {}, "b": {"proc": 1}}), 2)
self.assertEqual(running_count({}), 0)
class WorkerClaimGateTestCase(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.env = {"XCM_API_KEY": "test-placeholder"}
def tearDown(self):
self.tmp.cleanup()
def _worker(self, **kwargs):
kwargs.setdefault("env", self.env)
kwargs.setdefault("retry_backoff_seconds", 0)
kwargs.setdefault("max_retries", 0)
return Worker(self.db_path, "/fake/host.py", **kwargs)
def test_default_capacity_is_1(self):
worker = self._worker()
self.assertEqual(worker.capacity, 1)
def test_env_capacity_and_explicit_priority(self):
env = {"CLOUD_WORKER_CAPACITY": "3"}
self.assertEqual(self._worker(env=env).capacity, 3)
self.assertEqual(self._worker(env=env, capacity=2).capacity, 2)
def test_full_worker_refuses_claim_and_task_stays_queued(self):
_insert_task(self.db_path, "A")
_insert_task(self.db_path, "B")
worker = self._worker(capacity=1)
self.assertTrue(worker._claim("A"))
self.assertEqual(_status(self.db_path, "A"), "running")
self.assertFalse(worker._claim("B"))
self.assertEqual(_status(self.db_path, "B"), "queued")
self.assertEqual(set(worker._running_snapshot()), {"A"})
def test_release_restores_claim(self):
_insert_task(self.db_path, "A")
_insert_task(self.db_path, "B")
worker = self._worker(capacity=1)
self.assertTrue(worker._claim("A"))
self.assertFalse(worker._claim("B"))
worker._running_clear("A")
self.assertTrue(worker._claim("B"))
self.assertEqual(_status(self.db_path, "B"), "running")
def test_starting_task_counts_toward_capacity(self):
_insert_task(self.db_path, "B")
worker = self._worker(capacity=1)
# 空条目模拟“已认领、执行器还没起进程”的启动中任务。
worker._running_set("STARTING")
self.assertEqual(worker._running_count(), 1)
self.assertFalse(worker._claim("B"))
self.assertEqual(_status(self.db_path, "B"), "queued")
worker._running_clear("STARTING")
self.assertTrue(worker._claim("B"))
def test_capacity_2_allows_two_blocks_third(self):
for task_id in ("A", "B", "C"):
_insert_task(self.db_path, task_id)
worker = self._worker(capacity=2)
self.assertTrue(worker._claim("A"))
self.assertTrue(worker._claim("B"))
self.assertFalse(worker._claim("C"))
self.assertEqual(_status(self.db_path, "C"), "queued")
worker._running_clear("A")
self.assertTrue(worker._claim("C"))
def test_lost_cas_race_releases_slot(self):
_insert_task(self.db_path, "A")
_insert_task(self.db_path, "B")
first = self._worker(capacity=1)
second = self._worker(capacity=1)
self.assertTrue(first._claim("A"))
# second 的 CAS 抢输:返回 False,且它自己的占位已清,
# 不能留下一个空槽把自己堵死。
self.assertFalse(second._claim("A"))
self.assertEqual(second._running_snapshot(), {})
self.assertTrue(second._claim("B"))
def test_two_threads_claim_only_one_wins(self):
_insert_task(self.db_path, "A")
_insert_task(self.db_path, "B")
worker = self._worker(capacity=1)
barrier = threading.Barrier(2, timeout=5)
results = {}
def _try(task_id):
barrier.wait()
results[task_id] = worker._claim(task_id)
threads = [threading.Thread(target=_try, args=(t,))
for t in ("A", "B")]
for t in threads:
t.start()
for t in threads:
t.join(timeout=10)
self.assertEqual(sorted(results.values()), [False, True])
self.assertEqual(worker._running_count(), 1)
def test_missing_key_gate_releases_slot(self):
_insert_task(self.db_path, "A")
_insert_task(self.db_path, "B")
# env 里无 key:run_one 在缺 key 门提前返回 failed,
# 占位必须随之释放,否则 capacity=1 的 worker 会被堵死。
worker = self._worker(env={}, capacity=1)
result = worker.run_one()
self.assertTrue(result["dispatched"])
self.assertEqual(result["status"], "failed")
self.assertEqual(_status(self.db_path, "A"), "failed")
self.assertEqual(worker._running_snapshot(), {})
self.assertTrue(worker._claim("B"))
self.assertEqual(_status(self.db_path, "B"), "running")
if __name__ == "__main__":
unittest.main()
+125
View File
@@ -0,0 +1,125 @@
"""并发上限服务端面单测(/workers 占用展示)。
覆盖:GET /workers 行里含 capacity 与 running;心跳上报 running
后列表可见;心跳未给 running 时保留旧值;非法 running 400;
未认证 401;/workers 读权限口径是任何已认证用户(与既有
_workers_list 一致,不按项目成员过滤)。
只用标准库 http.client,不跑真实模型,不联网。
"""
import http.client
import json
import os
import sys
import tempfile
import threading
import unittest
from pathlib import Path
try:
from cloud.auth import AuthStore
from cloud.server import create_server
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.auth import AuthStore
from cloud.server import create_server
class CapsServerTestCase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db_path = os.path.join(self.tmp.name, "control.db")
self.auth = AuthStore(self.db_path)
self.owner_id, self.owner_key = self.auth.create_user("owner")
self.viewer_id, self.viewer_key = self.auth.create_user("viewer")
self.server = create_server(self.db_path, "127.0.0.1", 0)
self.port = self.server.server_address[1]
self.thread = threading.Thread(
target=self.server.serve_forever, daemon=True)
self.thread.start()
def tearDown(self):
self.server.shutdown()
self.server.server_close()
self.thread.join(timeout=5)
try:
self.auth.conn.close()
except Exception:
pass
self.tmp.cleanup()
def _req(self, method, path, body=None, key=None):
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
hdrs = {}
payload = None
if body is not None:
payload = json.dumps(body).encode("utf-8")
hdrs["Content-Type"] = "application/json"
if key is not None:
hdrs["X-API-Key"] = key
try:
conn.request(method, path, body=payload, headers=hdrs)
resp = conn.getresponse()
data = resp.read()
finally:
conn.close()
try:
parsed = json.loads(data.decode("utf-8")) if data else None
except ValueError:
parsed = None
return resp.status, parsed
def _workers(self, key):
status, body = self._req("GET", "/workers", key=key)
self.assertEqual(status, 200, body)
return {w["worker_id"]: w for w in body["workers"]}
def test_workers_list_shows_capacity_and_running(self):
status, body = self._req(
"POST", "/workers/heartbeat",
{"worker_id": "w1", "executor": "local",
"capacity": 4, "running": 2},
self.owner_key)
self.assertEqual(status, 200, body)
# 读权限:任何已认证用户可读(viewer 也能看到基础设施行)。
workers = self._workers(self.viewer_key)
self.assertEqual(workers["w1"]["capacity"], 4)
self.assertEqual(workers["w1"]["running"], 2)
def test_heartbeat_without_running_preserves_old(self):
self._req("POST", "/workers/heartbeat",
{"worker_id": "w1", "capacity": 2, "running": 1},
self.owner_key)
status, body = self._req(
"POST", "/workers/heartbeat", {"worker_id": "w1"},
self.owner_key)
self.assertEqual(status, 200, body)
self.assertEqual(body["running"], 1)
self.assertEqual(self._workers(self.owner_key)["w1"]["running"], 1)
def test_heartbeat_invalid_running_400(self):
status, _ = self._req(
"POST", "/workers/heartbeat",
{"worker_id": "w1", "running": "abc"}, self.owner_key)
self.assertEqual(status, 400)
status, _ = self._req(
"POST", "/workers/heartbeat",
{"worker_id": "w1", "running": -1}, self.owner_key)
self.assertEqual(status, 400)
def test_workers_requires_auth(self):
status, _ = self._req("GET", "/workers")
self.assertEqual(status, 401)
def test_heartbeat_registers_unknown_worker_with_defaults(self):
status, body = self._req(
"POST", "/workers/heartbeat", {"worker_id": "new1"},
self.owner_key)
self.assertEqual(status, 200, body)
self.assertEqual(body["capacity"], 1)
self.assertEqual(body["running"], 0)
if __name__ == "__main__":
unittest.main()
+6 -3
View File
@@ -240,12 +240,13 @@ class TaskExecutorFilterTestCase(unittest.TestCase):
def tearDown(self):
self.tmp.cleanup()
def _docker_worker(self):
def _docker_worker(self, capacity=None):
return Worker(
self.db_path, "host.py", config_path=None, repo=None,
executor=DockerExecutor(runner=lambda cmd: _result()),
env={"XCM_API_KEY": "test-placeholder"}, env_key=None,
task_timeout=1800, max_retries=0, retry_backoff_seconds=0)
task_timeout=1800, max_retries=0, retry_backoff_seconds=0,
capacity=capacity)
def test_claim_rejects_other_pool_task(self):
worker = self._docker_worker()
@@ -260,7 +261,9 @@ class TaskExecutorFilterTestCase(unittest.TestCase):
self.assertEqual(status, "queued")
def test_claim_accepts_unassigned_and_same_pool(self):
worker = self._docker_worker()
# 本测试连续认领两个任务, 测的是执行器过滤不是上限;
# caps 默认上限 1 会挡第二单, 故显式给 capacity=2。
worker = self._docker_worker(capacity=2)
self.assertTrue(worker._claim("t-none", executor="docker"))
self.assertTrue(worker._claim("t-docker", executor="docker"))
+134
View File
@@ -0,0 +1,134 @@
"""自动代码评审模块单测:cloud/review.py。
不跑真实模型、不联网:纯 sqlite 内存库 + 协议目标构造。
覆盖四种 ReviewTarget、非法组合、登记/队列/取单条/写回
状态机(findings 写回默认 completed、显式 failed 可覆盖)。
"""
import sqlite3
import sys
import unittest
from pathlib import Path
try:
from cloud.review import build_review_target
from cloud.review import create_review
from cloud.review import ensure_reviews_table
from cloud.review import get_review
from cloud.review import list_project_reviews
from cloud.review import list_reviews
from cloud.review import update_review
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.review import build_review_target
from cloud.review import create_review
from cloud.review import ensure_reviews_table
from cloud.review import get_review
from cloud.review import list_project_reviews
from cloud.review import list_reviews
from cloud.review import update_review
class ReviewTargetTestCase(unittest.TestCase):
def test_four_targets(self):
self.assertEqual(
build_review_target("uncommittedChanges"),
{"type": "uncommittedChanges"})
self.assertEqual(
build_review_target("baseBranch", branch="main"),
{"type": "baseBranch", "branch": "main"})
self.assertEqual(
build_review_target("commit", sha="abc123"),
{"type": "commit", "sha": "abc123"})
self.assertEqual(
build_review_target("custom", instructions="look closely"),
{"type": "custom", "instructions": "look closely"})
def test_target_validation(self):
with self.assertRaises(ValueError):
build_review_target("baseBranch")
with self.assertRaises(ValueError):
build_review_target("baseBranch", branch=" ")
with self.assertRaises(ValueError):
build_review_target("commit")
with self.assertRaises(ValueError):
build_review_target("custom")
with self.assertRaises(ValueError):
build_review_target("nope")
def test_whitespace_trimmed(self):
self.assertEqual(
build_review_target("baseBranch", branch=" main "),
{"type": "baseBranch", "branch": "main"})
class ReviewStoreTestCase(unittest.TestCase):
def setUp(self):
self.conn = sqlite3.connect(":memory:")
ensure_reviews_table(self.conn)
def tearDown(self):
self.conn.close()
def test_create_and_list(self):
first = create_review(
self.conn, "task-a", "proj-a", "uncommittedChanges",
"owner-1")
self.assertEqual(first["status"], "requested")
self.assertIsNone(first["findings"])
self.assertEqual(first["task_id"], "task-a")
second = create_review(
self.conn, "task-a", "proj-a", "baseBranch", "owner-1",
branch="main")
self.assertEqual(second["target_type"], "baseBranch")
self.assertEqual(second["branch"], "main")
queue = list_reviews(self.conn, "task-a")
self.assertEqual(
[r["id"] for r in queue], [first["id"], second["id"]])
self.assertEqual(list_reviews(self.conn, "task-other"), [])
proj = list_project_reviews(self.conn, "proj-a")
self.assertEqual(len(proj), 2)
def test_create_invalid_target_writes_nothing(self):
with self.assertRaises(ValueError):
create_review(
self.conn, "task-a", "proj-a", "commit", "owner-1")
self.assertEqual(list_reviews(self.conn, "task-a"), [])
def test_get_missing_returns_none(self):
self.assertIsNone(get_review(self.conn, "no-such-id"))
def test_findings_writeback_defaults_completed(self):
record = create_review(
self.conn, "task-a", "proj-a", "custom", "owner-1",
instructions="check it")
updated = update_review(
self.conn, record["id"], findings="two issues found")
self.assertEqual(updated["status"], "completed")
self.assertEqual(updated["findings"], "two issues found")
self.assertEqual(
get_review(self.conn, record["id"])["status"], "completed")
def test_explicit_failed_status_kept(self):
record = create_review(
self.conn, "task-a", "proj-a", "uncommittedChanges",
"owner-1")
updated = update_review(
self.conn, record["id"], status="failed",
findings="reviewer crashed")
self.assertEqual(updated["status"], "failed")
def test_update_validation(self):
record = create_review(
self.conn, "task-a", "proj-a", "uncommittedChanges",
"owner-1")
with self.assertRaises(ValueError):
update_review(self.conn, record["id"], status="bogus")
with self.assertRaises(ValueError):
update_review(self.conn, record["id"], findings=123)
self.assertIsNone(
update_review(self.conn, "no-such-id", findings="x"))
if __name__ == "__main__":
unittest.main()
+240
View File
@@ -0,0 +1,240 @@
"""自动代码评审服务端单测:POST /tasks/<id>/reviews、
GET /tasks/<id>/reviews、POST /tasks/<id>/reviews/<rid>/findings。
不跑真实模型、不联网:本地临时端口 + 临时 SQLite,与
test_peer_messages_server.py 同款骨架。覆盖鉴权、权限、
目标校验、队列查看、findings 写回与跨任务隔离。
"""
import http.client
import json
import os
import sys
import tempfile
import threading
import unittest
from pathlib import Path
try:
from cloud.auth import AuthStore
from cloud.server import create_server
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.auth import AuthStore
from cloud.server import create_server
class ReviewServerTestCase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db_path = os.path.join(self.tmp.name, "control.db")
self.auth = AuthStore(self.db_path)
self.owner_id, self.owner_key = self.auth.create_user("owner")
self.outsider_id, self.outsider_key = self.auth.create_user(
"outsider")
self.server = create_server(self.db_path, "127.0.0.1", 0)
self.port = self.server.server_address[1]
self.thread = threading.Thread(
target=self.server.serve_forever, daemon=True)
self.thread.start()
self.trace_dir = os.path.join(self.tmp.name, "host-traces")
os.makedirs(self.trace_dir, exist_ok=True)
self.server.trace_dir = self.trace_dir
status, self.project = self._req(
"POST", "/projects", {"name": "demo"}, self.owner_key)
self.assertEqual(status, 201)
status, self.project2 = self._req(
"POST", "/projects", {"name": "other"}, self.owner_key)
self.assertEqual(status, 201)
def tearDown(self):
self.server.shutdown()
self.server.server_close()
self.thread.join(timeout=5)
try:
self.auth.conn.close()
except Exception:
pass
self.tmp.cleanup()
def _req(self, method, path, body=None, key=None):
conn = http.client.HTTPConnection(
"127.0.0.1", self.port, timeout=10)
hdrs = {}
payload = None
if body is not None:
payload = json.dumps(body).encode("utf-8")
hdrs["Content-Type"] = "application/json"
if key is not None:
hdrs["X-API-Key"] = key
try:
conn.request(method, path, body=payload, headers=hdrs)
resp = conn.getresponse()
data = resp.read()
finally:
conn.close()
try:
parsed = json.loads(data.decode("utf-8")) if data else None
except ValueError:
parsed = None
return resp.status, parsed
def _create_task(self, project_id):
status, task = self._req(
"POST", "/tasks",
{"project_id": project_id, "prompt": "做事"},
self.owner_key)
self.assertEqual(status, 201, task)
return task
def test_create_default_and_list(self):
task = self._create_task(self.project["id"])
status, created = self._req(
"POST", "/tasks/%s/reviews" % task["id"], {},
self.owner_key)
self.assertEqual(status, 201, created)
self.assertEqual(created["status"], "requested")
self.assertIsNone(created["findings"])
self.assertEqual(created["target_type"], "uncommittedChanges")
status, body = self._req(
"GET", "/tasks/%s/reviews" % task["id"], None,
self.owner_key)
self.assertEqual(status, 200, body)
self.assertEqual(len(body["reviews"]), 1)
self.assertEqual(body["reviews"][0]["id"], created["id"])
def test_create_all_targets(self):
task = self._create_task(self.project["id"])
cases = [
({"target_type": "baseBranch", "branch": "main"}, "main"),
({"target_type": "commit", "sha": "abc123"}, "abc123"),
({"target_type": "custom", "instructions": "look"}, None),
]
for payload, _expected in cases:
status, body = self._req(
"POST", "/tasks/%s/reviews" % task["id"], payload,
self.owner_key)
self.assertEqual(status, 201, body)
status, body = self._req(
"GET", "/tasks/%s/reviews" % task["id"], None,
self.owner_key)
self.assertEqual(len(body["reviews"]), 3)
def test_create_target_validation(self):
task = self._create_task(self.project["id"])
bad = [
{"target_type": "baseBranch"},
{"target_type": "commit"},
{"target_type": "custom"},
{"target_type": "nope"},
{"target_type": "baseBranch", "branch": 123},
]
for payload in bad:
status, body = self._req(
"POST", "/tasks/%s/reviews" % task["id"], payload,
self.owner_key)
self.assertEqual(status, 400, (payload, body))
status, body = self._req(
"GET", "/tasks/%s/reviews" % task["id"], None,
self.owner_key)
self.assertEqual(body["reviews"], [])
def test_findings_writeback(self):
task = self._create_task(self.project["id"])
_status, review = self._req(
"POST", "/tasks/%s/reviews" % task["id"], {},
self.owner_key)
status, body = self._req(
"POST",
"/tasks/%s/reviews/%s/findings" % (task["id"], review["id"]),
{"findings": "two issues"}, self.owner_key)
self.assertEqual(status, 200, body)
self.assertEqual(body["status"], "completed")
self.assertEqual(body["findings"], "two issues")
status, body = self._req(
"GET", "/tasks/%s/reviews" % task["id"], None,
self.owner_key)
self.assertEqual(body["reviews"][0]["findings"], "two issues")
def test_findings_explicit_failed(self):
task = self._create_task(self.project["id"])
_status, review = self._req(
"POST", "/tasks/%s/reviews" % task["id"], {},
self.owner_key)
status, body = self._req(
"POST",
"/tasks/%s/reviews/%s/findings" % (task["id"], review["id"]),
{"status": "failed", "findings": "reviewer error"},
self.owner_key)
self.assertEqual(status, 200, body)
self.assertEqual(body["status"], "failed")
def test_findings_validation_and_isolation(self):
task = self._create_task(self.project["id"])
other = self._create_task(self.project["id"])
_status, review = self._req(
"POST", "/tasks/%s/reviews" % task["id"], {},
self.owner_key)
url = "/tasks/%s/reviews/%s/findings" % (
task["id"], review["id"])
status, _body = self._req("POST", url, {}, self.owner_key)
self.assertEqual(status, 400)
status, _body = self._req(
"POST", url, {"status": "bogus"}, self.owner_key)
self.assertEqual(status, 400)
status, _body = self._req(
"POST", url, {"findings": 123}, self.owner_key)
self.assertEqual(status, 400)
# 记录属于 task,挂在 other 任务下写回时 404,不泄露。
status, _body = self._req(
"POST",
"/tasks/%s/reviews/%s/findings" % (
other["id"], review["id"]),
{"findings": "x"}, self.owner_key)
self.assertEqual(status, 404)
status, _body = self._req(
"POST",
"/tasks/%s/reviews/%s/findings" % (
task["id"], "no-such-review"),
{"findings": "x"}, self.owner_key)
self.assertEqual(status, 404)
def test_auth_and_permissions(self):
task = self._create_task(self.project["id"])
status, _body = self._req(
"POST", "/tasks/%s/reviews" % task["id"], {})
self.assertEqual(status, 401)
status, _body = self._req(
"GET", "/tasks/%s/reviews" % task["id"], None)
self.assertEqual(status, 401)
status, _body = self._req(
"POST", "/tasks/%s/reviews" % task["id"], {},
self.outsider_key)
self.assertEqual(status, 403)
status, _body = self._req(
"GET", "/tasks/%s/reviews" % task["id"], None,
self.outsider_key)
self.assertEqual(status, 403)
_status, review = self._req(
"POST", "/tasks/%s/reviews" % task["id"], {},
self.owner_key)
status, _body = self._req(
"POST",
"/tasks/%s/reviews/%s/findings" % (
task["id"], review["id"]),
{"findings": "x"}, self.outsider_key)
self.assertEqual(status, 403)
def test_task_not_found(self):
status, _body = self._req(
"POST", "/tasks/no-such-task/reviews", {},
self.owner_key)
self.assertEqual(status, 404)
status, _body = self._req(
"GET", "/tasks/no-such-task/reviews", None,
self.owner_key)
self.assertEqual(status, 404)
if __name__ == "__main__":
unittest.main()
+66 -11
View File
@@ -147,6 +147,7 @@ from cloud.repos import RepoError, checkout_dir, resolve_repo
from cloud.sandbox import CloudExecutor, DockerExecutor, LocalExecutor
from cloud.uploads import upload_root as _upload_root
from cloud.sandbox import _default_runner as _sandbox_default_runner
from cloud.caps import effective_capacity, resolve_capacity
from cloud.workers import heartbeat_worker, register_worker
# 与 server.TERMINAL_STATUSES 对齐:这些状态的任务不再被认领或取消。
@@ -282,7 +283,7 @@ class Worker:
task_timeout=1800, max_retries=1,
retry_backoff_seconds=5.0, repos_dir="./repos",
repos_runner=None, heartbeat_file=None,
worker_id=None, capacity=1, fallback_models=None,
worker_id=None, capacity=None, fallback_models=None,
tenant_isolation=False, tenants_root=None,
tenant_template_home=None, tenant_template_repo=None,
egress_proxy_port=0, claim_executor_only=False,
@@ -412,10 +413,11 @@ class Worker:
# 自然变 stale,不要用随机 uuid 让注册表无法对应同一台机器。
self.worker_id = worker_id or "%s-%d" % (
socket.gethostname(), os.getpid())
try:
self.capacity = int(capacity)
except (TypeError, ValueError):
self.capacity = 1
# 并发上限配置来源(见 cloud/caps.py):显式构造参数 /
# CLI --worker-capacity > 环境变量 CLOUD_WORKER_CAPACITY >
# 默认 1;非法值归一为 1。self._env 在上面已定好,单测
# 注入的普通 dict 同样可读环境变量(无该键时默认 1)。
self.capacity = resolve_capacity(capacity, env=self._env)
# 常驻启动时的代码签名:run_loop 每轮边界比对,变了就自退
# (见 run_loop 的代码变化自退检查)。
self._code_signature = _code_signature()
@@ -436,7 +438,8 @@ class Worker:
try:
register_worker(
self.db_path, self.worker_id,
executor=self._executor_name(), capacity=self.capacity)
executor=self._executor_name(), capacity=self.capacity,
running=self._running_count())
except Exception as exc:
log_event(self.logger, "worker_register_failed",
message="worker register failed: %s" % exc)
@@ -446,7 +449,8 @@ class Worker:
try:
heartbeat_worker(
self.db_path, self.worker_id,
executor=self._executor_name(), capacity=self.capacity)
executor=self._executor_name(), capacity=self.capacity,
running=self._running_count())
except Exception as exc:
log_event(self.logger, "worker_registry_heartbeat_failed",
message="worker registry heartbeat failed: %s" % exc)
@@ -531,7 +535,23 @@ class Worker:
claim_executor_only 开启时(宿主级沙箱 worker 用):WHERE
只加 executor = ?,NULL 任务绝不认领(否则宿主 worker
会抢走本该进 Docker 的任务)。
并发上限门(caps):CAS 之前先在 _running_lock 下占位。
running 数达到 self.capacity 时直接返回 False,任务保持
queued,不碰数据库;占位成功才走 CAS。同一任务已在
_running 里时不重复计数(重复占位不增槽)。CAS 失败
(已被别的 worker 抢走)时,只清掉本次新建的占位,绝
不动已在跑任务的条目。占位条目在执行路径的 finally
里清(本地 _execute_via_process / Docker 同步路径),
认领后提前返回的三道门(缺 key / 仓库 / 配额)在
run_one 里显式清占位;CAS 语义本身一句未改。
"""
with self._running_lock:
already_running = task_id in self._running
if not already_running and \
len(self._running) >= effective_capacity(self.capacity):
return False
self._running.setdefault(task_id, {})
now = time.time()
conn = _db_connect(self.db_path)
try:
@@ -563,7 +583,10 @@ class Worker:
(now, now, self.host_name, task_id, executor),
)
conn.commit()
return cur.rowcount == 1
claimed = cur.rowcount == 1
if not claimed and not already_running:
self._running_clear(task_id)
return claimed
finally:
conn.close()
@@ -908,6 +931,16 @@ class Worker:
return {tid: dict(entry)
for tid, entry in self._running.items()}
def _running_count(self):
"""当前占用槽位数(_running_lock 下计数)。
启动中的任务已在 _claim 里占位(空条目),所以这里数
到的既包括已起进程 / 容器的任务,也包括已认领未起
进程的任务;注册表心跳与认领门共用这个口径。
"""
with self._running_lock:
return len(self._running)
def _wait_for_running(self, timeout=None):
"""等 _running 变空;timeout 为 None 时无限等。
@@ -2082,6 +2115,9 @@ class Worker:
project_id=task["project_id"], status="failed")
self._notify_task_finished(
task_id, task["project_id"], "failed", error=error)
# 认领占位在此释放:缺 key 门在执行器起进程之前返回,
# 不清占位会把槽位漏到 worker 退出(见 _claim 文档)。
self._running_clear(task_id)
return {"dispatched": True, "task_id": task_id,
"host_name": self.executor._host_name(task)
if hasattr(self.executor, "_host_name") else task_id,
@@ -2106,6 +2142,8 @@ class Worker:
project_id=task["project_id"], status="failed")
self._notify_task_finished(
task_id, task["project_id"], "failed", error=repo_error)
# 同缺 key 门:仓库解析失败在起进程之前返回,先清占位。
self._running_clear(task_id)
return {"dispatched": True, "task_id": task_id,
"host_name": self.executor._host_name(task)
if hasattr(self.executor, "_host_name") else task_id,
@@ -2129,6 +2167,8 @@ class Worker:
log_event(self.logger, "worker_quota_stop", task_id=task_id,
project_id=task["project_id"],
status="interrupted")
# 同缺 key 门:配额门在起进程之前返回,先清占位。
self._running_clear(task_id)
return {"dispatched": True, "task_id": task_id,
"host_name": self.executor._host_name(task)
if hasattr(self.executor, "_host_name") else task_id,
@@ -2736,6 +2776,11 @@ class Worker:
if stop_event is not None and stop_event.is_set():
break
self._heartbeat_running_tasks()
# 注册表心跳也每 30 秒刷一次:长任务跑满一轮时
# run_all 久久不返回,只靠轮次结束的心跳会让
# /workers 的 running 数长期停在旧值;这里刷的
# running 含启动中占位,与认领门口径一致。
self._registry_heartbeat()
hb_stop.wait(30)
hb_thread = threading.Thread(target=_hb_loop, daemon=True)
@@ -2795,6 +2840,14 @@ class Worker:
time.sleep(0.2)
finally:
hb_stop.set()
# 等心跳线程彻底退出再返回:否则它手头一次未写完的
# 注册表心跳可能在调用方清理 DB 目录时重建 control.db,
# 导致 tearDown rmtree 报 Directory not empty(全量套件
# 里 test_code_changed_exits_loop_before_claiming 偶发
# error 即此竞态)。daemon 线程 + set() 会唤醒 wait,
# 正常情况下 join 瞬间返回;5 秒超时兜底,避免写挂时
# 拖住退出。
hb_thread.join(timeout=5)
# 优雅排空:先等正在跑的任务自然结束,超时后才杀。
# 串行模式下 run_all 返回时任务已跑完,这里立刻通过;
# 并行模式下停机信号退出时后台线程可能还在跑,先等
@@ -2900,10 +2953,12 @@ def main(argv=None):
parser.add_argument("--worker-id", default=None,
help="多 worker 池身份(默认:主机名-进程号;"
"生产部署应配稳定 id,否则重启后旧行变 stale)")
parser.add_argument("--worker-capacity", type=int, default=1,
parser.add_argument("--worker-capacity", type=int, default=None,
help="worker 并发上限:同一 worker 内最多同时跑"
"几个任务(默认 1,即旧串行行为;>1 时用"
"线程池并行,认领仍是 CAS,成本写库持锁)")
"几个任务(未给时读环境变量"
" CLOUD_WORKER_CAPACITY,未设则默认 1,"
"即旧串行行为;>1 时用线程池并行,认领"
"仍是 CAS,成本写库持锁)")
parser.add_argument("--host-name", default=None,
help="本 worker 所在主机名(HA 阶段 1):认领任务"
"时写进 tasks.worktree_host,/pr 据此判断"
+71 -9
View File
@@ -62,10 +62,12 @@ def ensure_workers_table(db_path):
"worker_id TEXT PRIMARY KEY, "
"executor TEXT, "
"capacity INTEGER, "
"running INTEGER, "
"last_seen DOUBLE PRECISION, "
"created_at DOUBLE PRECISION)"
)
conn.commit()
_ensure_running_column(conn)
# 老库升精度只在确实是 Postgres REAL 列时做一次:每次心跳都
# ALTER 会反复拿 ACCESS EXCLUSIVE 锁,没必要。先查
# information_schema(SQLite 下这张表不存在,查询失败即跳过,
@@ -98,7 +100,59 @@ def ensure_workers_table(db_path):
conn.close()
def register_worker(db_path, worker_id, executor=None, capacity=1):
def _ensure_running_column(conn):
"""老库补 running 列(新表建表时已含该列)。
running 是 worker 上次心跳上报的当前占用槽数(启动中占位
也算,见 cloud/caps.py);列缺失时 _worker_to_dict 默认 0,
所以补列失败只回滚、不抛,绝不炸注册与心跳。
SQLite 用 PRAGMA 查列,Postgres 用 information_schema。
"""
try:
rows = conn.execute("PRAGMA table_info(workers)").fetchall()
names = set()
for row in rows:
try:
names.add(row[1])
except (IndexError, TypeError):
pass
if rows and "running" not in names:
conn.execute(
"ALTER TABLE workers ADD COLUMN running INTEGER DEFAULT 0")
conn.commit()
return
except Exception:
try:
conn.rollback()
except Exception:
pass
try:
row = conn.execute(
"SELECT COUNT(*) FROM information_schema.columns "
"WHERE table_name = 'workers' AND column_name = 'running'"
).fetchone()
exists = bool(row and row[0])
except Exception:
try:
conn.rollback()
except Exception:
pass
return
if not exists:
try:
conn.execute(
"ALTER TABLE workers ADD COLUMN IF NOT EXISTS "
"running INTEGER DEFAULT 0")
conn.commit()
except Exception:
try:
conn.rollback()
except Exception:
pass
def register_worker(db_path, worker_id, executor=None, capacity=1,
running=None):
"""注册(或刷新)一个 worker:不存在 INSERT,存在 UPDATE。
返回注册行 dict。worker_id 为空时抛 ValueError,调用方
@@ -112,14 +166,17 @@ def register_worker(db_path, worker_id, executor=None, capacity=1):
try:
cur = conn.execute(
"UPDATE workers SET executor = ?, capacity = ?,"
" last_seen = ? WHERE worker_id = ?",
(executor, capacity, now, worker_id),
" running = ?, last_seen = ? WHERE worker_id = ?",
(executor, capacity, running if running is not None else 0,
now, worker_id),
)
if cur.rowcount == 0:
conn.execute(
"INSERT INTO workers(worker_id, executor, capacity,"
" last_seen, created_at) VALUES(?, ?, ?, ?, ?)",
(worker_id, executor, capacity, now, now),
" running, last_seen, created_at)"
" VALUES(?, ?, ?, ?, ?, ?)",
(worker_id, executor, capacity,
running if running is not None else 0, now, now),
)
conn.commit()
row = conn.execute(
@@ -130,7 +187,8 @@ def register_worker(db_path, worker_id, executor=None, capacity=1):
return _worker_to_dict(row)
def heartbeat_worker(db_path, worker_id, executor=None, capacity=None):
def heartbeat_worker(db_path, worker_id, executor=None, capacity=None,
running=None):
"""心跳:就是带 last_seen 刷新的注册。
executor/capacity 未给时保留旧值(已注册的 worker 只刷时间,
@@ -148,17 +206,19 @@ def heartbeat_worker(db_path, worker_id, executor=None, capacity=None):
if row is None:
conn.execute(
"INSERT INTO workers(worker_id, executor, capacity,"
" last_seen, created_at) VALUES(?, ?, ?, ?, ?)",
" running, last_seen, created_at)"
" VALUES(?, ?, ?, ?, ?, ?)",
(worker_id, executor, capacity if capacity is not None
else 1, now, now),
else 1, running if running is not None else 0, now, now),
)
else:
old = _worker_to_dict(row)
conn.execute(
"UPDATE workers SET executor = ?, capacity = ?,"
" last_seen = ? WHERE worker_id = ?",
" running = ?, last_seen = ? WHERE worker_id = ?",
(executor if executor is not None else old["executor"],
capacity if capacity is not None else old["capacity"],
running if running is not None else old["running"],
now, worker_id),
)
conn.commit()
@@ -197,10 +257,12 @@ def _worker_to_dict(row):
return row[name]
except (KeyError, IndexError, TypeError):
return None
running = _get("running")
return {
"worker_id": _get("worker_id"),
"executor": _get("executor"),
"capacity": _get("capacity"),
"running": running if running is not None else 0,
"last_seen": _get("last_seen"),
"created_at": _get("created_at"),
}
+47
View File
@@ -511,6 +511,18 @@ class Server:
raise RuntimeError(f"thread/resume 失败: {resp}")
return resp["result"]
def start_review(self, thread_id, target, timeout=600):
# 协议 review/start:params 必填 threadId 与 target,
# target 四种见 protocol v2 的 ReviewTarget(uncommitted
# Changes / baseBranch / commit / custom);delivery 缺
# 省即 inline,不发 detached(已废弃)。失败直接抛错,
# 调用方不得把 error 响应当成功打印。
params = {"threadId": thread_id, "target": target}
resp = self.request("review/start", params, timeout=timeout)
if not resp or "error" in resp:
raise RuntimeError(f"review/start 失败: {resp}")
return resp["result"]
def fork_thread(self, thread_id, cwd=None, sandbox=None):
params = {"threadId": thread_id}
if cwd:
@@ -2012,6 +2024,13 @@ def main():
p = sub.add_parser("archive"); p.add_argument("--task", required=True)
p = sub.add_parser("fork"); p.add_argument("--task", required=True); p.add_argument("--name", required=True); p.add_argument("--sandbox", default=None)
p = sub.add_parser("compact"); p.add_argument("--task", required=True)
p = sub.add_parser("code-review"); p.add_argument("--task", required=True)
p.add_argument("--base-branch", default=None,
help="评审目标:与该分支的差异(baseBranch)")
p.add_argument("--commit", default=None,
help="评审目标:指定提交 sha(commit)")
p.add_argument("--instructions", default=None,
help="评审目标:自定义评审指令(custom)")
p = sub.add_parser("run"); p.add_argument("--task", required=True); p.add_argument("--prompt", required=True)
p = sub.add_parser("interrupt"); p.add_argument("--task", required=True)
p = sub.add_parser("delegate")
@@ -2460,6 +2479,34 @@ def main():
resp = srv.request("thread/compact/start",
{"threadId": t["thread_id"]}, timeout=120)
print(f"compact {args.task}: {resp}")
elif args.cmd == "code-review":
# 注意命名:host.py 已有另一条线的 review 子命令
# (PR 评审评论),本命令换名 code-review,不覆盖它。
# 目标优先级:commit > instructions > base-branch >
# 缺省 uncommittedChanges;与 cloud/review.py 的
# build_review_target 同款组合,非法组合直接报错。
# compact 同款原因先 resume 再 review(未加载的旧
# thread 直接 review 会报 thread not found)。
# app-server 返回 error 时只打印失败并以非零退出,
# 绝不把失败结果当成评审结论。
t = state["tasks"][args.task]
if args.commit:
target = {"type": "commit", "sha": args.commit}
elif args.instructions:
target = {"type": "custom",
"instructions": args.instructions}
elif args.base_branch:
target = {"type": "baseBranch",
"branch": args.base_branch}
else:
target = {"type": "uncommittedChanges"}
try:
srv.resume_thread(t["thread_id"])
result = srv.start_review(t["thread_id"], target)
except Exception as exc:
print(f"code-review failed: {exc}")
raise SystemExit(1)
print(f"code-review {args.task}: {result}")
elif args.cmd == "archive":
t = state["tasks"][args.task]
srv.archive_thread(t["thread_id"])