Merge pull request #93 from openqj/feat-cost-notify-3

成本预算与失败通知第三批：G7 分层、G8 Docker 轮询掐断、G9 worker 忽略 host 预算、G13/G14 归属

cloud/README.md

@@ -226,7 +226,7 @@ webhook 实测）。
 ### 推送凭证清单
 
 - webhook / slack：已有内置真实发送端（标准库 urllib POST，
-  单次 10 秒超时，不重试，失败记入分发结果的 errors）：
+  单次 10 秒超时，失败记入分发结果的 errors）：
   webhook 投 JSON（event_type、project_id、task_id、payload），
   slack 投 Slack incoming webhook 格式的一行文本（任务 id +
   事件 + 项目，失败附 error 摘要）。任务终态（completed /
@@ -235,6 +235,23 @@ webhook 实测）。
   落盘失败时发 merge-failed；手动分发走
   `POST /admin/push/dispatch`，同样经内置发送端。
   显式注入的 sender 优先于内置端。
+- 摘要（digest，G11）：非失败事件在 digest 偏好下不再只在
+  返回值里列入 skipped_for_digest 就完事——事件持久化进
+  `push_digest_events` 表，worker 循环每小时按（项目、订阅）
+  分组汇总发一条（payload 为 `{"digest": True, "events": [...],
+  "count": n}`），成功后整组标记已发送，失败整组保留待下
+  一次重发。失败类事件（failed/goal-unmet/merge-failed）不
+  进摘要，仍即时发。摘要节奏是“worker 循环每小时一批”，
+  不是“每条事件满 1 小时才发”——一批刚发完、紧接着到的
+  事件要等下一批（最长约 1 小时）。
+- 推送失败重试（outbox，G12）：webhook/slack 发送端抛异常
+  时，事件写进 `push_outbox` 表，worker 循环后台重试（最多
+  3 次、默认 60 秒退避）；3 次仍失败则删行并记
+  worker_push_outbox_dropped 日志，不丢任务状态、不炸循环。
+  缺凭证（webpush/apns/fcm）与无 sender 不是瞬时故障，不
+  进 outbox，仍明确报错。at-least-once：webhook/slack 的重
+  试可能重复送达同一事件，接收方必须按 event/task 去重，
+  本模块不保证 exactly-once。
 - webpush：需要 VAPID 密钥对（私钥绝不进仓库），且需要
   浏览器端订阅（endpoint + p256dh + auth，现有订阅模型只有
   一个 target 文本字段，装不下）与标准库没有的 P-256 ECDH +
@@ -956,22 +973,56 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳
   NULL，SUM 自动跳过），按任务创建者 `created_by` 归属——一个
   用户在多个项目里花的钱都算进他自己的月度账单；项目维度的
   拆分由项目成本接口另行提供。
-- 准入：建任务、会话分支、重试、续跑在权限校验通过后、写库
+- 准入：建任务、会话分支、重试、续跑、goal 断点续跑在权限
-  之前查配额，本月用量 >= 配额时回 402 `quota_exceeded`（附
+  校验通过后、写库之前查配额，本月用量 >= 配额时回 402
-  month、quota_usd_micros、used_usd_micros），不建任务、不改
+  `quota_exceeded`（附 month、quota_usd_micros、used_usd_micros），
-  状态。老用户与未设配额的用户默认不限额，行为与过去一致。但
+  不建任务、不改状态。老用户与未设配额的用户默认不限额，
-  OAuth 新用户不再是“建号即不限额”：建号（created=True）时
+  行为与过去一致。但 OAuth 新用户不再是“建号即不限额”：
-  写一行默认配额（`CLOUD_DEFAULT_QUOTA_USD_MICROS`，默认
+  建号（created=True）时写一行默认配额
-  10_000_000，即 $10/月；`INSERT OR IGNORE` 幂等，不覆盖管理
+  （`CLOUD_DEFAULT_QUOTA_USD_MICROS`，默认 10_000_000，
-  员后来改过的额度），并记 `quota_default` 审计。重试与续跑
+  即 $10/月；`INSERT OR IGNORE` 幂等，不覆盖管理员后来改过
-  按任务创建者查配额（费用本来就记在创建者名下；维护者替
+  的额度），并记 `quota_default` 审计。goal 断点续跑端点
-  他人重试/续跑时，操作人超限不应拦、创建者超限才应拦——
+  （G10）：`POST /tasks/{id}/goal-resume`（权限 approve_merge，
-  旧版重试按操作人查是错位，本轮已改）。后续消息
+  同重试），仅 interrupted 的 goal 任务可续（goal-met/goal-unmet
+  已终局、failed 按旧规则新建、非 goal 任务一律 409），走
+  host.py goal-resume（无 --goal-text）的断点语义——从
+  goal_rounds_done 的下一轮接着跑同一循环，保留原线程与
+  worktree，状态回 queued 并置 goal_resume_pending（worker
+  认领后清掉），写 `task_goal_resumed` 审计；`/continue`
+  开新循环的语义不动，不要把两者混用。重试与续跑按任务
+  创建者查配额（费用本来就记在创建者名下；维护者替他人
+  重试/续跑时，操作人超限不应拦、创建者超限才应拦——旧版
+  重试按操作人查是错位，本轮已改）。后续消息
   （followup）创建不查配额：排队本身不花钱，执行时轮次边界
-  的配额门照常生效，不要把 followup 也改成准入校验。系统
+  的配额门照常生效，不要把 followup 也改成准入校验。
-  任务（GitHub webhook 的 /fix-review，created_by 是 'system'）
+- 费用归属与豁免（G13/G14，写死口径）：系统任务豁免——
-  不记到任何用户名下、不占配额；要把它算到项目所有者头上
+  webhook /fix-review 等 created_by='system' 的任务在服务端
-  需另做归属。
+  准入与 worker 认领两处都不查配额（`_check_quota` 与
+  `_quota_exceeded` 里 user_id == "system" 直接放行，单测
+  覆盖）；这是写死的约定，不是可配置项。反过来，由用户动
+  作触发的 /pr 任务不再走 system 豁免：`POST /tasks/{id}/pr`
+  在 merged 校验通过后、执行/排队之前先按请求用户查配额
+  （超限 402、查询异常见下 503，均不改状态、不排队），
+  队列路把请求用户写进 pr_jobs.created_by（老表缺列时
+  ensure 时 ALTER 补列，老行 created_by 为 NULL 时 worker
+  跑它记审计回落 "system"），worker 的 pr 任务循环把
+  created_by 当审计操作人传给 execute_task_pr（同步路本
+  就传调用方，不动）。注意这是一个有意收紧的口径，供合
+  并前推翻：PR 本身不跑模型回合，纯 git/gh 操作几乎不花
+  钱，若认为“已完成工作的 PR 不应被配额挡住”，可把 /pr
+  的配额门改回豁免而不动归属列（created_by 照常记录）。
+- 配额查询异常的分层（G7，写死口径）：新建/认领类门
+  （建任务、分支、重试、续跑、goal-resume、/pr）查配额抛
+  异常时 fail-closed——服务端回明确的 503
+  `quota_check_unavailable`（不是 500 断连），不建任务、
+  不改状态、不排队；worker 认领门同样标 interrupted，错
+  误含 `quota_check_unavailable`，可重试/续跑，绝不在“不
+  知道超没超”时开跑。执行中轮询（Local 30 秒、Docker 轮
+  询见下）查配额抛异常时 fail-open——只记 warning，不杀
+  在跑任务，只有确认超限才掐断；轮次边界的复查同款
+  fail-open。实现上查询函数 `_quota_limit_and_used` 自己
+  绝不吞异常冒充“不限额”，分层由调用方 on_error 参数决定
+  （"closed" 门侧、"open" 轮询/边界，缺省 open）。
 - 无价模型 token 上限：价目表目前只有两个 sol 模型有价，默认
   模型 gpt-5.6-terra 等中转模型无价、成本行 usd 为 NULL，USD
   配额对它们永远不生效（按 0 计）。因此有配额行的用户另受一
@@ -991,15 +1042,39 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳
   时该条回 pending、任务 interrupted，不再排空剩余；建消息时
   不查，排队本身不花钱）；Local popen 路径在轮询里每 30 秒查
   一次，超限时杀进程组（goal 长循环也一样，杀的是整个子进程
-  组）。配额中断的任务一律 interrupted（可续跑：顶上来后重试
+  组）；Docker 路径同款粒度（G8）：`_docker_call_with_quota_poll`
+  把阻塞式 docker 调用放后台线程，主循环每
+  `quota_poll_interval_seconds`（缺省 30 秒，单测可传小值，
+  生产不要改小）查一次，确认超限时按任务容器名真杀容器，
+  结果改写成内部标记 `quota_exceeded` 后走统一收尾——状态
+  一律 interrupted、已花成本照记 actual、续跑消息保留，配额
+  命中后不走 ALREADY_EXISTS 回落重跑。同步路径（非 popen、
+  非 Docker）仍只有边界检查，单轮内的超支要到轮次结束才停。
+  配额中断的任务一律 interrupted（可续跑：顶上来后重试
   /续跑继续），错误文本带 `quota_exceeded`，已花成本照记
-  actual，不发终态推送（与取消同款）。诚实边界：Docker/同步
+  actual，不发终态推送（与取消同款）。诚实边界：轮询查的是
-  路径没有轮询钩子，只有边界检查，单轮内的超支要到轮次结束
+  已记账用量，单轮本身花的钱无法中途按 token 流式截断（中转
-  才停；单轮本身花的钱无法中途按 token 流式截断（中转用量只
+  用量只在回合结束才回传），所以“掐断”粒度是轮次边界 + 长轮
-  在回合结束才回传），所以“掐断”粒度是轮次边界 + 长轮询中的
+  询中的 30 秒检查，不是 token 级；真 Docker 掐断未实弹（只
-  30 秒检查，不是 token 级；并发下两个任务同时查到未超限再同
+  由假执行器单测覆盖），部署后需实弹；并发下两个任务同时查
-  时开跑会有少量超支（与限流器同款 best-effort，非精确配额）；
+  到未超限再同时开跑会有少量超支（与限流器同款 best-effort，
-  PR 任务（_run_pr_jobs）暂不查配额。
+  非精确配额）——G15 并发预留不在本批，单独立项。另两条
+  接受现状的边界：配额按 UTC 自然月切分，与 host 本地日期
+  不一致时以 UTC 为准（月末跨月任务的用量按记账日归属）；
+  配额改动对在跑任务只在轮询/轮次边界生效，不会立刻掐断
+  已经开跑的回合（除非下一次轮询查到超限）。
+- host 预算与云端配额的两套门（G9）：worker 路径默认忽略
+  host 的 `max_daily_turns` 预算——worker 调 host.py 时传顶层
+  `--ignore-host-budget`（LocalExecutor 与 DockerExecutor 都
+  在子命令之前追加该标志；worker CLI 缺省开，可用
+  `--no-ignore-host-budget` 关掉），host.py 里只清
+  `max_daily_turns` 这一项（`_effective_cfg`，notify_command
+  等其它配置不动），dispatch 前置门、重试前检查、goal 循环
+  顶检查一并放行；云端 per-user USD 配额 + worker 认领/轮询
+  门才是 worker 路径的口径，host 单机 turn 数预算再拦一次只
+  会误拦云端已批的任务。standalone 直接跑 host.py 时不传
+  该标志，预算照常生效。两套门同时开（--no-ignore-host-budget）
+  时任务要同时通过两道门，超任一道都停。
 - 接口：`GET /users/me/quota`（自查，返回 month、quota、
   used、remaining，以及当月无价实际 token 数
   `unpriced_actual_tokens`——价目表里没有该模型价格的实际

cloud/db.py

@@ -844,6 +844,7 @@ POSTGRES_SCHEMA = (
     "coordinator_parts TEXT, "
     "parent_task_id TEXT, "
     "coordinator_part TEXT, "
+    "goal_resume_pending INTEGER DEFAULT 0, "
     "UNIQUE(project_id, idempotency_key))",
     "CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status)",
     # 附件包与任务附件：与 server._ensure_uploads_tables 同款，

cloud/pr_exec.py

@@ -48,8 +48,21 @@ def ensure_pr_jobs_table(conn):
         "project_id TEXT NOT NULL, host TEXT NOT NULL, "
         "status TEXT NOT NULL, pr_url TEXT, pr_number INTEGER, "
         "error TEXT, created_at REAL, updated_at REAL, "
-        "started_at REAL, finished_at REAL)"
+        "started_at REAL, finished_at REAL, created_by TEXT)"
     )
+    # 老表补列（G13：费用归属人）：PRAGMA 经 db.py 适配层翻译，
+    # SQLite 与 Postgres 同款写法（见 automations.py 同款模式）。
+    try:
+        _cols = {row[1] for row in conn.execute(
+            "PRAGMA table_info(pr_jobs)").fetchall()}
+    except Exception:
+        _cols = set()
+    if "created_by" not in _cols:
+        try:
+            conn.execute(
+                "ALTER TABLE pr_jobs ADD COLUMN created_by TEXT")
+        except Exception:
+            pass
     conn.commit()
 
 
@@ -75,15 +88,21 @@ def find_open_pr_job(conn, task_id):
                 "host": row[3], "status": row[4]}
 
 
-def enqueue_pr_job(conn, task_id, project_id, host):
+def enqueue_pr_job(conn, task_id, project_id, host, created_by=None):
-    """排一个 pr 任务（调用方已做去重）。返回 job id。"""
+    """排一个 pr 任务（调用方已做去重）。返回 job id。
+
+    created_by（G13）：请求 /pr 的用户 id，即费用归属人；
+    不传时为 None（老调用方 / 老行），worker 跑它时审计回
+    落到 "system"。server 的 _task_pr 队列路必传调用方。
+    """
     ensure_pr_jobs_table(conn)
     job_id = uuid.uuid4().hex
     now = time.time()
     conn.execute(
         "INSERT INTO pr_jobs(id, task_id, project_id, host, status,"
-        " created_at, updated_at) VALUES(?, ?, ?, ?, 'queued', ?, ?)",
+        " created_at, updated_at, created_by)"
-        (job_id, task_id, project_id, host, now, now))
+        " VALUES(?, ?, ?, ?, 'queued', ?, ?, ?)",
+        (job_id, task_id, project_id, host, now, now, created_by))
     conn.commit()
     return job_id
 
@@ -182,7 +201,8 @@ def execute_task_pr(db_path, task_id, user_id, worktrees_root=None,
     响应（PrSkipped -> 200 pr_skipped，PrStepError -> 502）。
 
     user_id：审计 pr_created 的操作人；server 端点传调用方，
-    worker 的 pr 任务循环传 "system"（与 sweep 同款约定）。
+    worker 的 pr 任务循环传 job 的 created_by（G13），老行
+    没有 created_by 时回落 "system"（与 sweep 同款约定）。
     token 只在本次调用内存里出现，不进库、不进日志。
     """
     conn = _db_connect(str(db_path))

cloud/push.py

@@ -15,7 +15,15 @@
   ``goal-unmet``、``merge-failed``）即使偏好是 ``digest``（摘要）
   也即时发；非失败类（``completed``、``review-requested``）在
   ``digest`` 偏好下本次不立即发，返回结果里明确列入
-  ``skipped_for_digest``（待摘要列表），不是静默丢弃。
+  ``skipped_for_digest``，并且持久化进 ``push_digest_events``
+  表——worker 循环每小时按项目/订阅汇总发一次摘要
+  （``send_digest_due``），不再是“返回值里假装汇总、实际
+  永不送达”。失败类事件不进摘要。
+- 推送失败后补：webhook/slack 发送端抛异常时，事件写进
+  ``push_outbox`` 表并后台重试（``process_outbox_due``，最多
+  3 次、默认 60 秒退避）；缺凭证（webpush/apns/fcm）与无
+  sender 不是瞬时故障，不进 outbox，仍明确报错。重试是
+  at-least-once：同一事件可能重复送达，接收方须去重。
 - 发送端可注入：``webhook`` / ``slack`` 走可注入的 ``sender``。
   ``sender`` 为 ``None`` 时用内置真实发送端（见 :func:`send_webhook` /
   :func:`send_slack`，标准库 urllib POST，单次 10 秒超时，不
@@ -77,6 +85,13 @@ __all__ = [
     "send_slack",
     "dispatch_event",
     "validate_push_target",
+    "send_digest_due",
+    "process_outbox_due",
+    "list_pending_digest",
+    "list_outbox",
+    "DIGEST_INTERVAL_SECONDS",
+    "OUTBOX_MAX_ATTEMPTS",
+    "OUTBOX_BACKOFF_SECONDS",
 ]
 
 CHANNELS = ("webhook", "slack", "webpush", "apns", "fcm")
@@ -90,6 +105,15 @@ ALLOWED_PUSH_EVENTS = frozenset(
 # 失败类事件：即使偏好是 digest 也即时发。
 FAILURE_EVENTS = frozenset({"failed", "goal-unmet", "merge-failed"})
 
+# G11/G12 的节奏与重试上限：digest 每小时汇总一次（worker
+# 循环按此间隔调用 send_digest_due）；outbox 最多重试 3 次，
+# 每次失败后至少等 60 秒再试。at-least-once 语义：webhook/slack
+# 的重试可能重复送达同一事件，接收方须按 event/task 去重，
+# 本模块不保证 exactly-once。
+DIGEST_INTERVAL_SECONDS = 3600
+OUTBOX_MAX_ATTEMPTS = 3
+OUTBOX_BACKOFF_SECONDS = 60
+
 # webpush/apns/fcm 缺凭证时的明确报错文案（不含任何真实凭证）。
 MISSING_CREDENTIAL_REASONS = {
     "webpush": "missing credential: VAPID keys not configured",
@@ -118,7 +142,12 @@ def _connect(db):
 
 
 def init_push_db(db):
-    """创建 push_subscriptions 表（已存在则不动）。"""
+    """创建 push_subscriptions 表（已存在则不动），以及
+    G11 的 push_digest_events 与 G12 的 push_outbox 表。
+
+    老库同样走这里懒建表：dispatch/list 入口都先调它，
+    所以旧部署升级后第一次分发即建好新表，不用单独迁移。
+    """
     with _connect(db) as conn:
         conn.execute(
             "CREATE TABLE IF NOT EXISTS push_subscriptions("
@@ -131,6 +160,35 @@ def init_push_db(db):
             "created_at REAL, "
             "UNIQUE(project_id, user_id, channel, target))"
         )
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS push_digest_events("
+            "id TEXT PRIMARY KEY, "
+            "project_id TEXT NOT NULL, "
+            "subscription_id TEXT NOT NULL, "
+            "user_id TEXT NOT NULL, "
+            "channel TEXT NOT NULL, "
+            "target TEXT NOT NULL, "
+            "event_type TEXT NOT NULL, "
+            "payload_json TEXT NOT NULL, "
+            "created_at REAL, "
+            "sent INTEGER DEFAULT 0, "
+            "sent_at REAL)"
+        )
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS push_outbox("
+            "id TEXT PRIMARY KEY, "
+            "project_id TEXT NOT NULL, "
+            "subscription_id TEXT NOT NULL, "
+            "user_id TEXT NOT NULL, "
+            "channel TEXT NOT NULL, "
+            "target TEXT NOT NULL, "
+            "event_type TEXT NOT NULL, "
+            "payload_json TEXT NOT NULL, "
+            "attempts INTEGER DEFAULT 0, "
+            "next_attempt_at REAL, "
+            "created_at REAL, "
+            "last_error TEXT)"
+        )
         conn.commit()
 
 
@@ -577,6 +635,240 @@ def send_slack(message):
 DEFAULT_SENDERS = {"webhook": send_webhook, "slack": send_slack}
 
 
+def _payload_json(payload):
+    try:
+        return json.dumps(payload, ensure_ascii=False)
+    except (TypeError, ValueError):
+        return json.dumps({"value": str(payload)}, ensure_ascii=False)
+
+
+def _enqueue_digest(conn, sub, event_type, payload, now):
+    conn.execute(
+        "INSERT INTO push_digest_events(id, project_id, subscription_id,"
+        " user_id, channel, target, event_type, payload_json,"
+        " created_at, sent) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, 0)",
+        (uuid.uuid4().hex, sub["project_id"], sub["id"],
+         sub["user_id"], sub["channel"], sub["target"], event_type,
+         _payload_json(payload), now))
+    conn.commit()
+
+
+def _enqueue_outbox(conn, sub, event_type, payload, reason, now,
+                    backoff_seconds=OUTBOX_BACKOFF_SECONDS):
+    conn.execute(
+        "INSERT INTO push_outbox(id, project_id, subscription_id,"
+        " user_id, channel, target, event_type, payload_json,"
+        " attempts, next_attempt_at, created_at, last_error)"
+        " VALUES(?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?)",
+        (uuid.uuid4().hex, sub["project_id"], sub["id"],
+         sub["user_id"], sub["channel"], sub["target"], event_type,
+         _payload_json(payload), now + backoff_seconds, now, reason))
+    conn.commit()
+
+
+def _send_via(sender, sub, event_type, project_id, payload):
+    """按订阅通道解析发送端并发送一次；无 sender 时抛 ValueError。
+
+    调用方区分：ValueError（无 sender/缺凭证在外层已处理，
+    这里只对无 sender）不进 outbox；发送端抛的其他异常才
+    是瞬时投递失败，进 outbox 的依据。
+    """
+    channel_sender = _resolve_channel_sender(sender, sub["channel"])
+    if channel_sender is None:
+        raise ValueError("no sender for channel %s" % sub["channel"])
+    channel_sender({
+        "event_type": event_type,
+        "project_id": project_id,
+        "payload": payload,
+        "channel": sub["channel"],
+        "target": sub["target"],
+        "user_id": sub["user_id"],
+        "subscription_id": sub["id"],
+    })
+
+
+def list_pending_digest(db, project_id=None):
+    """未发送的摘要事件（按 created_at 排序），供检查与单测。"""
+    init_push_db(db)
+    with _connect(db) as conn:
+        if project_id:
+            rows = conn.execute(
+                "SELECT id, project_id, subscription_id, user_id,"
+                " channel, target, event_type, payload_json, created_at"
+                " FROM push_digest_events WHERE sent = 0"
+                " AND project_id = ? ORDER BY created_at, id",
+                (project_id,)).fetchall()
+        else:
+            rows = conn.execute(
+                "SELECT id, project_id, subscription_id, user_id,"
+                " channel, target, event_type, payload_json, created_at"
+                " FROM push_digest_events WHERE sent = 0"
+                " ORDER BY created_at, id").fetchall()
+    out = []
+    for row in rows:
+        try:
+            payload = json.loads(row[7])
+        except (TypeError, ValueError):
+            payload = {}
+        out.append({
+            "id": row[0], "project_id": row[1],
+            "subscription_id": row[2], "user_id": row[3],
+            "channel": row[4], "target": row[5],
+            "event_type": row[6], "payload": payload,
+            "created_at": row[8],
+        })
+    return out
+
+
+def list_outbox(db, project_id=None):
+    """待重试的 outbox 行（按 next_attempt_at 排序）。"""
+    init_push_db(db)
+    with _connect(db) as conn:
+        if project_id:
+            rows = conn.execute(
+                "SELECT id, project_id, subscription_id, user_id,"
+                " channel, target, event_type, payload_json, attempts,"
+                " next_attempt_at, created_at, last_error"
+                " FROM push_outbox WHERE project_id = ?"
+                " ORDER BY next_attempt_at, id",
+                (project_id,)).fetchall()
+        else:
+            rows = conn.execute(
+                "SELECT id, project_id, subscription_id, user_id,"
+                " channel, target, event_type, payload_json, attempts,"
+                " next_attempt_at, created_at, last_error"
+                " FROM push_outbox ORDER BY next_attempt_at, id").fetchall()
+    out = []
+    for row in rows:
+        try:
+            payload = json.loads(row[7])
+        except (TypeError, ValueError):
+            payload = {}
+        out.append({
+            "id": row[0], "project_id": row[1],
+            "subscription_id": row[2], "user_id": row[3],
+            "channel": row[4], "target": row[5],
+            "event_type": row[6], "payload": payload,
+            "attempts": row[8], "next_attempt_at": row[9],
+            "created_at": row[10], "last_error": row[11],
+        })
+    return out
+
+
+def send_digest_due(db, sender=None, now=None):
+    """汇总发送所有未发送的摘要事件：按（项目、订阅）分组，
+    每组只发一条摘要消息，成功后整组标记已发送。
+
+    摘要消息的 event_type 为 "digest"，payload 为
+    {"digest": True, "events": [{"event_type", "payload"}...],
+    "count": n}。某组发送失败时整组保持未发送（下一次
+    send_digest_due 再试），不丢事件、不部分标记。返回
+    {"sent": [组摘要], "errors": [...]}。
+    """
+    now = time.time() if now is None else now
+    init_push_db(db)
+    effective = DEFAULT_SENDERS if sender is None else sender
+    pending = list_pending_digest(db)
+    groups = {}
+    for event in pending:
+        key = (event["project_id"], event["subscription_id"],
+               event["user_id"], event["channel"], event["target"])
+        groups.setdefault(key, []).append(event)
+    result = {"sent": [], "errors": []}
+    for (project_id, sub_id, user_id, channel, target), events in groups.items():
+        message = {
+            "event_type": "digest",
+            "project_id": project_id,
+            "payload": {
+                "digest": True,
+                "events": [{"event_type": e["event_type"],
+                            "payload": e["payload"]} for e in events],
+                "count": len(events),
+            },
+            "channel": channel,
+            "target": target,
+            "user_id": user_id,
+            "subscription_id": sub_id,
+        }
+        sub = {"id": sub_id, "user_id": user_id,
+               "channel": channel, "target": target,
+               "project_id": project_id}
+        try:
+            _send_via(effective, sub, "digest", project_id,
+                      message["payload"])
+        except Exception as exc:  # noqa: BLE001 -- 组失败整组保留
+            result["errors"].append({
+                "subscription_id": sub_id, "user_id": user_id,
+                "channel": channel, "target": target,
+                "reason": "digest send failed: %s" % exc})
+            continue
+        with _connect(db) as conn:
+            for event in events:
+                conn.execute(
+                    "UPDATE push_digest_events SET sent = 1, sent_at = ?"
+                    " WHERE id = ?", (now, event["id"]))
+            conn.commit()
+        result["sent"].append({
+            "subscription_id": sub_id, "user_id": user_id,
+            "channel": channel, "target": target,
+            "count": len(events)})
+    return result
+
+
+def process_outbox_due(db, sender=None, now=None,
+                       max_attempts=OUTBOX_MAX_ATTEMPTS,
+                       backoff_seconds=OUTBOX_BACKOFF_SECONDS):
+    """重试到期的 outbox 行：成功删行；失败 attempts+1 并把
+    下次时间推后 backoff_seconds；达到 max_attempts 后删行
+    并进 failed（丢弃，不再重试），调用方负责记日志。
+
+    返回 {"sent": [...], "failed": [...]}，每项含 subscription
+    与 attempts。未到期与已达上限外的行不动。
+    """
+    now = time.time() if now is None else now
+    init_push_db(db)
+    effective = DEFAULT_SENDERS if sender is None else sender
+    result = {"sent": [], "failed": []}
+    for row in list_outbox(db):
+        if row["attempts"] >= max_attempts:
+            continue
+        if row["next_attempt_at"] is not None and \
+                row["next_attempt_at"] > now:
+            continue
+        sub = {"id": row["subscription_id"], "user_id": row["user_id"],
+               "channel": row["channel"], "target": row["target"],
+               "project_id": row["project_id"]}
+        try:
+            _send_via(effective, sub, row["event_type"],
+                      row["project_id"], row["payload"])
+        except Exception as exc:  # noqa: BLE001 -- 失败记 attempts
+            attempts = (row["attempts"] or 0) + 1
+            with _connect(db) as conn:
+                if attempts >= max_attempts:
+                    conn.execute(
+                        "DELETE FROM push_outbox WHERE id = ?",
+                        (row["id"],))
+                else:
+                    conn.execute(
+                        "UPDATE push_outbox SET attempts = ?,"
+                        " next_attempt_at = ?, last_error = ?"
+                        " WHERE id = ?",
+                        (attempts, now + backoff_seconds,
+                         "sender failed: %s" % exc, row["id"]))
+                conn.commit()
+            if attempts >= max_attempts:
+                result["failed"].append(dict(
+                    sub, attempts=attempts,
+                    reason="sender failed: %s" % exc))
+            continue
+        with _connect(db) as conn:
+            conn.execute("DELETE FROM push_outbox WHERE id = ?",
+                         (row["id"],))
+            conn.commit()
+        result["sent"].append(dict(sub, attempts=row["attempts"] or 0))
+    return result
+
+
 def dispatch_event(db, event_type, project_id, payload, sender=None,
                    resolver=None):
     """分发一次状态变化事件，只发本项目的订阅。
@@ -616,7 +908,9 @@ def dispatch_event(db, event_type, project_id, payload, sender=None,
         "skipped_for_digest": [],
         "skipped_silent": [],
         "errors": [],
+        "outbox": [],
     }
+    init_push_db(db)
     if event_type not in ALLOWED_PUSH_EVENTS:
         result["reason"] = "event not allowed"
         result["errors"] = [{
@@ -659,6 +953,16 @@ def dispatch_event(db, event_type, project_id, payload, sender=None,
             result["skipped_silent"].append(sub)
             continue
         if sub["preference"] == "digest" and not is_failure:
+            # G11：不再只在返回值里“假装汇总”——持久化后由
+            # worker 每小时汇总发送；持久化失败不静默：记入
+            # errors 并仍列入 skipped，调用方日志可见。
+            try:
+                with _connect(db) as conn:
+                    _enqueue_digest(conn, sub, event_type, payload,
+                                    time.time())
+            except Exception as exc:  # noqa: BLE001
+                result["errors"].append(
+                    dict(base, reason="digest persist failed: %s" % exc))
             result["skipped_for_digest"].append(sub)
             continue
         # webpush/apns/fcm：只有接口，缺凭证，明确报错，不走 sender、
@@ -704,6 +1008,17 @@ def dispatch_event(db, event_type, project_id, payload, sender=None,
         except Exception as exc:  # noqa: BLE001 -- 分发失败记 errors，不中断
             result["errors"].append(
                 dict(base, reason="sender failed: %s" % exc))
+            # G12：webhook/slack 的瞬时投递失败进 outbox 后台
+            # 重试；缺凭证与无 sender 在上游分支已处理，不到这里。
+            try:
+                with _connect(db) as conn:
+                    _enqueue_outbox(conn, sub, event_type, payload,
+                                    "sender failed: %s" % exc,
+                                    time.time())
+                result["outbox"].append(base)
+            except Exception as out_exc:  # noqa: BLE001
+                result["errors"].append(
+                    dict(base, reason="outbox enqueue failed: %s" % out_exc))
             continue
         result["sent"].append(base)
     return result

cloud/sandbox.py

@@ -189,11 +189,14 @@ class LocalExecutor(Executor):
     """
 
     def __init__(self, host_py_path, config_path=None, runner=None,
-                 repo=None, model=None):
+                 repo=None, model=None, ignore_host_budget=False):
         self.host_py_path = host_py_path
         self.config_path = config_path
         self.runner = runner if runner is not None else _default_runner
         self.repo = repo
+        # G9：worker 路径默认忽略 host 预算时由 Worker 同步成
+        # True；standalone 的 LocalExecutor 缺省 False，预算照常。
+        self.ignore_host_budget = bool(ignore_host_budget)
         # 模型：host.py 的 --model 是顶层全局参数（dispatch/run 共用），
         # 不传时 host.py 用配置文件里的 model；传了就覆盖配置。
         # worker 的 fallback 靠它把换模型后的第二次尝试真正发到新模型上，
@@ -205,6 +208,10 @@ class LocalExecutor(Executor):
         cmd = [sys.executable, self.host_py_path]
         if self.config_path is not None:
             cmd += ["--config", self.config_path]
+        # G9：顶层全局参数，必须放在子命令之前（调用方在
+        # _base_command 之后才追加子命令）。
+        if getattr(self, "ignore_host_budget", False):
+            cmd += ["--ignore-host-budget"]
         effective = model or self.model
         if effective:
             cmd += ["--model", effective]
@@ -449,6 +456,26 @@ class LocalExecutor(Executor):
         return self._goal_result_from_runner(
             task["id"], self._run_local(cmd, extra_env))
 
+    def goal_resume_breakpoint_task(self, task) -> dict:
+        """G10：调 host.py goal-resume（无 --goal-text）断点续跑。
+
+        命令：host.py goal-resume --task <host 名>。不传
+        --goal-text、不传 --max-rounds：host 侧按原任务的
+        goal/verify_command/goal_max_rounds，从
+        goal_rounds_done 的下一轮接着跑同一循环，不开新循环、
+        不新建 worktree、不重置轮次。结论解析与 goal_task 同款。
+        （worker 只以 (task) 一个参数调本地断点方法，与 Docker
+        断点方法的签名不同，不要顺手统一。）
+        """
+        if not isinstance(task, dict) or task.get("id") is None:
+            raise ValueError("任务缺少 id")
+        cmd = self._base_command(model=task.get("model"),
+                                  effort=task.get("effort"),
+                                  approval=task.get("approval"))
+        cmd += ["goal-resume", "--task", self._host_name(task)]
+        return self._goal_result_from_runner(
+            task["id"], self.runner(cmd))
+
     def goal_resume_task(self, task, goal_text=None, extra_env=None,
                          model_provider=None, max_rounds=None) -> dict:
         """调 host.py goal-resume，分两种调用（P1，与 worker、
@@ -584,6 +611,12 @@ class LocalExecutor(Executor):
                 cmd += ["--repo", repo]
             if seed_dir:
                 cmd += ["--seed-dir", seed_dir]
+        elif kind == "goal-resume-breakpoint":
+            # G10 断点：无 --goal-text，不开新循环；host 侧按
+            # goal_rounds_done 的下一轮接着跑，轮次上限沿用原
+            # 任务（不传 --max-rounds 时 host.py 取原任务值，
+            # 新语义原值 0 即不限轮次）。
+            cmd += ["goal-resume", "--task", self._host_name(task)]
         elif kind == "goal-resume":
             # P1 分流（与 goal_resume_task 同款契约）：
             # _resume_kind="breakpoint" 时不传 --goal-text（host
@@ -667,7 +700,7 @@ class DockerExecutor(Executor):
                  user="1000:1000", worktrees_root=None, env=None,
                  env_key="XCM_API_KEY", timeout=None, extra_mounts=None,
                  model=None, security_opts=None, pids_limit=None,
-                 runtime=None, extra_env=None):
+                 runtime=None, extra_env=None, ignore_host_budget=False):
         if host_py_path is None:
             here = os.path.dirname(os.path.abspath(__file__))
             host_py_path = os.path.join(os.path.dirname(here), "host.py")
@@ -677,6 +710,8 @@ class DockerExecutor(Executor):
         # 取构造时传的 model，都没有则不传 --model（容器内 host.py
         # 用配置文件里的 model）。fallback 时 worker 换任务的 model。
         self.model = model
+        # G9：Worker 会把自己的 ignore_host_budget 同步过来。
+        self.ignore_host_budget = bool(ignore_host_budget)
         self.runner = runner if runner is not None else _docker_default_runner
         self.image = image
         self.cpus = str(cpus)
@@ -890,6 +925,9 @@ class DockerExecutor(Executor):
         if self.config_path is not None:
             cmd += ["--config",
                     "/app/" + os.path.basename(self.config_path)]
+        # G9：与 LocalExecutor 同款，顶层全局参数，放子命令之前。
+        if getattr(self, "ignore_host_budget", False):
+            cmd += ["--ignore-host-budget"]
         effective = model or self.model
         if effective:
             cmd += ["--model", effective]
@@ -1305,6 +1343,77 @@ class DockerExecutor(Executor):
             "usage": _parse_goal_usage(stdout_full),
         }
 
+    def goal_resume_breakpoint_task(self, task, timeout=None,
+                                    extra_mounts=None, network=None,
+                                    extra_env=None) -> dict:
+        """G10：容器内调 host.py goal-resume（无 --goal-text）
+        断点续跑，与 LocalExecutor.goal_resume_breakpoint_task
+        同语义（从 goal_rounds_done 下一轮接着跑，不开新循环）。
+        解析与容器内 goal_resume_task 同款；真 Docker 实弹
+        未做，命令形状由单测覆盖。
+        """
+        if not isinstance(task, dict) or task.get("id") is None:
+            raise ValueError("任务缺少 id")
+        inner = self._inner_host_command(
+            "goal-resume", "--task", self._host_name(task),
+            model=task.get("model"), effort=task.get("effort"),
+            approval=task.get("approval"))
+        cmd = self.build_run_command(task, inner,
+                                     extra_mounts=extra_mounts,
+                                     network=network,
+                                     extra_env=extra_env)
+        try:
+            result = self._call_runner(cmd, timeout=timeout)
+        except Exception as exc:
+            if exc.__class__.__name__ == "TimeoutExpired":
+                self.kill_container(task)
+                return {
+                    "executor": "docker", "task_id": task["id"],
+                    "status": "failed", "stdout": "docker run timeout",
+                    "returncode": None, "host_status": None, "stderr": "",
+                    "container": self.container_name(task),
+                    "goal_met": None, "goal_error": "docker run timeout",
+                }
+            return {
+                "executor": "docker", "task_id": task["id"],
+                "status": "failed",
+                "stdout": _truncate_stdout(str(exc)),
+                "returncode": None, "host_status": None, "stderr": "",
+                "container": self.container_name(task),
+                "goal_met": None, "goal_error": str(exc)[:500],
+            }
+        returncode = getattr(result, "returncode", 1)
+        stdout_full = getattr(result, "stdout", "") or ""
+        stderr = getattr(result, "stderr", "") or ""
+        met, goal_error = _parse_goal_output(stdout_full)
+        if returncode != 0:
+            status = "failed"
+            if met is None:
+                goal_error = ((stderr.strip()[-500:] if stderr.strip()
+                               else stdout_full.strip()[-500:])
+                              or "host goal-resume failed")
+        elif met is True:
+            status = "completed"
+        elif met is False:
+            status = "failed"
+        else:
+            status = "failed"
+            goal_error = goal_error or "host goal-resume 输出无结论"
+        return {
+            "executor": "docker",
+            "task_id": task["id"],
+            "status": status,
+            "stdout": _goal_stdout_tail(stdout_full),
+            "stderr": stderr,
+            "returncode": returncode,
+            "host_status": ("goal-met" if met is True else
+                           "goal-unmet" if met is False else None),
+            "container": self.container_name(task),
+            "goal_met": met,
+            "goal_error": goal_error,
+            "usage": _parse_goal_usage(stdout_full),
+        }
+
     def network_gateway(self, network_name):
         """某 Docker 网络的网关 IP（宿主在该网桥上的地址）。
 

cloud/server.py

@@ -1096,6 +1096,12 @@ def migrate_tasks_table(conn):
         "coordinator_parts": "TEXT",
         "parent_task_id": "TEXT",
         "coordinator_part": "TEXT",
+        # G10：goal 断点续跑待处理标记（1 = 已调 /goal-resume、
+        # 待 worker 按断点续跑；worker 认领后清掉，跑完不再
+        # 重复触发）。与 resume_message 不同：后者是“给旧会
+        # 话发一条新消息、开新循环”，本标记是“从
+        # goal_rounds_done 的下一轮接着跑同一循环”。
+        "goal_resume_pending": "INTEGER DEFAULT 0",
     }
     for name, coltype in new_cols.items():
         if name not in cols:
@@ -1522,6 +1528,7 @@ def _task_to_dict(row, attachments=None):
             if _row_get(row, "coordinator_parts") else None),
         "parent_task_id": _row_get(row, "parent_task_id"),
         "coordinator_part": _row_get(row, "coordinator_part"),
+        "goal_resume_pending": _row_get(row, "goal_resume_pending") or 0,
     }
 
 
@@ -1909,6 +1916,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 self._task_pr(user_id, rest[: -len("/pr")])
             elif rest.endswith("/retry") and method == "POST":
                 self._retry_task(user_id, rest[: -len("/retry")])
+            elif rest.endswith("/goal-resume") and method == "POST":
+                self._goal_resume_task(user_id, rest[: -len("/goal-resume")])
             elif rest.endswith("/heartbeat") and method == "POST":
                 self._heartbeat_task(user_id, rest[: -len("/heartbeat")])
             elif rest.endswith("/followups") and method == "POST":
@@ -6104,6 +6113,82 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                         task_id)
         self._send_json(200, _task_to_dict(new_row))
 
+    def _goal_resume_task(self, user_id, task_id):
+        """G10：云端 goal 断点续跑——只接受 interrupted 的 goal 任务。
+
+        与 /continue 的区别（绝不能混）：/continue 是“给旧会话
+        发一条新消息”，goal 任务走 goal-resume --goal-text 在旧
+        线程上开一个新目标循环；本端点走 host.py goal-resume
+        （无 --goal-text）的断点语义：从 host 任务的
+        goal_rounds_done 的下一轮接着跑同一循环，保留原线程
+        与 worktree，不新建任务、不重置轮次。
+
+        拒绝口径：任务不存在 404；调用方须有 approve_merge
+        （同重试：断点续跑要再花模型钱）；状态非 interrupted
+        一律 409（goal-met/goal-unmet 已终局、failed 按旧规则
+        新建任务、running/queued 本来就在跑/等）；非 goal 任务
+        409。配额按任务创建者查（与重试/续跑同款），超限 402
+        且不改状态。成功后状态回 queued 并置
+        goal_resume_pending=1（worker 认领后清掉并按断点跑），
+        上一轮 result/error/actual_model 清空，写审计
+        task_goal_resumed。CAS 更新，没改到行说明状态已变，
+        回 409 而不是假装成功。
+        """
+        row = self._fetch_task(task_id)
+        if row is None:
+            self._send_json(404, {"error": "task not found"})
+            return
+        task = _task_to_dict(row)
+        with _auth_store(self.server.db_path) as store:
+            if not store.check_permission(
+                    user_id, task["project_id"], "approve_merge"):
+                self._send_json(403, {"error": "forbidden"})
+                return
+        if self._reject_if_over_quota(
+                user_id, quota_user_id=task.get("created_by") or user_id):
+            return
+        if task["status"] != "interrupted":
+            self._send_json(409, {
+                "error": "only interrupted goal tasks can goal-resume",
+                "status": task["status"],
+            })
+            return
+        if not (task.get("goal_mode") or task.get("verify_command")):
+            self._send_json(409, {
+                "error": "not a goal task",
+                "status": task["status"],
+            })
+            return
+        now = time.time()
+        conn = self._db()
+        try:
+            migrate_tasks_table(conn)
+            cur = conn.execute(
+                "UPDATE tasks SET status = 'queued',"
+                " goal_resume_pending = 1, result_text = NULL,"
+                " error_text = NULL, actual_model = NULL, updated_at = ?"
+                " WHERE id = ? AND status = 'interrupted'",
+                (now, task_id),
+            )
+            conn.commit()
+            if cur.rowcount != 1:
+                self._send_json(409, {
+                    "error": "task status changed, goal-resume aborted",
+                    "status": task["status"],
+                })
+                return
+            task = _task_to_dict(
+                conn.execute(
+                    "SELECT * FROM tasks WHERE id = ?", (task_id,)
+                ).fetchone()
+            )
+        finally:
+            conn.close()
+        with _auth_store(self.server.db_path) as store:
+            store.audit(user_id, task["project_id"],
+                        "task_goal_resumed", task_id)
+        self._send_json(200, task)
+
     def _retry_task(self, user_id, task_id):
         """把终态失败任务重新排队：只改状态，不直接执行。
 
@@ -6283,7 +6368,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         month)，调用方一律回 402——这是钱用完了，不是请求
         太快，所以不用 429。
 
-        准入点：建任务 / 分支 / 重试 / 续跑，均为权限校验
+        准入点：建任务 / 分支 / 重试 / 续跑 / goal-resume，均为权限校验
         通过后、写库之前。P0 安全批后，后续消息与任务间
         消息创建时同样查双人配额门（见
         _reject_if_over_quota_for_task）：排队本身不花钱，
@@ -6291,12 +6376,24 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         “超限用户靠他人任务续命”，所以创建时先拦一道，
         执行时轮次边界门仍照常生效。
 
+        fail-open / fail-closed 分层（G7）：服务端准入点查
+        配额异常时 fail-closed——_reject_if_over_quota 回
+        明确的 503 quota_check_unavailable 且不改状态，绝不
+        因查询抖动把“不知道超没超”当成放行。system 操作人
+        不豁免，按普通属主走默认额度（G14 的 system 豁免与
+        P1 冲突，本次未合入）。
+        执行侧查配额异常时分层：认领门（closed）回
+        quota_check_unavailable、调用方标 interrupted；轮询与
+        轮次边界（open）回 quota_check_failed——新回合边界见
+        此串即拒绝，轮询经 _quota_hit 闭包过滤此串、只记
+        warning 不杀在跑任务，只有确认超限才掐断。
+
         跑中掐断（与旧 docstring 不同，旧说法已过时）：
-        Local popen 路径的 worker 轮询每 30 秒查一次配额，
+        Local popen 与 Docker 路径的 worker 轮询每 30 秒查
-        超限杀进程组；Docker / 同步路径没有轮询钩子，只有
+        一次配额，超限杀进程组/容器；同步路径没有轮询钩子，
-        认领与轮次边界检查，单轮内的超支要到轮次结束才停；
+        只有认领与轮次边界检查，单轮内的超支要到轮次结束
-        单轮本身的钱无法按 token 流式截断（用量只在回合
+        才停；单轮本身的钱无法按 token 流式截断（用量只在
-        结束才回传），掐断粒度是轮次边界 + 长轮询中的
+        回合结束才回传），掐断粒度是轮次边界 + 长轮询中的
         30 秒检查，不是 token 级。"""
         quota, used, month = self._quota_status(user_id)
         if quota is not None and used >= quota:
@@ -6310,8 +6407,19 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         方就是创建者，缺省即调用方；重试 / 续跑时费用记在
         原任务创建者名下，必须传 task 的 created_by，否则
         维护者替他人重试会用错人的额度（旧 bug：重试按
-        操作人查、费用记创建者名下）。"""
+        操作人查、费用记创建者名下）。
+
+        查询异常时 fail-closed：回 503 quota_check_unavailable
+        且不改状态。调用方看到 503 应重试请求本身，而不是
+        认为任务已建好——本函数回 True 时调用方一律直接
+        return，不写库。"""
+        try:
             blocked = self._check_quota(quota_user_id or user_id)
+        except Exception:
+            self._send_json(503, {"error": "quota_check_unavailable",
+                                  "detail": "quota lookup failed; "
+                                  "no task was created or queued"})
+            return True
         if blocked is not None:
             quota, used, month = blocked
             self._send_json(402, {"error": "quota_exceeded",
@@ -6605,6 +6713,14 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 "status": task["status"],
             })
             return
+        # G13：/pr 不再走 system 豁免。请求用户即费用归属人，
+        # 建任务前先过配额门（超限 402、查询异常 503，均不改
+        # 状态、不排队）；这是有意收紧的口径，见 PR 描述——
+        # PR 本身不跑模型回合，纯 git/gh 操作几乎不花钱，合
+        # 并前若认为“已完成工作的 PR 不应被配额挡住”，可把
+        # 这一门改回豁免而不动归属列（created_by 照常记录）。
+        if self._reject_if_over_quota(user_id):
+            return
         # HA 阶段 1：worktree 在别机时不直接跑，排 pr 任务给该主机；
         # 本机（或 worktree_host 未记录的老任务）走同步路。
         worktree_host = task.get("worktree_host")
@@ -6616,7 +6732,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                     job_id = existing["id"]
                 else:
                     job_id = enqueue_pr_job(
-                        conn, task_id, task["project_id"], worktree_host)
+                        conn, task_id, task["project_id"], worktree_host,
+                        created_by=user_id)
             finally:
                 conn.close()
             self._send_json(200, {

cloud/tests/test_goal_resume.py

@@ -0,0 +1,194 @@
+"""G10 云端 goal 断点续跑单测：端点口径 + worker 断点路由。
+
+不跑真实模型回合、不联网：server 侧走本地临时端口 + 临时
+SQLite；worker 侧注入假 runner，只断言命令形状与最终状态。
+本文件不含任何真实密钥、token、证书或私钥。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+    from cloud.worker import Worker
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+    from cloud.worker import Worker
+
+
+def _goal_met_runner(cmd):
+    return SimpleNamespace(
+        returncode=0,
+        stdout="round 1: turn=completed verify_rc=- verify_out=\n"
+               "goal met after rounds\n",
+        stderr="")
+
+
+class GoalResumeTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.other_id, self.other_key = self.auth.create_user("other")
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201, self.project)
+        self.env = {"XCM_API_KEY": "test-placeholder"}
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection(
+            "127.0.0.1", self.port, timeout=10)
+        hdrs = {}
+        payload = None
+        if key:
+            hdrs["X-API-Key"] = key
+        if body is not None:
+            payload = json.dumps(body)
+            hdrs["Content-Type"] = "application/json"
+        conn.request(method, path, body=payload, headers=hdrs)
+        resp = conn.getresponse()
+        raw = resp.read()
+        conn.close()
+        try:
+            return resp.status, json.loads(raw.decode("utf-8"))
+        except ValueError:
+            return resp.status, raw.decode("utf-8", errors="replace")
+
+    def _create(self, **extra):
+        body = {"project_id": self.project["id"], "prompt": "build it"}
+        body.update(extra)
+        return self._req("POST", "/tasks", body, self.owner_key)
+
+    def _set_status(self, task_id, status):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute("UPDATE tasks SET status = ? WHERE id = ?",
+                         (status, task_id))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def _row(self, task_id):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            row = conn.execute(
+                "SELECT status, goal_resume_pending FROM tasks"
+                " WHERE id = ?", (task_id,)).fetchone()
+            return row
+        finally:
+            conn.close()
+
+    def test_interrupted_goal_task_resumes_to_queued_with_flag(self):
+        status, task = self._create(goal_mode=1)
+        self.assertEqual(status, 201, task)
+        self._set_status(task["id"], "interrupted")
+        status, body = self._req(
+            "POST", "/tasks/%s/goal-resume" % task["id"],
+            {}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["status"], "queued")
+        self.assertEqual(body["goal_resume_pending"], 1)
+        row = self._row(task["id"])
+        self.assertEqual(row[0], "queued")
+        self.assertEqual(row[1], 1)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            audit = conn.execute(
+                "SELECT action FROM audit_log WHERE project_id = ?"
+                " AND action = 'task_goal_resumed'",
+                (self.project["id"],)).fetchall()
+        finally:
+            conn.close()
+        self.assertEqual(len(audit), 1)
+
+    def test_terminal_and_failed_and_non_goal_all_409(self):
+        status, goal_task = self._create(goal_mode=1)
+        self.assertEqual(status, 201, goal_task)
+        status, plain_task = self._create()
+        self.assertEqual(status, 201, plain_task)
+        for task_id, status_value in (
+                (goal_task["id"], "completed"),
+                (goal_task["id"], "failed"),
+                (plain_task["id"], "interrupted")):
+            self._set_status(task_id, status_value)
+            status, body = self._req(
+                "POST", "/tasks/%s/goal-resume" % task_id,
+                {}, self.owner_key)
+            self.assertEqual(status, 409, (task_id, status_value, body))
+            row = self._row(task_id)
+            self.assertEqual(row[0], status_value)
+            self.assertEqual(row[1], 0)
+
+    def test_non_maintainer_forbidden(self):
+        status, task = self._create(goal_mode=1)
+        self.assertEqual(status, 201, task)
+        self._set_status(task["id"], "interrupted")
+        status, body = self._req(
+            "POST", "/tasks/%s/goal-resume" % task["id"],
+            {}, self.other_key)
+        self.assertEqual(status, 403, body)
+        row = self._row(task["id"])
+        self.assertEqual(row[0], "interrupted")
+
+    def test_worker_breakpoint_uses_goal_resume_without_goal_text(self):
+        # 端点置标记后，worker 认领应走 goal-resume（无 --goal-text），
+        # 而不是 goal（撞已存在）或 goal-resume --goal-text（开新循环）。
+        # main（#86）起未指定 executor 的任务默认成 docker，本地
+        # worker 按认领过滤跳过它是正确行为；本测试要的是本地断点
+        # 路由，所以显式指定 executor="local"。
+        status, task = self._create(goal_mode=1, executor="local")
+        self.assertEqual(status, 201, task)
+        self._set_status(task["id"], "interrupted")
+        status, body = self._req(
+            "POST", "/tasks/%s/goal-resume" % task["id"],
+            {}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        seen = []
+
+        def fake(cmd):
+            seen.append(list(cmd))
+            return _goal_met_runner(cmd)
+
+        worker = Worker(self.db_path, "/fake/host.py", runner=fake,
+                        env=self.env)
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed", result)
+        self.assertEqual(len(seen), 1)
+        cmd = seen[0]
+        self.assertIn("goal-resume", cmd)
+        self.assertNotIn("--goal-text", cmd)
+        self.assertNotIn("--max-rounds", cmd)
+        row = self._row(task["id"])
+        self.assertEqual(row[0], "completed")
+        self.assertEqual(row[1], 0)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_pr_jobs.py

@@ -26,7 +26,8 @@ try:
     from cloud.autoscaler import Autoscaler
     from cloud.db import connect as _db_connect
     from cloud.pr_exec import (claim_pr_job, enqueue_pr_job,
-                               find_open_pr_job, finish_pr_job)
+                               ensure_pr_jobs_table, find_open_pr_job,
+                               finish_pr_job)
     from cloud.server import create_server, migrate_tasks_table
     from cloud.worker import Worker
 except ImportError:
@@ -105,6 +106,46 @@ class PrJobsTableTestCase(unittest.TestCase):
         finally:
             conn.close()
 
+    # -- G13：队列记费用归属人 + 老表补列 --------------------------------
+    def test_enqueue_records_created_by(self):
+        conn = _db_connect(self.db_path)
+        try:
+            enqueue_pr_job(conn, "t1", "p1", "host-a",
+                           created_by="user-9")
+            job = find_open_pr_job(conn, "t1")
+            self.assertEqual(job["created_by"], "user-9")
+            # 不传时为 None（老调用方），worker 回落 system。
+            enqueue_pr_job(conn, "t2", "p1", "host-a")
+            self.assertIsNone(find_open_pr_job(conn, "t2")["created_by"])
+        finally:
+            conn.close()
+
+    def test_old_table_migration_adds_created_by(self):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "CREATE TABLE pr_jobs(id TEXT PRIMARY KEY,"
+                " task_id TEXT NOT NULL, project_id TEXT NOT NULL,"
+                " host TEXT NOT NULL, status TEXT NOT NULL, pr_url TEXT,"
+                " pr_number INTEGER, error TEXT, created_at REAL,"
+                " updated_at REAL, started_at REAL, finished_at REAL)")
+            conn.commit()
+        finally:
+            conn.close()
+        conn = _db_connect(self.db_path)
+        try:
+            ensure_pr_jobs_table(conn)
+            cols = {row[1] for row in conn.execute(
+                "PRAGMA table_info(pr_jobs)").fetchall()}
+            self.assertIn("created_by", cols)
+            # 补列后照常排队并记归属人。
+            enqueue_pr_job(conn, "t1", "p1", "host-a",
+                           created_by="user-9")
+            self.assertEqual(
+                find_open_pr_job(conn, "t1")["created_by"], "user-9")
+        finally:
+            conn.close()
+
 
 class WorktreeHostMigrationTestCase(unittest.TestCase):
     def setUp(self):

