推送真发送：webhook/slack 内置真实发送端 + worker 终态自动通知

此前订阅与分发框架都在，但分发端点传 sender=None，
webhook/slack 永远报无 sender、从未真发过。本轮：
- push：send_webhook/send_slack（标准库 urllib POST，
  单次 10 秒超时，不重试，失败记 errors）；sender=None
  时用内置端，显式 sender 优先；webpush/apns/fcm 仍报
  缺凭证（原因写明：webpush 需浏览器端三字段订阅与标准库
  没有的 P-256 ECDH + AES-GCM；apns 需 HTTP/2；fcm 需
  Firebase 服务账号，三者均无现成接收方，不造假发送）
- worker：run_one 终态（completed/failed，缺 key 与仓库
  失败两条早退路径也在内）后 best-effort 调分发，通知
  失败只记日志、绝不影响任务状态；取消（interrupted）
  不在事件集合里，不通知
- 测试：真实发送端 6 个（本地回环服务器）、worker 终态
  通知 2 个，旧无 sender 断言按新语义改写

cloud/README.md

@@ -222,12 +222,24 @@ webhook 实测）。
 
 ### 推送凭证清单
 
-- webhook / slack：需要可注入 sender（worker 侧），HTTP 层没有
-  sender 时明确报错，不静默成功。
-- webpush：需要 VAPID 密钥对（私钥绝不进仓库）。
-- apns：需要 APNs 证书。
-- fcm：需要 FCM key。
-以上三者目前只有接口 + 缺凭证报错，没有真实发送端。
+- webhook / slack：已有内置真实发送端（标准库 urllib POST，
+  单次 10 秒超时，不重试，失败记入分发结果的 errors）：
+  webhook 投 JSON（event_type、project_id、task_id、payload），
+  slack 投 Slack incoming webhook 格式的一行文本（任务 id +
+  事件 + 项目，失败附 error 摘要）。任务终态（completed /
+  failed）由 worker 在 run_one 的终态写回后自动通知订阅者；
+  手动分发走 `POST /admin/push/dispatch`，同样经内置发送端。
+  显式注入的 sender 优先于内置端。
+- webpush：需要 VAPID 密钥对（私钥绝不进仓库），且需要
+  浏览器端订阅（endpoint + p256dh + auth，现有订阅模型只有
+  一个 target 文本字段，装不下）与标准库没有的 P-256 ECDH +
+  AES-GCM——只有接口 + 缺凭证报错，没有真实发送端。
+- apns：需要 APNs 证书与标准库没有的 HTTP/2——只有接口 +
+  缺凭证报错。
+- fcm：需要 FCM key 与用户另建的 Firebase 服务账号——只有
+  接口 + 缺凭证报错。
+以上三者目前只有接口 + 缺凭证报错，没有真实发送端，也
+不编造推送。
 
 ### SQLite 并发上限
 
@@ -374,7 +386,7 @@ psycopg，写明“需安装 psycopg”），绝不静默回落到 SQLite，
 | `POST /projects/{id}/push/subscriptions` | 该项目 `create_task` | 记订阅 body `{channel, target, preference?}`，channel 为 webhook/slack/webpush/apns/fcm，preference 为 instant/digest/silent，返回 `201` |
 | `GET /projects/{id}/push/subscriptions` | 该项目 `read` | 只列调用者自己的订阅（不泄露他人 target） |
 | `DELETE /projects/{id}/push/subscriptions/{sid}` | 该项目 `create_task` | 只能删自己的订阅，项目 owner 可删任意；不存在 `404`，他人订阅 `403` |
-| `POST /admin/push/dispatch` | 该项目 `read` | body `{event_type, project_id, payload?}`；只发状态变化事件（completed/failed/goal-unmet/merge-failed/review-requested）；HTTP 层无注入 sender：webhook/slack 报“无 sender”，webpush/apns/fcm 报缺凭证，都不静默成功；真实发送端由 worker 注入 sender 后调 `push.dispatch_event` |
+| `POST /admin/push/dispatch` | 该项目 `read` | body `{event_type, project_id, payload?}`；只发状态变化事件（completed/failed/goal-unmet/merge-failed/review-requested）；sender=None 时用内置真实发送端（webhook/slack 真 POST，webpush/apns/fcm 报缺凭证），都不静默成功；worker 在任务终态（completed/failed）时自动调分发，不依赖手动调本端点 |
 | `POST /webhooks/github` | GitHub 签名 | 见“webhook 端点” |
 | `POST /webhooks/slack` | Slack 签名 | 见“webhook 端点” |
 

