From 5883169c829d54c2e5f1a847f490875b55efc73b Mon Sep 17 00:00:00 2001 From: host Date: Tue, 27 Oct 2026 09:19:00 +0000 Subject: [PATCH 1/3] =?UTF-8?q?feat:=20goal=20=E6=96=AD=E7=82=B9=E7=BB=AD?= =?UTF-8?q?=E8=B7=91=E7=AB=AF=E7=82=B9=EF=BC=88G10=EF=BC=89+=20=E6=91=98?= =?UTF-8?q?=E8=A6=81=E6=8C=81=E4=B9=85=E5=8C=96=E4=B8=8E=E6=8E=A8=E9=80=81?= =?UTF-8?q?=20outbox=EF=BC=88G11/G12=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit G10：POST /tasks/{id}/goal-resume，仅 interrupted 的 goal 任务， 走 host.py goal-resume（无 --goal-text）从 goal_rounds_done 下一轮 断点续跑，保留原线程与 worktree，状态回 queued 并置 goal_resume_pending（worker 认领后清掉），写 task_goal_resumed 审计；goal-met/unmet、failed、非 goal 一律 409；/continue 开新 循环语义不动。Local/Docker 两路新增 goal_resume_breakpoint_task， 不支持断点的执行器明确失败，绝不回落成 goal 或开新循环。 G11：digest 非失败事件持久化进 push_digest_events，worker 循环 每小时按（项目、订阅）分组汇总发一条，失败类仍即时发。 G12：webhook/slack 投递失败写 push_outbox，worker 循环后台重试 （最多 3 次、默认 60 秒退避），3 次仍失败删行并记日志，不丢 任务状态；缺凭证与无 sender 不进 outbox；文档写明 at-least-once 下 webhook/slack 可能重复，接收方须去重。 测试：cloud 769（4 跳过）、host 68、deploy 19 全过。 --- cloud/README.md | 50 +++-- cloud/db.py | 1 + cloud/push.py | 319 +++++++++++++++++++++++++++++++- cloud/sandbox.py | 97 ++++++++++ cloud/server.py | 85 +++++++++ cloud/tests/test_goal_resume.py | 194 +++++++++++++++++++ cloud/tests/test_push.py | 115 ++++++++++++ cloud/worker.py | 118 +++++++++++- 8 files changed, 958 insertions(+), 21 deletions(-) create mode 100644 cloud/tests/test_goal_resume.py diff --git a/cloud/README.md b/cloud/README.md index 64fad24..f727683 100644 --- a/cloud/README.md +++ b/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,17 +973,26 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 NULL，SUM 自动跳过），按任务创建者 `created_by` 归属——一个 用户在多个项目里花的钱都算进他自己的月度账单；项目维度的 拆分由项目成本接口另行提供。 -- 准入：建任务、会话分支、重试、续跑在权限校验通过后、写库 - 之前查配额，本月用量 >= 配额时回 402 `quota_exceeded`（附 - month、quota_usd_micros、used_usd_micros），不建任务、不改 - 状态。老用户与未设配额的用户默认不限额，行为与过去一致。但 - OAuth 新用户不再是“建号即不限额”：建号（created=True）时 - 写一行默认配额（`CLOUD_DEFAULT_QUOTA_USD_MICROS`，默认 - 10_000_000，即 $10/月；`INSERT OR IGNORE` 幂等，不覆盖管理 - 员后来改过的额度），并记 `quota_default` 审计。重试与续跑 - 按任务创建者查配额（费用本来就记在创建者名下；维护者替 - 他人重试/续跑时，操作人超限不应拦、创建者超限才应拦—— - 旧版重试按操作人查是错位，本轮已改）。后续消息 +- 准入：建任务、会话分支、重试、续跑、goal 断点续跑在权限 + 校验通过后、写库之前查配额，本月用量 >= 配额时回 402 + `quota_exceeded`（附 month、quota_usd_micros、used_usd_micros）， + 不建任务、不改状态。老用户与未设配额的用户默认不限额， + 行为与过去一致。但 OAuth 新用户不再是“建号即不限额”： + 建号（created=True）时写一行默认配额 + （`CLOUD_DEFAULT_QUOTA_USD_MICROS`，默认 10_000_000， + 即 $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 也改成准入校验。系统 任务（GitHub webhook 的 /fix-review，created_by 是 'system'） diff --git a/cloud/db.py b/cloud/db.py index 80345c6..78c8baf 100644 --- a/cloud/db.py +++ b/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 同款， diff --git a/cloud/push.py b/cloud/push.py index 1835212..2cc2024 100644 --- a/cloud/push.py +++ b/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 diff --git a/cloud/sandbox.py b/cloud/sandbox.py index e27677e..ebd5a58 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -449,6 +449,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 。不传 + --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 +604,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 @@ -1305,6 +1331,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（宿主在该网桥上的地址）。 diff --git a/cloud/server.py b/cloud/server.py index e9ca411..e448c9d 100644 --- a/cloud/server.py +++ b/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): """把终态失败任务重新排队：只改状态，不直接执行。 diff --git a/cloud/tests/test_goal_resume.py b/cloud/tests/test_goal_resume.py new file mode 100644 index 0000000..cc133df --- /dev/null +++ b/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() diff --git a/cloud/tests/test_push.py b/cloud/tests/test_push.py index 6cdf61a..d2d1d39 100644 --- a/cloud/tests/test_push.py +++ b/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() diff --git a/cloud/worker.py b/cloud/worker.py index cf1ba8a..4ec20a4 100644 --- a/cloud/worker.py +++ b/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,23 @@ 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): 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 + self._last_digest_sent = 0.0 # env 只用于 key 存在性检查：默认接进程环境，单测可注入 # 一个普通 dict；worker 永远不把 env 里的值写出。 self._env = env if env is not None else os.environ @@ -757,7 +771,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( @@ -1590,7 +1605,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"): @@ -2277,7 +2293,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,7 +2334,16 @@ 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.executor.goal_resume_breakpoint_task( + task, + timeout=effective_docker_timeout, + extra_mounts=tenant_mounts, + network=tenant_network, + extra_env=proxy_env) + elif task.get("_resume"): # P1 断点续跑传 None（不传 --goal-text， # host 沿用原 goal+verify）；开新目标 # 才传本轮 prompt 当新 goal_text。 @@ -2362,7 +2390,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" \ @@ -2926,6 +2972,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 +3013,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"): @@ -3800,6 +3869,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 +3980,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: From e31fcc6cba2728e19cd5d31f8612e712a7e4bd36 Mon Sep 17 00:00:00 2001 From: host Date: Tue, 27 Oct 2026 10:07:44 +0000 Subject: [PATCH 2/3] =?UTF-8?q?feat:=20G7=20=E5=88=86=E5=B1=82=20fail-open?= =?UTF-8?q?/fail-closed=20+=20G8=20Docker=20=E8=BD=AE=E8=AF=A2=E6=8E=90?= =?UTF-8?q?=E6=96=AD=20+=20G9=20worker=20=E5=BF=BD=E7=95=A5=20host=20?= =?UTF-8?q?=E9=A2=84=E7=AE=97=20+=20G13/G14=20=E5=BD=92=E5=B1=9E=EF=BC=88?= =?UTF-8?q?=E7=AC=AC=E4=B8=89=E6=89=B9=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - G7：认领门 closed 时查配额异常回 quota_check_unavailable、标 interrupted（可重试/续跑）；轮询与轮次边界用 open，异常时回 quota_check_failed——新回合边界拒绝，执行中轮询经 _quota_hit 闭包过滤此串、只记 warning 不杀在跑任务，确认超限才掐断；查询函数本身不吞异常冒充不限额。 - G8：Docker 路径加 _docker_call_with_quota_poll（后台线程跑阻塞式 docker 调用，主循环每 quota_poll_interval_seconds 缺省 30 秒查一次，确认超限按容器名真杀容器，结果改写 quota_exceeded 走统一收尾：interrupted、成本照记、续跑消息保留，配额命中后不走 ALREADY_EXISTS 回落）；同步路径保持边界检查。 - G9：worker 路径默认忽略 host max_daily_turns（host.py 顶层 --ignore-host-budget 只清该项，Local/Docker 执行器在子命令前追加，worker CLI 缺省开、--no-ignore-host-budget 关）；standalone 直接跑 host.py 时预算照常。 - G13/G14：/pr 不再走 system 豁免，建任务前按请求用户查配额（402/503 均不排队不改状态），队列路 pr_jobs.created_by 记请求用户（老表 ALTER 补列），worker 跑 pr 任务时审计操作人用 created_by（老行回落 system）。注意：G14 原案的 system 豁免（服务端与 worker 写死）与 P1 冲突（P1：system 按普通属主走默认额度，不另设无限额系统预算），本次 rebase 未合入该豁免，system 豁免需另行决策。 - 配额口径与 P0/P1 对齐：无配额行按默认额度（非不限额），只有显式 NULL 行才不限额；无价上限只计 source='actual' 行；缺属主（含父任务回查仍无）拒绝新回合。 - README 配额节同步以上口径与 G15 接受现状（best-effort 超支、UTC 月切分、在跑任务只在轮询/轮次边界响应配额改动）。G4 与 G15 不在本批。 --- cloud/README.md | 73 +++++- cloud/pr_exec.py | 32 ++- cloud/sandbox.py | 16 +- cloud/server.py | 50 +++- cloud/tests/test_pr_jobs.py | 43 ++- cloud/tests/test_task_options.py | 33 +++ cloud/tests/test_task_quotas.py | 96 ++++++- cloud/tests/test_worker_quota.py | 142 ++++++++++ cloud/worker.py | 437 +++++++++++++++++++++---------- host.py | 23 ++ tests/test_host.py | 17 ++ 11 files changed, 786 insertions(+), 176 deletions(-) diff --git a/cloud/README.md b/cloud/README.md index f727683..baed554 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -994,10 +994,35 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 重试/续跑时，操作人超限不应拦、创建者超限才应拦——旧版 重试按操作人查是错位，本轮已改）。后续消息 （followup）创建不查配额：排队本身不花钱，执行时轮次边界 - 的配额门照常生效，不要把 followup 也改成准入校验。系统 - 任务（GitHub webhook 的 /fix-review，created_by 是 'system'） - 不记到任何用户名下、不占配额；要把它算到项目所有者头上 - 需另做归属。 + 的配额门照常生效，不要把 followup 也改成准入校验。 +- 费用归属与豁免（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 计）。因此有配额行的用户另受一 @@ -1017,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/同步 - 路径没有轮询钩子，只有边界检查，单轮内的超支要到轮次结束 - 才停；单轮本身花的钱无法中途按 token 流式截断（中转用量只 - 在回合结束才回传），所以“掐断”粒度是轮次边界 + 长轮询中的 - 30 秒检查，不是 token 级；并发下两个任务同时查到未超限再同 - 时开跑会有少量超支（与限流器同款 best-effort，非精确配额）； - PR 任务（_run_pr_jobs）暂不查配额。 + actual，不发终态推送（与取消同款）。诚实边界：轮询查的是 + 已记账用量，单轮本身花的钱无法中途按 token 流式截断（中转 + 用量只在回合结束才回传），所以“掐断”粒度是轮次边界 + 长轮 + 询中的 30 秒检查，不是 token 级；真 Docker 掐断未实弹（只 + 由假执行器单测覆盖），部署后需实弹；并发下两个任务同时查 + 到未超限再同时开跑会有少量超支（与限流器同款 best-effort， + 非精确配额）——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`——价目表里没有该模型价格的实际 diff --git a/cloud/pr_exec.py b/cloud/pr_exec.py index 9d3c22a..b85fb40 100644 --- a/cloud/pr_exec.py +++ b/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): - """排一个 pr 任务（调用方已做去重）。返回 job id。""" +def enqueue_pr_job(conn, task_id, project_id, host, created_by=None): + """排一个 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', ?, ?)", - (job_id, task_id, project_id, host, now, now)) + " created_at, updated_at, created_by)" + " 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)) diff --git a/cloud/sandbox.py b/cloud/sandbox.py index ebd5a58..f5f54f1 100644 --- a/cloud/sandbox.py +++ b/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] @@ -693,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") @@ -703,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) @@ -916,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] diff --git a/cloud/server.py b/cloud/server.py index e448c9d..29b41f2 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -6368,7 +6368,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): month)，调用方一律回 402——这是钱用完了，不是请求 太快，所以不用 429。 - 准入点：建任务 / 分支 / 重试 / 续跑，均为权限校验 + 准入点：建任务 / 分支 / 重试 / 续跑 / goal-resume，均为权限校验 通过后、写库之前。P0 安全批后，后续消息与任务间 消息创建时同样查双人配额门（见 _reject_if_over_quota_for_task）：排队本身不花钱， @@ -6376,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 秒查一次配额， - 超限杀进程组；Docker / 同步路径没有轮询钩子，只有 - 认领与轮次边界检查，单轮内的超支要到轮次结束才停； - 单轮本身的钱无法按 token 流式截断（用量只在回合 - 结束才回传），掐断粒度是轮次边界 + 长轮询中的 + Local popen 与 Docker 路径的 worker 轮询每 30 秒查 + 一次配额，超限杀进程组/容器；同步路径没有轮询钩子， + 只有认领与轮次边界检查，单轮内的超支要到轮次结束 + 才停；单轮本身的钱无法按 token 流式截断（用量只在 + 回合结束才回传），掐断粒度是轮次边界 + 长轮询中的 30 秒检查，不是 token 级。""" quota, used, month = self._quota_status(user_id) if quota is not None and used >= quota: @@ -6395,8 +6407,19 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): 方就是创建者，缺省即调用方；重试 / 续跑时费用记在 原任务创建者名下，必须传 task 的 created_by，否则 维护者替他人重试会用错人的额度（旧 bug：重试按 - 操作人查、费用记创建者名下）。""" - blocked = self._check_quota(quota_user_id or user_id) + 操作人查、费用记创建者名下）。 + + 查询异常时 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", @@ -6690,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") @@ -6701,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, { diff --git a/cloud/tests/test_pr_jobs.py b/cloud/tests/test_pr_jobs.py index 9c6a672..0d77378 100644 --- a/cloud/tests/test_pr_jobs.py +++ b/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): diff --git a/cloud/tests/test_task_options.py b/cloud/tests/test_task_options.py index 06610a9..d8afa54 100644 --- a/cloud/tests/test_task_options.py +++ b/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() diff --git a/cloud/tests/test_task_quotas.py b/cloud/tests/test_task_quotas.py index fa2ef4a..bf1ab0d 100644 --- a/cloud/tests/test_task_quotas.py +++ b/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,96 @@ 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_exempt_server(self): + 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) + self.assertIsNone(handler._check_quota("system")) + self.assertFalse(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() diff --git a/cloud/tests/test_worker_quota.py b/cloud/tests/test_worker_quota.py index 556abe6..aac3159 100644 --- a/cloud/tests/test_worker_quota.py +++ b/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,144 @@ 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(self): + _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")): + # 默认 on_error="open"：异常不冒充超限，放行。 + self.assertIsNone(worker._quota_exceeded(task, None)) + self.assertIsNone(worker._quota_exceeded(task, None, + on_error="open")) + + # -- G13：system 豁免（worker 侧） ---------------------------------- + def test_system_task_exempt_from_quota(self): + _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"], "completed") + self.assertEqual(_status(self.db_path, "T1"), "completed") + + # -- 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() diff --git a/cloud/worker.py b/cloud/worker.py index 4ec20a4..8076283 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -372,7 +372,9 @@ class Worker: upload_dir=None, digest_interval_seconds=3600, outbox_backoff_seconds=60, - outbox_max_attempts=3): + 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 @@ -385,6 +387,17 @@ class Worker: 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 里的值写出。 @@ -416,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 可注入（单测用它返回一个假长跑 @@ -961,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）一致：按任务创建者算， @@ -970,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" 的系统任务按普通属主算（无行时同样走默认 - 额度，不另设无限额的系统预算）。 - - 无配额行：不再是不限额，按默认配额 - _default_quota_usd_micros()（$10/月）算；只有显式 - NULL 行才是不限额（与 P0 的服务端口径一致，管理员 - 显式设 NULL 的行仍被尊重）。 - - 查库失败：返回 "quota_check_failed: ..."（fail-closed）， - 绝不返回 None 放行。调用方分流：新回合边界（认领门、 - 重试/fallback/后续消息边界）见此串即拒绝新回合并记 - worker_quota_stop 审计；执行中轮询（popen 轮询、 - Docker 配额看门狗）见此串只记警告、不中途杀当前轮 - ——查询抖动不应杀掉已在跑的回合，但下一轮边界仍会 - 拒绝（见 run_one 的 _quota_hit 过滤）。查库失败的 - 审计由边界调用方记，这里只记日志，避免轮询高频写审计。 + "system" 的系统任务按普通属主算，不另设无限额的 + 系统预算（G14 的 system 豁免与 P1 冲突，本次未合入）。 + - 无配额行：按默认配额 _default_quota_usd_micros() + （$10/月）算；只有显式 NULL 行才是不限额。 + 查库失败分层（G7 + P1）：认领门传 on_error="closed"， + 返回 "quota_check_unavailable: ..."（调用方标 + interrupted，可重试/续跑）；默认 "open" 返回 + "quota_check_failed: ..."——新回合边界（直接调用） + 见此串即拒绝，执行中轮询经 _quota_hit 闭包过滤此串、 + 只记警告不杀当前轮。查询抖动不应杀掉已在跑的回合， + 但下一轮边界仍会拒绝。 """ owner = (task or {}).get("created_by") if not owner and (task or {}).get("parent_task_id"): @@ -1017,95 +1102,45 @@ class Worker: return ("quota_exceeded: 任务缺少属主，无法查配额，" "拒绝新回合（请归属真实用户后再排队）") try: - 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: - # 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( - model, usage_total.get("input_tokens", 0), - usage_total.get("cached_input_tokens", 0), - usage_total.get("output_tokens", 0)) - used += micros or 0 - if micros is None: - # 无价模型：本轮 token 计入无价上限（USD 口 - # 径里它是 0，不会在上面那道里被拦住）。 - unpriced_used += ( - max(0, int(usage_total.get("input_tokens", 0))) - + max(0, int(usage_total.get("output_tokens", 0)))) - if used >= quota: - return ("quota_exceeded: 本月配额已用完（已用 %d / " - "配额 %d 微美元，%s）" % (used, quota, month)) - cap = unpriced_token_cap() - if unpriced_used >= cap: - return ("quota_exceeded: 本月无价模型 token 已用完" - "（已用 %d / 上限 %d，%s）" - % (unpriced_used, cap, month)) - return None + quota, used, month, unpriced_used = \ + self._quota_limit_and_used(owner) except Exception as exc: - # P1 fail-closed：查库失败不再返回 None。调用方按 - # 前缀分流（见本函数 docstring）：quota_check_failed - # 在新回合边界 = 拒绝，在执行中轮询 = 只警告不杀轮。 + # 查库失败分层（见本函数 docstring）：closed（认领门） + # 回 quota_check_unavailable，调用方标 interrupted； + # open 回 quota_check_failed，新回合边界拒绝，执行中 + # 轮询经 _quota_hit 闭包过滤、只警告不杀轮。 try: self.logger.warning("quota check failed: %s", exc) except Exception: pass + if on_error == "closed": + return "quota_check_unavailable: 配额查询失败，任务未开跑" return ("quota_check_failed: 配额查询失败，拒绝新回合" "（查库异常，非配额用完）") + if quota is None: + return None + if usage_total: + model = (task or {}).get("model") or self.model + micros = price_usd_micros( + model, usage_total.get("input_tokens", 0), + usage_total.get("cached_input_tokens", 0), + usage_total.get("output_tokens", 0)) + used += micros or 0 + if micros is None: + # 无价模型：本轮 token 计入无价上限（USD 口 + # 径里它是 0，不会在上面那道里被拦住）。 + unpriced_used += ( + max(0, int(usage_total.get("input_tokens", 0))) + + max(0, int(usage_total.get("output_tokens", 0)))) + if used >= quota: + return ("quota_exceeded: 本月配额已用完（已用 %d / " + "配额 %d 微美元，%s）" % (used, quota, month)) + cap = unpriced_token_cap() + if unpriced_used >= cap: + return ("quota_exceeded: 本月无价模型 token 已用完" + "（已用 %d / 上限 %d，%s）" + % (unpriced_used, cap, month)) + return None def _ready_tasks(self): """当前就绪任务列表（pick_next 的批量版）。 @@ -1511,11 +1546,13 @@ class Worker: 的标记，调用方据此保持 interrupted；deadline_seconds 到期时杀进程组并返回 status='timeout' 的标记，调用方 据此置 failed（超时是失败，不是取消）。quota_check 不为 - 空时每 30 秒查一次配额（轮询本身 50 毫秒一圈，逐圈查会 - 把库打爆），超限时同样杀进程组并返回 status= - 'quota_exceeded' 的内部标记——调用方据此走配额中断收尾， - 绝不能把这个标记原样写进 tasks.status（不在合法状态集 - 里）。Docker/同步路径没有这个轮询钩子，只走边界检查。 + 空时每 quota_poll_interval_seconds（缺省 30 秒）查一次 + 配额（轮询本身 50 毫秒一圈，逐圈查会把库打爆），超限时 + 同样杀进程组并返回 status='quota_exceeded' 的内部标记 + ——调用方据此走配额中断收尾，绝不能把这个标记原样写 + 进 tasks.status（不在合法状态集里）。Docker 路径见 + _docker_call_with_quota_poll（同款轮询）；同步路径没有 + 轮询钩子，只走边界检查。 """ task_id = task["id"] gateway_provider, gateway_env = self._ensure_gateway(task) @@ -1539,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: @@ -1567,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() @@ -2205,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 门、重试、成本与最终审计）。 @@ -2337,12 +2452,16 @@ class Worker: if task.get("_goal_resume_breakpoint"): # G10 断点：Docker 路用无 --goal-text 的 # goal-resume-breakpoint 命令。 - result = self.executor.goal_resume_breakpoint_task( + result = self._docker_call_with_quota_poll( task, - timeout=effective_docker_timeout, - extra_mounts=tenant_mounts, - network=tenant_network, - extra_env=proxy_env) + 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）；开新目标 @@ -2350,31 +2469,41 @@ class Worker: _resume_text = None if task.get( "_resume_kind") == "breakpoint" \ else task["prompt"] - result = 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) + 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), + 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), + quota_check) + else: + 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), seed_host_dir=seed_host_dir, - model_provider=gateway_provider) - else: - result = self.executor.dispatch_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: # seed 参数只传给 LocalExecutor：CloudExecutor # 基类与测试假执行器的 goal_task/dispatch_task @@ -2434,19 +2563,23 @@ 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( - task, timeout=effective_docker_timeout, - extra_mounts=tenant_mounts - if self.tenant_isolation else None, - network=tenant_network - if self.tenant_isolation else None, - extra_env=_merge_env( - proxy_env if self.egress_proxy_port - else None, gateway_env), - model_provider=gateway_provider) + 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, + network=tenant_network + if self.tenant_isolation else None, + extra_env=_merge_env( + proxy_env if self.egress_proxy_port + else None, gateway_env), + model_provider=gateway_provider), + quota_check) else: if gateway_provider and not isinstance( self.executor, LocalExecutor): @@ -3092,8 +3225,11 @@ class Worker: # 再开跑——直接标 interrupted + 配额原因，不记成本（还没花 # 钱），续跑消息不消费，顶上来后重试/续跑可再跑。与 key # 门、仓库门同款提前返回形状，区别是状态为 interrupted - #（可续跑）而非 failed。 - quota_error = self._quota_exceeded(task, None) + #（可续跑）而非 failed。G7 分层：认领门用 closed—— + # 查询异常时同样标 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( @@ -3696,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", @@ -4121,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"], @@ -4219,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, diff --git a/host.py b/host.py index 81d77d3..6a1eed1 100644 --- a/host.py +++ b/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") diff --git a/tests/test_host.py b/tests/test_host.py index cb88340..688e9e2 100644 --- a/tests/test_host.py +++ b/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: From 9f49b97cd3143d1dd06a8cff7456bba941104c5e Mon Sep 17 00:00:00 2001 From: host Date: Wed, 28 Oct 2026 11:25:17 +0800 Subject: [PATCH 3/3] =?UTF-8?q?test:=20G14/open=20=E6=97=A7=E5=8F=A3?= =?UTF-8?q?=E5=BE=84=E6=B5=8B=E8=AF=95=E6=8C=89=20P1=20=E5=8F=A3=E5=BE=84?= =?UTF-8?q?=E6=9B=B4=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - test_system_quota_exempt_server → test_system_quota_not_exempt_server：system 按普通属主算，quota 0 时照常拦（402）。 - test_system_task_exempt_from_quota → test_system_task_not_exempt_from_quota：system quota 0 时认领门拦，标 interrupted。 - test_poll_quota_lookup_failure_open → test_poll_quota_lookup_failure_open_returns_check_failed：open 异常时回 quota_check_failed（边界拒绝、轮询经 _quota_hit 过滤只警告），closed 回 quota_check_unavailable。 --- cloud/tests/test_task_quotas.py | 10 +++++++--- cloud/tests/test_worker_quota.py | 28 +++++++++++++++++++--------- 2 files changed, 26 insertions(+), 12 deletions(-) diff --git a/cloud/tests/test_task_quotas.py b/cloud/tests/test_task_quotas.py index bf1ab0d..aea99a2 100644 --- a/cloud/tests/test_task_quotas.py +++ b/cloud/tests/test_task_quotas.py @@ -375,7 +375,10 @@ class ServerQuotasTestCase(unittest.TestCase): self.assertEqual(n, 0) # -- G13：system 豁免（server 侧） ----------------------------------- - def test_system_quota_exempt_server(self): + 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( @@ -386,8 +389,9 @@ class ServerQuotasTestCase(unittest.TestCase): conn.close() handler = object.__new__(CloudControlPlaneHandler) handler.server = SimpleNamespace(db_path=self.db_path) - self.assertIsNone(handler._check_quota("system")) - self.assertFalse(handler._reject_if_over_quota("system")) + 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): diff --git a/cloud/tests/test_worker_quota.py b/cloud/tests/test_worker_quota.py index aac3159..545cec5 100644 --- a/cloud/tests/test_worker_quota.py +++ b/cloud/tests/test_worker_quota.py @@ -360,7 +360,12 @@ class WorkerQuotaTestCase(unittest.TestCase): conn.close() self.assertEqual(rows, []) - def test_poll_quota_lookup_failure_open(self): + 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}, @@ -369,21 +374,26 @@ class WorkerQuotaTestCase(unittest.TestCase): with mock.patch.object( Worker, "_quota_limit_and_used", side_effect=RuntimeError("db down")): - # 默认 on_error="open"：异常不冒充超限，放行。 - self.assertIsNone(worker._quota_exceeded(task, None)) - self.assertIsNone(worker._quota_exceeded(task, None, - on_error="open")) + 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 侧） ---------------------------------- - def test_system_task_exempt_from_quota(self): + # -- 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"], "completed") - self.assertEqual(_status(self.db_path, "T1"), "completed") + self.assertEqual(result["status"], "interrupted") + self.assertEqual(_status(self.db_path, "T1"), "interrupted") # -- G8 Docker 轮询掐断 --------------------------------------------- def test_docker_quota_poll_kills_container(self):