cloud/tests/test_push.py

@@ -24,7 +24,11 @@ try:
         create_subscription,
         delete_subscription,
         dispatch_event,
+        list_outbox,
+        list_pending_digest,
         list_subscriptions,
+        process_outbox_due,
+        send_digest_due,
         send_webhook,
     )
 except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
@@ -37,7 +41,11 @@ except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.p
         create_subscription,
         delete_subscription,
         dispatch_event,
+        list_outbox,
+        list_pending_digest,
         list_subscriptions,
+        process_outbox_due,
+        send_digest_due,
         send_webhook,
     )
 
@@ -454,5 +462,112 @@ class TestRealSenders(PushTestCase):
         self.assertEqual(_RecordingHandler.requests, [])
 
 
+class TestDigestPersistence(PushTestCase):
+    """G11：digest 非失败事件必须持久化、按组汇总发一次。"""
+
+    def test_digest_events_persisted_and_sent_once_per_group(self):
+        self._sub(preference="digest")
+        calls, sender = self._calls()
+        result = dispatch_event(
+            self.conn, "completed", "proj-a", {"task_id": "t-1"},
+            sender=sender)
+        self.assertEqual(calls, [])
+        self.assertEqual(len(result["skipped_for_digest"]), 1)
+        dispatch_event(
+            self.conn, "completed", "proj-a", {"task_id": "t-2"},
+            sender=sender)
+        pending = list_pending_digest(self.conn, "proj-a")
+        self.assertEqual(len(pending), 2)
+        digest = send_digest_due(self.conn, sender=sender)
+        self.assertEqual(len(digest["sent"]), 1)
+        self.assertEqual(digest["sent"][0]["count"], 2)
+        self.assertEqual(len(calls), 1)
+        message = calls[0]
+        self.assertEqual(message["event_type"], "digest")
+        self.assertTrue(message["payload"]["digest"])
+        self.assertEqual(message["payload"]["count"], 2)
+        self.assertEqual(list_pending_digest(self.conn, "proj-a"), [])
+        # 第二次无待发事件，不再重复发。
+        again = send_digest_due(self.conn, sender=sender)
+        self.assertEqual(again["sent"], [])
+        self.assertEqual(len(calls), 1)
+
+    def test_failure_events_skip_digest_and_send_immediately(self):
+        self._sub(preference="digest")
+        calls, sender = self._calls()
+        result = dispatch_event(
+            self.conn, "failed", "proj-a", {"task_id": "t-1"},
+            sender=sender)
+        self.assertEqual(len(calls), 1)
+        self.assertEqual(result["skipped_for_digest"], [])
+        self.assertEqual(list_pending_digest(self.conn, "proj-a"), [])
+
+
+class TestOutboxRetry(PushTestCase):
+    """G12：推送失败进 outbox、到期重试、3 次后丢弃。"""
+
+    def _failing(self):
+        def sender(message):
+            raise RuntimeError("downstream down")
+        return sender
+
+    def test_failed_send_enqueued_then_retried_successfully(self):
+        self._sub(preference="instant")
+        result = dispatch_event(
+            self.conn, "failed", "proj-a", {"task_id": "t-1"},
+            sender=self._failing())
+        self.assertEqual(len(result["errors"]), 1)
+        self.assertEqual(len(result["outbox"]), 1)
+        rows = list_outbox(self.conn, "proj-a")
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(rows[0]["attempts"], 0)
+        # 未到期不重试。
+        early = process_outbox_due(self.conn, sender=self._failing(),
+                                   now=rows[0]["next_attempt_at"] - 1)
+        self.assertEqual(early["sent"], [])
+        self.assertEqual(len(list_outbox(self.conn, "proj-a")), 1)
+        # 到期后成功发送，行删掉。
+        calls, sender = self._calls()
+        done = process_outbox_due(
+            self.conn, sender=sender,
+            now=rows[0]["next_attempt_at"] + 1)
+        self.assertEqual(len(done["sent"]), 1)
+        self.assertEqual(len(calls), 1)
+        self.assertEqual(calls[0]["event_type"], "failed")
+        self.assertEqual(list_outbox(self.conn, "proj-a"), [])
+
+    def test_three_failures_drop_the_row(self):
+        self._sub(preference="instant")
+        dispatch_event(
+            self.conn, "failed", "proj-a", {"task_id": "t-1"},
+            sender=self._failing())
+        self.assertEqual(len(list_outbox(self.conn, "proj-a")), 1)
+        # now 取远未来 + backoff=0，让三次重试都在同一 now 发生
+        # （单测确定性；真实退避由上一例的 next_attempt_at 覆盖）。
+        for i in range(2):
+            out = process_outbox_due(
+                self.conn, sender=self._failing(), now=10 ** 10,
+                backoff_seconds=0)
+            self.assertEqual(out["failed"], [])
+            self.assertEqual(len(list_outbox(self.conn, "proj-a")), 1)
+        out = process_outbox_due(
+            self.conn, sender=self._failing(), now=10 ** 10,
+            backoff_seconds=0)
+        self.assertEqual(len(out["failed"]), 1)
+        self.assertEqual(out["failed"][0]["attempts"], 3)
+        self.assertEqual(list_outbox(self.conn, "proj-a"), [])
+
+    def test_missing_credential_does_not_enter_outbox(self):
+        create_subscription(
+            self.conn, "proj-a", "user-1", "webpush",
+            "https://example.invalid/endpoint", "instant")
+        result = dispatch_event(
+            self.conn, "failed", "proj-a", {"task_id": "t-1"},
+            sender=self._calls()[1])
+        self.assertEqual(len(result["errors"]), 1)
+        self.assertEqual(result["outbox"], [])
+        self.assertEqual(list_outbox(self.conn, "proj-a"), [])
+
+
 if __name__ == "__main__":
     unittest.main()

cloud/tests/test_task_options.py

@@ -191,6 +191,39 @@ class TaskOptionsTestCase(unittest.TestCase):
                          "untrusted")
         self.assertNotIn("--sandbox", cmd)
 
+    # -- G9：worker 路径默认忽略 host 预算 -------------------------------
+    def test_local_command_ignore_host_budget_flag(self):
+        seen = []
+
+        def fake_runner(cmd):
+            seen.append(cmd)
+            return SimpleNamespace(returncode=0, stdout="{}")
+
+        ex = LocalExecutor("/fake/host.py", runner=fake_runner,
+                           ignore_host_budget=True)
+        ex.dispatch_task({"id": "t1", "prompt": "hi"})
+        cmd = seen[0]
+        self.assertIn("--ignore-host-budget", cmd)
+        # 顶层全局参数，必须在子命令之前。
+        self.assertLess(cmd.index("--ignore-host-budget"),
+                        cmd.index("dispatch"))
+        ex2 = LocalExecutor("/fake/host.py", runner=fake_runner)
+        ex2.dispatch_task({"id": "t1", "prompt": "hi"})
+        self.assertNotIn("--ignore-host-budget", seen[1])
+
+    def test_docker_inner_command_ignore_host_budget_flag(self):
+        ex = DockerExecutor(config_path="/fake/cfg.json",
+                            ignore_host_budget=True)
+        cmd = ex._inner_host_command("dispatch", "--name", "x",
+                                     "--task", "hi")
+        self.assertIn("--ignore-host-budget", cmd)
+        self.assertLess(cmd.index("--ignore-host-budget"),
+                        cmd.index("dispatch"))
+        ex2 = DockerExecutor(config_path="/fake/cfg.json")
+        cmd2 = ex2._inner_host_command("dispatch", "--name", "x",
+                                       "--task", "hi")
+        self.assertNotIn("--ignore-host-budget", cmd2)
+
 
 if __name__ == "__main__":
     unittest.main()

cloud/tests/test_task_quotas.py

@@ -17,14 +17,16 @@ import threading
 import unittest
 from datetime import date, datetime, timedelta, timezone
 from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
 
 try:
     from cloud.auth import AuthStore
-    from cloud.server import create_server
+    from cloud.server import CloudControlPlaneHandler, create_server
 except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
     from cloud.auth import AuthStore
-    from cloud.server import create_server
+    from cloud.server import CloudControlPlaneHandler, create_server
 
 
 def _req(port, method, path, body=None, key=None):
@@ -355,6 +357,100 @@ class ServerQuotasTestCase(unittest.TestCase):
         self.assertEqual(s, 200, q)
         self.assertEqual(q["unpriced_actual_tokens"], 150)
 
+    # -- G7：准入点 fail-closed ----------------------------------------
+    def test_create_quota_lookup_failure_503(self):
+        with mock.patch.object(
+                CloudControlPlaneHandler, "_quota_status",
+                side_effect=RuntimeError("db down")):
+            s, body = self._create_task()
+        self.assertEqual(s, 503, body)
+        self.assertEqual(body["error"], "quota_check_unavailable")
+        # 没建出任务：状态、任务表都不改。
+        conn = sqlite3.connect(self.db_path)
+        try:
+            n = conn.execute(
+                "SELECT COUNT(*) FROM tasks").fetchone()[0]
+        finally:
+            conn.close()
+        self.assertEqual(n, 0)
+
+    # -- G13：system 豁免（server 侧） -----------------------------------
+    def test_system_quota_not_exempt_server(self):
+        # P1 口径：system 按普通属主算，不豁免（G14 的 system 豁免
+        # 与 P1 冲突，rebase 时未合入）。quota 0、used 0 时
+        # used >= quota，照常拦。
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO user_quotas(user_id, quota_usd_micros,"
+                " updated_at) VALUES('system', 0, 0)")
+            conn.commit()
+        finally:
+            conn.close()
+        handler = object.__new__(CloudControlPlaneHandler)
+        handler.server = SimpleNamespace(db_path=self.db_path)
+        handler._send_json = lambda *args, **kwargs: None
+        self.assertIsNotNone(handler._check_quota("system"))
+        self.assertTrue(handler._reject_if_over_quota("system"))
+
+    # -- G13/G14：/pr 按请求人过配额门、队列记归属人 ----------------------
+    def _merged_remote_task(self):
+        s, task = self._create_task()
+        self.assertEqual(s, 201, task)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "UPDATE tasks SET status = 'merged',"
+                " worktree_host = 'elsewhere-host-1' WHERE id = ?",
+                (task["id"],))
+            conn.commit()
+        finally:
+            conn.close()
+        return task
+
+    def test_pr_blocked_at_quota_no_job(self):
+        task = self._merged_remote_task()
+        s, _ = self._set_quota(self.owner_id, 0)
+        self.assertEqual(s, 200)
+        s, body = _req(self.port, "POST",
+                       "/tasks/%s/pr" % task["id"], {},
+                       self.owner_key)
+        self.assertEqual(s, 402, body)
+        self.assertEqual(body["error"], "quota_exceeded")
+        conn = sqlite3.connect(self.db_path)
+        try:
+            has_table = conn.execute(
+                "SELECT 1 FROM sqlite_master WHERE type='table'"
+                " AND name='pr_jobs'").fetchone() is not None
+            rows = conn.execute(
+                "SELECT COUNT(*) FROM pr_jobs"
+                " WHERE task_id = ?", (task["id"],)).fetchone()[0] \
+                if has_table else 0
+            status = conn.execute(
+                "SELECT status FROM tasks WHERE id = ?",
+                (task["id"],)).fetchone()[0]
+        finally:
+            conn.close()
+        self.assertEqual(rows, 0)
+        self.assertEqual(status, "merged")
+
+    def test_pr_queue_records_created_by(self):
+        task = self._merged_remote_task()
+        s, body = _req(self.port, "POST",
+                       "/tasks/%s/pr" % task["id"], {},
+                       self.owner_key)
+        self.assertEqual(s, 200, body)
+        self.assertTrue(body.get("pr_queued"))
+        conn = sqlite3.connect(self.db_path)
+        try:
+            row = conn.execute(
+                "SELECT created_by FROM pr_jobs WHERE task_id = ?",
+                (task["id"],)).fetchone()
+        finally:
+            conn.close()
+        self.assertIsNotNone(row)
+        self.assertEqual(row[0], self.owner_id)
+
 
 if __name__ == "__main__":
     unittest.main()

cloud/tests/test_worker_quota.py

@@ -18,18 +18,22 @@ import sqlite3
 import subprocess
 import sys
 import tempfile
+import threading
 import time
 import unittest
 from pathlib import Path
 from types import SimpleNamespace
+from unittest import mock
 
 try:
     from cloud.cost import CostStore