cloud/push.py

@@ -17,21 +17,36 @@
   ``digest`` 偏好下本次不立即发，返回结果里明确列入
   ``skipped_for_digest``（待摘要列表），不是静默丢弃。
 - 发送端可注入：``webhook`` / ``slack`` 走可注入的 ``sender``。
-  ``sender`` 为 ``None`` 时对这两个通道也返回“无 sender”错误，
-  不静默成功。``sender`` 可以是可调用对象（收到一个 dict，见
+  ``sender`` 为 ``None`` 时用内置真实发送端（见 :func:`send_webhook` /
+  :func:`send_slack`，标准库 urllib POST，单次 10 秒超时，不
+  重试，失败记入 errors）；显式传入的 sender 优先于内置端。
+  ``sender`` 可以是可调用对象（收到一个 dict，见
   :func:`dispatch_event`），也可以是按通道分的 dict
   （``{"webhook": callable, "slack": callable}``），或带 ``send``
-  方法的对象。
+  方法的对象。显式 sender 里没有某通道时，该通道仍报
+  “无 sender”，不静默成功。
 - ``webpush`` / ``apns`` / ``fcm`` 只有接口，没有真实发送端：
   一律明确报错“缺凭证（VAPID / APNs 证书 / FCM key）”，不编造
   真实推送、不生成或提交任何私钥到仓库、不向真实推送服务发
