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 项。
2026-11-04 20:54:50 +00:00
1 parent 17fa493da7
commit facc2a249a
@@ -161,7 +161,13 @@ from cloud.coordinator import parse_parts as _coord_parse_parts
 from cloud.coordinator import parts_by_name as _coord_parts_by_name
 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
@@ -317,17 +323,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):
@@ -2978,21 +2977,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 懒建：全新库上它还不存
@@ -3004,21 +2996,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()
@@ -3107,11 +3086,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 的批量版）。
@@ -6321,7 +6302,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 秒看门狗互补，绝不抢杀）：
@@ -6409,7 +6390,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 已死：按任务派生容器名在本机真杀。
@@ -9311,7 +9297,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 被信号处理置位、
@@ -9427,7 +9413,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)
@@ -9708,7 +9694,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"
                             "秒刷心跳，不会被误杀")