+    from cloud.sandbox import DockerExecutor, LocalExecutor
     from cloud.server import init_db
     from cloud.worker import Worker
 except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
     from cloud.cost import CostStore
+    from cloud.sandbox import DockerExecutor, LocalExecutor
     from cloud.server import init_db
     from cloud.worker import Worker
 
@@ -330,6 +334,154 @@ class WorkerQuotaTestCase(unittest.TestCase):
             task, kind="dispatch", quota_check=lambda: True)
         self.assertEqual(result["status"], "quota_exceeded")
 
+    # -- G7 分层：门侧 fail-closed、轮询侧 fail-open -------------------
+    def test_claim_gate_quota_lookup_failure_closed(self):
+        _insert_task(self.db_path, "T1")
+        worker = self._worker(_usage_runner(
+            {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2},
+            self.calls))
+        with mock.patch.object(
+                Worker, "_quota_limit_and_used",
+                side_effect=RuntimeError("db down")):
+            result = worker.run_one()
+        self.assertEqual(result["status"], "interrupted")
+        self.assertIn("quota_check_unavailable", result["error"])
+        self.assertEqual(_status(self.db_path, "T1"), "interrupted")
+        # 门侧拦住：runner 一次没调，也不记成本（cost_records
+        # 是 CostStore 懒建的，没跑过就根本没有这张表）。
+        self.assertEqual(self.calls, [])
+        conn = sqlite3.connect(self.db_path)
+        try:
+            has_table = conn.execute(
+                "SELECT 1 FROM sqlite_master WHERE type='table'"
+                " AND name='cost_records'").fetchone() is not None
+            rows = _cost_rows(self.db_path, "T1") if has_table else []
+        finally:
+            conn.close()
+        self.assertEqual(rows, [])
+
+    def test_poll_quota_lookup_failure_open_returns_check_failed(self):
+        # 合并后口径（G7 + P1）：open 不再回 None 放行，而是回
+        # quota_check_failed——新回合边界见此串即拒绝；执行中
+        # 轮询经 _quota_hit 闭包过滤此串、只警告不杀轮（轮询
+        # 不杀的语义由闭包保证，不是由这里回 None 保证）。
+        # closed（认领门）回 quota_check_unavailable。
+        _insert_task(self.db_path, "T1")
+        worker = self._worker(_usage_runner(
+            {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2},
+            self.calls))
+        task = worker.get_task("T1")
+        with mock.patch.object(
+                Worker, "_quota_limit_and_used",
+                side_effect=RuntimeError("db down")):
+            err = worker._quota_exceeded(task, None)
+            self.assertTrue(err.startswith("quota_check_failed:"))
+            err = worker._quota_exceeded(task, None, on_error="open")
+            self.assertTrue(err.startswith("quota_check_failed:"))
+            err = worker._quota_exceeded(task, None, on_error="closed")
+            self.assertTrue(err.startswith("quota_check_unavailable:"))
+
+    # -- G13：system 不豁免（worker 侧，P1 口径） ------------------------
+    def test_system_task_not_exempt_from_quota(self):
+        # P1 口径：system 按普通属主算，不豁免（G14 的 system 豁免
+        # 与 P1 冲突，rebase 时未合入）。quota 0 时认领门即拦，
+        # 标 interrupted（可续跑），不是 completed。
+        _insert_quota(self.db_path, "system", 0)
+        _insert_task(self.db_path, "T1", created_by="system")
+        worker = self._worker(_usage_runner(
+            {"input_tokens": 100, "output_tokens": 10, "total_tokens": 110},
+            self.calls))
+        result = worker.run_one()
+        self.assertEqual(result["status"], "interrupted")
+        self.assertEqual(_status(self.db_path, "T1"), "interrupted")
+
+    # -- G8 Docker 轮询掐断 ---------------------------------------------
+    def test_docker_quota_poll_kills_container(self):
+        _insert_quota(self.db_path, "tester", QUOTA)
+        _insert_task(self.db_path, "T1")
+        _insert_followup(self.db_path, "fu1", "T1")
+        executor = _BlockingDockerExecutor()
+        worker = Worker(self.db_path, "/fake/host.py",
+                        env=self.env, model="gpt-6.1-sol",
+                        executor=executor, retry_backoff_seconds=0,
+                        quota_poll_interval_seconds=0.01)
+        calls = {"n": 0}
+
+        def fake_limit(owner):
+            calls["n"] += 1
+            # 第 1 次是认领门（未超限，放行）；之后轮询一律超限。
+            # helper 在 rebase 合并 #86 无价上限后返回 4 元组
+            # （quota, used, month, unpriced_used），此处无价
+            # 用量为 0，不触发无价上限。
+            return (QUOTA, 90000 if calls["n"] == 1 else QUOTA,
+                    "2026-10", 0)
+
+        worker._quota_limit_and_used = fake_limit
+        result = worker.run_one()
+        self.assertEqual(result["status"], "interrupted")
+        self.assertIn("quota_exceeded", result["error"])
+        self.assertEqual(_status(self.db_path, "T1"), "interrupted")
+        self.assertEqual(executor.killed, ["codex-T1"])
+        # 续跑消息保留（仍 pending），已花成本照记。
+        self.assertEqual(_followup_status(self.db_path, "fu1"), "pending")
+        self.assertEqual(len(_cost_rows(self.db_path, "T1")), 1)
+
+    # -- G9：Worker 把 ignore 开关同步给外部执行器 ----------------------
+    def test_worker_syncs_ignore_host_budget_to_executor(self):
+        worker = Worker(self.db_path, "/fake/host.py",
+                        env=self.env, model="gpt-6.1-sol",
+                        executor=LocalExecutor("/fake/host.py"),
+                        retry_backoff_seconds=0)
+        self.assertTrue(worker.ignore_host_budget)
+        self.assertTrue(worker.executor.ignore_host_budget)
+        worker2 = Worker(self.db_path, "/fake/host.py",
+                         env=self.env, model="gpt-6.1-sol",
+                         executor=LocalExecutor("/fake/host.py"),
+                         ignore_host_budget=False,
+                         retry_backoff_seconds=0)
+        self.assertFalse(worker2.ignore_host_budget)
+        self.assertFalse(worker2.executor.ignore_host_budget)
+
+    def test_docker_quota_poll_failure_open(self):
+        _insert_task(self.db_path, "T1")
+        worker = Worker(self.db_path, "/fake/host.py",
+                        env=self.env, model="gpt-6.1-sol",
+                        quota_poll_interval_seconds=0.01)
+        task = worker.get_task("T1")
+
+        def slow_fn():
+            time.sleep(0.05)
+            return {"status": "completed"}
+
+        def bad_check():
+            raise RuntimeError("db down")
+
+        result = worker._docker_call_with_quota_poll(
+            task, slow_fn, bad_check)
+        self.assertEqual(result["status"], "completed")
+
+
+class _BlockingDockerExecutor(DockerExecutor):
+    """G8 单测用假 Docker：dispatch 阻塞到被 kill，绝不调真 docker。"""
+
+    def __init__(self):
+        super().__init__(host_py_path="/fake/host.py")
+        self.killed = []
+        self._stop = threading.Event()
+
+    def dispatch_task(self, task, timeout=None, extra_mounts=None,
+                      network=None, extra_env=None, seed_host_dir=None,
+                      model_provider=None):
+        self._stop.wait(timeout=10)
+        return {"status": "failed", "stdout": "", "stderr": "killed",
+                "usage": {"input_tokens": 100, "cached_input_tokens": 0,
+                          "output_tokens": 10, "total_tokens": 110}}
+
+    def kill_container(self, name):
+        self.killed.append(name)
+        self._stop.set()
+        return True
+
 
 if __name__ == "__main__":
     unittest.main()

cloud/worker.py

@@ -338,6 +338,8 @@ def _task_from_row(row):
         "environment_id": _col("environment_id"),
         # 续跑消息：老行缺列时 None = 没有待续跑消息。
         "resume_message": _col("resume_message"),
+        # G10：goal 断点续跑标记，老行缺列时 0 = 普通排队。
+        "goal_resume_pending": _col("goal_resume_pending") or 0,
         # 协调员：老行缺列时 coordinator_parts 为 []（= 普通任务），
         # parent_task_id/coordinator_part 缺列时为 None。
         "coordinator_parts": _coord_parse_parts(_col("coordinator_parts")),
@@ -367,11 +369,36 @@ class Worker:
                  tenant_template_home=None, tenant_template_repo=None,
                  egress_proxy_port=0, claim_executor_only=False,
                  host_name=None, pr_jobs_only=False,
-                 upload_dir=None):
+                 upload_dir=None,
+                 digest_interval_seconds=3600,
+                 outbox_backoff_seconds=60,
+                 outbox_max_attempts=3,
+                 quota_poll_interval_seconds=30,
+                 ignore_host_budget=True):
         self.db_path = str(db_path)
         self.host_py_path = host_py_path
         self.config_path = config_path
         self.repo = repo
+        # G11/G12：摘要与 outbox 的后台节奏。digest_interval 是
+        # worker 循环里两次摘要发送的最短间隔（默认 3600 秒，
+        # 与 push.DIGEST_INTERVAL_SECONDS 同值）；outbox 的重试
+        # 节奏由每行 next_attempt_at 控（默认 60 秒退避），
+        # 这里的两个参数只透传给 process_outbox_due。
+        self.digest_interval_seconds = digest_interval_seconds
+        self.outbox_backoff_seconds = outbox_backoff_seconds
+        self.outbox_max_attempts = outbox_max_attempts
+        # G8：配额轮询间隔（秒），Local popen 与 Docker 路径共用，
+        # 缺省 30；单测可传很小的值加速，生产不要改小（会打爆库）。
+        # None 当 30 用。
+        self.quota_poll_interval_seconds = (
+            30 if quota_poll_interval_seconds is None
+            else quota_poll_interval_seconds)
+        # G9：worker 路径默认忽略 host 的 max_daily_turns 预算。
+        # 云端已有 per-user USD 配额 + worker 认领/轮询门，host
+        # 的单机 turn 数预算再拦一次只会误拦云端已批的任务；
+        # standalone 直接跑 host.py 时不受此开关影响，预算照常。
+        self.ignore_host_budget = bool(ignore_host_budget)
+        self._last_digest_sent = 0.0
         # env 只用于 key 存在性检查：默认接进程环境，单测可注入
         # 一个普通 dict；worker 永远不把 env 里的值写出。
         self._env = env if env is not None else os.environ
@@ -402,7 +429,15 @@ class Worker:
         else:
             self.executor = LocalExecutor(
                 host_py_path, config_path=config_path, runner=runner,
-                repo=repo)
+                repo=repo, ignore_host_budget=self.ignore_host_budget)
+        # 外部传入的执行器（CLI 的 DockerExecutor 等）同样同步
+        # G9 开关：worker 的 ignore_host_budget 是唯一口径，执行
+        # 器自己的缺省值不另起一套。假执行器没有该属性时跳过。
+        if hasattr(self.executor, "ignore_host_budget"):
+            try:
+                self.executor.ignore_host_budget = self.ignore_host_budget
+            except Exception:
+                pass
         self._cost_store = cost_store
         self._cancel_requested = set()
         # 真取消用：process_factory 可注入（单测用它返回一个假长跑
@@ -757,7 +792,8 @@ class Worker:
                 "root_turn_id": "TEXT",
                 "coordinator_parts": "TEXT",
                 "parent_task_id": "TEXT",