-  任何请求。真实凭证与真实发送端是后续待接线的工作，见模块
-  末尾的缺口说明。
-- 本模块不联网、不调用真实推送服务、不跑真实模型回合。
-  密钥、token、证书一律不由本模块读取、写入或返回。
+  任何请求。原因不是“没空写”：webpush 需要浏览器端订阅
+  （endpoint + p256dh + auth 三字段，现有订阅模型只有一个
+  target 文本字段，装不下）与标准库没有的 P-256 ECDH +
+  AES-GCM；apns 需要标准库没有的 HTTP/2；fcm 需要用户
+  另建 Firebase 项目与服务账号——三者都没有现成接收方，
+  本模块不为此造一套半吊子加密或假发送。真实凭证与真实
+  发送端是后续待接线的工作，见模块末尾的缺口说明。
+- 本模块不跑真实模型回合。webhook/slack 的真实发送只 POST
+  到订阅者自己填的 target URL（创建订阅已要求项目
+  create_task 权限；能建订阅的成员本来就能经任务执行代码，
+  因此不对 target 做内网屏蔽——屏蔽既拦不住提权，也会误伤
+  本地接收器）。密钥、token、证书一律不由本模块读取、写入
+  或返回。
 """
 
+import json
 import sqlite3
+import urllib.error
+import urllib.request
 from cloud.db import connect as _db_connect
 from cloud.db import PgConnection as _PgConnection
 import time
@@ -44,10 +59,13 @@ __all__ = [
     "ALLOWED_PUSH_EVENTS",
     "FAILURE_EVENTS",
     "MISSING_CREDENTIAL_REASONS",
+    "DEFAULT_SENDERS",
     "init_push_db",
     "create_subscription",
     "delete_subscription",
     "list_subscriptions",
+    "send_webhook",
+    "send_slack",
     "dispatch_event",
 ]
 
@@ -254,6 +272,91 @@ def _resolve_channel_sender(sender, channel):
     return send if callable(send) else None
 
 
+# 真实发送端：单次投递超时（秒），不重试——失败记入 errors，
+# 下一次事件自然重发，不在这里循环撞墙。
+_PUSH_SEND_TIMEOUT = 10
+
+
+def _post_json(url, payload):
+    """POST JSON 到 url：返回 (ok, detail)，2xx 为 ok。
+
+    HTTP 状态错误与网络异常一律返回 (False, reason)，不抛
+    （调用方据此记 errors）；只有调用方传了非法 URL 时
+    Request 构造本身可能抛 ValueError，由发送端先校验 scheme
+    挡住，不走到这里。
+    """
+    data = json.dumps(payload).encode("utf-8")
+    request = urllib.request.Request(
+        url, data=data, method="POST",
+        headers={"Content-Type": "application/json",
+                 "User-Agent": "codex-app-server"})
+    try:
+        with urllib.request.urlopen(
+                request, timeout=_PUSH_SEND_TIMEOUT) as response:
+            response.read()
+            return True, "HTTP %s" % response.status
+    except urllib.error.HTTPError as e:
+        return False, "HTTP %s" % e.code
+    except Exception as exc:  # noqa: BLE001 -- 网络异常转 reason，不抛
+        return False, "%s: %s" % (type(exc).__name__, exc)
+
+
+def _require_http_url(target, channel):
+    """发送端 target 校验：只接受 http(s) URL，其他一律 ValueError
+    （dispatch 记入 errors，不静默成功、不乱 POST）。"""
+    if not isinstance(target, str) or not (
+            target.startswith("http://") or target.startswith("https://")):
+        raise ValueError(
+            "%s target must be an http(s) URL" % channel)
+    return target
+
+
+def send_webhook(message):
+    """webhook 通道真实发送：POST JSON 到订阅的 target URL。
+
+    body 固定为 {event_type, project_id, task_id, payload}，
+    task_id 从 payload 里取（没有则为 None）。非 2xx 抛
+    RuntimeError（dispatch 记入 errors，不算 sent）。
+    """
+    target = _require_http_url(message["target"], "webhook")
+    payload = message["payload"] if isinstance(
+        message["payload"], dict) else {}
+    ok, detail = _post_json(target, {
+        "event_type": message["event_type"],
+        "project_id": message["project_id"],
+        "task_id": payload.get("task_id"),
+        "payload": message["payload"],
+    })
+    if not ok:
+        raise RuntimeError("webhook delivery failed: %s" % detail)
+
+
+def send_slack(message):
+    """slack 通道真实发送：POST Slack incoming webhook 格式
+    （{"text": ...}）到 target URL（Slack webhook 地址本身
+    就是 http(s) URL，与 webhook 通道同一投递路径）。
+
+    文本一行讲清事件：任务 id + 事件类型 + 项目，失败类
+    事件附 error 摘要（截断 500 字）。非 2xx 抛 RuntimeError。
+    """
+    target = _require_http_url(message["target"], "slack")
+    payload = message["payload"] if isinstance(
+        message["payload"], dict) else {}
+    text = "[codex] task %s %s (project %s)" % (
+        payload.get("task_id", "?"), message["event_type"],
+        message["project_id"])
+    if payload.get("error"):
+        text += "\nerror: %s" % str(payload["error"])[:500]
+    ok, detail = _post_json(target, {"text": text})
+    if not ok:
+        raise RuntimeError("slack delivery failed: %s" % detail)
+
+
+# sender=None 时的内置真实发送端：只覆盖 webhook/slack；
+# webpush/apns/fcm 不在此列，仍走缺凭证报错。
+DEFAULT_SENDERS = {"webhook": send_webhook, "slack": send_slack}
+
+
 def dispatch_event(db, event_type, project_id, payload, sender=None):
     """分发一次状态变化事件，只发本项目的订阅。
 
@@ -265,9 +368,10 @@ def dispatch_event(db, event_type, project_id, payload, sender=None):
         project_id: 只查该项目的订阅，跨项目不混。
         payload: 原事件载荷（dict 或其他），原样放入发送给 sender
             的消息 dict 的 ``payload`` 字段。
-        sender: 可注入发送端（见模块 docstring）。webhook/slack
-            通道在 sender 为 None 时返回“无 sender”错误，不静默成功；
-            webpush/apns/fcm 通道不走 sender，一律返回缺凭证错误。
+        sender: 可注入发送端（见模块 docstring）；为 None 时用
+            内置真实发送端（webhook/slack 真 POST，webpush/apns/fcm
+            仍报缺凭证）；显式传入的 sender 优先于内置端，显式
+            sender 里没有某通道时该通道报“无 sender”，不静默成功。
 
     返回 dict，键固定为::
 
@@ -314,6 +418,11 @@ def dispatch_event(db, event_type, project_id, payload, sender=None):
         }]
         return result
 
+    if sender is None:
+        # 内置真实发送端：webhook/slack 真 POST；显式 sender
+        # 传入时不走这里（注入优先），webpush/apns/fcm 永远
+        # 不走 sender，仍在循环里报缺凭证。
+        sender = DEFAULT_SENDERS
     subscriptions = list_subscriptions(db, project_id)
     is_failure = event_type in FAILURE_EVENTS
     for sub in subscriptions:
@@ -336,7 +445,8 @@ def dispatch_event(db, event_type, project_id, payload, sender=None):
             result["errors"].append(
                 dict(base, reason=MISSING_CREDENTIAL_REASONS[sub["channel"]]))
             continue
-        # webhook/slack：必须有可注入 sender，否则明确报错。
+        # webhook/slack：sender=None 时已换成内置真实发送端；
+        # 显式 sender 里没有该通道时明确报错，不静默成功。
         channel_sender = _resolve_channel_sender(sender, sub["channel"])
         if channel_sender is None:
             result["errors"].append(

cloud/server.py

@@ -2428,9 +2428,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             if not auth.check_permission(user_id, project_id, "read"):
                 self._send_json(403, {"error": "forbidden"})
                 return
-        # HTTP 层无可注入 sender：webhook/slack 会报“无 sender”，
-        # webpush/apns/fcm 会报缺凭证——都不静默成功。真实发送端
-        # 由 worker 注入 sender 后调 push.dispatch_event。
+        # 分发：sender=None 时用 push 的内置真实发送端
+        # （webhook/slack 真 POST，webpush/apns/fcm 仍报缺凭证）；
+        # 显式注入的 sender 优先于内置端（见 push.dispatch_event）。
         result = _push_dispatch(
             self.server.db_path, event_type, project_id,
             body.get("payload") or {}, sender=None)

cloud/tests/test_push.py

@@ -1,14 +1,17 @@
 """push.py 单测：订阅增删查、偏好过滤、分发错误不静默丢。
 
-只用标准库，不联网、不向真实推送服务发任何请求：所有发送都用
-假 sender（可调用对象或按通道分的 dict）注入；webpush/apns/fcm
-只断言缺凭证报错，不调用任何 sender。
+不连外网、不向真实推送服务发任何请求：注入 sender 的测试
+用假 sender；真实发送端的测试用本地回环 HTTP 服务器
+（127.0.0.1 临时端口）收 POST，断言方法、头与 body。
 本文件不含任何真实密钥、token、证书或私钥。
 """
 
+import http.server
+import json
 import os
 import sqlite3
 import sys
+import threading
 import unittest
 from pathlib import Path
 
@@ -22,6 +25,7 @@ try:
         delete_subscription,
         dispatch_event,
         list_subscriptions,
+        send_webhook,
     )
 except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
@@ -34,6 +38,7 @@ except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.p
         delete_subscription,
         dispatch_event,
         list_subscriptions,
+        send_webhook,
     )
 
 
@@ -184,10 +189,13 @@ class TestDispatchPreferences(PushTestCase):
 
 
 class TestDispatchSendersAndCredentials(PushTestCase):
-    def test_webhook_without_sender_returns_no_sender_error_not_silent_ok(self):
+    def test_explicit_sender_missing_channel_returns_no_sender_error_not_silent_ok(self):
+        # sender=None 时内置真实发送端会真 POST（见 TestRealSenders），
+        # “无 sender”只发生在显式 sender 里没有该通道时，仍不静默成功。
         self._sub(channel="webhook")
         result = dispatch_event(
-            self.conn, "completed", "proj-a", {"task_id": "t-1"}, sender=None)
+            self.conn, "completed", "proj-a", {"task_id": "t-1"},
+            sender={})
         self.assertEqual(result["sent"], [])
         self.assertEqual(len(result["errors"]), 1)
         self.assertIn("no sender", result["errors"][0]["reason"])
@@ -300,5 +308,141 @@ class TestDispatchIsolation(PushTestCase):
             len(list_subscriptions(self.conn, "proj-b", "user-1")), 1)
 
 
+class _RecordingHandler(http.server.BaseHTTPRequestHandler):
+    """本地回环接收器：记录每次 POST 的头与 body，按类属性回状态。"""
+    requests = []
+    status = 200
+
+    def do_POST(self):
+        length = int(self.headers.get("Content-Length") or 0)
+        body = self.rfile.read(length) if length else b""
+        _RecordingHandler.requests.append({
+            "path": self.path,
+            "content_type": self.headers.get("Content-Type"),
+            "body": body,
+        })
+        self.send_response(_RecordingHandler.status)
+        self.send_header("Content-Length", "2")
+        self.end_headers()
+        self.wfile.write(b"{}")
+
+    def log_message(self, *args):
+        pass
+
+
+class TestRealSenders(PushTestCase):
+    """内置真实发送端：对本地回环服务器真 POST，断言方法/头/body；
+    不连外网。"""
+
+    def setUp(self):
+        super().setUp()
+        _RecordingHandler.requests = []
+        _RecordingHandler.status = 200
+        self.server = http.server.ThreadingHTTPServer(
+            ("127.0.0.1", 0), _RecordingHandler)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        super().tearDown()
+
+    def _url(self, path="/hook"):
+        return "http://127.0.0.1:%d%s" % (self.port, path)
+
+    def test_webhook_posts_json_via_default_sender(self):
+        create_subscription(
+            self.conn, "proj-a", "user-1", "webhook", self._url(),
+            "instant")
+        result = dispatch_event(
+            self.conn, "completed", "proj-a", {"task_id": "t-1"},
+            sender=None)
+        self.assertEqual(len(result["sent"]), 1)
+        self.assertEqual(result["errors"], [])
+        self.assertEqual(len(_RecordingHandler.requests), 1)
+        req = _RecordingHandler.requests[0]
+        self.assertEqual(req["content_type"], "application/json")
+        body = json.loads(req["body"].decode("utf-8"))
+        self.assertEqual(body["event_type"], "completed")
+        self.assertEqual(body["project_id"], "proj-a")
+        self.assertEqual(body["task_id"], "t-1")
+        self.assertEqual(body["payload"], {"task_id": "t-1"})
+
+    def test_webhook_non_2xx_is_error_not_sent(self):
+        _RecordingHandler.status = 500
+        create_subscription(
+            self.conn, "proj-a", "user-1", "webhook", self._url(),
+            "instant")
+        result = dispatch_event(
+            self.conn, "failed", "proj-a", {"task_id": "t-1"},
+            sender=None)
+        self.assertEqual(result["sent"], [])
+        self.assertEqual(len(result["errors"]), 1)
+        self.assertIn("delivery failed", result["errors"][0]["reason"])
+        self.assertEqual(len(_RecordingHandler.requests), 1)
+
+    def test_webhook_non_http_target_is_error(self):
+        create_subscription(
+            self.conn, "proj-a", "user-1", "webhook",
+            "ftp://example.invalid/hook", "instant")
+        with self.assertRaises(ValueError):
+            send_webhook({
+                "event_type": "completed", "project_id": "proj-a",
+                "payload": {"task_id": "t-1"},
+                "channel": "webhook",
+                "target": "ftp://example.invalid/hook",
+                "user_id": "user-1", "subscription_id": "s-1"})
+        result = dispatch_event(
+            self.conn, "completed", "proj-a", {"task_id": "t-1"},
+            sender=None)
+        self.assertEqual(result["sent"], [])
+        self.assertIn("http(s) URL", result["errors"][0]["reason"])
+
+    def test_slack_posts_text_with_task_and_error(self):
+        create_subscription(
+            self.conn, "proj-a", "user-1", "slack", self._url(),
+            "instant")
+        result = dispatch_event(
+            self.conn, "failed", "proj-a",
+            {"task_id": "t-9", "error": "boom happened"},
+            sender=None)
+        self.assertEqual(len(result["sent"]), 1)
+        body = json.loads(_RecordingHandler.requests[0]["body"].decode())
+        self.assertIn("t-9", body["text"])
+        self.assertIn("failed", body["text"])
+        self.assertIn("boom happened", body["text"])
+
+    def test_webpush_still_missing_credential_with_default_sender(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=None)
+        self.assertEqual(result["sent"], [])
+        self.assertIn("missing credential", result["errors"][0]["reason"])
+        self.assertEqual(_RecordingHandler.requests, [])
+
+    def test_explicit_sender_wins_over_default(self):
+        create_subscription(
+            self.conn, "proj-a", "user-1", "webhook", self._url(),
+            "instant")
+        calls = []
+
+        def fake(message):
+            calls.append(message)
+
+        result = dispatch_event(
+            self.conn, "completed", "proj-a", {"task_id": "t-1"},
+            sender={"webhook": fake})
+        self.assertEqual(len(calls), 1)
+        self.assertEqual(len(result["sent"]), 1)
+        self.assertEqual(_RecordingHandler.requests, [])
+
+
 if __name__ == "__main__":
     unittest.main()

cloud/tests/test_server_oauth_push.py

@@ -126,7 +126,7 @@ class OAuthPushEndpointsTestCase(unittest.TestCase):
         pid = self.project["id"]
         status, sub = self._req(
             "POST", "/projects/%s/push/subscriptions" % pid,
-            {"channel": "webhook", "target": "https://example.invalid/hook",
+            {"channel": "webhook", "target": "http://127.0.0.1:1/hook",
              "preference": "instant"},
             self.owner_key)
         self.assertEqual(status, 201, sub)
@@ -135,15 +135,17 @@ class OAuthPushEndpointsTestCase(unittest.TestCase):
             key=self.owner_key)
         self.assertEqual(status, 200)
         self.assertEqual(len(body["subscriptions"]), 1)
-        # HTTP 层无注入 sender：分发应报“无 sender”，不静默成功。
+        # 内置真实发送端会真 POST：目标不可达时分发应报投递失败
+        # （不是“无 sender”，也不是静默成功），sent 为空。
         status, body = self._req(
             "POST", "/admin/push/dispatch",
             {"event_type": "failed", "project_id": pid,
              "payload": {"task_id": "t1"}},
             self.owner_key)
         self.assertEqual(status, 200)
+        self.assertEqual(body["sent"], [])
         self.assertEqual(len(body["errors"]), 1)
-        self.assertIn("no sender", body["errors"][0]["reason"])
+        self.assertIn("delivery failed", body["errors"][0]["reason"])
         # outsider 不能往别人项目里订阅/分发。
         status, _ = self._req(
             "POST", "/projects/%s/push/subscriptions" % pid,

cloud/tests/test_worker.py

@@ -126,6 +126,103 @@ class WorkerTestCase(unittest.TestCase):
         # B 的依赖未完成，始终不可派，保持 queued。
         self.assertEqual(_status(self.db_path, "B"), "queued")
 
+    # -- 终态推送 --------------------------------------------------------
+    def test_notify_on_completed_and_failed(self):
+        """任务终态自动通知订阅：completed 与 failed 各发一次真 POST
+        到本地回环 webhook，body 含事件类型与任务 id；取消
+        （interrupted）不在事件集合里，不通知（见下个测试）。"""
+        import http.server
+        import threading
+
+        requests = []
+
+        class Handler(http.server.BaseHTTPRequestHandler):
+            def do_POST(self):
+                length = int(self.headers.get("Content-Length") or 0)
+                body = self.rfile.read(length) if length else b""
+                requests.append(json.loads(body.decode("utf-8")))
+                self.send_response(200)
+                self.send_header("Content-Length", "2")
+                self.end_headers()
+                self.wfile.write(b"{}")
+
+            def log_message(self, *args):
+                pass
+
+        server = http.server.ThreadingHTTPServer(
+            ("127.0.0.1", 0), Handler)
+        thread = threading.Thread(
+            target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            from cloud.push import create_subscription
+            url = "http://127.0.0.1:%d/hook" % server.server_address[1]
+            create_subscription(
+                self.db_path, "p1", "user-1", "webhook", url, "instant")
+            _insert_task(self.db_path, "A")
+            _insert_task(self.db_path, "B")
+
+            def failing(cmd):
+                if "dispatch" in cmd and cmd[cmd.index("--name") + 1] == "B":
+                    return SimpleNamespace(returncode=1, stdout="boom",
+                                           stderr="")
+                return _completed_runner(cmd)
+
+            worker = self._worker(runner=failing)
+            results = worker.run_all()
+            self.assertEqual(
+                sorted(r["status"] for r in results),
+                ["completed", "failed"])
+            by_task = {b["task_id"]: b for b in requests}
+            self.assertEqual(set(by_task), {"A", "B"})
+            self.assertEqual(by_task["A"]["event_type"], "completed")
+            self.assertEqual(by_task["B"]["event_type"], "failed")
+            self.assertIn("boom", json.dumps(by_task["B"]["payload"]))
+        finally:
+            server.shutdown()
+            server.server_close()
+            thread.join(timeout=5)
+
+    def test_cancel_does_not_notify(self):
+        """取消只标 interrupted，不在推送事件集合里，不发通知。"""
+        import http.server
+        import threading
+
+        requests = []
+
+        class Handler(http.server.BaseHTTPRequestHandler):
+            def do_POST(self):
+                length = int(self.headers.get("Content-Length") or 0)
+                self.rfile.read(length) if length else b""
+                requests.append(True)
+                self.send_response(200)
+                self.send_header("Content-Length", "2")
+                self.end_headers()
+                self.wfile.write(b"{}")
+
+            def log_message(self, *args):
+                pass
+
+        server = http.server.ThreadingHTTPServer(
+            ("127.0.0.1", 0), Handler)
+        thread = threading.Thread(
+            target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            from cloud.push import create_subscription
+            url = "http://127.0.0.1:%d/hook" % server.server_address[1]
+            create_subscription(
+                self.db_path, "p1", "user-1", "webhook", url, "instant")
+            _insert_task(self.db_path, "A")
+            worker = self._worker(runner=_completed_runner)
+            self.assertTrue(worker.cancel("A"))
+            self.assertEqual(worker.run_all(), [])
+            self.assertEqual(requests, [])
+        finally:
+            server.shutdown()
+            server.server_close()
+            thread.join(timeout=5)
+
     # -- 取消 ------------------------------------------------------------
     def test_cancel_queued_task_marks_interrupted(self):
         _insert_task(self.db_path, "A")

cloud/worker.py

@@ -640,6 +640,34 @@ class Worker:
         except Exception:
             pass
 
+    def _notify_task_finished(self, task_id, project_id, event_type,
+                              error=None):
+        """任务终态推送：best-effort，绝不因通知失败影响任务状态。
+
+        只发 completed/failed（push.ALLOWED_PUSH_EVENTS 的子集；
+        interrupted 不在事件集合里，取消不通知）。无订阅时
+        dispatch 直接返回空 sent，不联网。同步发送、单次 10 秒
+        超时：慢订阅会拖住 worker 循环最多几秒，换取投递结果
+        可观测（失败记日志，不静默丢）；后台队列是后续优化，
+        不在本轮。
+        """
+        try:
+            from cloud.push import dispatch_event
+            result = dispatch_event(
+                self.db_path, event_type, project_id,
+                {"task_id": task_id, "status": event_type,
+                 "executor": self._executor_name(), "error": error})
+            if result.get("errors"):
+                log_event(self.logger, "worker_notify_failed",
+                          task_id=task_id, project_id=project_id,
+                          status=event_type,
+                          message="notify errors: %s" % (result["errors"],))
+        except Exception as exc:  # noqa: BLE001 -- 通知失败绝不炸任务
+            log_event(self.logger, "worker_notify_failed",
+                      task_id=task_id, project_id=project_id,
+                      status=event_type,
+                      message="notify exception: %s" % exc)
+
     # -- 认领 ----------------------------------------------------------
     def _budget_exceeded(self):
         return budget_exceeded(self.usage_turns, self.max_turns,
@@ -1346,6 +1374,8 @@ class Worker:
             self._audit(task["project_id"], "worker_missing_key", error)
             log_event(self.logger, "worker_missing_key", task_id=task_id,
                       project_id=task["project_id"], status="failed")
+            self._notify_task_finished(
+                task_id, task["project_id"], "failed", error=error)
             return {"dispatched": True, "task_id": task_id,
                     "host_name": self.executor._host_name(task)
                     if hasattr(self.executor, "_host_name") else task_id,
@@ -1365,6 +1395,8 @@ class Worker:
             self._audit(task["project_id"], "worker_repo_failed", repo_error)
             log_event(self.logger, "worker_repo_failed", task_id=task_id,
                       project_id=task["project_id"], status="failed")
+            self._notify_task_finished(
+                task_id, task["project_id"], "failed", error=repo_error)
             return {"dispatched": True, "task_id": task_id,
                     "host_name": self.executor._host_name(task)
                     if hasattr(self.executor, "_host_name") else task_id,
@@ -1537,6 +1569,9 @@ class Worker:
                       else "worker_failed",
                       task_id=task_id, project_id=task["project_id"],
                       status=final)
+            if final in ("completed", "failed"):
+                self._notify_task_finished(
+                    task_id, task["project_id"], final, error=error)
         else:
             # 已被取消/回收：DB 状态以 interrupted 为准，不再
             # 补一条 worker_completed/failed 审计误导排查。