feat: cloud 门禁/超时/成本/沙箱收敛为单一判定源

- 新增标准库叶子层 timeouts/gates/quota/sandbox_policy:
  审批 3600 与单轮 600 各有名常量与环境覆盖层,已建项
  expires_at 永不重算;目录/控制/合并状态门禁统一判定,
  目录违规无任何覆盖参数;配额取数与比较唯一来源
  (USD 先于无价、命中均为 >=、无价只计 actual 行);
  /tmp 隔离矩阵与失联清理判定纯函数化。
- 调用点委托:server/worker/automations/auto_memory 的
  配额 SQL 与比较、server/tree/marker 的目录解析、
  marker 的控制门、sandbox 的任务容器名正则、两处 300 秒
  陈旧阈值默认值均收敛,别名与错误文案不变。
- 统一两个 P0 口径分歧:服务端无价计数由全部 usd IS NULL
  行改为 actual-only;显式不限额用户在服务端准入亦不受
  无价上限约束(与 worker 运行时一致)。
- 不新增门禁种类与 record 模式(主线无此实现);不改
  skill_seed/skills/SKILL.md;不改 v1.1 契约与分支保护。
- 新增 test_arch_converge.py 18 项。
This commit is contained in:
host committed 2026-11-03 02:49:25 +00:00
1 parent 2720acef46
commit 41e8c98512
16 files changed
+1068 -301

No files matched your search

+18 -16
View File
@@ -64,19 +64,24 @@
- item/tool/call 与 account/chatgptAuthTokens/refresh:前者是
动态工具执行,后者是凭证刷新,都不是用户审批,排除。
本模块只依赖标准库,不 import cloud 其他模块,避免循环
import;服务端与未来 worker/hook 共用。占位符用 ?,建表与
列补齐写法与 peer_messages 同款(db 适配层负责翻译)。
本模块只依赖标准库与同为标准库叶子的 cloud.timeouts,
不 import 其他 cloud 模块,避免循环 import;服务端与未来
worker/hook 共用。占位符用 ?,建表与列补齐写法与
peer_messages 同款(db 适配层负责翻译)。
"""
import json
import time
import uuid
from cloud.timeouts import APPROVAL_DEFAULT_SECONDS as _APPROVAL_DEFAULT
from cloud.timeouts import expires_at_for as _expires_at_for
from cloud.timeouts import resolve_module_timeout as _resolve_timeout
KINDS = ("elicitation", "command", "file_change", "permissions",
"user_input")
DECISIONS = ("approved", "denied")
DEFAULT_TIMEOUT_SECONDS = 3600
DEFAULT_TIMEOUT_SECONDS = _APPROVAL_DEFAULT
MAX_SUMMARY_CHARS = 2000
MAX_DETAIL_CHARS = 20000
@@ -167,12 +172,15 @@ def approval_to_dict(row):
def create_approval(conn, task_id, project_id, kind="elicitation",
summary="", detail=None, requested_by=None,
timeout_seconds=DEFAULT_TIMEOUT_SECONDS, now=None):
timeout_seconds=None, now=None):
"""创建一条待审批项,返回 dict。非法输入抛 ValueError,
调用方据此回 400;成功时 commit。
timeout_seconds:None 用默认值;0 表示不过期(只允许
模块级调用,服务端入口另有限制);负数非法。"""
timeout_seconds:None 用默认值(cloud/timeouts.py 的
环境变量感知默认值,未设环境变量时为 3600;签名默认
故意写 None 而不是 3600,避免默认值在 import 时绑定死、
环境覆盖层传不进去);0 表示不过期(只允许模块级调用,
服务端入口另有限制);负数非法。"""
if kind not in KINDS:
raise ValueError("kind must be one of: " + ", ".join(KINDS))
if not isinstance(summary, str) or not summary.strip():
@@ -180,17 +188,11 @@ def create_approval(conn, task_id, project_id, kind="elicitation",
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")
# 不传 default:None 时走 timeouts 的环境变量感知默认值。
timeout_seconds = _resolve_timeout(timeout_seconds)
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
expires_at = _expires_at_for(now, timeout_seconds)
item_id = uuid.uuid4().hex
ensure_approvals_table(conn)
conn.execute(
+11 -21
View File
@@ -59,6 +59,9 @@ import time
from cloud.db import connect as _db_connect
from cloud.identity import IdentityStore
from cloud.quota import default_quota_usd_micros as _quota_default
from cloud.quota import monthly_usage as _quota_monthly_usage
from cloud.quota import quota_for_owner as _quota_for_owner
DEFAULT_SETTINGS = {
"generate_memories": 0,
@@ -113,12 +116,9 @@ def redact_text(text):
def _default_quota_usd_micros():
"""与 worker/server 同款默认配额解析,避免循环 import。"""
try:
return max(0, int(os.environ.get(
"CLOUD_DEFAULT_QUOTA_USD_MICROS", "10000000")))
except (TypeError, ValueError):
return 10000000
"""默认配额解析:收敛到 cloud/quota.py(quota 模块是
标准库叶子,不存在旧注释担心的循环 import)。"""
return _quota_default()
def quota_remaining_percent(db_path, owner):
@@ -131,27 +131,17 @@ def quota_remaining_percent(db_path, owner):
month = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m")
conn = _db_connect(db_path)
try:
conn.execute(
"CREATE TABLE IF NOT EXISTS user_quotas("
"user_id TEXT PRIMARY KEY, quota_usd_micros BIGINT, "
"updated_at DOUBLE PRECISION)")
# 配额行与当月 USD 用量取数收敛到 cloud/quota.py
#(无价部分本函数不用,只取 USD 一项;缺 cost_records
# 表时仍按旧行为计 0,其它查库异常仍抛给调用方)。
quota = _quota_for_owner(conn, owner)
conn.commit()
row = conn.execute(
"SELECT quota_usd_micros FROM user_quotas WHERE user_id = ?",
(owner,)).fetchone()
quota = _default_quota_usd_micros() if row is None else row[0]
if quota is None:
return None
if quota <= 0:
return 0.0
try:
used_row = conn.execute(
"SELECT COALESCE(SUM(c.usd_micros), 0)"
" FROM cost_records c JOIN tasks t"
" ON c.task_id = t.id"
" WHERE t.created_by = ? AND c.day LIKE ?",
(owner, month + "%")).fetchone()
used = used_row[0] if used_row else 0
used, _unpriced = _quota_monthly_usage(conn, owner, month)
except Exception as exc:
if "no such table" in str(exc):
used = 0
+24 -51
View File
@@ -38,9 +38,10 @@
id 记在 automation_runs.task_id 里,派发与任务绑定,不用
为自动化另起一套通知系统)。
本模块只依赖标准库,不 import cloud 其它模块,避免循环
import;表迁移写法仿 peer_messages(CREATE TABLE IF NOT
EXISTS + PRAGMA 查列补齐,SQLite 与 Postgres 适配层同款)。
本模块只依赖标准库与同为标准库叶子的 cloud.quota,不
import 其他 cloud 模块,避免循环 import;表迁移写法仿
peer_messages(CREATE TABLE IF NOT EXISTS + PRAGMA 查列补齐,
SQLite 与 Postgres 适配层同款)。
"""
import os
@@ -48,6 +49,11 @@ import time
import uuid
from datetime import datetime, timezone
from cloud.quota import decide as _quota_decide
from cloud.quota import default_quota_usd_micros as _quota_default
from cloud.quota import monthly_usage as _quota_monthly_usage
from cloud.quota import quota_for_owner as _quota_for_owner
MIN_INTERVAL_SECONDS = 60
MAX_INTERVAL_SECONDS = 30 * 86400
MAX_NAME_LEN = 100
@@ -520,13 +526,9 @@ def record_automation_run(conn, automation_id, project_id, task_id=None,
def _default_quota_usd_micros():
"""默认月配额(微美元):与服务端 _default_quota_usd_micros
同值同环境变量(CLOUD_DEFAULT_QUOTA_USD_MICROS,默认 $10)。"""
try:
return max(0, int(os.environ.get(
"CLOUD_DEFAULT_QUOTA_USD_MICROS", "10000000")))
except (TypeError, ValueError):
return 10000000
"""默认月配额(微美元):解析收敛到 cloud/quota.py
(同值同环境变量 CLOUD_DEFAULT_QUOTA_USD_MICROS,默认 $10)。"""
return _quota_default()
def _unpriced_token_cap():
@@ -549,29 +551,17 @@ def _automation_quota_exceeded(conn, owner):
过处理,绝不建任务)。cost_records 表不存在时视为零花费
(CostStore 首次记账时才建表,不存在即从未记过账),其
它查库异常才 fail-closed——全新库上不能因此误拒全部运行。
本模块只依赖标准库,不 import cloud 其它模块,故此处就
地实现 SQL 版口径,注释里标明三处同口径点,改口径时三
处要一起改(server._quota_status、worker._quota_exceeded、
本函数)。
取数与比较判定收敛到 cloud/quota.py(本函数只保留
表存在探针与 PG 事务回滚处理,因其与自动化执行器的
事务形态绑定,不宜塞进通用取数层)。
"""
if not owner:
return ("quota_exceeded: 自动化缺少创建人,无法查配额,"
"拒绝建任务")
try:
conn.execute(
"CREATE TABLE IF NOT EXISTS user_quotas("
"user_id TEXT PRIMARY KEY, quota_usd_micros BIGINT, "
" updated_at DOUBLE PRECISION)")
row = conn.execute(
"SELECT quota_usd_micros FROM user_quotas"
" WHERE user_id = ?",
(owner,)).fetchone()
if row is None:
quota = _default_quota_usd_micros()
else:
quota = row_get(row, "quota_usd_micros")
if quota is None:
return None
quota = _quota_for_owner(conn, owner)
if quota is None:
return None
month = datetime.now(timezone.utc).strftime("%Y-%m")
try:
conn.execute("SELECT COUNT(*) FROM cost_records").fetchone()
@@ -585,22 +575,8 @@ def _automation_quota_exceeded(conn, owner):
pass
table_exists = False
if table_exists:
used_row = conn.execute(
"SELECT COALESCE(SUM(c.usd_micros), 0) AS total"
" FROM cost_records c JOIN tasks t"
" ON c.task_id = t.id"
" WHERE t.created_by = ? AND c.day LIKE ?",
(owner, month + "%")).fetchone()
used = row_get(used_row, "total") or 0
unpriced_row = conn.execute(
"SELECT COALESCE(SUM(c.input_tokens + c.output_tokens), 0)"
" AS total"
" FROM cost_records c JOIN tasks t"
" ON c.task_id = t.id"
" WHERE t.created_by = ? AND c.day LIKE ?"
" AND c.usd_micros IS NULL AND c.source = 'actual'",
(owner, month + "%")).fetchone()
unpriced_used = row_get(unpriced_row, "total") or 0
used, unpriced_used = _quota_monthly_usage(
conn, owner, month)
else:
used = 0
unpriced_used = 0
@@ -612,13 +588,10 @@ def _automation_quota_exceeded(conn, owner):
except Exception:
pass
return "quota_check_failed: %s" % exc
if used >= quota:
return ("quota_exceeded: 本月配额已用完(已用 %d / "
"配额 %d 微美元,%s)" % (used, quota, month))
if unpriced_used >= _unpriced_token_cap():
return ("quota_exceeded: 本月无价模型 token 已用完"
"(已用 %d / 上限 %d,%s)"
% (unpriced_used, _unpriced_token_cap(), month))
decision = _quota_decide(quota, used, unpriced_used,
_unpriced_token_cap(), month)
if decision is not None:
return decision["message"]
return None
+4 -1
View File
@@ -32,8 +32,11 @@
import json
from cloud.timeouts import LONG_TIMEOUT_SECONDS as _LONG_TIMEOUT
MAX_PARTS = 20
DEFAULT_TIMEOUT_SECONDS = 3600
# 超时数值收敛到 cloud/timeouts.py(别名保留,调用点不变)。
DEFAULT_TIMEOUT_SECONDS = _LONG_TIMEOUT
DEFAULT_POLL_INTERVAL = 5.0
REPORT_RESULT_CHARS = 500
EFFORTS = {"low", "medium", "high", "xhigh", "max"}
+125
View File
@@ -0,0 +1,125 @@
"""门禁层:调用方只传事实,本层只出决定。
收敛对象是主线上已经存在的硬门禁,不是新发明的门禁
种类。主线(origin/main)上不存在 merge_gate 的
record/enforce 模式、分支保护门禁与全自动门禁——那些只
在未合并的分支里;本轨不从分支搬功能、不新增门禁种类、
不把硬门禁降级为“只记录不阻断”(那会削弱安全语义,
不属于收敛)。本层收敛的是三类已有判定:
1. 目录门(resolve_inside):worktree 内相对路径的安全
解析。服务端 _safe_worktree_path、cloud/tree.py 的
resolve_tree_path、cloud/marker_tools.py 的
safe_worktree_path(worker 经 marker_tools 调用)此前
各写了一份 realpath + commonpath,仅在 commonpath
的 ValueError 处理上不一致(服务端不捕,另两处捕)。
现统一为本函数:绝对路径、.. 逃逸、符号链接逃逸、
跨盘符 ValueError 一律拒绝;空串返回原 worktree。
2. 控制门(decide_control):marker_tools.check_control_gate
的纯函数判定(cancel / retry 的目标状态与派生关系),
现为本层 canonical 实现,marker_tools 只保留同名薄壳
(调用点与测试 import 的名字不变)。
3. 合并状态门(merge_state_decision):服务端 _merge_task
的状态判定——只有 completed 可标记 merged,其它状态
一律拒绝(含已 merged 的重复调用)。权限判定
(approve_merge)仍在服务端,不进本层:本层只看事实
(任务状态),不查权限。
不可覆盖规则(冻结):
- 目录门的拒绝没有任何 confirm / override / bypass 参数。
本模块的函数签名里 deliberately 没有这类参数:显式
确认即使存在,也只能覆盖“保护类风险”,绝不能覆盖
目录违规——而主线上目录违规本来就没有确认通道,
任何调用方都不得以“用户已确认”为由绕过本层。
测试钉住:resolve_inside 与 decide_control 的签名中
不含 confirm / override / bypass 字样。
- 合并状态门同样没有覆盖参数:非 completed 任务不存在
“确认后合并”的路径,重复 merged 调用仍是拒绝。
返回形态沿用 codebase 的元组风格:(allowed, reason),
reason 为空表示放行;目录门另有 resolve_inside 直接
返回路径或 None(旧调用点的形态不变)。
"""
import os
TERMINAL_STATUSES = ("completed", "failed", "interrupted", "merged",
"merge-failed")
RETRYABLE_STATUSES = ("failed", "interrupted", "merge-failed")
def resolve_inside(base, rel):
"""worktree 内相对路径的安全解析:越界回 None。
空串返回 base 本身(旧三处行为一致);绝对路径、
.. 逃逸、符号链接逃逸、commonpath 跨盘符 ValueError
一律 None。目录内指回内部的符号链接放行到真实位置
(调用方仍只拿到 worktree 内的内容)。
"""
if not rel:
return base
if os.path.isabs(rel):
return None
real_base = os.path.realpath(base)
target = os.path.realpath(os.path.join(real_base, rel))
try:
if os.path.commonpath([real_base, target]) != real_base:
return None
except ValueError:
return None
return target
def is_goal_task(task):
"""goal 任务判定:verify_command 非空或 goal_mode 为真。
与 worker run_one 的 goal 分支同款;goal 字段本身是
goal_mode=1 时 prompt 的副本,不单独当判定条件。
"""
if not task:
return False
return bool(task.get("verify_command") or task.get("goal_mode"))
def decide_control(target, sender, relation, kind):
"""task-cancel / task-retry 的纯函数门禁,返回 (ok, reason)。
kind 只能是 "cancel" 或 "retry",其它值一律拒绝。
权限层不在这里:worker 侧另查 create_task /
approve_merge;本函数只管目标状态与关系,单测不需
AuthStore。goal 任务不可取消或重试(goal 循环自有轮次)。
"""
if kind not in ("cancel", "retry"):
return False, "未知的控制操作"
if target is None:
return False, "目标任务不存在"
if (target.get("project_id") is not None
and sender.get("project_id") is not None
and target.get("project_id") != sender.get("project_id")):
return False, "跨项目任务不可操作"
if target.get("id") == sender.get("id"):
return False, "不能操作自己"
if is_goal_task(target):
return False, "goal 任务不可取消或重试(goal 循环自有轮次)"
status = target.get("status")
if kind == "cancel":
if status in TERMINAL_STATUSES:
return False, "目标已终态(%s),不可取消" % status
else:
if status not in RETRYABLE_STATUSES:
return False, ("仅 failed / interrupted / merge-failed"
" 可重试,当前状态为 %s" % status)
if relation not in ("descendant", "same_creator"):
return False, "目标不在发送方派生链上且创建人不同"
return True, ""
def merge_state_decision(status):
"""合并标记的状态判定:仅 completed 放行,返回 (ok, reason)。
含已 merged 的重复调用在内,其它状态一律拒绝;文案与
服务端旧 409 错误一致。权限判定不在本层。
"""
if status == "completed":
return True, ""
return False, "only completed tasks can be marked merged"
+3 -1
View File
@@ -75,6 +75,7 @@ from dataclasses import dataclass, field
from pathlib import Path
from typing import Protocol, TypedDict
from cloud.timeouts import LONG_TIMEOUT_SECONDS as _LONG_TIMEOUT
from .harness2 import (
EditResult,
Rollout,
@@ -93,7 +94,8 @@ from .harness2 import (
TERMINAL_STATUSES = frozenset(
{"completed", "failed", "interrupted", "merged", "merge-failed"}
)
DEFAULT_TASK_TIMEOUT_SECONDS = 3600
# 超时数值收敛到 cloud/timeouts.py(别名保留,调用点不变)。
DEFAULT_TASK_TIMEOUT_SECONDS = _LONG_TIMEOUT
DEFAULT_POLL_SECONDS = 10
TRACE_MAX_CHARS = 50000
+23 -54
View File
@@ -125,7 +125,8 @@ docs/marker-tools.md 新增的两节。
同创建人-only 的目标一律拒绝,并把原因写进报告——绝不
在查不到权限时静默放宽。
本模块只依赖标准库,不 import cloud 其它模块,避免循环
本模块只依赖标准库与同为标准库叶子的 cloud.gates /
cloud.timeouts,不 import 其他 cloud 模块,避免循环
import。DB 助手函数只用 ? 占位符,与 peer_messages 同款写法。
"""
@@ -137,6 +138,13 @@ import uuid
from datetime import datetime, timedelta, timezone
from typing import Any
from cloud.gates import RETRYABLE_STATUSES as _GATE_RETRYABLE
from cloud.gates import TERMINAL_STATUSES as _GATE_TERMINAL
from cloud.gates import decide_control as _gate_decide_control
from cloud.gates import is_goal_task as _gate_is_goal_task
from cloud.gates import resolve_inside as _gate_resolve_inside
from cloud.timeouts import LONG_TIMEOUT_SECONDS as _LONG_TIMEOUT
# ZoneInfo 在极老 Python 上不存在:先声明为 Any 再 try import,
# 这样 except 分支的 None 赋值不再是“给 type 赋值”(mypy
# misc + assignment),运行时仍是“有则类、无则 None”,
@@ -164,12 +172,13 @@ MAX_SPAWN_DEPTH = 3
MAX_DIRECT_CHILDREN = 20
DEFAULT_POLL_INTERVAL_SECONDS = 5
DEFAULT_POLL_TIMEOUT_SECONDS = 300
MAX_POLL_TIMEOUT_SECONDS = 3600
# 轮询上限与状态集收敛到 cloud/gates.py / cloud/timeouts.py
#(别名保留,调用点与测试 import 的名字不变)。
MAX_POLL_TIMEOUT_SECONDS = _LONG_TIMEOUT
DEFAULT_SNAPSHOT_CHARS = 2000
MAX_SNAPSHOT_CHARS = 8000
TERMINAL_STATUSES = ("completed", "failed", "interrupted", "merged",
"merge-failed")
RETRYABLE_STATUSES = ("failed", "interrupted", "merge-failed")
TERMINAL_STATUSES = _GATE_TERMINAL
RETRYABLE_STATUSES = _GATE_RETRYABLE
EFFORTS = ("low", "medium", "high", "xhigh", "max")
EXECUTORS = ("local", "docker", "cloud")
TODO_STATUSES = ("pending", "in_progress", "blocked", "completed",
@@ -619,21 +628,10 @@ def extract_task_files(text):
def safe_worktree_path(worktree, rel):
"""worktree 内相对路径的安全解析:与服务端
_safe_worktree_path 同款——绝对路径、.. 逃逸、符号链接
逃逸一律返回 None;空串返回 worktree 本身。"""
if not rel:
return worktree
if os.path.isabs(rel):
return None
base = os.path.realpath(worktree)
target = os.path.realpath(os.path.join(base, rel))
try:
if os.path.commonpath([base, target]) != base:
return None
except ValueError:
return None
return target
"""worktree 内相对路径的安全解析:判定收敛到 cloud/gates.py
(与服务端 _safe_worktree_path 同款;绝对路径、.. 逃逸、
符号链接逃逸一律返回 None;空串返回 worktree 本身)。"""
return _gate_resolve_inside(worktree, rel)
_SPAWN_SUBJECT_ZH = ("新建的任务", "派生任务", "新建任务",
@@ -1123,45 +1121,16 @@ def ancestry_relation(conn, sender, target):
def is_goal_task(task):
"""goal 任务判定:verify_command 非空或 goal_mode 为真。
与 worker run_one 的 goal 分支同款(verify_command 或
goal_mode);goal 字段本身是 goal_mode=1 时 prompt 的副本,
不单独当判定条件,避免把普通任务的 goal 文本误判。"""
if not task:
return False
return bool(task.get("verify_command") or task.get("goal_mode"))
"""goal 任务判定:薄壳,判定在 cloud/gates.py。"""
return _gate_is_goal_task(task)
def check_control_gate(target, sender, relation, kind):
"""task-cancel / task-retry 的纯函数门禁,返回 (ok, reason)。
kind 只能是 "cancel" 或 "retry",其它值一律拒绝。权限层
不在这里:worker 侧另查 create_task / approve_merge,见
模块头;本函数只管目标状态与关系,单测不需要 AuthStore。"""
if kind not in ("cancel", "retry"):
return False, "未知的控制操作"
if target is None:
return False, "目标任务不存在"
if (target.get("project_id") is not None
and sender.get("project_id") is not None
and target.get("project_id") != sender.get("project_id")):
return False, "跨项目任务不可操作"
if target.get("id") == sender.get("id"):
return False, "不能操作自己"
if is_goal_task(target):
return False, "goal 任务不可取消或重试(goal 循环自有轮次)"
status = target.get("status")
if kind == "cancel":
if status in TERMINAL_STATUSES:
return False, "目标已终态(%s),不可取消" % status
else:
if status not in RETRYABLE_STATUSES:
return False, ("仅 failed / interrupted / merge-failed"
" 可重试,当前状态为 %s" % status)
if relation not in ("descendant", "same_creator"):
return False, "目标不在发送方派生链上且创建人不同"
return True, ""
薄壳:判定在 cloud/gates.py(权限层仍在 worker 侧,
本函数只管目标状态与关系,单测不需要 AuthStore)。"""
return _gate_decide_control(target, sender, relation, kind)
def _heartbeat_age_text(snapshot, now=None):
+164
View File
@@ -0,0 +1,164 @@
"""成本与配额的唯一判定源。
收敛前同一口径写在三处,注释里明文写着“改口径时三处
要一起改”(server._quota_status、worker._quota_exceeded、
automations._automation_quota_exceeded),且三处已经
不一致——这是本轨的 P0:
- 无价 token 上限的计数口径:worker 与 automations 只计
source='actual' 的行,服务端 _quota_unpriced_tokens 计
全部 usd IS NULL 的行(含 estimated 估算行);而
cost.py 的 P1 决议、CostStore._report 的
unpriced_actual_tokens、服务端自己的项目成本报表
(actual-only)都站在 actual-only 一边。服务端是孤例。
本轨以 actual-only 为 canonical:估算行不是真实花费,
绝不能触发上限;服务端 _quota_unpriced_tokens 改为本
模块口径(行为变化仅限含 estimated 无价行的用户:此
前可能被估算行误拦,此后不再误拦;真实花费的拦阻不变)。
- 默认配额解析:server、worker、automations 各一份
CLOUD_DEFAULT_QUOTA_USD_MICROS 解析,现统一为
default_quota_usd_micros(默认值 $10/月 = 10_000_000
微美元,非法值回落,无配额行按默认值、显式 NULL 行
才是不限额——三处旧语义一致,收敛不改语义)。
- 比较判定:used >= quota(USD)先于 unpriced >= cap
(无价),两者任一命中即阻断;该顺序与三处旧代码一致,
现收为 decide() 一处。调用方只负责取数与映射错误码
(服务端 402、worker interrupted、自动化 skipped_quota),
不得在调用点另写一套 >= 比较。
本模块只依赖标准库,是叶子模块:调用方传一个 DBAPI
连接(? 占位符,Postgres 适配层同款翻译),不 import
cloud 其它模块,避免循环。无价上限的数值本身仍由
cost.unpriced_token_cap() 拥有(环境变量
CLOUD_QUOTA_UNPRICED_TOKEN_CAP),decide() 只接受
cap 作参数,不再另起一套上限解析。
调阈边界(写清,不越界):
- 本模块只判定、不建议、不生效。没有任何函数会算
“建议把上限调到 X”,也没有任何函数会写回配额行或
改环境变量。阈值变化的唯一生效路径是运维改配置:
改 user_quotas 行(立即对下次查询生效)或改环境变量
后重启进程(进程内 env 按调用时读取)。未来如需
“自动算建议”,必须是独立的只读函数,其输出绝不能
被 decide() 或任何准入点自动采纳。
- 归属口径冻结:按任务创建者 created_by 算,不按项目、
不按操作人;协调员子任务无 created_by 时跟父任务走
(该回查仍在 worker 侧,属主解析不是本模块的职责)。
- 窗口口径冻结:UTC 自然月,与 cost_records.day 同口径
(day LIKE 'YYYY-MM%')。
"""
import os
from datetime import datetime, timezone
DEFAULT_QUOTA_USD_MICROS = 10000000
def default_quota_usd_micros():
"""默认月配额(微美元):环境变量非法时回落 $10。"""
try:
return max(0, int(os.environ.get(
"CLOUD_DEFAULT_QUOTA_USD_MICROS", "10000000")))
except (TypeError, ValueError):
return DEFAULT_QUOTA_USD_MICROS
def utc_month(now=None):
"""UTC 自然月 'YYYY-MM'(与 cost_records.day 同口径)。"""
moment = datetime.now(timezone.utc) if now is None else now
return moment.strftime("%Y-%m")
def _row_value(row, name):
try:
return row[name]
except (KeyError, IndexError, TypeError):
return None
def quota_for_owner(conn, owner):
"""属主的配额行:无行回默认值,显式 NULL 行回 None(不限额)。
只读:无行时不在此写行(写行由服务端的
_apply_default_quota / _backfill_default_quota 负责)。
表不存在时先幂等建表(与三处旧代码同款)。
"""
conn.execute(
"CREATE TABLE IF NOT EXISTS user_quotas("
"user_id TEXT PRIMARY KEY, quota_usd_micros BIGINT,"
" updated_at DOUBLE PRECISION)")
row = conn.execute(
"SELECT quota_usd_micros FROM user_quotas"
" WHERE user_id = ?",
(owner,)).fetchone()
if row is None:
return default_quota_usd_micros()
return _row_value(row, "quota_usd_micros")
def monthly_usage(conn, owner, month):
"""当月已记账用量:(used_usd_micros, unpriced_actual_tokens)。
USD:cost_records 与 tasks 按 created_by 连接,SUM
usd_micros(estimated 行 usd 为 NULL,SUM 自动跳过,
与旧三处一致)。无价:usd IS NULL 且 source='actual'
的行 input + output 之和——estimated 行一律排除(见
模块头 P0)。调用方须保证 cost_records 表已存在
(服务端/worker 走 CostStore 懒建,自动化走表存在探针)。
"""
used_row = conn.execute(
"SELECT COALESCE(SUM(c.usd_micros), 0)"
" FROM cost_records c JOIN tasks t ON c.task_id = t.id"
" WHERE t.created_by = ? AND c.day LIKE ?",
(owner, month + "%")).fetchone()
used = _row_value(used_row, 0)
if used is None:
# sqlite3.Row 按列名取不到时按位置取。
try:
used = used_row[0]
except (IndexError, TypeError):
used = 0
unpriced_row = conn.execute(
"SELECT COALESCE(SUM(c.input_tokens + c.output_tokens), 0)"
" FROM cost_records c JOIN tasks t ON c.task_id = t.id"
" WHERE t.created_by = ? AND c.day LIKE ?"
" AND c.usd_micros IS NULL AND c.source = 'actual'",
(owner, month + "%")).fetchone()
unpriced = _row_value(unpriced_row, 0)
if unpriced is None:
try:
unpriced = unpriced_row[0]
except (IndexError, TypeError):
unpriced = 0
return int(used or 0), int(unpriced or 0)
def decide(quota, used_usd, unpriced_used, cap, month):
"""唯一比较判定:放行回 None,否则回决定 dict。
quota 为 None(显式不限额)时永远放行,即使无价
已超——与服务端、worker 旧行为一致(不限额用户不
受无价上限约束)。USD 门先于无价门;命中条件均为
>=(用满即拦,不是超过才拦)。
"""
if quota is None:
return None
if used_usd >= quota:
return {
"kind": "usd",
"quota_usd_micros": quota,
"used_usd_micros": used_usd,
"month": month,
"message": ("quota_exceeded: 本月配额已用完(已用 %d / "
"配额 %d 微美元,%s)" % (used_usd, quota, month)),
}
if unpriced_used >= cap:
return {
"kind": "unpriced",
"unpriced_token_cap": cap,
"unpriced_tokens_used": unpriced_used,
"month": month,
"message": ("quota_exceeded: 本月无价模型 token 已用完"
"(已用 %d / 上限 %d,%s)"
% (unpriced_used, cap, month)),
}
return None
+3 -1
View File
@@ -22,6 +22,7 @@ import subprocess
import sys
import time
from cloud.sandbox_policy import TASK_CONTAINER_RE as _POLICY_TASK_RE
from cloud.marker_tools import extract_quota_status as _extract_quota_status
from cloud.marker_tools import extract_automation_creates as \
_extract_automation_creates
@@ -1314,7 +1315,8 @@ class DockerExecutor(Executor):
# 容器也叫 codex-*(例如 codex-postgres),按前缀删会把数据库
# 容器当成残留任务容器杀掉(2026-10-26 实测:worker 重启时
# codex-postgres 就是这样被删掉,导致控制面连不上 Postgres)。
_TASK_CONTAINER_RE = re.compile(r"^codex-[0-9a-f]{32}$")
# 正则收敛到 cloud/sandbox_policy.py(类属性别名保留)。
_TASK_CONTAINER_RE = _POLICY_TASK_RE
def cleanup_stale_containers(self, prefix="codex-"):
"""清理上一轮残留的任务容器:只删 codex-<32 位 hex 任务 id>。
+88
View File
@@ -0,0 +1,88 @@
"""沙箱与清理的单一口径(纯函数层,只依赖标准库)。
收敛前分层口径散在各处注释里,本模块把它钉成可单测
的函数,调用点只传事实、不再各写一套:
1. /tmp 隔离矩阵(Docker 外层 vs 宿主 bwrap 内层):
- Docker 执行器:容器 /tmp 与 /var/tmp 固定为 tmpfs
(sandbox.build_run_command 在额外挂载之后追加
--tmpfs,且 /tmp 目标的额外挂载在解析时即被拒绝),
容器内 /tmp 绝不落在宿主 /tmp。
- 宿主 bwrap / 本地执行器:/tmp 不隔离。宿主级 bwrap
的 /tmp 会落到宿主 /tmp 上,内层 codex 的 bwrap
--proc 挂载亦不受本仓库控制——本模块只如实标注,
绝不宣称“内层沙箱已修”。
- 任务容器名正则(codex-<32 位 hex 任务 id>)此前只
活在 sandbox.cleanup_stale_containers 的注释与正则里;
现收为 TASK_CONTAINER_RE,sandbox 直接复用它。
清理绝不按 "codex-" 前缀删:基础设施容器
(如 codex-postgres)名字不匹配正则,永远不被删。
2. 失联清理判定(cleanup_decision):worker
_reap_kill_requests 的状态与心跳两道门——任务必须
已是 interrupted,且心跳停滞超过阈值(心跳取
COALESCE(last_heartbeat, updated_at),与
server.sweep_stale_tasks 同口径;无心跳时间戳时按
已停滞处理,与 _reap 旧代码的 fall-through 一致;
恰等于阈值时不杀,旧代码是 hb >= now - threshold 即
跳过)。请求 pending、执行器有 kill_container、容器
名由任务 id 派生这三道门仍在 worker 侧,不进本层:
本层只判“该不该杀”,不执行杀。
3. Worker 寻址(worker_address):workers 注册表只有
worker_id 一列身份,默认形为“主机名-进程号”
(worker.py,未给 --worker-id 时)。本函数从该串
派生 (host, pid):取最后一个连字符后全为数字的一段
作 pid,否则 pid 为 None。注册表本身不加 host/pid
列——加列需 PG 迁移与 test_p2_migrations 登记,超出
本轨收敛范围;寻址先用派生函数,不动表结构。
单一状态来源仍是:workers 表(注册与心跳)+
tasks.last_heartbeat(陈旧判定)+ kill_requests
(清理意图),三者缺一不可,不得另起一套状态表。
"""
import re
TMPFS_PATHS = ("/tmp", "/var/tmp")
TASK_CONTAINER_RE = re.compile(r"^codex-[0-9a-f]{32}$")
def tmpfs_isolated(executor_kind, path):
"""某执行器下某路径是否走 tmpfs 隔离:(isolated, reason)。"""
if executor_kind == "docker" and path in TMPFS_PATHS:
return True, "docker --tmpfs 隔离,容器内路径不落宿主"
if executor_kind == "docker":
return False, "docker 下仅 /tmp 与 /var/tmp 走 tmpfs"
if executor_kind in ("host-bwrap", "local"):
return False, "宿主 bwrap / 本地执行器 /tmp 不隔离,会落宿主"
return False, "未知执行器,按不隔离处理(fail-closed 口径)"
def cleanup_decision(status, heartbeat_ts, now, threshold_seconds):
"""失联清理判定:放行回 (True, reason),否则 (False, reason)。
status 必须为 'interrupted';心跳停滞须严格超过阈值
(恰等于阈值不杀);heartbeat_ts 为 None 时按已停滞
处理(与 _reap 旧代码一致:无时间戳不阻拦收殓)。
"""
if status != "interrupted":
return False, "任务不在 interrupted,不杀(防复活任务被旧请求误杀)"
if heartbeat_ts is None:
return True, "无心跳时间戳,按已停滞处理"
try:
age = now - float(heartbeat_ts)
except (TypeError, ValueError):
return False, "心跳时间戳不可解析,不杀"
if age > threshold_seconds:
return True, "心跳停滞 %.0f 秒,超过阈值 %s 秒" % (
age, threshold_seconds)
return False, "owner 可能还活着(心跳 %d 秒前),留待下轮" % age
def worker_address(worker_id):
"""从 worker_id 派生 (host, pid):pid 取末段纯数字,否则 None。"""
if not worker_id or not isinstance(worker_id, str):
return "", None
if "-" in worker_id:
host, _, tail = worker_id.rpartition("-")
if host and tail.isdigit():
return host, int(tail)
return worker_id, None
+90 -91
View File
@@ -64,6 +64,15 @@ from cloud.auth import AuthStore
from cloud.coordinator import parse_parts as _coord_parse_parts
from cloud.coordinator import validate_parts as _coord_validate_parts
from cloud.cost import CostStore, unpriced_token_cap
from cloud.gates import merge_state_decision as _merge_state_decision
from cloud.gates import resolve_inside as _gate_resolve_inside
from cloud.quota import decide as _quota_decide
from cloud.quota import default_quota_usd_micros as _quota_default
from cloud.quota import monthly_usage as _quota_monthly_usage
from cloud.quota import quota_for_owner as _quota_for_owner
from cloud.timeouts import STALE_SWEEP_SECONDS as _STALE_SWEEP
from cloud.timeouts import approval_default_seconds as _approval_default
from cloud.timeouts import validate_server_timeout as _validate_timeout
from cloud.db import connect as _db_connect
from cloud.db import create_database as _create_database
from cloud.db import init_postgres_schema as _init_postgres_schema
@@ -547,19 +556,11 @@ def _task_worktree_for_files(server, task):
def _safe_worktree_path(worktree, rel):
"""worktree 内相对路径的安全解析:空串返回 worktree 本身;
绝对路径、.. 逃逸、符号链接逃逸一律返回 None。realpath 后
再校验 commonpath,目录内的符号链接指回内部时放行(读到
的仍是 worktree 内的内容),指出去时拦住。"""
if not rel:
return worktree
if os.path.isabs(rel):
return None
base = os.path.realpath(worktree)
target = os.path.realpath(os.path.join(base, rel))
if os.path.commonpath([base, target]) != base:
return None
return target
"""worktree 内相对路径的安全解析:判定收敛到 cloud/gates.py
(空串返回 worktree 本身;绝对路径、.. 逃逸、符号链接
逃逸一律返回 None;目录内的符号链接指回内部时放行,
指出去时拦住;目录违规无任何覆盖通道)。"""
return _gate_resolve_inside(worktree, rel)
def _extract_thread_list(result):
@@ -836,7 +837,8 @@ def _build_agent_chain(task, trace_dirs):
[s.get("ts"), done.get("ts") if done else None]),
})
return agents
DEFAULT_SWEEP_THRESHOLD_SECONDS = 300
# 陈旧回收阈值默认值收敛到 cloud/timeouts.py(别名保留)。
DEFAULT_SWEEP_THRESHOLD_SECONDS = _STALE_SWEEP
# 限流默认值:每 IP / 每 API key 每分钟 120 次;环境变量
# CLOUD_RATE_LIMIT_PER_MINUTE / CLOUD_RATE_LIMIT_WINDOW_SECONDS 可调。
# 限流默认内存滑动窗口;Postgres 下改走 DB 共享计数(多 server
@@ -1600,12 +1602,9 @@ def _default_quota_usd_micros():
新号现在建号即写一行默认配额(老用户无配额行不受影响,
行为不变)。环境变量 CLOUD_DEFAULT_QUOTA_USD_MICROS 可
覆盖,默认 10_000_000($10/月);非法值按默认计。
解析收敛到 cloud/quota.py(别名保留)。
"""
try:
return max(0, int(os.environ.get(
"CLOUD_DEFAULT_QUOTA_USD_MICROS", "10000000")))
except (TypeError, ValueError):
return 10000000
return _quota_default()
def _apply_default_quota(conn, user_id):
@@ -2055,7 +2054,7 @@ def request_kill(db_path, task_id, requested_by):
conn.close()
def sweep_stale_tasks(db_path, threshold_seconds=300):
def sweep_stale_tasks(db_path, threshold_seconds=_STALE_SWEEP):
"""把 running 且心跳过期的任务改为 interrupted,返回被扫任务 id 列表。
心跳取 last_heartbeat,无心跳则用 updated_at;早于
@@ -8096,9 +8095,11 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
user_id, task["project_id"], "approve_merge"):
self._send_json(403, {"error": "forbidden"})
return
if task["status"] != "completed":
_merge_ok, _merge_reason = _merge_state_decision(
task["status"])
if not _merge_ok:
self._send_json(409, {
"error": "only completed tasks can be marked merged",
"error": _merge_reason,
"status": task["status"],
})
return
@@ -8211,7 +8212,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
权限:项目 create_task(建审批项是写动作,与后续消息
同级)。kind 缺省 elicitation;summary 必填;
timeout_seconds 只接受 60 到 86400 秒整数,缺省 3600,
超出范围回 400。成功记审计 approval_created。"""
超出范围回 400(校验收敛到 cloud/timeouts.py,文案
不变)。成功记审计 approval_created。"""
row = self._fetch_task(task_id)
if row is None:
self._send_json(404, {"error": "task not found"})
@@ -8233,9 +8235,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
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:
timeout = body.get("timeout_seconds", _approval_default())
try:
timeout = _validate_timeout(timeout)
except ValueError:
self._send_json(
400, {"error": "timeout_seconds must be 60..86400"})
return
@@ -8634,77 +8637,61 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
from datetime import datetime, timezone
return datetime.now(timezone.utc).strftime("%Y-%m")
def _quota_status(self, user_id):
"""返回 (quota_usd_micros|None, used_usd_micros, month)。
P0 安全批后:无配额行不再等于不限额——缺行时按默认
配额处理(与 _default_quota_usd_micros() 同值,只在
返回值里合成,不在这里写行;写行由 _apply_default_quota
/ _backfill_default_quotas 负责)。只有管理员显式把
行写成 NULL 时,quota 才为 None(不限额)。"""
month = self._quota_month()
conn = self._db()
try:
_ensure_quotas_table(conn)
conn.commit()
# cost_records 由 CostStore 懒建:全新库上它还不存在,
# 直接 JOIN 会报 no such table——先实例化一次建表
# (与 _project_cost 同款做法),再跑配额查询。
_cost_store = CostStore(self.server.db_path)
try:
_cost_store._conn.close()
except Exception:
pass
row = conn.execute(
"SELECT quota_usd_micros FROM user_quotas"
" WHERE user_id = ?",
(user_id,)).fetchone()
if row is None:
quota = _default_quota_usd_micros()
else:
quota = row[0]
used = conn.execute(
"SELECT COALESCE(SUM(c.usd_micros), 0)"
" FROM cost_records c JOIN tasks t ON c.task_id = t.id"
" WHERE t.created_by = ? AND c.day LIKE ?",
(user_id, month + "%")).fetchone()[0]
return quota, int(used or 0), month
finally:
conn.close()
def _quota_snapshot(self, user_id, month=None):
"""配额快照:(quota, used_usd, unpriced_actual, month)。
def _quota_unpriced_tokens(self, user_id, month=None):
"""当月无价模型 token 数(usd_micros IS NULL 的成本行,
input + output 之和;cached 已含在 input 里,不重复计)。
合并说明(#86 × #90):两边曾各自定义过一次(#90 的
只要 user_id 且只计 source='actual',#86 的要 month 且
计全部 usd IS NULL 行),自动合并后后一个覆盖前一个,
上报调用点会 TypeError。这里统一成一个方法,以 main
(#86)的口径为准:计全部 usd IS NULL 行(含 estimated
行),上报与上限共用同一口径,不再留两套数字;month
缺省为当月,兼容两边的调用点。与 _quota_status 同款
CostStore 懒建表做法,全新库上 cost_records 不存在时
先建表再查。无配额行时调用方不应以此挡人(见
_check_unpriced_token_cap)。
取数收敛到 cloud/quota.py:配额行(无行按默认值、
显式 NULL 行为不限额)与当月用量(USD 全量 +
无价 actual-only)各只有一处 SQL。CostStore 懒建表
步骤保留(全新库上 cost_records 不存在时先建表再查)。
本函数只读:无行时不在此写行(写行仍由
_apply_default_quota / _backfill_default_quota 负责)。
"""
if month is None:
month = self._quota_month()
conn = self._db()
try:
_ensure_quotas_table(conn)
conn.commit()
_cost_store = CostStore(self.server.db_path)
try:
_cost_store._conn.close()
except Exception:
pass
row = conn.execute(
"SELECT COALESCE(SUM(c.input_tokens + c.output_tokens), 0)"
" FROM cost_records c JOIN tasks t ON c.task_id = t.id"
" WHERE t.created_by = ? AND c.day LIKE ?"
" AND c.usd_micros IS NULL",
(user_id, month + "%")).fetchone()
return int(row[0] if row is not None else 0)
quota = _quota_for_owner(conn, user_id)
used, unpriced = _quota_monthly_usage(conn, user_id, month)
return quota, used, unpriced, month
finally:
conn.close()
def _quota_status(self, user_id):
"""返回 (quota_usd_micros|None, used_usd_micros, month)。
P0 安全批后:无配额行不再等于不限额——缺行时按默认
配额处理(只在返回值里合成,不在这里写行;写行由
_apply_default_quota / _backfill_default_quota 负责)。
只有管理员显式把行写成 NULL 时,quota 才为 None
(不限额)。取数见 _quota_snapshot。"""
quota, used, _unpriced, month = self._quota_snapshot(user_id)
return quota, used, month
def _quota_unpriced_tokens(self, user_id, month=None):
"""当月无价模型 token 数(actual-only,见下)。
收敛说明(改动点,见 cloud/quota.py 模块头):
旧实现计全部 usd IS NULL 的行(含 estimated 估算行),
与 cost.py 的 P1 决议、CostStore 报表、worker、
自动化三处 actual-only 口径不一致;现统一为
actual-only:估算行不是真实花费,绝不触发上限。
input + output 之和;cached 已含在 input 里,不重复计。
month 缺省为当月。与 _quota_status 同款 CostStore
懒建表做法。无配额行时调用方不应以此挡人(见
_check_unpriced_token_cap 与 quota.decide 的
不限额语义)。
"""
_quota, _used, unpriced, _month = self._quota_snapshot(
user_id, month=month)
return unpriced
def _quota_payload(self, user_id, quota, used, month,
unpriced_actual_tokens=None):
payload = {"user_id": user_id, "month": month,
@@ -8718,14 +8705,23 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
def _check_unpriced_token_cap(self, user_id):
"""无价 token 上限准入:超限返回 (cap, used, month),否则 None。
P0 安全批后:无配额行用户同样受限(无价 token 上限与
配额行存在与否无关);USD 配额本身不计量无价模型,所
判定收敛到 cloud/quota.py:无配额行用户按默认配额受限;
显式 NULL(不限额)行用户不受限——与 worker 运行时门
一致(旧实现不论配额行一律拦,是三处不一致之一,
本轨随收敛统一)。USD 配额本身不计量无价模型,所
以这道是补 USD 口径的洞,不是第二套计费。"""
month = self._quota_month()
used = self._quota_unpriced_tokens(user_id, month)
cap = unpriced_token_cap()
if used >= cap:
return cap, used, month
# 取数经 _quota_status / _quota_unpriced_tokens(两者
# 内部共用 _quota_snapshot),比较经 quota.decide。
# _quota_status 在此保留为可 mock 的接缝(503 测试依赖
# 它抛异常时准入点 fail-closed),故 _check_* 不直连
# _quota_snapshot。
quota, _used, month = self._quota_status(user_id)
unpriced = self._quota_unpriced_tokens(user_id, month)
decision = _quota_decide(quota, _used, unpriced,
unpriced_token_cap(), month)
if decision is not None and decision["kind"] == "unpriced":
return (decision["unpriced_token_cap"],
decision["unpriced_tokens_used"], month)
return None
def _check_quota(self, user_id):
@@ -8761,7 +8757,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
回合结束才回传),掐断粒度是轮次边界 + 长轮询中的
30 秒检查,不是 token 级。"""
quota, used, month = self._quota_status(user_id)
if quota is not None and used >= quota:
unpriced = self._quota_unpriced_tokens(user_id, month)
decision = _quota_decide(quota, used, unpriced,
unpriced_token_cap(), month)
if decision is not None and decision["kind"] == "usd":
return quota, used, month
return None
+344
View File
@@ -0,0 +1,344 @@
"""架构收敛 A 线单测:超时 / 门禁 / 成本配额 / 沙箱清理。
只覆盖本轨新增的四个叶子模块与调用点的委托关系,不跑
模型、不联网、不碰生产:SQLite 内存库 + 临时目录。
钉住的冻结语义(改动这些断言即改语义,必须先改设计):
- 审批默认 3600、服务端入口 60..86400、已建项 expires_at
永不因默认值变化而重算;600 秒是单轮护栏,不是审批超时。
- 目录门无 confirm / override / bypass 参数;合并状态门
仅 completed 放行;控制门与旧 marker 行为逐字一致。
- 配额判定唯一来源是 quota.decide:USD 门先于无价门,
命中均为 >=;无价只计 source='actual',估算行不触发上限;
无配额行按默认值,显式 NULL 行才是不限额。
- 失联清理只在 interrupted + 心跳停滞严格超阈时放行;
容器清理只认 codex-<32 位 hex>,绝不按前缀删。
"""
import inspect
import os
import sqlite3
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
try:
from cloud import gates, quota, sandbox_policy, timeouts
from cloud.approvals import DEFAULT_TIMEOUT_SECONDS, create_approval
from cloud.automations import (
_automation_quota_exceeded,
_unpriced_token_cap as _auto_cap,
)
from cloud.coordinator import DEFAULT_TIMEOUT_SECONDS as _COORD_TIMEOUT
from cloud.cost import unpriced_token_cap as _cost_cap
from cloud.harness2_tasks import (
DEFAULT_TASK_TIMEOUT_SECONDS as _HARNESS_TIMEOUT,
)
from cloud.marker_tools import (
MAX_POLL_TIMEOUT_SECONDS as _MARKER_POLL_MAX,
)
from cloud.marker_tools import check_control_gate, safe_worktree_path
from cloud.sandbox import DockerExecutor
from cloud.server import _safe_worktree_path as _server_safe_path
from cloud.tree import resolve_tree_path
from cloud.workers import DEFAULT_ALIVE_SECONDS as _WORKER_ALIVE
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud import gates, quota, sandbox_policy, timeouts
from cloud.approvals import DEFAULT_TIMEOUT_SECONDS, create_approval
from cloud.automations import (
_automation_quota_exceeded,
_unpriced_token_cap as _auto_cap,
)
from cloud.coordinator import DEFAULT_TIMEOUT_SECONDS as _COORD_TIMEOUT
from cloud.cost import unpriced_token_cap as _cost_cap
from cloud.harness2_tasks import (
DEFAULT_TASK_TIMEOUT_SECONDS as _HARNESS_TIMEOUT,
)
from cloud.marker_tools import (
MAX_POLL_TIMEOUT_SECONDS as _MARKER_POLL_MAX,
)
from cloud.marker_tools import check_control_gate, safe_worktree_path
from cloud.sandbox import DockerExecutor
from cloud.server import _safe_worktree_path as _server_safe_path
from cloud.tree import resolve_tree_path
from cloud.workers import DEFAULT_ALIVE_SECONDS as _WORKER_ALIVE
def _quota_conn():
"""配额测试用的最小库:tasks + cost_records + user_quotas。"""
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute("CREATE TABLE tasks(id TEXT PRIMARY KEY,"
" created_by TEXT)")
conn.execute("CREATE TABLE cost_records(task_id TEXT,"
" usd_micros INTEGER, input_tokens INTEGER,"
" output_tokens INTEGER, day TEXT, source TEXT)")
conn.execute("CREATE TABLE user_quotas(user_id TEXT PRIMARY KEY,"
" quota_usd_micros BIGINT, updated_at DOUBLE PRECISION)")
conn.execute("INSERT INTO tasks(id, created_by) VALUES('t1', 'u1')")
conn.commit()
return conn
def _cost_row(conn, task_id, usd, inp, out, source, day="2026-10-31"):
conn.execute(
"INSERT INTO cost_records(task_id, usd_micros, input_tokens,"
" output_tokens, day, source) VALUES(?, ?, ?, ?, ?, ?)",
(task_id, usd, inp, out, day, source))
conn.commit()
class TimeoutConvergeTestCase(unittest.TestCase):
def test_constants_match_old_literals(self):
self.assertEqual(timeouts.APPROVAL_DEFAULT_SECONDS, 3600)
self.assertEqual(timeouts.APPROVAL_MIN_SECONDS, 60)
self.assertEqual(timeouts.APPROVAL_MAX_SECONDS, 86400)
self.assertEqual(timeouts.TURN_SECONDS, 600)
self.assertEqual(timeouts.STALE_SWEEP_SECONDS, 300)
self.assertEqual(timeouts.WORKER_ALIVE_SECONDS, 120)
# 各模块别名与单一来源同值(调用点 import 的名字不变)。
self.assertEqual(DEFAULT_TIMEOUT_SECONDS, 3600)
self.assertEqual(_COORD_TIMEOUT, timeouts.LONG_TIMEOUT_SECONDS)
self.assertEqual(_HARNESS_TIMEOUT, timeouts.LONG_TIMEOUT_SECONDS)
self.assertEqual(_MARKER_POLL_MAX, timeouts.LONG_TIMEOUT_SECONDS)
self.assertEqual(_WORKER_ALIVE, timeouts.WORKER_ALIVE_SECONDS)
def test_server_validation_keeps_old_bounds_and_message(self):
for good in (60, 3600, 86400):
self.assertEqual(timeouts.validate_server_timeout(good), good)
for bad in (None, True, 59, 86401, 3.5, "3600"):
with self.assertRaises(ValueError) as ctx:
timeouts.validate_server_timeout(bad)
self.assertEqual(str(ctx.exception),
"timeout_seconds must be 60..86400")
def test_module_timeout_semantics(self):
self.assertEqual(timeouts.resolve_module_timeout(None), 3600)
self.assertIsNone(timeouts.resolve_module_timeout(0))
self.assertEqual(timeouts.resolve_module_timeout(60), 60)
for bad in (-1, True, "60"):
with self.assertRaises(ValueError):
timeouts.resolve_module_timeout(bad)
self.assertIsNone(timeouts.expires_at_for(1000, None))
self.assertEqual(timeouts.expires_at_for(1000, 3600), 4600)
def test_env_override_and_invalid_fallback(self):
with mock.patch.dict(os.environ,
{"CLOUD_APPROVAL_TIMEOUT_SECONDS": "1800"}):
self.assertEqual(timeouts.approval_default_seconds(), 1800)
with mock.patch.dict(os.environ,
{"CLOUD_APPROVAL_TIMEOUT_SECONDS": "30"}):
self.assertEqual(timeouts.approval_default_seconds(), 3600)
with mock.patch.dict(os.environ,
{"CLOUD_APPROVAL_TIMEOUT_SECONDS": "abc"}):
self.assertEqual(timeouts.approval_default_seconds(), 3600)
def test_existing_approval_expires_at_never_recomputed(self):
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
try:
with mock.patch.dict(os.environ, {}, clear=False):
os.environ.pop("CLOUD_APPROVAL_TIMEOUT_SECONDS", None)
old = create_approval(conn, "t1", "p1", summary="s",
now=1000.0)
self.assertEqual(old["expires_at"], 4600.0)
with mock.patch.dict(os.environ,
{"CLOUD_APPROVAL_TIMEOUT_SECONDS": "60"}):
new = create_approval(conn, "t1", "p1", summary="s",
now=2000.0)
self.assertEqual(new["expires_at"], 2060.0)
# 旧行不受默认值变化影响:不回填、不重算。
from cloud.approvals import get_approval
self.assertEqual(get_approval(conn, old["id"])["expires_at"],
4600.0)
finally:
conn.close()
class GatesConvergeTestCase(unittest.TestCase):
def test_path_gate_cases_across_all_wrappers(self):
with tempfile.TemporaryDirectory() as tmp:
base = os.path.join(tmp, "w")
os.makedirs(os.path.join(base, "sub"))
# 目录内指回内部的符号链接放行,指出去的拦住。
os.symlink(os.path.join(base, "sub"),
os.path.join(base, "inner-link"))
os.symlink("/etc", os.path.join(base, "escape-link"))
for fn in (gates.resolve_inside, resolve_tree_path,
safe_worktree_path, _server_safe_path):
self.assertEqual(fn(base, ""), base)
self.assertEqual(fn(base, "sub"),
os.path.realpath(os.path.join(base, "sub")))
self.assertIsNone(fn(base, "../etc/passwd"))
self.assertIsNone(fn(base, "/etc/passwd"))
self.assertIsNotNone(fn(base, "inner-link"))
self.assertIsNone(fn(base, "escape-link"))
def test_path_and_control_gates_have_no_bypass_parameter(self):
for fn in (gates.resolve_inside, gates.decide_control,
gates.merge_state_decision):
params = set(inspect.signature(fn).parameters)
self.assertFalse(params & {"confirm", "override", "bypass",
"force", "confirmed"})
def test_merge_state_gate(self):
self.assertEqual(gates.merge_state_decision("completed"),
(True, ""))
for status in ("queued", "running", "failed", "interrupted",
"merged", "merge-failed"):
ok, reason = gates.merge_state_decision(status)
self.assertFalse(ok)
self.assertEqual(reason,
"only completed tasks can be marked merged")
def test_control_gate_matches_old_marker_behavior(self):
sender = {"id": "s1", "project_id": "p1"}
target = {"id": "t1", "project_id": "p1", "status": "running"}
self.assertEqual(check_control_gate(target, sender, "descendant",
"cancel"), (True, ""))
self.assertEqual(
check_control_gate(target, sender, "descendant", "retry")[0],
False)
target2 = dict(target, status="failed")
self.assertEqual(check_control_gate(target2, sender, "same_creator",
"retry"), (True, ""))
self.assertEqual(
check_control_gate(target2, sender, "other", "retry")[0], False)
self.assertEqual(
check_control_gate(target2, {"id": "t1", "project_id": "p1"},
"descendant", "cancel")[0], False)
goal = dict(target2, goal_mode=1)
self.assertEqual(
check_control_gate(goal, sender, "descendant", "retry")[0],
False)
cross = dict(target2, project_id="p2")
self.assertEqual(
check_control_gate(cross, sender, "descendant", "retry")[0],
False)
class QuotaConvergeTestCase(unittest.TestCase):
def test_decide_precedence_and_equality(self):
self.assertIsNone(quota.decide(None, 999999, 999999, 100, "2026-10"))
self.assertIsNone(quota.decide(1000, 999, 99, 100, "2026-10"))
both = quota.decide(1000, 1000, 1000, 100, "2026-10")
self.assertEqual(both["kind"], "usd")
only_unpriced = quota.decide(1000, 10, 100, 100, "2026-10")
self.assertEqual(only_unpriced["kind"], "unpriced")
self.assertIn("quota_exceeded", only_unpriced["message"])
def test_quota_for_owner_rows(self):
conn = _quota_conn()
try:
with mock.patch.dict(os.environ, {}, clear=False):
os.environ.pop("CLOUD_DEFAULT_QUOTA_USD_MICROS", None)
self.assertEqual(quota.quota_for_owner(conn, "u1"),
10000000)
conn.execute("INSERT INTO user_quotas VALUES('u1', NULL, 0)")
conn.commit()
self.assertIsNone(quota.quota_for_owner(conn, "u1"))
conn.execute("UPDATE user_quotas SET quota_usd_micros = 5000"
" WHERE user_id = 'u1'")
conn.commit()
self.assertEqual(quota.quota_for_owner(conn, "u1"), 5000)
finally:
conn.close()
def test_monthly_usage_actual_only(self):
conn = _quota_conn()
try:
_cost_row(conn, "t1", 300, 10, 5, "actual")
_cost_row(conn, "t1", None, 100, 50, "actual")
_cost_row(conn, "t1", None, 999900, 99, "estimated")
used, unpriced = quota.monthly_usage(conn, "u1", "2026-10")
self.assertEqual(used, 300)
# 估算行 999999 token 不计入无价上限口径。
self.assertEqual(unpriced, 150)
finally:
conn.close()
def test_automation_gate_uses_actual_only(self):
conn = _quota_conn()
try:
conn.execute("INSERT INTO user_quotas VALUES('u1', 100000000, 0)")
conn.commit()
_cost_row(conn, "t1", None, 999900, 99, "estimated")
with mock.patch.dict(
os.environ,
{"CLOUD_QUOTA_UNPRICED_TOKEN_CAP": "1000"}):
# 估算行再大也不拦。
self.assertIsNone(_automation_quota_exceeded(conn, "u1"))
_cost_row(conn, "t1", None, 600, 500, "actual")
blocked = _automation_quota_exceeded(conn, "u1")
self.assertIsNotNone(blocked)
self.assertIn("quota_exceeded", blocked)
finally:
conn.close()
def test_unpriced_cap_env_alias_matches_cost(self):
with mock.patch.dict(
os.environ, {"CLOUD_QUOTA_UNPRICED_TOKEN_CAP": "12345"}):
self.assertEqual(_auto_cap(), 12345)
self.assertEqual(_cost_cap(), 12345)
with mock.patch.dict(
os.environ, {"CLOUD_QUOTA_UNPRICED_TOKEN_CAP": "junk"}):
self.assertEqual(_auto_cap(), _cost_cap())
class SandboxPolicyTestCase(unittest.TestCase):
def test_tmpfs_matrix(self):
self.assertTrue(sandbox_policy.tmpfs_isolated("docker", "/tmp")[0])
self.assertTrue(
sandbox_policy.tmpfs_isolated("docker", "/var/tmp")[0])
self.assertFalse(
sandbox_policy.tmpfs_isolated("host-bwrap", "/tmp")[0])
self.assertFalse(sandbox_policy.tmpfs_isolated("local", "/tmp")[0])
self.assertFalse(
sandbox_policy.tmpfs_isolated("docker", "/work")[0])
def test_task_container_regex_shared_with_sandbox(self):
self.assertIs(sandbox_policy.TASK_CONTAINER_RE,
DockerExecutor._TASK_CONTAINER_RE)
good = "codex-" + "a" * 32
self.assertTrue(sandbox_policy.TASK_CONTAINER_RE.match(good))
for bad in ("codex-postgres", "codex-xyz", "codex-" + "a" * 31,
"codex-" + "A" * 32):
self.assertIsNone(sandbox_policy.TASK_CONTAINER_RE.match(bad))
def test_cleanup_decision(self):
now = 10000.0
self.assertEqual(
sandbox_policy.cleanup_decision("running", 1.0, now, 300)[0],
False)
self.assertEqual(
sandbox_policy.cleanup_decision("interrupted", now - 301,
now, 300)[0], True)
# 恰等于阈值不杀(旧 _reap:hb >= now - threshold 即跳过)。
self.assertEqual(
sandbox_policy.cleanup_decision("interrupted", now - 300,
now, 300)[0], False)
self.assertEqual(
sandbox_policy.cleanup_decision("interrupted", now - 10,
now, 300)[0], False)
self.assertEqual(
sandbox_policy.cleanup_decision("interrupted", None,
now, 300)[0], True)
self.assertEqual(
sandbox_policy.cleanup_decision("interrupted", "junk",
now, 300)[0], False)
def test_worker_address(self):
self.assertEqual(sandbox_policy.worker_address("my-host-42"),
("my-host", 42))
self.assertEqual(sandbox_policy.worker_address("plain"),
("plain", None))
self.assertEqual(sandbox_policy.worker_address("host-abc"),
("host-abc", None))
self.assertEqual(sandbox_policy.worker_address(""), ("", None))
if __name__ == "__main__":
unittest.main()
+124
View File
@@ -0,0 +1,124 @@
"""超时默认值的单一配置源。
背景:主线(origin/main)上同一个“3600 秒”以字面量散落在
多处——审批队列(cloud/approvals.py)、协调员
(cloud/coordinator.py)、接线层任务超时
(cloud/harness2_tasks.py)、标记轮询上限
(cloud/marker_tools.py)、服务端审批入口(cloud/server.py),
另有单轮护栏 600 秒(host.py 的 turn_timeout 默认值,见下)
与陈旧回收 300 秒(server.sweep_stale_tasks /
worker._reap_kill_requests)。改一处漏一处是真实风险:
本次收敛前,审批入口与队列模块的上下界就各写了一套。
本模块只依赖标准库,是叶子模块:任何 cloud 模块都可
import 它而不引入循环。每个用途各有名常量,值即使相同
也不共用一个名字——改审批默认绝不能顺带改掉轮询上限。
各常量与现状(改值前必须先读本段):
- APPROVAL_DEFAULT_SECONDS = 3600:新审批项的默认等待
秒数。覆盖层:环境变量 CLOUD_APPROVAL_TIMEOUT_SECONDS,
仅当解析为 60..86400 的整数时生效,否则回落默认值。
- APPROVAL_MIN_SECONDS / APPROVAL_MAX_SECONDS = 60 / 86400:
服务端审批入口的校验区间(保持与旧行为一致)。
- LONG_TIMEOUT_SECONDS = 3600:协调员、接线层任务超时、
标记轮询上限三处共用的数值,各模块仍保留自己的常量名
作为别名(调用点与测试 import 的名字不变)。
- TURN_SECONDS = 600:host.py 单轮护栏的默认值。host.py
不在 cloud/ 内,本轨文件边界不改它;这里只登记数值与
含义,杜绝“审批 600 秒”的误读——主线上不存在 600 秒的
审批超时,600 秒是单轮护栏,不是审批等待。
- STALE_SWEEP_SECONDS = 300:陈旧回收与失联收殓共用的
阈值默认值(server 与 worker 各保留自己的参数名)。
- WORKER_ALIVE_SECONDS = 120:worker 注册表存活阈值。
已建审批项的迁移语义(冻结,不可悄悄改):
- 本模块没有任何回填、改写 expires_at 的函数。默认值或
环境变量变化只影响之后新建的审批项;已建项的 expires_at
在创建时已落库,之后一律不动。过期扫荡仍按行上的
expires_at 判定,绝不按当前默认值重算。
- 服务端入口的校验与队列模块的创建语义各有分工,不可
混用:服务端入口只接受 60..86400 的整数(见
validate_server_timeout,显式 null、布尔值、浮点数、
越界一律拒绝,与旧行为一致);队列模块级创建允许
timeout_seconds=None(用默认值)与 0(不过期,见
resolve_module_timeout),服务端入口永远传不出 0。
"""
import os
APPROVAL_DEFAULT_SECONDS = 3600
APPROVAL_MIN_SECONDS = 60
APPROVAL_MAX_SECONDS = 86400
LONG_TIMEOUT_SECONDS = 3600
TURN_SECONDS = 600
STALE_SWEEP_SECONDS = 300
WORKER_ALIVE_SECONDS = 120
def approval_default_seconds(override=None):
"""新审批项的默认等待秒数:显式参数 > 环境变量 > 默认值。
环境变量 CLOUD_APPROVAL_TIMEOUT_SECONDS 只有解析为
60..86400 的整数时才生效;非法、越界、布尔语义一律
回落默认值,绝不抛异常(默认值解析失败不能炸掉建项)。
"""
if override is not None:
try:
value = int(override)
except (TypeError, ValueError):
return APPROVAL_DEFAULT_SECONDS
if isinstance(override, bool):
return APPROVAL_DEFAULT_SECONDS
if APPROVAL_MIN_SECONDS <= value <= APPROVAL_MAX_SECONDS:
return value
return APPROVAL_DEFAULT_SECONDS
raw = os.environ.get("CLOUD_APPROVAL_TIMEOUT_SECONDS") or ""
try:
value = int(raw)
except (TypeError, ValueError):
return APPROVAL_DEFAULT_SECONDS
if APPROVAL_MIN_SECONDS <= value <= APPROVAL_MAX_SECONDS:
return value
return APPROVAL_DEFAULT_SECONDS
def validate_server_timeout(value):
"""服务端审批入口的超时校验:返回合法整数,否则 ValueError。
与旧入口行为逐字一致:缺省由调用方先填默认值;显式
null、布尔值、浮点数、非整数、越界一律拒绝,错误文案
保持 "timeout_seconds must be 60..86400" 不变。
"""
if isinstance(value, bool) or not isinstance(value, int):
raise ValueError("timeout_seconds must be 60..86400")
if value < APPROVAL_MIN_SECONDS or value > APPROVAL_MAX_SECONDS:
raise ValueError("timeout_seconds must be 60..86400")
return value
def resolve_module_timeout(value, default=None):
"""队列模块级创建的超时解析:返回秒数或 None(不过期)。
None 用默认值;0 表示不过期(只允许模块级调用,服务
端入口另有限制,永远传不出 0);负数、布尔值、非数字
一律 ValueError。浮点数沿用旧行为:接受并按原值参与
expires_at 计算(服务端入口不走这条路,不受影响)。
"""
if default is None:
default = approval_default_seconds()
if value is None:
return default
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError("timeout_seconds invalid")
if value < 0:
raise ValueError("timeout_seconds invalid")
if not value:
return None
return value
def expires_at_for(now, timeout_seconds):
"""由创建时刻与解析后的超时算 expires_at:不过期回 None。"""
if not timeout_seconds:
return None
return now + timeout_seconds
+5 -12
View File
@@ -28,6 +28,8 @@ max_depth 从请求根算起:1 只列根的直接条目,2 再列一层
import os
from cloud.gates import resolve_inside as _resolve_inside
__all__ = [
"DEFAULT_IGNORE",
"DEFAULT_MAX_DEPTH",
@@ -64,19 +66,10 @@ def resolve_tree_path(root, rel):
放行到其真实位置——调用方仍只拿到 worktree 内的内容,
但 build_tree 遇到链接本身时不跟进(见下),所以这条
放行只影响“显式 path 指向内部链接目录”时的根解析。
判定收敛到 cloud/gates.py(与服务端、marker_tools
同款,目录违规无任何覆盖通道)。
"""
if not rel:
return root
if os.path.isabs(rel):
return None
base = os.path.realpath(root)
target = os.path.realpath(os.path.join(base, rel))
try:
if os.path.commonpath([base, target]) != base:
return None
except ValueError:
return None
return target
return _resolve_inside(root, rel)
def build_tree(root, path="", max_depth=DEFAULT_MAX_DEPTH,
+39 -52
View File
@@ -142,7 +142,13 @@ from cloud.coordinator import ensure_columns as _coord_ensure_columns
from cloud.coordinator import parse_parts as _coord_parse_parts
from cloud.coordinator import nonterminal_children as _coord_nonterminal
from cloud.cost import CostStore, estimate_tokens, price_usd_micros, \
quota_cutoff_reason, sum_usage
sum_usage, unpriced_token_cap
from cloud.quota import decide as _quota_decide
from cloud.quota import default_quota_usd_micros as _quota_default
from cloud.quota import monthly_usage as _quota_monthly_usage
from cloud.quota import quota_for_owner as _quota_for_owner
from cloud.sandbox_policy import cleanup_decision as _cleanup_decision
from cloud.timeouts import STALE_SWEEP_SECONDS as _STALE_SWEEP
from cloud.db import connect as _db_connect
from cloud.goal_command import parse_goal_command as _parse_goal_command
from cloud.logging_config import get_logger, log_event, setup_logging
@@ -279,17 +285,10 @@ def _worker_host_name(executor, task):
def _default_quota_usd_micros():
"""无配额行时的默认配额(微美元),与 server 侧同值。
server.py 的 _default_quota_usd_micros() 在另一路文件里,
worker 不跨模块 import 它(避免循环 import),这里复刻
同款解析:环境变量 CLOUD_DEFAULT_QUOTA_USD_MICROS,非法
值回落 10_000_000。"""
try:
return max(0, int(os.environ.get(
"CLOUD_DEFAULT_QUOTA_USD_MICROS", "10000000")))
except (TypeError, ValueError):
return 10000000
"""无配额行时的默认配额(微美元):解析收敛到
cloud/quota.py(别名保留;quota 模块是标准库叶子,
不存在旧注释担心的循环 import)。"""
return _quota_default()
def _code_signature(host_py_path=None):
@@ -2081,21 +2080,14 @@ class Worker:
month = datetime.now(timezone.utc).strftime("%Y-%m")
conn = _db_connect(self.db_path)
try:
conn.execute(
"CREATE TABLE IF NOT EXISTS user_quotas("
"user_id TEXT PRIMARY KEY, quota_usd_micros BIGINT,"
" updated_at DOUBLE PRECISION)")
conn.commit()
row = conn.execute(
"SELECT quota_usd_micros FROM user_quotas"
" WHERE user_id = ?",
(owner,)).fetchone()
if row is None:
# P0/P1 口径:无行按默认额度,不是不限额(只有
# 显式 NULL 行才是不限额,见下)。
quota = _default_quota_usd_micros()
else:
quota = row[0]
# 取数收敛到 cloud/quota.py(配额行口径与当月
# 用量 SQL 各只有一处;无行按默认额度,显式 NULL
# 行才是不限额,见 quota_for_owner)。
quota = _quota_for_owner(conn, owner)
try:
conn.commit()
except Exception:
pass
if quota is None:
return None, 0, month, 0
# cost_records 由 CostStore 懒建:全新库上它还不存
@@ -2107,21 +2099,8 @@ class Worker:
_cost_store._conn.close()
except Exception:
pass
used_row = conn.execute(
"SELECT COALESCE(SUM(c.usd_micros), 0)"
" FROM cost_records c JOIN tasks t"
" ON c.task_id = t.id"
" WHERE t.created_by = ? AND c.day LIKE ?",
(owner, month + "%")).fetchone()
used = used_row[0] if used_row else 0
unpriced_row = conn.execute(
"SELECT COALESCE(SUM(c.input_tokens + c.output_tokens), 0)"
" FROM cost_records c JOIN tasks t"
" ON c.task_id = t.id"
" WHERE t.created_by = ? AND c.day LIKE ?"
" AND c.usd_micros IS NULL AND c.source = 'actual'",
(owner, month + "%")).fetchone()
unpriced_used = unpriced_row[0] if unpriced_row else 0
used, unpriced_used = _quota_monthly_usage(
conn, owner, month)
return quota, int(used or 0), month, int(unpriced_used or 0)
finally:
conn.close()
@@ -2210,11 +2189,13 @@ class Worker:
unpriced_used += (
max(0, int(usage_total.get("input_tokens", 0)))
+ max(0, int(usage_total.get("output_tokens", 0))))
# 硬切断判定收敛到 cost.quota_cutoff_reason:USD 配额与
# 无价上限任一超限即回 quota_exceeded 文本,调用方(认领
# 门、轮次边界、30 秒轮询)据此停掉任务、不再开新回合;
# 文本与旧分支逐字一致,旧单测按此前缀断言。
return quota_cutoff_reason(quota, used, unpriced_used, month)
# 比较判定收敛到 cloud/quota.py(USD 门先于无价门,
# 命中条件均为 >=;文案与旧分支逐字一致)。
decision = _quota_decide(quota, used, unpriced_used,
unpriced_token_cap(), month)
if decision is not None:
return decision["message"]
return None
def _ready_tasks(self):
"""当前就绪任务列表(pick_next 的批量版)。
@@ -5249,7 +5230,7 @@ class Worker:
conn.close()
return marked
def _reap_kill_requests(self, sweep_threshold=300):
def _reap_kill_requests(self, sweep_threshold=_STALE_SWEEP):
"""跨 worker 真 kill 的收殓器(run loop 每轮调一次)。
分工(与 owner 活着时的 1 秒看门狗互补,绝不抢杀):
@@ -5337,7 +5318,12 @@ class Worker:
hb = row["last_heartbeat"]
if hb is None:
hb = row["updated_at"]
if hb is not None and hb >= now - sweep_threshold:
# 状态与心跳两道门收敛到 sandbox_policy:
# 非 interrupted 已在上一步终结请求;这里只
# 判心跳是否停滞超阈(恰等于阈值不杀)。
_reap_ok, _reap_reason = _cleanup_decision(
"interrupted", hb, now, sweep_threshold)
if not _reap_ok:
# owner 还活着:留待下轮,绝不抢杀。
continue
# owner 已死:按任务派生容器名在本机真杀。
@@ -6755,7 +6741,7 @@ class Worker:
return polls
def run_loop(self, interval=300, max_seconds=None, stop_event=None,
drain_timeout=60, sweep_threshold=300):
drain_timeout=60, sweep_threshold=_STALE_SWEEP):
"""常驻循环:每 interval 秒跑一轮 run_all,空闲也不退出。
退出条件:stop_event 被置位、_loop_stop 被信号处理置位、
@@ -6871,7 +6857,7 @@ class Worker:
try:
self._reap_kill_requests(
sweep_threshold
if sweep_threshold is not None else 300)
if sweep_threshold is not None else _STALE_SWEEP)
except Exception as exc:
log_event(self.logger, "worker_reap_failed",
message="worker reap failed: %s" % exc)
@@ -7152,7 +7138,8 @@ def main(argv=None):
"后先等在跑任务自然结束,超时后才杀残留"
"并标 interrupted;默认 60 秒(systemd"
"默认 90 秒停机时限内能完成)")
parser.add_argument("--sweep-threshold", type=float, default=300,
parser.add_argument("--sweep-threshold", type=float,
default=_STALE_SWEEP,
help="自动陈旧回收阈值(秒):running 任务心跳"
"超过该秒数未更新即被回收;活任务每 30"
"秒刷心跳,不会被误杀")
+3 -1
View File
@@ -67,8 +67,10 @@ import socket
import time
from cloud.db import connect as _db_connect
from cloud.timeouts import WORKER_ALIVE_SECONDS as _WORKER_ALIVE
DEFAULT_ALIVE_SECONDS = 120
# 存活阈值数值收敛到 cloud/timeouts.py(别名保留)。
DEFAULT_ALIVE_SECONDS = _WORKER_ALIVE
# workers 表新增列:(列名, 类型)。pid 用 BIGINT:SQLite 下为
# INTEGER 亲和性,Postgres 下为 8 字节整数,均可容纳进程号;