feat: goal 断点续跑端点（G10）+ 摘要持久化与推送 outbox（G11/G12）

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 全过。
2026-10-28 11:09:29 +08:00
1 parent ac553509ea
commit 5883169c82
@@ -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