diff --git a/cloud/README.md b/cloud/README.md index 483c021..de1f8a9 100644 --- a/cloud/README.md +++ b/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 端点” | diff --git a/cloud/push.py b/cloud/push.py index ff41c34..9002c17 100644 --- a/cloud/push.py +++ b/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( diff --git a/cloud/server.py b/cloud/server.py index e58f5a8..29da652 100644 --- a/cloud/server.py +++ b/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) diff --git a/cloud/tests/test_push.py b/cloud/tests/test_push.py index 14fedef..fcda543 100644 --- a/cloud/tests/test_push.py +++ b/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() diff --git a/cloud/tests/test_server_oauth_push.py b/cloud/tests/test_server_oauth_push.py index 0a5ff45..242d2e1 100644 --- a/cloud/tests/test_server_oauth_push.py +++ b/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, diff --git a/cloud/tests/test_worker.py b/cloud/tests/test_worker.py index 1b82e28..e1f9075 100644 --- a/cloud/tests/test_worker.py +++ b/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") diff --git a/cloud/worker.py b/cloud/worker.py index 7f67f86..87774ba 100644 --- a/cloud/worker.py +++ b/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 审计误导排查。