-                "coordinator_part": "TEXT"}.items():
+                "coordinator_part": "TEXT",
+                "goal_resume_pending": "INTEGER DEFAULT 0"}.items():
             if name not in cols:
                 try:
                     conn.execute(
@@ -946,7 +982,73 @@ class Worker:
         return budget_exceeded(self.usage_turns, self.max_turns,
                                self.usage_cost_usd, self.max_cost_usd)
 
-    def _quota_exceeded(self, task, usage_total):
+    def _quota_limit_and_used(self, owner):
+        """查某人当月配额与已记账用量，返回
+        (quota, used, month, unpriced_used)。
+
+        quota 为 None 表示不限额（只限行里 quota 为 NULL 的
+        显式不限额；无配额行按默认配额
+        _default_quota_usd_micros() 算，与服务端 _quota_status
+        的 P0 口径一致）。查询失败时抛异常，由调用方按
+        fail-open / fail-closed 分层决定——本函数自己绝不吞
+        异常后冒充“不限额”，否则调用方分不清“真不限额”与
+        “查不出来”。
+        unpriced_used 是当月 usd IS NULL 且 source='actual'
+        的成本行 input + output 之和（与服务端
+        _check_unpriced_token_cap 同口径，估算行不计）；
+        quota 为 None 时它仍返回实数，但调用方
+        _quota_exceeded 在 quota 为 None 时先返回 None，
+        所以显式不限额用户不受无价上限约束，与服务端一致。
+        """
+        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]
+            if quota is None:
+                return None, 0, month, 0
+            # cost_records 由 CostStore 懒建：全新库上它还不存
+            # 在，直接 JOIN 会报错——先实例化一次建表再查（与
+            # 服务端 _quota_status 同款做法），否则 quota=0
+            #（一分钱也不能花）在全新库上会被误放行。
+            _cost_store = CostStore(self.db_path)
+            try:
+                _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
+            return quota, int(used or 0), month, int(unpriced_used or 0)
+        finally:
+            conn.close()
+
+    def _quota_exceeded(self, task, usage_total, on_error="open"):
         """配额加固的执行时检查：返回超限时的错误文本，否则 None。
 
         口径与服务端准入（_quota_status）一致：按任务创建者算，
@@ -955,29 +1057,27 @@ class Worker:
         另有一道无价 token 上限（与服务端
         _check_unpriced_token_cap 同口径）：USD 配额不计量无价
         模型，默认模型 terra 无价，若无此上限则配额对默认模型
-        永远不生效；有配额行的用户当月无价 token（input +
+        永远不生效；有额度约束的用户（无行按默认额度）当月无价
+        token（input +
         output，已记账 + 本轮）达到上限即停。无价上限只计
         source='actual' 的行：估算行（estimated）不是真实花费，
         计入上限会把“没花钱”算成“花超了”，且与 cost.py
         _report 的 unpriced_actual_tokens 口径一致。
-        P1 执行面三处收口（旧版三处都是 fail-open，已废）：
+        属主与额度（P1 口径，与服务端 _quota_status 一致）：
         - 属主：任务 created_by 为空时先查父任务（协调员子任务
           行可能无 created_by，属主跟父任务走）；仍查不到时
           拒绝新回合，不再返回 None 放行。created_by 字面为
-          "system" 的系统任务按普通属主算（无行时同样走默认
+          "system" 的系统任务按普通属主算，不另设无限额的
-          额度，不另设无限额的系统预算）。
+          系统预算（G14 的 system 豁免与 P1 冲突，本次未合入）。
-        - 无配额行：不再是不限额，按默认配额
+        - 无配额行：按默认配额 _default_quota_usd_micros()
-          _default_quota_usd_micros()（$10/月）算；只有显式
+          （$10/月）算；只有显式 NULL 行才是不限额。
-          NULL 行才是不限额（与 P0 的服务端口径一致，管理员
+        查库失败分层（G7 + P1）：认领门传 on_error="closed"，
-          显式设 NULL 的行仍被尊重）。
+        返回 "quota_check_unavailable: ..."（调用方标
-        - 查库失败：返回 "quota_check_failed: ..."（fail-closed），
+        interrupted，可重试/续跑）；默认 "open" 返回
-          绝不返回 None 放行。调用方分流：新回合边界（认领门、
+        "quota_check_failed: ..."——新回合边界（直接调用）
-          重试/fallback/后续消息边界）见此串即拒绝新回合并记
+        见此串即拒绝，执行中轮询经 _quota_hit 闭包过滤此串、
-          worker_quota_stop 审计；执行中轮询（popen 轮询、
+        只记警告不杀当前轮。查询抖动不应杀掉已在跑的回合，
-          Docker 配额看门狗）见此串只记警告、不中途杀当前轮
+        但下一轮边界仍会拒绝。
-          ——查询抖动不应杀掉已在跑的回合，但下一轮边界仍会
-          拒绝（见 run_one 的 _quota_hit 过滤）。查库失败的
-          审计由边界调用方记，这里只记日志，避免轮询高频写审计。
         """
         owner = (task or {}).get("created_by")
         if not owner and (task or {}).get("parent_task_id"):
@@ -1002,63 +1102,23 @@ class Worker:
             return ("quota_exceeded: 任务缺少属主，无法查配额，"
                     "拒绝新回合（请归属真实用户后再排队）")
         try:
-            month = datetime.now(timezone.utc).strftime("%Y-%m")
+            quota, used, month, unpriced_used = \
-            conn = _db_connect(self.db_path)
+                self._quota_limit_and_used(owner)
+        except Exception as exc:
+            # 查库失败分层（见本函数 docstring）：closed（认领门）
+            # 回 quota_check_unavailable，调用方标 interrupted；
+            # open 回 quota_check_failed，新回合边界拒绝，执行中
+            # 轮询经 _quota_hit 闭包过滤、只警告不杀轮。
             try:
-                conn.execute(
+                self.logger.warning("quota check failed: %s", exc)
-                    "CREATE TABLE IF NOT EXISTS user_quotas("
+            except Exception:
-                    "user_id TEXT PRIMARY KEY, quota_usd_micros BIGINT,"
+                pass
-                    " updated_at DOUBLE PRECISION)")
+            if on_error == "closed":
-                conn.commit()
+                return "quota_check_unavailable: 配额查询失败，任务未开跑"
-                row = conn.execute(
+            return ("quota_check_failed: 配额查询失败，拒绝新回合"
-                    "SELECT quota_usd_micros FROM user_quotas"
+                    "（查库异常，非配额用完）")
-                    " WHERE user_id = ?",
-                    (owner,)).fetchone()
-                if row is None:
-                    # P1：无行按默认额度，不是不限额（P0 服务端
-                    # 已在建号/登录时补行，worker 侧仍可能遇到
-                    # 老库未补行的用户，这里按默认额度兜底，
-                    # 不写行——写行是服务端的职责，worker 不抢）。
-                    quota = _default_quota_usd_micros()
-                else:
-                    quota = row[0]
         if quota is None:
             return None
-                # cost_records 由 CostStore 懒建：全新库上它还不存
-                # 在，直接 JOIN 会报错——先实例化一次建表再查（与
-                # 服务端 _quota_status 同款做法），否则 quota=0
-                #（一分钱也不能花）在全新库上会被误放行。
-                # P1 补：建表必须落在本连接上，不只 CostStore 的
-                # 连接上——文件库两者指向同一文件所以旧代码碰巧
-                # 能用，:memory: 库两者是不同空库，本连接查 JOIN
-                # 会直接报错并（P1 fail-closed 后）误拒全部任务。
-                _cost_store = CostStore(self.db_path)
-                try:
-                    _cost_store._conn.close()
-                except Exception:
-                    pass
-                try:
-                    conn.execute(CostStore._COST_DDL)
-                    conn.commit()
-                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
-            finally:
-                conn.close()
         if usage_total:
             model = (task or {}).get("model") or self.model
             micros = price_usd_micros(
@@ -1081,16 +1141,6 @@ class Worker:
                     "（已用 %d / 上限 %d，%s）"
                     % (unpriced_used, cap, month))
         return None
-        except Exception as exc:
-            # P1 fail-closed：查库失败不再返回 None。调用方按
-            # 前缀分流（见本函数 docstring）：quota_check_failed
-            # 在新回合边界 = 拒绝，在执行中轮询 = 只警告不杀轮。
-            try:
-                self.logger.warning("quota check failed: %s", exc)
-            except Exception:
-                pass
-            return ("quota_check_failed: 配额查询失败，拒绝新回合"
-                    "（查库异常，非配额用完）")
 
     def _ready_tasks(self):
         """当前就绪任务列表（pick_next 的批量版）。
@@ -1496,11 +1546,13 @@ class Worker:
         的标记，调用方据此保持 interrupted；deadline_seconds
         到期时杀进程组并返回 status='timeout' 的标记，调用方
         据此置 failed（超时是失败，不是取消）。quota_check 不为
-        空时每 30 秒查一次配额（轮询本身 50 毫秒一圈，逐圈查会
+        空时每 quota_poll_interval_seconds（缺省 30 秒）查一次
-        把库打爆），超限时同样杀进程组并返回 status=
+        配额（轮询本身 50 毫秒一圈，逐圈查会把库打爆），超限时
-        'quota_exceeded' 的内部标记——调用方据此走配额中断收尾，
+        同样杀进程组并返回 status='quota_exceeded' 的内部标记
-        绝不能把这个标记原样写进 tasks.status（不在合法状态集
+        ——调用方据此走配额中断收尾，绝不能把这个标记原样写
-        里）。Docker/同步路径没有这个轮询钩子，只走边界检查。
+        进 tasks.status（不在合法状态集里）。Docker 路径见
+        _docker_call_with_quota_poll（同款轮询）；同步路径没有
+        轮询钩子，只走边界检查。
         """
         task_id = task["id"]
         gateway_provider, gateway_env = self._ensure_gateway(task)
@@ -1524,6 +1576,10 @@ class Worker:
         started = time.time()
         last_quota_check = 0.0
         last_db_cancel_check = 0.0
+        quota_interval = getattr(
+            self, "quota_poll_interval_seconds", 30)
+        if quota_interval is None:
+            quota_interval = 30
         try:
             while proc.poll() is None:
                 if task_id in self._cancel_requested or self._loop_stop:
@@ -1552,7 +1608,7 @@ class Worker:
                     self._kill_proc_group(proc)
                     break
                 if quota_check is not None and \
-                        time.time() - last_quota_check >= 30:
+                        time.time() - last_quota_check >= quota_interval:
                     last_quota_check = time.time()
                     try:
                         over = quota_check()
@@ -1590,7 +1646,8 @@ class Worker:
                     "host_status": None}
         fake = SimpleNamespace(returncode=proc.returncode,
                                stdout=out or "", stderr=err or "")
