From 5883169c829d54c2e5f1a847f490875b55efc73b Mon Sep 17 00:00:00 2001 From: host Date: Tue, 27 Oct 2026 09:19:00 +0000 Subject: [PATCH] =?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: