From 71fbfc9c52980b1eedc535cd6792a08fd13dbfda Mon Sep 17 00:00:00 2001 From: rookiestar28 Date: Tue, 28 Apr 2026 19:24:48 +0800 Subject: [PATCH] feat(scheduler): normalize delivery targets --- api/schedules.py | 21 +- docs/release/api_contract.md | 10 + services/scheduler/delivery_contract.py | 224 ++++++++++++++++ services/scheduler/models.py | 8 +- tests/test_f73_schedule_delivery_contract.py | 264 +++++++++++++++++++ 5 files changed, 522 insertions(+), 5 deletions(-) create mode 100644 services/scheduler/delivery_contract.py create mode 100644 tests/test_f73_schedule_delivery_contract.py diff --git a/api/schedules.py b/api/schedules.py index a691869..1fc540d 100644 --- a/api/schedules.py +++ b/api/schedules.py @@ -16,12 +16,20 @@ from typing import Optional # can silently import another pack's module and break auth/approval semantics. if __package__ and "." in __package__: from ..services.aiohttp_compat import import_aiohttp_web + from ..services.scheduler.delivery_contract import ( + DeliveryContractError, + normalize_schedule_delivery, + ) from ..services.scheduler.models import Schedule, TriggerType from ..services.scheduler.storage import get_schedule_store from ..services.templates import is_template_allowed from ..services.webhook_auth import AuthError else: # pragma: no cover (test-only import mode) from services.aiohttp_compat import import_aiohttp_web # type: ignore + from services.scheduler.delivery_contract import ( # type: ignore + DeliveryContractError, + normalize_schedule_delivery, + ) from services.scheduler.models import Schedule, TriggerType # type: ignore from services.scheduler.storage import get_schedule_store # type: ignore from services.templates import is_template_allowed # type: ignore @@ -31,6 +39,10 @@ logger = logging.getLogger("ComfyUI-OpenClaw.api.schedules") web = import_aiohttp_web() +def _delivery_error_response(exc: DeliveryContractError) -> web.Response: + return web.json_response({"error": str(exc), "code": exc.code}, status=400) + + def _get_scheduler_runner(): if __package__ and "." in __package__: from ..services.scheduler.runner import get_scheduler_runner @@ -157,6 +169,8 @@ class ScheduleHandlers: timezone=data.get("timezone", "local"), enabled=data.get("enabled", True), ) + except DeliveryContractError as e: + return _delivery_error_response(e) except ValueError as e: return web.json_response({"error": str(e)}, status=400) @@ -209,7 +223,10 @@ class ScheduleHandlers: if "inputs" in data: existing.inputs = data["inputs"] if "delivery" in data: - existing.delivery = data["delivery"] + try: + existing.delivery = normalize_schedule_delivery(data["delivery"]) + except DeliveryContractError as e: + return _delivery_error_response(e) if "timezone" in data: existing.timezone = data["timezone"] if "enabled" in data: @@ -218,6 +235,8 @@ class ScheduleHandlers: # Re-validate try: existing.validate() + except DeliveryContractError as e: + return _delivery_error_response(e) except ValueError as e: return web.json_response({"error": str(e)}, status=400) diff --git a/docs/release/api_contract.md b/docs/release/api_contract.md index d3ebeb6..a29cc6f 100644 --- a/docs/release/api_contract.md +++ b/docs/release/api_contract.md @@ -185,6 +185,16 @@ Model-manager contract notes: | `POST` | `/approvals/{id}/approve` | Approve a pending request. | | `POST` | `/approvals/{id}/reject` | Reject a pending request. | +Schedule `delivery` is normalized before persistence. Supported fields are +`platform`, `target_id` (legacy aliases such as `channel_id` are accepted), +`thread_id` (aliases such as `thread_ts`, `topic_id`, and `message_thread_id` +are accepted), `workspace_id`, `account_id`, `mode`, and `failure_alert`. +Omitting `delivery` on update preserves the existing target, `delivery: null` +clears it, and `{"enabled": false}` or `{"mode": "none"}` records explicit +no-delivery. Invalid delivery targets are rejected before persistence with +bounded codes: `delivery_malformed`, `delivery_ambiguous`, or +`delivery_unsupported`. + ### 1.6 Bridge (Sidecar) **Base Path**: `/bridge/` diff --git a/services/scheduler/delivery_contract.py b/services/scheduler/delivery_contract.py new file mode 100644 index 0000000..11ffdaf --- /dev/null +++ b/services/scheduler/delivery_contract.py @@ -0,0 +1,224 @@ +""" +Canonical scheduler delivery contract. + +Schedules persist a stable delivery shape so CRUD, manual run, and background +execution do not depend on loose connector-specific dictionaries. +""" + +from __future__ import annotations + +import re +from typing import Any, Dict, Iterable, Optional + +from ..chatops.targets import TargetValidationError, parse_target +from ..chatops.transport_contract import TransportType + +DELIVERY_MALFORMED = "delivery_malformed" +DELIVERY_AMBIGUOUS = "delivery_ambiguous" +DELIVERY_UNSUPPORTED = "delivery_unsupported" + +SUPPORTED_PLATFORMS = { + "custom", + "discord", + "feishu", + "lark", + "slack", + "telegram", + "webhook", +} + +_SAFE_ID_PATTERN = re.compile(r"^[a-zA-Z0-9_.:@/-]{1,200}$") +_FEISHU_TARGET_PATTERN = re.compile(r"^[a-zA-Z0-9_-]{1,128}$") +_FEISHU_THREAD_PATTERN = re.compile(r"^[a-zA-Z0-9_.-]{1,128}$") +_WEBHOOK_SAFE_TARGET_PATTERN = re.compile(r"^[a-zA-Z0-9_.-]{1,128}$") + + +class DeliveryContractError(ValueError): + """Raised when schedule delivery cannot be normalized safely.""" + + def __init__(self, code: str, message: str): + super().__init__(message) + self.code = code + + +def normalize_schedule_delivery(value: Any) -> Optional[Dict[str, Any]]: + """ + Normalize loose schedule delivery data into the canonical persisted contract. + + Patch semantics are handled by callers: + - omitted delivery: caller preserves existing value + - ``None``: clear delivery + - ``{"enabled": false}`` or ``{"mode": "none"}``: explicit no-delivery + """ + if value is None: + return None + + if not isinstance(value, dict): + raise DeliveryContractError(DELIVERY_MALFORMED, "delivery must be an object") + + raw = dict(value) + if _is_explicit_no_delivery(raw): + return {"enabled": False, "mode": "none"} + + platform = _resolve_platform(raw) + target_id = _resolve_alias( + raw, ("target_id", "channel_id", "chat_id", "room_id", "callback_url", "url") + ) + thread_id = _resolve_alias( + raw, ("thread_id", "thread_ts", "message_thread_id", "topic_id") + ) + workspace_id = _resolve_alias(raw, ("workspace_id", "team_id")) + account_id = _resolve_alias(raw, ("account_id",)) + mode = _normalize_mode(raw.get("mode", "reply")) + failure_alert = _normalize_bool(raw.get("failure_alert", True), "failure_alert") + + if not target_id: + raise DeliveryContractError( + DELIVERY_MALFORMED, "delivery.target_id is required" + ) + + _validate_platform_target(platform, target_id, thread_id, mode) + _validate_optional_safe_id(workspace_id, "workspace_id") + _validate_optional_safe_id(account_id, "account_id") + + normalized: Dict[str, Any] = { + "enabled": True, + "platform": platform, + "target_id": target_id, + } + if thread_id: + normalized["thread_id"] = thread_id + if workspace_id: + normalized["workspace_id"] = workspace_id + if account_id: + normalized["account_id"] = account_id + normalized["mode"] = mode + normalized["failure_alert"] = failure_alert + return normalized + + +def _is_explicit_no_delivery(raw: Dict[str, Any]) -> bool: + enabled = raw.get("enabled") + if isinstance(enabled, bool) and not enabled: + return True + mode = raw.get("mode") + return isinstance(mode, str) and mode.strip().lower() == "none" + + +def _resolve_platform(raw: Dict[str, Any]) -> str: + platform = _resolve_alias(raw, ("platform", "transport")) + if not platform and "url" in raw: + platform = "webhook" + if not platform: + raise DeliveryContractError(DELIVERY_MALFORMED, "delivery.platform is required") + + platform = platform.lower() + if platform not in SUPPORTED_PLATFORMS: + raise DeliveryContractError( + DELIVERY_UNSUPPORTED, f"unsupported delivery platform: {platform}" + ) + return "feishu" if platform == "lark" else platform + + +def _resolve_alias(raw: Dict[str, Any], aliases: Iterable[str]) -> Optional[str]: + values: list[tuple[str, str]] = [] + for key in aliases: + if key not in raw: + continue + value = raw.get(key) + if value is None: + continue + if isinstance(value, bool): + raise DeliveryContractError( + DELIVERY_MALFORMED, f"delivery.{key} must be a string" + ) + text = str(value).strip() + if not text: + continue + values.append((key, text)) + + unique_values = {text for _, text in values} + if len(unique_values) > 1: + fields = ", ".join(key for key, _ in values) + raise DeliveryContractError( + DELIVERY_AMBIGUOUS, + f"delivery fields conflict: {fields}", + ) + return values[0][1] if values else None + + +def _normalize_mode(value: Any) -> str: + if value is None: + return "reply" + if not isinstance(value, str): + raise DeliveryContractError( + DELIVERY_MALFORMED, "delivery.mode must be a string" + ) + mode = value.strip().lower() + if mode == "none": + raise DeliveryContractError( + DELIVERY_MALFORMED, + "delivery.mode=none must be represented as explicit no-delivery", + ) + return mode or "reply" + + +def _normalize_bool(value: Any, field_name: str) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, str): + text = value.strip().lower() + if text in {"1", "true", "yes", "on"}: + return True + if text in {"0", "false", "no", "off"}: + return False + raise DeliveryContractError( + DELIVERY_MALFORMED, f"delivery.{field_name} must be boolean" + ) + + +def _validate_platform_target( + platform: str, target_id: str, thread_id: Optional[str], mode: str +) -> None: + if platform == "feishu": + if not _FEISHU_TARGET_PATTERN.match(target_id): + raise DeliveryContractError( + DELIVERY_MALFORMED, "invalid delivery.target_id for feishu" + ) + if thread_id and not _FEISHU_THREAD_PATTERN.match(thread_id): + raise DeliveryContractError( + DELIVERY_MALFORMED, "invalid delivery.thread_id for feishu" + ) + _validate_mode_for_scheduler(mode) + return + + if platform == "webhook": + if target_id.startswith(("http://", "https://")): + return + if not _WEBHOOK_SAFE_TARGET_PATTERN.match(target_id): + raise DeliveryContractError( + DELIVERY_MALFORMED, "invalid delivery.target_id for webhook" + ) + _validate_mode_for_scheduler(mode) + return + + try: + parse_target(TransportType(platform), target_id, thread_id, mode) + except (TargetValidationError, ValueError) as exc: + raise DeliveryContractError(DELIVERY_MALFORMED, str(exc)) from exc + + +def _validate_mode_for_scheduler(mode: str) -> None: + if mode not in {"reply", "new_thread", "dm", "broadcast"}: + raise DeliveryContractError( + DELIVERY_MALFORMED, f"invalid delivery.mode: {mode}" + ) + + +def _validate_optional_safe_id(value: Optional[str], field_name: str) -> None: + if value is None: + return + if not _SAFE_ID_PATTERN.match(value): + raise DeliveryContractError( + DELIVERY_MALFORMED, f"invalid delivery.{field_name}" + ) diff --git a/services/scheduler/models.py b/services/scheduler/models.py index da92e07..260f2f9 100644 --- a/services/scheduler/models.py +++ b/services/scheduler/models.py @@ -11,6 +11,8 @@ from datetime import datetime, timezone from enum import Enum from typing import Any, Dict, List, Optional +from .delivery_contract import normalize_schedule_delivery + logger = logging.getLogger("ComfyUI-OpenClaw.services.scheduler") @@ -71,6 +73,7 @@ class Schedule: def __post_init__(self): """Validation after initialization.""" + self.delivery = normalize_schedule_delivery(self.delivery) self.validate() def validate(self) -> None: @@ -108,10 +111,7 @@ class Schedule: # Delivery validation (if present) if self.delivery: - if "url" in self.delivery: - url = self.delivery["url"] - if not url.startswith(("http://", "https://")): - raise ValueError("delivery.url must be http(s)") + self.delivery = normalize_schedule_delivery(self.delivery) if self.compute_error_count < 0: raise ValueError("compute_error_count must be >= 0") diff --git a/tests/test_f73_schedule_delivery_contract.py b/tests/test_f73_schedule_delivery_contract.py new file mode 100644 index 0000000..b7e98c2 --- /dev/null +++ b/tests/test_f73_schedule_delivery_contract.py @@ -0,0 +1,264 @@ +import json +import unittest +from unittest.mock import AsyncMock, MagicMock, patch + +try: + from aiohttp import web # noqa: F401 + + AIOHTTP_AVAILABLE = True +except ModuleNotFoundError: + AIOHTTP_AVAILABLE = False + + +class TestScheduleDeliveryContractModel(unittest.TestCase): + def test_legacy_delivery_dict_migrates_to_canonical_contract(self): + from services.scheduler.models import Schedule, TriggerType + + schedule = Schedule.from_dict( + { + "schedule_id": "sched_legacy_delivery", + "name": "Legacy delivery", + "template_id": "tmpl_1", + "trigger_type": TriggerType.INTERVAL.value, + "interval_sec": 300, + "delivery": { + "platform": "telegram", + "channel_id": "-1001234567890", + "message_thread_id": 456, + }, + } + ) + + self.assertEqual( + schedule.delivery, + { + "enabled": True, + "platform": "telegram", + "target_id": "-1001234567890", + "thread_id": "456", + "mode": "reply", + "failure_alert": True, + }, + ) + + def test_explicit_no_delivery_contract_is_preserved(self): + from services.scheduler.models import Schedule, TriggerType + + schedule = Schedule( + schedule_id="sched_no_delivery", + name="No delivery", + template_id="tmpl_1", + trigger_type=TriggerType.INTERVAL, + interval_sec=300, + delivery={"mode": "none"}, + ) + + self.assertEqual(schedule.delivery, {"enabled": False, "mode": "none"}) + + +@unittest.skipIf(not AIOHTTP_AVAILABLE, "aiohttp not available") +class TestScheduleDeliveryContractApi(unittest.IsolatedAsyncioTestCase): + async def test_create_normalizes_telegram_topic_delivery_before_persistence(self): + from api.schedules import ScheduleHandlers + + store = MagicMock() + store.add.return_value = True + + with patch("api.schedules.get_schedule_store", return_value=store): + handlers = ScheduleHandlers(template_checker=lambda template_id: True) + + request = AsyncMock() + request.json = AsyncMock( + return_value={ + "name": "Topic delivery", + "template_id": "tmpl_1", + "trigger_type": "interval", + "interval_sec": 300, + "delivery": { + "platform": "telegram", + "channel_id": "-1001234567890", + "message_thread_id": 456, + }, + } + ) + + response = await handlers.create_schedule(request) + + self.assertEqual(response.status, 201) + created_schedule = store.add.call_args.args[0] + self.assertEqual( + created_schedule.delivery, + { + "enabled": True, + "platform": "telegram", + "target_id": "-1001234567890", + "thread_id": "456", + "mode": "reply", + "failure_alert": True, + }, + ) + body = json.loads(response.body) + self.assertEqual(body["schedule"]["delivery"], created_schedule.delivery) + + async def test_update_delivery_patch_semantics_preserve_clear_and_no_delivery(self): + from api.schedules import ScheduleHandlers + from services.scheduler.models import Schedule, TriggerType + + existing = Schedule( + schedule_id="sched_patch", + name="Patch delivery", + template_id="tmpl_1", + trigger_type=TriggerType.INTERVAL, + interval_sec=300, + delivery={ + "platform": "telegram", + "target_id": "-1001234567890", + "thread_id": "456", + }, + ) + store = MagicMock() + store.get.return_value = existing + store.update.return_value = True + + with patch("api.schedules.get_schedule_store", return_value=store): + handlers = ScheduleHandlers(template_checker=lambda template_id: True) + + preserve_request = AsyncMock() + preserve_request.match_info = {"schedule_id": "sched_patch"} + preserve_request.json = AsyncMock(return_value={"name": "Renamed"}) + + response = await handlers.update_schedule(preserve_request) + self.assertEqual(response.status, 200) + self.assertEqual(existing.delivery["thread_id"], "456") + + clear_request = AsyncMock() + clear_request.match_info = {"schedule_id": "sched_patch"} + clear_request.json = AsyncMock(return_value={"delivery": None}) + + response = await handlers.update_schedule(clear_request) + self.assertEqual(response.status, 200) + self.assertIsNone(existing.delivery) + + no_delivery_request = AsyncMock() + no_delivery_request.match_info = {"schedule_id": "sched_patch"} + no_delivery_request.json = AsyncMock( + return_value={"delivery": {"enabled": False}} + ) + + response = await handlers.update_schedule(no_delivery_request) + self.assertEqual(response.status, 200) + self.assertEqual(existing.delivery, {"enabled": False, "mode": "none"}) + + async def test_malformed_delivery_target_returns_bounded_error_code(self): + from api.schedules import ScheduleHandlers + + store = MagicMock() + + with patch("api.schedules.get_schedule_store", return_value=store): + handlers = ScheduleHandlers(template_checker=lambda template_id: True) + + request = AsyncMock() + request.json = AsyncMock( + return_value={ + "name": "Bad delivery", + "template_id": "tmpl_1", + "trigger_type": "interval", + "interval_sec": 300, + "delivery": { + "platform": "telegram", + "target_id": "-1001234567890", + "thread_id": "not-a-topic", + }, + } + ) + + response = await handlers.create_schedule(request) + + self.assertEqual(response.status, 400) + body = json.loads(response.body) + self.assertEqual(body["code"], "delivery_malformed") + store.add.assert_not_called() + + async def test_conflicting_delivery_target_aliases_return_bounded_error_code(self): + from api.schedules import ScheduleHandlers + + store = MagicMock() + + with patch("api.schedules.get_schedule_store", return_value=store): + handlers = ScheduleHandlers(template_checker=lambda template_id: True) + + request = AsyncMock() + request.json = AsyncMock( + return_value={ + "name": "Ambiguous delivery", + "template_id": "tmpl_1", + "trigger_type": "interval", + "interval_sec": 300, + "delivery": { + "platform": "feishu", + "target_id": "oc_abc123", + "channel_id": "oc_other", + "workspace_id": "tenant-alpha", + }, + } + ) + + response = await handlers.create_schedule(request) + + self.assertEqual(response.status, 400) + body = json.loads(response.body) + self.assertEqual(body["code"], "delivery_ambiguous") + store.add.assert_not_called() + + +class TestScheduleDeliveryContractRunner(unittest.TestCase): + def test_execute_schedule_submits_normalized_delivery(self): + from services.scheduler.models import Schedule, TriggerType + from services.scheduler.runner import SchedulerRunner + + submitted = {} + + async def submit_fn(**kwargs): + submitted.update(kwargs) + return {"prompt_id": "prompt_1"} + + schedule = Schedule( + schedule_id="sched_runner_delivery", + name="Runner delivery", + template_id="tmpl_1", + trigger_type=TriggerType.INTERVAL, + interval_sec=300, + delivery={ + "platform": "slack", + "channel_id": "C123456789", + "thread_ts": "1234567890.123456", + "workspace_id": "T123456789", + }, + ) + runner = SchedulerRunner(submit_fn=submit_fn, tick_interval=10) + runner._store = MagicMock() + runner._store.update.return_value = True + + with patch("services.scheduler.runner.get_run_history") as mock_history: + history = MagicMock() + history.is_processed.return_value = False + mock_history.return_value = history + + runner._execute_schedule(schedule, 1000.0) + + self.assertEqual( + submitted["delivery"], + { + "enabled": True, + "platform": "slack", + "target_id": "C123456789", + "thread_id": "1234567890.123456", + "workspace_id": "T123456789", + "mode": "reply", + "failure_alert": True, + }, + ) + + +if __name__ == "__main__": + unittest.main()