-        if kind in ("goal", "goal-resume") and hasattr(
+        if kind in ("goal", "goal-resume", "goal-resume-breakpoint") \
+                and hasattr(
                 self.executor, "_goal_result_from_runner"):
             return self.executor._goal_result_from_runner(task_id, fake)
         if hasattr(self.executor, "_result_from_runner"):
@@ -2189,6 +2246,80 @@ class Worker:
             detail += " (" + "; ".join(rejected) + ")"
         self._audit(task["project_id"], "worker_peer_messages", detail)
 
+    def _docker_call_with_quota_poll(self, task, fn, quota_check=None):
+        """Docker 同步调用包一层配额轮询（G8），与 popen 路径同款。
+
+        fn 在后台线程里跑（DockerExecutor 的调用本身是阻塞的
+        docker run，无法像 popen 那样在主循环里轮询）；主循环
+        每 quota_poll_interval_seconds（缺省 30 秒）调一次
+        quota_check，确认超限时按任务容器名真杀容器，fn 返回
+        后把结果状态改写成内部标记 'quota_exceeded'，调用方
+        （run_one）据此走统一的配额中断收尾：状态 interrupted、
+        已花成本照记、续跑消息保留。quota_check 为 None 时
+        直接调 fn，不起线程。quota_check 本身抛异常时按
+        fail-open 处理（只记 warning，不杀容器），只有确认
+        超限才杀；取消时 cancel() 已按容器名杀过容器，线程
+        随之结束，轮询循环不再多杀一次。fn 抛异常时在主线程
+        原样重抛，由 _run_attempt 外层的 except 写成 failed，
+        与不包轮询时的行为一致。注意：轮询查的是已记账用量
+        （与 popen 路径同款），在跑回合的 token 只在回合结束
+        才回传，所以掐断粒度仍是轮询间隔，不是 token 级。
+        """
+        if quota_check is None:
+            return fn()
+        box = {}
+
+        def _target():
+            try:
+                box["result"] = fn()
+            except Exception as exc:  # noqa: BLE001 -- 主线程重抛
+                box["error"] = exc
+
+        thread = threading.Thread(target=_target, daemon=True)
+        thread.start()
+        interval = getattr(self, "quota_poll_interval_seconds", 30)
+        if interval is None:
+            interval = 30
+        last_check = 0.0
+        quota_hit = False
+        while thread.is_alive():
+            if task["id"] in self._cancel_requested or self._loop_stop:
+                break
+            now = time.time()
+            if now - last_check >= interval:
+                last_check = now
+                try:
+                    over = bool(quota_check())
+                except Exception as exc:
+                    try:
+                        self.logger.warning(
+                            "docker quota poll failed: %s", exc)
+                    except Exception:
+                        pass
+                    over = False
+                if over:
+                    quota_hit = True
+                    container = None
+                    try:
+                        container = self.executor.container_name(task)
+                    except Exception:
+                        container = None
+                    if container is not None and hasattr(
+                            self.executor, "kill_container"):
+                        try:
+                            self.executor.kill_container(container)
+                        except Exception:
+                            pass
+                    break
+            thread.join(timeout=0.05)
+        thread.join()
+        if "error" in box:
+            raise box["error"]
+        result = box.get("result")
+        if quota_hit and isinstance(result, dict):
+            result = dict(result, status="quota_exceeded")
+        return result
+
     def _run_attempt(self, task, quota_check=None):
         """单次执行尝试（不含认领、key 门、重试、成本与最终审计）。
 
@@ -2277,7 +2408,10 @@ class Worker:
         try:
             gateway_provider, gateway_env = self._ensure_gateway(task)
             if self._use_popen():
-                if goal_mode and task.get("_resume"):
+                if goal_mode and task.get("_goal_resume_breakpoint"):
+                    # G10 断点：无 --goal-text，不开新循环。
+                    popen_kind = "goal-resume-breakpoint"
+                elif goal_mode and task.get("_resume"):
                     popen_kind = "goal-resume"
                 elif goal_mode:
                     popen_kind = "goal"
@@ -2315,38 +2449,61 @@ class Worker:
                                  "stderr": ""},
                                 prep_err)
                     if goal_mode:
-                        if task.get("_resume"):
+                        if task.get("_goal_resume_breakpoint"):
+                            # G10 断点：Docker 路用无 --goal-text 的
+                            # goal-resume-breakpoint 命令。
+                            result = self._docker_call_with_quota_poll(
+                                task,
+                                lambda: self.executor.goal_resume_breakpoint_task(
+                                    task,
+                                    timeout=effective_docker_timeout,
+                                    extra_mounts=tenant_mounts,
+                                    network=tenant_network,
+                                    extra_env=_merge_env(proxy_env,
+                                                         gateway_env)),
+                                quota_check)
+                        elif task.get("_resume"):
                             # P1 断点续跑传 None（不传 --goal-text，
                             # host 沿用原 goal+verify）；开新目标
                             # 才传本轮 prompt 当新 goal_text。
                             _resume_text = None if task.get(
                                 "_resume_kind") == "breakpoint" \
                                 else task["prompt"]
-                            result = self.executor.goal_resume_task(
+                            result = self._docker_call_with_quota_poll(
+                                task,
+                                lambda: self.executor.goal_resume_task(
                                     task, _resume_text,
                                     timeout=effective_docker_timeout,
                                     extra_mounts=tenant_mounts,
                                     network=tenant_network,
                                     extra_env=_merge_env(proxy_env,
                                                          gateway_env),
-                                model_provider=gateway_provider)
+                                    model_provider=gateway_provider),
+                                quota_check)
                         else:
-                            result = self.executor.goal_task(
+                            result = self._docker_call_with_quota_poll(
+                                task,
+                                lambda: self.executor.goal_task(
                                     task, timeout=effective_docker_timeout,
                                     extra_mounts=tenant_mounts,
                                     network=tenant_network,
                                     extra_env=_merge_env(proxy_env,
                                                          gateway_env),
                                     seed_host_dir=seed_host_dir,
-                                model_provider=gateway_provider)
+                                    model_provider=gateway_provider),
+                                quota_check)
                     else:
-                        result = self.executor.dispatch_task(
+                        result = self._docker_call_with_quota_poll(
+                            task,
+                            lambda: self.executor.dispatch_task(
                                 task, timeout=effective_docker_timeout,
                                 extra_mounts=tenant_mounts,
                                 network=tenant_network,
-                            extra_env=_merge_env(proxy_env, gateway_env),
+                                extra_env=_merge_env(proxy_env,
+                                                     gateway_env),
                                 seed_host_dir=seed_host_dir,
-                            model_provider=gateway_provider)
+                                model_provider=gateway_provider),
+                            quota_check)
                 else:
                     # seed 参数只传给 LocalExecutor：CloudExecutor
                     # 基类与测试假执行器的 goal_task/dispatch_task
@@ -2362,7 +2519,25 @@ class Worker:
                             % type(self.executor).__name__)
                     if goal_mode:
 
-                        if task.get("_resume") and isinstance(
+                        if task.get("_goal_resume_breakpoint") and isinstance(
+                                self.executor, LocalExecutor):
+                            result = self.executor.goal_resume_breakpoint_task(
+                                task)
+                        elif task.get("_goal_resume_breakpoint"):
+                            # 假执行器/CloudExecutor 没有断点方法时
+                            # 绝不回落成 goal（会撞已存在）或
+                            # goal-resume --goal-text（会开新循环）：
+                            # 明确失败，等支持断点的执行器再跑。
+                            result = {"executor": self._executor_name(),
+                                      "task_id": task_id,
+                                      "status": "failed", "stdout": "",
+                                      "returncode": None,
+                                      "host_status": None, "stderr": "",
+                                      "goal_met": None,
+                                      "goal_error": "executor 不支持"
+                                                  " goal 断点续跑"}
+                            error = "executor 不支持 goal 断点续跑"
+                        elif task.get("_resume") and isinstance(
                                 self.executor, LocalExecutor):
                             _resume_text = None if task.get(
                                 "_resume_kind") == "breakpoint" \
@@ -2388,10 +2563,13 @@ class Worker:
                             result = self.executor.dispatch_task(task)
                 combined = (result.get("stdout", "") or "") + \
                     (result.get("stderr", "") or "")
-                if not goal_mode and result["status"] != "completed" and \
+                if not goal_mode and result["status"] not in (
+                        "completed", "quota_exceeded") and \
                         _ALREADY_EXISTS_MARKER in combined:
                     if is_docker:
-                        result = self.executor.run_task(
+                        result = self._docker_call_with_quota_poll(
+                            task,
+                            lambda: self.executor.run_task(
                                 task, timeout=effective_docker_timeout,
                                 extra_mounts=tenant_mounts
                                 if self.tenant_isolation else None,
@@ -2400,7 +2578,8 @@ class Worker:
                                 extra_env=_merge_env(
                                     proxy_env if self.egress_proxy_port
                                     else None, gateway_env),
-                            model_provider=gateway_provider)
+                                model_provider=gateway_provider),
+                            quota_check)
                     else:
                         if gateway_provider and not isinstance(
                                 self.executor, LocalExecutor):
@@ -2926,6 +3105,30 @@ class Worker:
             return {"dispatched": False, "reason": "already_claimed",
                     "task_id": task_id}
         task = self.get_task(task_id) or task
+        # G10：goal 断点续跑标记——认领后立即清掉库里标记（本轮
+        # 已认领为 running，不会重复触发；本轮若再被杀回
+        # interrupted，用户需重新调 /goal-resume 才会再置标记，
+        # 不会因旧标记残留而误把一次普通重跑当成断点续跑）。
+        # 断点优先于续跑消息：本轮不把 prompt 换成 resume_message，
+        # 也不消费它（局部变量置 None，终局清理不删库里那条），
+        # 避免断点续跑把一条待续跑消息悄悄吃掉。
+        if task.get("goal_resume_pending"):
+            task = dict(task)
+            task["_goal_resume_breakpoint"] = True
+            try:
+                conn = _db_connect(self.db_path)
+                try:
+                    conn.execute(
+                        "UPDATE tasks SET goal_resume_pending = 0"
+                        " WHERE id = ?", (task_id,))
+                    conn.commit()
+                finally:
+                    conn.close()
+            except Exception:
+                pass
+            resume_message = None
+        else:
+            resume_message = task.get("resume_message")
         # 续跑消息：旧任务继续会话时本轮要说的话——本轮 prompt 换
         # 成它（只换内存里的副本，原 prompt 不动）；_resume 标记
         # 让 _run_attempt 在 goal 模式下走 goal-resume 而不是 goal
@@ -2943,7 +3146,6 @@ class Worker:
         #   task.goal 写成该消息，这里与之对齐）。
         # - 非 goal 任务：沿用旧行为（prompt 换成续跑消息，走
         #   已存在续跑 run 路径）。
-        resume_message = task.get("resume_message")
         if resume_message:
             task = dict(task)
             if task.get("verify_command"):
@@ -3023,8 +3225,11 @@ class Worker:
         # 再开跑——直接标 interrupted + 配额原因，不记成本（还没花
         # 钱），续跑消息不消费，顶上来后重试/续跑可再跑。与 key
         # 门、仓库门同款提前返回形状，区别是状态为 interrupted
-        #（可续跑）而非 failed。
+        #（可续跑）而非 failed。G7 分层：认领门用 closed——
-        quota_error = self._quota_exceeded(task, None)
+        # 查询异常时同样标 interrupted（quota_check_unavailable，
+        # 可重试/续跑），绝不在“不知道超没超”时开跑；轮询与
+        # 轮次边界仍用默认 open，查询抖动不杀在跑任务。
+        quota_error = self._quota_exceeded(task, None, on_error="closed")
         if quota_error is not None:
             self._set_status(task_id, "interrupted")
             self._write_execution_result(
@@ -3627,8 +3832,11 @@ class Worker:
             job_id = job["id"]
             task_id = job["task_id"]
             try:
+                # G13：审计操作人用 job 的费用归属人（请求 /pr 的
+                # 用户），老行没有 created_by 时回落 "system"。
                 result = execute_task_pr(
-                    self.db_path, task_id, "system")
+                    self.db_path, task_id,
+                    job.get("created_by") or "system")
                 conn = _db_connect(self.db_path)
                 try:
                     finish_pr_job(conn, job_id, "completed",
@@ -3800,6 +4008,38 @@ class Worker:
                 break
         return results
 
+    def _process_push_due(self, now=None):
+        """G11/G12 后台投递：每轮循环调用，绝不中断在跑任务。
+
+        outbox 每轮都试到期的行（重试节奏由 next_attempt_at
+        控，默认 60 秒退避、最多 3 次，到上限后删行并记日志，
+        不丢任务状态）；摘要只在距上次发送满 digest_interval
+        后才发一次（默认每小时），避免每轮都把摘要提前发掉。
+        全部异常只记日志，不抛、不炸循环。force 语义由调用方
+        通过把 _last_digest_sent 置 0 实现（单测用）。
+        """
+        now = time.time() if now is None else now
+        try:
+            from cloud.push import process_outbox_due, send_digest_due
+            out = process_outbox_due(
+                self.db_path, now=now,
+                max_attempts=self.outbox_max_attempts,
+                backoff_seconds=self.outbox_backoff_seconds)
+            for failed in out.get("failed", []):
+                log_event(self.logger, "worker_push_outbox_dropped",
+                          message="push outbox dropped after retries: %s"
+                                  % failed)
+            if now - self._last_digest_sent >= \
+                    self.digest_interval_seconds:
+                self._last_digest_sent = now
+                digest = send_digest_due(self.db_path, now=now)
+                for err in digest.get("errors", []):
+                    log_event(self.logger, "worker_push_digest_failed",
+                              message="push digest send failed: %s" % err)
+        except Exception as exc:
+            log_event(self.logger, "worker_push_due_failed",
+                      message="push due processing failed: %s" % exc)
+
     def run_loop(self, interval=300, max_seconds=None, stop_event=None,
                  drain_timeout=60, sweep_threshold=300):
         """常驻循环：每 interval 秒跑一轮 run_all，空闲也不退出。
@@ -3879,6 +4119,9 @@ class Worker:
                               message="worker_loop_error: %s" % exc)
                     results = []
                 all_results.extend(results)
+                # G11/G12 后台投递：摘要（每小时）与 outbox 重试，
+                # 轮次边界做，不中断在跑任务，失败只记日志。
+                self._process_push_due()
                 # 自动陈旧回收：死 worker 的 running 任务在这里被
                 # 标 interrupted；活任务心跳新鲜，不会被扫到。
                 if sweep_threshold is not None:
@@ -4017,6 +4260,15 @@ def main(argv=None):
                              "列表里的第一个模型再跑一次，401/400"
                              "绝不 fallback，取消绝不 fallback）")
     parser.add_argument("--max-turns", type=int, default=None)
+    parser.add_argument("--ignore-host-budget", dest="ignore_host_budget",
+                        action="store_true", default=True,
+                        help="G9（默认开）：worker 路径忽略 host 的 "
+                             "max_daily_turns 预算，云端 per-user USD "
+                             "配额 + worker 门才是口径")
+    parser.add_argument("--no-ignore-host-budget",
+                        dest="ignore_host_budget", action="store_false",
+                        help="G9：关闭忽略，worker 调 host.py 时 host "
+                             "的 max_daily_turns 预算照常生效（双重门）")
     parser.add_argument("--max-tasks", type=int, default=None)
     parser.add_argument("--executor", default="local",
                         choices=["local", "docker", "cloud"],
@@ -4115,6 +4367,7 @@ def main(argv=None):
                          args.fallback_models.split(",")]
                         if args.fallback_models else []),
                     max_turns=args.max_turns,
+                    ignore_host_budget=args.ignore_host_budget,
                     executor=_build_cli_executor(args, host_py),
                     task_timeout=args.task_timeout,
                     max_retries=args.max_retries,

host.py

@@ -2057,6 +2057,19 @@ def _load_config_file(path):
     return {}
 
 
+def _effective_cfg(cfg, ignore_host_budget):
+    """G9：--ignore-host-budget 时的生效配置。
+
+    只清 max_daily_turns 一项（其它配置原样），返回新 dict
+    不改原 cfg；不传开关时原样返回。main 用它，单测直接
+    测它，不用拉起 host.py 子进程。"""
+    if not ignore_host_budget:
+        return cfg
+    out = dict(cfg or {})
+    out["max_daily_turns"] = None
+    return out
+
+
 def _resolve_path(base_dir, p):
     if not p:
         return p
@@ -2099,6 +2112,12 @@ def main():
                          "打一行 turn progress 并往链路文件记一条 "
                          "kind=progress；0 表示关闭进度输出；不传时用配置 "
                          "progress_interval，未配则默认 30")
+    ap.add_argument("--ignore-host-budget", action="store_true",
+                    help="G9：忽略 host 的 max_daily_turns 预算（云端 "
+                         "worker 路径默认传它：云端已有 per-user USD "
+                         "配额 + worker 认领/轮询门，host 单机 turn "
+                         "数预算再拦一次只会误拦云端已批的任务；"
+                         "standalone 直接跑 host.py 时不传，预算照常）")
     sub = ap.add_subparsers(dest="cmd", required=True)
 
     p = sub.add_parser("init")
@@ -2241,6 +2260,10 @@ def main():
         print(f"配置文件已写：{config_path}，请按需改 repo/model 后再用")
         return
     # 有效配置（init/config 之外的命令都用它）
+    # G9：只清预算这一项，notify_command 等其它配置不动；
+    # 本进程内所有 budget_exceeded(state, cfg) 检查随之失效，
+    # dispatch 前置门、重试前检查、goal 循环顶检查一并放行。
+    cfg = _effective_cfg(cfg, getattr(args, "ignore_host_budget", False))
     args.model = args.model or cfg.get("model") or "gpt-5.6-terra"
     args.effort = args.effort or cfg.get("effort") or "low"
     args.sandbox = args.sandbox or cfg.get("sandbox")

tests/test_host.py

@@ -14,6 +14,23 @@ sys.path.insert(0, str(REPO_ROOT))
 import host
 
 
+class EffectiveCfgTest(unittest.TestCase):
+    # G9：--ignore-host-budget 只清 max_daily_turns，其它配置不动。
+    def test_ignore_host_budget_clears_only_budget(self):
+        cfg = {"max_daily_turns": 10, "notify_command": "echo hi",
+               "model": "gpt-6.1-sol"}
+        out = host._effective_cfg(cfg, True)
+        self.assertIsNone(out["max_daily_turns"])
+        self.assertEqual(out["notify_command"], "echo hi")
+        self.assertEqual(out["model"], "gpt-6.1-sol")
+        # 不改原 cfg。
+        self.assertEqual(cfg["max_daily_turns"], 10)
+
+    def test_no_ignore_returns_cfg_unchanged(self):
+        cfg = {"max_daily_turns": 10}
+        self.assertEqual(host._effective_cfg(cfg, False), cfg)
+
+
 class HostPureFunctionsTest(unittest.TestCase):
     def test_save_state_merges_tasks_and_preserves_disk_tasks(self):
         with tempfile.TemporaryDirectory() as tmp: