diff --git a/cloud/approvals.py b/cloud/approvals.py index 5396cab..375f4ee 100644 --- a/cloud/approvals.py +++ b/cloud/approvals.py @@ -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( diff --git a/cloud/auto_memory.py b/cloud/auto_memory.py index 9e4f6cd..c324b01 100644 --- a/cloud/auto_memory.py +++ b/cloud/auto_memory.py @@ -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 diff --git a/cloud/automations.py b/cloud/automations.py index 3320da2..291e582 100644 --- a/cloud/automations.py +++ b/cloud/automations.py @@ -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 diff --git a/cloud/coordinator.py b/cloud/coordinator.py index e3dd37a..964893f 100644 --- a/cloud/coordinator.py +++ b/cloud/coordinator.py @@ -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"} diff --git a/cloud/gates.py b/cloud/gates.py new file mode 100644 index 0000000..840a690 --- /dev/null +++ b/cloud/gates.py @@ -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" diff --git a/cloud/harness2_tasks.py b/cloud/harness2_tasks.py index 1e17f97..1622363 100644 --- a/cloud/harness2_tasks.py +++ b/cloud/harness2_tasks.py @@ -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 diff --git a/cloud/marker_tools.py b/cloud/marker_tools.py index e161085..2d4af28 100644 --- a/cloud/marker_tools.py +++ b/cloud/marker_tools.py @@ -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): diff --git a/cloud/quota.py b/cloud/quota.py new file mode 100644 index 0000000..462fb10 --- /dev/null +++ b/cloud/quota.py @@ -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 diff --git a/cloud/sandbox.py b/cloud/sandbox.py index c522367..1205c8d 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -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>。 diff --git a/cloud/sandbox_policy.py b/cloud/sandbox_policy.py new file mode 100644 index 0000000..39e5c1b --- /dev/null +++ b/cloud/sandbox_policy.py @@ -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 diff --git a/cloud/server.py b/cloud/server.py index 29ed1c5..13cb459 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -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 diff --git a/cloud/tests/test_arch_converge.py b/cloud/tests/test_arch_converge.py new file mode 100644 index 0000000..c4070a0 --- /dev/null +++ b/cloud/tests/test_arch_converge.py @@ -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() diff --git a/cloud/timeouts.py b/cloud/timeouts.py new file mode 100644 index 0000000..860cbb7 --- /dev/null +++ b/cloud/timeouts.py @@ -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 diff --git a/cloud/tree.py b/cloud/tree.py index 981d2a6..54f1394 100644 --- a/cloud/tree.py +++ b/cloud/tree.py @@ -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, diff --git a/cloud/worker.py b/cloud/worker.py index 92a446a..a88c4a5 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -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" "秒刷心跳，不会被误杀") diff --git a/cloud/workers.py b/cloud/workers.py index dc60836..868b5ff 100644 --- a/cloud/workers.py +++ b/cloud/workers.py @@ -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 字节整数，均可容纳进程号；