diff --git a/frontend-web/src/core/scheduled-tasks/api.ts b/frontend-web/src/core/scheduled-tasks/api.ts index 88ef55a9..6d8a1942 100644 --- a/frontend-web/src/core/scheduled-tasks/api.ts +++ b/frontend-web/src/core/scheduled-tasks/api.ts @@ -2,10 +2,12 @@ import { apiFetch } from "@/core/api"; import { getBackendBaseURL } from "@/core/config"; import type { + PublishedScheduledTask, ScheduledTask, ScheduledTaskDeliveryProfile, ScheduledTaskResultFileContent, ScheduledTaskRun, + ScheduledTaskSubscription, SchedulerThread, } from "./types"; @@ -214,3 +216,57 @@ export async function triggerScheduledTask(taskId: string) { }), ); } + +export async function listPublishedScheduledTasks(params: { limit?: number; offset?: number } = {}) { + const query = new URLSearchParams(); + if (params.limit != null) query.set("limit", String(params.limit)); + if (params.offset != null) query.set("offset", String(params.offset)); + const suffix = query.toString(); + return parseResponse( + await apiFetch(`${baseUrl()}/published${suffix ? `?${suffix}` : ""}`), + ); +} + +export async function publishScheduledTask(taskId: string) { + return parseResponse( + await apiFetch(`${baseUrl()}/${encodeURIComponent(taskId)}/publish`, { + method: "POST", + }), + ); +} + +export async function unpublishScheduledTask(taskId: string) { + return parseResponse( + await apiFetch(`${baseUrl()}/${encodeURIComponent(taskId)}/unpublish`, { + method: "POST", + }), + ); +} + +export async function subscribeScheduledTask(taskId: string, notifyElement: boolean) { + return parseResponse( + await apiFetch(`${baseUrl()}/${encodeURIComponent(taskId)}/subscribe`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ notify_element: notifyElement }), + }), + ); +} + +export async function updateScheduledTaskSubscription(taskId: string, notifyElement: boolean) { + return parseResponse( + await apiFetch(`${baseUrl()}/${encodeURIComponent(taskId)}/subscribe`, { + method: "PATCH", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ notify_element: notifyElement }), + }), + ); +} + +export async function unsubscribeScheduledTask(taskId: string) { + return parseResponse<{ success: boolean }>( + await apiFetch(`${baseUrl()}/${encodeURIComponent(taskId)}/subscribe`, { + method: "DELETE", + }), + ); +} diff --git a/frontend-web/src/core/scheduled-tasks/hooks.ts b/frontend-web/src/core/scheduled-tasks/hooks.ts index c01bdab8..9851af38 100644 --- a/frontend-web/src/core/scheduled-tasks/hooks.ts +++ b/frontend-web/src/core/scheduled-tasks/hooks.ts @@ -7,13 +7,19 @@ import { getSchedulerThread, getScheduledTaskDeliveryProfile, getScheduledTaskRun, + listPublishedScheduledTasks, listScheduledTaskRuns, listScheduledTasks, pauseScheduledTask, + publishScheduledTask, resumeScheduledTask, + subscribeScheduledTask, triggerScheduledTask, + unpublishScheduledTask, + unsubscribeScheduledTask, updateScheduledTaskDeliveryProfile, updateScheduledTask, + updateScheduledTaskSubscription, } from "./api"; export function useSchedulerThread() { @@ -58,6 +64,15 @@ export function useScheduledTaskDeliveryProfile() { }); } +export function usePublishedScheduledTasks(enabled: boolean = true) { + return useQuery({ + queryKey: ["scheduled-tasks", "published"], + queryFn: () => listPublishedScheduledTasks({ limit: 100 }), + enabled, + refetchOnWindowFocus: false, + }); +} + export function useScheduledTaskActions() { const queryClient = useQueryClient(); const invalidate = () => @@ -107,5 +122,27 @@ export function useScheduledTaskActions() { queryKey: ["scheduled-tasks", "delivery-profile"], }), }), + publish: useMutation({ + mutationFn: publishScheduledTask, + onSuccess: invalidate, + }), + unpublish: useMutation({ + mutationFn: unpublishScheduledTask, + onSuccess: invalidate, + }), + subscribe: useMutation({ + mutationFn: ({ taskId, notifyElement }: { taskId: string; notifyElement: boolean }) => + subscribeScheduledTask(taskId, notifyElement), + onSuccess: invalidate, + }), + updateSubscription: useMutation({ + mutationFn: ({ taskId, notifyElement }: { taskId: string; notifyElement: boolean }) => + updateScheduledTaskSubscription(taskId, notifyElement), + onSuccess: invalidate, + }), + unsubscribe: useMutation({ + mutationFn: unsubscribeScheduledTask, + onSuccess: invalidate, + }), }; } diff --git a/frontend-web/src/core/scheduled-tasks/types.ts b/frontend-web/src/core/scheduled-tasks/types.ts index 4ae42bf6..e8009dc1 100644 --- a/frontend-web/src/core/scheduled-tasks/types.ts +++ b/frontend-web/src/core/scheduled-tasks/types.ts @@ -1,3 +1,14 @@ +export type ScheduledTaskRole = "owner" | "subscriber"; + +export type ScheduledTaskSubscription = { + id: string; + task_id: string; + user_id: string; + notify_element: boolean; + created_at?: string | null; + updated_at?: string | null; +}; + export type ScheduledTask = { task_id: string; name: string; @@ -12,6 +23,20 @@ export type ScheduledTask = { source_agent_name?: string | null; execution_agent_name?: string | null; scheduler_thread_id: string; + published?: boolean; + published_at?: string | null; + role?: ScheduledTaskRole; + _subscription?: ScheduledTaskSubscription | null; +}; + +export type PublishedScheduledTask = { + task_id: string; + name: string; + prompt_summary: string; + schedule_text: string; + timezone: string; + published_at?: string | null; + owner_user_id: string; }; export type SchedulerThread = { diff --git a/frontend-web/src/pages/ScheduledTasksPage.tsx b/frontend-web/src/pages/ScheduledTasksPage.tsx index f489d650..a85c7b69 100644 --- a/frontend-web/src/pages/ScheduledTasksPage.tsx +++ b/frontend-web/src/pages/ScheduledTasksPage.tsx @@ -2,9 +2,15 @@ import { useEffect, useMemo, useState } from "react"; import { + BellOff, + BellRing, Download, Edit3, Eye, + Globe, + Heart, + HeartOff, + Lock, Pause, Play, Plus, @@ -21,6 +27,8 @@ import { Dialog, DialogContent, DialogFooter, DialogHeader, DialogTitle } from " import { Input } from "@/components/ui/input"; import { ScrollArea } from "@/components/ui/scroll-area"; import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select"; +import { Switch } from "@/components/ui/switch"; +import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs"; import { Textarea } from "@/components/ui/textarea"; import { WorkspaceBody, @@ -30,11 +38,13 @@ import { import { useAgents } from "@/core/agents"; import { downloadScheduledTaskRun, + type PublishedScheduledTask, type ScheduledTask, useScheduledTaskActions, useScheduledTaskDeliveryProfile, useScheduledTaskRuns, useScheduledTasks, + usePublishedScheduledTasks, } from "@/core/scheduled-tasks"; import { cn } from "@/lib/utils"; @@ -217,6 +227,9 @@ export default function ScheduledTasksPage() { const { data: deliveryProfile } = useScheduledTaskDeliveryProfile(); const { agents } = useAgents(); const actions = useScheduledTaskActions(); + const [activeTab, setActiveTab] = useState<"mine" | "plaza">("mine"); + const { data: published = [], isLoading: publishedLoading } = + usePublishedScheduledTasks(activeTab === "plaza"); const [selectedTaskId, setSelectedTaskId] = useState(null); const [elementUserId, setElementUserId] = useState(""); const [editing, setEditing] = useState(null); @@ -224,6 +237,8 @@ export default function ScheduledTasksPage() { const [form, setForm] = useState(emptyForm); const [scheduleConfig, setScheduleConfig] = useState(DEFAULT_SCHEDULE_CONFIG); const [runsTimeFilter, setRunsTimeFilter] = useState<"all" | "today" | "7d" | "30d">("all"); + const [subscribeTarget, setSubscribeTarget] = useState(null); + const [subscribeNotify, setSubscribeNotify] = useState(false); const sortedTasks = useMemo( () => @@ -377,95 +392,232 @@ export default function ScheduledTasksPage() { 保存 + setActiveTab(v as typeof activeTab)}> + + 我的任务 + 广场 + + -
- {isLoading && ( -
正在加载任务...
- )} - {!isLoading && sortedTasks.length === 0 && ( -
- 还没有定时任务。点击新建创建第一个任务。 -
- )} - {sortedTasks.map((task) => ( - + + + + + + ) : ( + <> + + + + )} +
+ + ); + })} + + ) : ( +
+ {publishedLoading && ( +
正在加载广场任务...
+ )} + {!publishedLoading && published.length === 0 && ( +
+ 广场上还没有发布的任务。 +
+ )} + {published.map((item) => { + const subscribed = tasks.some((t) => t.task_id === item.task_id); + return ( +
+
+
+
{item.name}
+
+ {item.schedule_text} · {item.timezone} +
+
+ {subscribed && ( + + 已收藏 + + )} +
+
+ {item.prompt_summary || "（无摘要）"} +
+
+
- - {task.enabled ? "启用" : "暂停"} - -
-
-
下次执行：{formatDate(task.next_run_at)}
-
上次执行：{formatDate(task.last_run_at)}
-
最近状态：{statusLabel(task.last_status)}
-
-
- - - - -
- - ))} - + ); + })} + + )} @@ -566,6 +718,75 @@ export default function ScheduledTasksPage() { + { + if (!open) { + setSubscribeTarget(null); + setSubscribeNotify(false); + } + }} + > + + + 收藏到我的任务 + +
+
+
{subscribeTarget?.name}
+
+ {subscribeTarget?.schedule_text} · {subscribeTarget?.timezone} +
+
+
+ {subscribeTarget?.prompt_summary || "（无摘要）"} +
+
+
+
同步接收 Element 通知
+
+ 开启后，每次执行结果将复制一份发送到你设置的 Element 用户。 + {!deliveryProfile?.element_user_id && ( + 你尚未配置 Element 用户 ID。 + )} +
+
+ +
+
+ + + + + + + !open && closeDialog()}> diff --git a/offline-backend-20260512/backend/app/gateway/routers/scheduled_tasks.py b/offline-backend-20260512/backend/app/gateway/routers/scheduled_tasks.py index 2895c6f6..617a52a0 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/scheduled_tasks.py +++ b/offline-backend-20260512/backend/app/gateway/routers/scheduled_tasks.py @@ -43,6 +43,10 @@ class ScheduledTaskDeliveryProfileUpdateRequest(BaseModel): element_user_id: str | None = None +class ScheduledTaskSubscribeRequest(BaseModel): + notify_element: bool = False + + async def _user_id(request: Request) -> str: user_id = await get_current_user(request) if not user_id: @@ -325,3 +329,63 @@ async def trigger_scheduled_task(task_id: str, request: Request) -> dict: if task is None: raise HTTPException(status_code=404, detail=f"Task {task_id} not found") return {"success": True, "task": task} + + +# --- Publish / subscribe --- + + +@router.get("/published") +async def list_published_scheduled_tasks(request: Request, limit: int = 50, offset: int = 0) -> list[dict]: + service = get_scheduled_task_service() + await _user_id(request) + return await service.list_published(limit=max(1, min(limit, 200)), offset=max(0, offset)) + + +@router.post("/{task_id}/publish") +async def publish_scheduled_task(task_id: str, request: Request) -> dict: + service = get_scheduled_task_service() + user_id = await _user_id(request) + task = await service.publish_task(user_id, task_id) + if task is None: + raise HTTPException(status_code=404, detail=f"Task {task_id} not found") + return task + + +@router.post("/{task_id}/unpublish") +async def unpublish_scheduled_task(task_id: str, request: Request) -> dict: + service = get_scheduled_task_service() + user_id = await _user_id(request) + task = await service.unpublish_task(user_id, task_id) + if task is None: + raise HTTPException(status_code=404, detail=f"Task {task_id} not found") + return task + + +@router.post("/{task_id}/subscribe") +async def subscribe_scheduled_task(task_id: str, body: ScheduledTaskSubscribeRequest, request: Request) -> dict: + service = get_scheduled_task_service() + user_id = await _user_id(request) + sub = await service.subscribe_task(user_id, task_id, notify_element=body.notify_element) + if sub is None: + raise HTTPException(status_code=404, detail="Task is not available for subscription") + return sub + + +@router.patch("/{task_id}/subscribe") +async def update_scheduled_task_subscription(task_id: str, body: ScheduledTaskSubscribeRequest, request: Request) -> dict: + service = get_scheduled_task_service() + user_id = await _user_id(request) + sub = await service.subscribe_task(user_id, task_id, notify_element=body.notify_element) + if sub is None: + raise HTTPException(status_code=404, detail="Subscription not found") + return sub + + +@router.delete("/{task_id}/subscribe") +async def delete_scheduled_task_subscription(task_id: str, request: Request) -> dict: + service = get_scheduled_task_service() + user_id = await _user_id(request) + deleted = await service.unsubscribe_task(user_id, task_id) + if not deleted: + raise HTTPException(status_code=404, detail="Subscription not found") + return {"success": True} diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260513_02_publish_subscribe_scheduled_tasks.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260513_02_publish_subscribe_scheduled_tasks.py new file mode 100644 index 00000000..0f98c349 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260513_02_publish_subscribe_scheduled_tasks.py @@ -0,0 +1,66 @@ +"""Publish/subscribe support for scheduled tasks. + +Revision ID: 20260513_02 +Revises: 20260513_01 +Create Date: 2026-05-13 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op +from sqlalchemy import inspect + + +revision = "20260513_02" +down_revision = "20260513_01" +branch_labels = None +depends_on = None + + +def _has_column(table_name: str, column_name: str) -> bool: + inspector = inspect(op.get_bind()) + return any(column["name"] == column_name for column in inspector.get_columns(table_name)) + + +def _has_table(table_name: str) -> bool: + inspector = inspect(op.get_bind()) + return table_name in inspector.get_table_names() + + +def upgrade() -> None: + if not _has_column("scheduled_tasks", "published"): + op.add_column( + "scheduled_tasks", + sa.Column("published", sa.Boolean(), nullable=False, server_default=sa.false()), + ) + op.create_index("ix_scheduled_tasks_published", "scheduled_tasks", ["published"]) + if not _has_column("scheduled_tasks", "published_at"): + op.add_column( + "scheduled_tasks", + sa.Column("published_at", sa.DateTime(timezone=True), nullable=True), + ) + + if not _has_table("scheduled_task_subscriptions"): + op.create_table( + "scheduled_task_subscriptions", + sa.Column("id", sa.String(64), primary_key=True), + sa.Column("task_id", sa.String(64), nullable=False), + sa.Column("user_id", sa.String(64), nullable=False), + sa.Column("notify_element", sa.Boolean(), nullable=False, server_default=sa.false()), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.UniqueConstraint("task_id", "user_id", name="uq_scheduled_task_subscriptions_task_user"), + ) + op.create_index("ix_scheduled_task_subscriptions_task_id", "scheduled_task_subscriptions", ["task_id"]) + op.create_index("ix_scheduled_task_subscriptions_user_id", "scheduled_task_subscriptions", ["user_id"]) + + +def downgrade() -> None: + if _has_table("scheduled_task_subscriptions"): + op.drop_table("scheduled_task_subscriptions") + if _has_column("scheduled_tasks", "published_at"): + op.drop_column("scheduled_tasks", "published_at") + if _has_column("scheduled_tasks", "published"): + op.drop_index("ix_scheduled_tasks_published", table_name="scheduled_tasks") + op.drop_column("scheduled_tasks", "published") diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py index 0839465d..2c428108 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py @@ -19,7 +19,13 @@ from deerflow.persistence.agents.model import AgentRow from deerflow.persistence.feedback.model import FeedbackRow from deerflow.persistence.models.run_event import RunEventRow from deerflow.persistence.run.model import RunRow -from deerflow.persistence.scheduled_tasks.model import ScheduledTaskDeliveryProfileRow, ScheduledTaskRow, ScheduledTaskRunRow, SchedulerThreadRow +from deerflow.persistence.scheduled_tasks.model import ( + ScheduledTaskDeliveryProfileRow, + ScheduledTaskRow, + ScheduledTaskRunRow, + ScheduledTaskSubscriptionRow, + SchedulerThreadRow, +) from deerflow.persistence.thread_meta.model import ThreadMetaRow from deerflow.persistence.user.model import UserRow @@ -31,6 +37,7 @@ __all__ = [ "ScheduledTaskDeliveryProfileRow", "ScheduledTaskRow", "ScheduledTaskRunRow", + "ScheduledTaskSubscriptionRow", "SchedulerThreadRow", "ThreadMetaRow", "UserRow", diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/__init__.py index c11f40fc..aecb39aa 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/__init__.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/__init__.py @@ -4,7 +4,13 @@ from __future__ import annotations from deerflow.persistence.scheduled_tasks.base import ScheduledTaskStore from deerflow.persistence.scheduled_tasks.memory import MemoryScheduledTaskStore -from deerflow.persistence.scheduled_tasks.model import ScheduledTaskDeliveryProfileRow, ScheduledTaskRow, ScheduledTaskRunRow, SchedulerThreadRow +from deerflow.persistence.scheduled_tasks.model import ( + ScheduledTaskDeliveryProfileRow, + ScheduledTaskRow, + ScheduledTaskRunRow, + ScheduledTaskSubscriptionRow, + SchedulerThreadRow, +) from deerflow.persistence.scheduled_tasks.sql import ScheduledTaskRepository __all__ = [ @@ -14,6 +20,7 @@ __all__ = [ "ScheduledTaskRow", "ScheduledTaskRunRow", "ScheduledTaskStore", + "ScheduledTaskSubscriptionRow", "SchedulerThreadRow", "make_scheduled_task_store", ] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/base.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/base.py index 4697538a..d3d89cbd 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/base.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/base.py @@ -63,3 +63,46 @@ class ScheduledTaskStore(ABC): @abstractmethod async def update_delivery_profile(self, user_id: str, data: dict[str, Any]) -> dict[str, Any]: raise NotImplementedError + + # --- Publish / subscribe --- + + @abstractmethod + async def get_task_any(self, task_id: str) -> dict[str, Any] | None: + """Return a task by id without restricting by owner. Used for publish/subscribe lookups.""" + raise NotImplementedError + + @abstractmethod + async def get_run_any(self, run_id: str) -> dict[str, Any] | None: + """Return a run by id without restricting by owner. Used for subscriber visibility.""" + raise NotImplementedError + + @abstractmethod + async def set_task_published(self, task_id: str, owner_user_id: str, *, published: bool) -> dict[str, Any] | None: + """Toggle published flag. Returns updated task or ``None`` if missing / not owned.""" + raise NotImplementedError + + @abstractmethod + async def list_published_tasks(self, *, limit: int = 50, offset: int = 0) -> list[dict[str, Any]]: + raise NotImplementedError + + @abstractmethod + async def list_subscribed_tasks(self, user_id: str) -> list[dict[str, Any]]: + """Return tasks the user has subscribed to. Each item carries ``_subscription`` dict.""" + raise NotImplementedError + + @abstractmethod + async def get_subscription(self, task_id: str, user_id: str) -> dict[str, Any] | None: + raise NotImplementedError + + @abstractmethod + async def upsert_subscription(self, task_id: str, user_id: str, *, notify_element: bool) -> dict[str, Any]: + raise NotImplementedError + + @abstractmethod + async def delete_subscription(self, task_id: str, user_id: str) -> bool: + raise NotImplementedError + + @abstractmethod + async def list_notify_subscribers(self, task_id: str) -> list[dict[str, Any]]: + """Subscribers with ``notify_element=True`` for the given task.""" + raise NotImplementedError diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/memory.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/memory.py index 094cd10f..c6eee52b 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/memory.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/memory.py @@ -2,7 +2,8 @@ from __future__ import annotations -from datetime import datetime +import uuid +from datetime import UTC, datetime from typing import Any from deerflow.persistence.scheduled_tasks.base import ScheduledTaskStore @@ -15,6 +16,7 @@ class MemoryScheduledTaskStore(ScheduledTaskStore): self._runs_by_key: set[tuple[str, str]] = set() self._runs: dict[str, dict[str, Any]] = {} self._profiles: dict[str, dict[str, Any]] = {} + self._subscriptions: dict[tuple[str, str], dict[str, Any]] = {} async def get_or_create_scheduler_thread(self, user_id: str, thread_id_factory) -> dict: if user_id not in self._threads: @@ -105,3 +107,73 @@ class MemoryScheduledTaskStore(ScheduledTaskStore): if "element_room_id" in data: profile["element_room_id"] = data["element_room_id"] or None return dict(profile) + + # --- Publish / subscribe --- + + async def get_task_any(self, task_id: str) -> dict[str, Any] | None: + task = self._tasks.get(task_id) + return dict(task) if task is not None else None + + async def get_run_any(self, run_id: str) -> dict[str, Any] | None: + run = self._runs.get(run_id) + return dict(run) if run is not None else None + + async def set_task_published(self, task_id: str, owner_user_id: str, *, published: bool) -> dict[str, Any] | None: + task = self._tasks.get(task_id) + if task is None or task["user_id"] != owner_user_id: + return None + task["published"] = bool(published) + task["published_at"] = datetime.now(UTC) if published else None + return dict(task) + + async def list_published_tasks(self, *, limit: int = 50, offset: int = 0) -> list[dict[str, Any]]: + tasks = [dict(task) for task in self._tasks.values() if task.get("published")] + tasks.sort(key=lambda t: (t.get("published_at") or t.get("created_at") or ""), reverse=True) + return tasks[max(0, offset) : max(0, offset) + max(1, limit)] + + async def list_subscribed_tasks(self, user_id: str) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + for (task_id, sub_user_id), sub in self._subscriptions.items(): + if sub_user_id != user_id: + continue + task = self._tasks.get(task_id) + if task is None: + continue + payload = dict(task) + payload["_subscription"] = dict(sub) + out.append(payload) + out.sort(key=lambda item: item["_subscription"].get("created_at") or "", reverse=True) + return out + + async def get_subscription(self, task_id: str, user_id: str) -> dict[str, Any] | None: + sub = self._subscriptions.get((task_id, user_id)) + return dict(sub) if sub is not None else None + + async def upsert_subscription(self, task_id: str, user_id: str, *, notify_element: bool) -> dict[str, Any]: + key = (task_id, user_id) + existing = self._subscriptions.get(key) + now = datetime.now(UTC) + if existing is None: + sub = { + "id": str(uuid.uuid4()), + "task_id": task_id, + "user_id": user_id, + "notify_element": bool(notify_element), + "created_at": now, + "updated_at": now, + } + self._subscriptions[key] = sub + return dict(sub) + existing["notify_element"] = bool(notify_element) + existing["updated_at"] = now + return dict(existing) + + async def delete_subscription(self, task_id: str, user_id: str) -> bool: + return self._subscriptions.pop((task_id, user_id), None) is not None + + async def list_notify_subscribers(self, task_id: str) -> list[dict[str, Any]]: + return [ + dict(sub) + for (tid, _uid), sub in self._subscriptions.items() + if tid == task_id and sub.get("notify_element") + ] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/model.py index 2b89ef44..b53070be 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/model.py @@ -39,6 +39,20 @@ class ScheduledTaskRow(Base): last_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) last_status: Mapped[str | None] = mapped_column(String(32)) execution_context_json: Mapped[dict] = mapped_column(JSON, default=dict) + published: Mapped[bool] = mapped_column(Boolean, default=False, index=True) + published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC)) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC)) + + +class ScheduledTaskSubscriptionRow(Base): + __tablename__ = "scheduled_task_subscriptions" + __table_args__ = (UniqueConstraint("task_id", "user_id", name="uq_scheduled_task_subscriptions_task_user"),) + + id: Mapped[str] = mapped_column(String(64), primary_key=True) + task_id: Mapped[str] = mapped_column(String(64), index=True) + user_id: Mapped[str] = mapped_column(String(64), index=True) + notify_element: Mapped[bool] = mapped_column(Boolean, default=False) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC)) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC)) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py index 61e45383..1195b50b 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py @@ -5,17 +5,25 @@ from __future__ import annotations from datetime import UTC, datetime from typing import Any -from sqlalchemy import select, update +import uuid + +from sqlalchemy import delete, select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from deerflow.persistence.scheduled_tasks.base import ScheduledTaskStore -from deerflow.persistence.scheduled_tasks.model import ScheduledTaskDeliveryProfileRow, ScheduledTaskRow, ScheduledTaskRunRow, SchedulerThreadRow +from deerflow.persistence.scheduled_tasks.model import ( + ScheduledTaskDeliveryProfileRow, + ScheduledTaskRow, + ScheduledTaskRunRow, + ScheduledTaskSubscriptionRow, + SchedulerThreadRow, +) def _to_dict(row) -> dict[str, Any]: data = row.to_dict() - for key in ("created_at", "updated_at", "next_run_at", "last_run_at", "scheduled_for", "started_at", "finished_at"): + for key in ("created_at", "updated_at", "next_run_at", "last_run_at", "scheduled_for", "started_at", "finished_at", "published_at"): value = data.get(key) if isinstance(value, datetime): data[key] = value.isoformat() @@ -194,3 +202,107 @@ class ScheduledTaskRepository(ScheduledTaskStore): await session.commit() await session.refresh(row) return _to_dict(row) + + # --- Publish / subscribe --- + + async def get_task_any(self, task_id: str) -> dict[str, Any] | None: + async with self._sf() as session: + row = await session.get(ScheduledTaskRow, task_id) + return _to_dict(row) if row is not None else None + + async def get_run_any(self, run_id: str) -> dict[str, Any] | None: + async with self._sf() as session: + row = await session.get(ScheduledTaskRunRow, run_id) + return _to_dict(row) if row is not None else None + + async def set_task_published(self, task_id: str, owner_user_id: str, *, published: bool) -> dict[str, Any] | None: + async with self._sf() as session: + row = await session.get(ScheduledTaskRow, task_id) + if row is None or row.user_id != owner_user_id: + return None + row.published = bool(published) + row.published_at = datetime.now(UTC) if published else None + row.updated_at = datetime.now(UTC) + await session.commit() + await session.refresh(row) + return _to_dict(row) + + async def list_published_tasks(self, *, limit: int = 50, offset: int = 0) -> list[dict[str, Any]]: + async with self._sf() as session: + result = await session.execute( + select(ScheduledTaskRow) + .where(ScheduledTaskRow.published.is_(True)) + .order_by(ScheduledTaskRow.published_at.desc().nullslast(), ScheduledTaskRow.created_at.desc()) + .offset(max(0, offset)) + .limit(max(1, limit)) + ) + return [_to_dict(row) for row in result.scalars()] + + async def list_subscribed_tasks(self, user_id: str) -> list[dict[str, Any]]: + async with self._sf() as session: + result = await session.execute( + select(ScheduledTaskRow, ScheduledTaskSubscriptionRow) + .join(ScheduledTaskSubscriptionRow, ScheduledTaskSubscriptionRow.task_id == ScheduledTaskRow.task_id) + .where(ScheduledTaskSubscriptionRow.user_id == user_id) + .order_by(ScheduledTaskSubscriptionRow.created_at.desc()) + ) + tasks: list[dict[str, Any]] = [] + for task_row, sub_row in result.all(): + task = _to_dict(task_row) + task["_subscription"] = _to_dict(sub_row) + tasks.append(task) + return tasks + + async def get_subscription(self, task_id: str, user_id: str) -> dict[str, Any] | None: + async with self._sf() as session: + result = await session.execute( + select(ScheduledTaskSubscriptionRow) + .where(ScheduledTaskSubscriptionRow.task_id == task_id, ScheduledTaskSubscriptionRow.user_id == user_id) + .limit(1) + ) + row = result.scalar_one_or_none() + return _to_dict(row) if row is not None else None + + async def upsert_subscription(self, task_id: str, user_id: str, *, notify_element: bool) -> dict[str, Any]: + async with self._sf() as session: + result = await session.execute( + select(ScheduledTaskSubscriptionRow) + .where(ScheduledTaskSubscriptionRow.task_id == task_id, ScheduledTaskSubscriptionRow.user_id == user_id) + .limit(1) + ) + row = result.scalar_one_or_none() + if row is None: + row = ScheduledTaskSubscriptionRow( + id=str(uuid.uuid4()), + task_id=task_id, + user_id=user_id, + notify_element=bool(notify_element), + ) + session.add(row) + else: + row.notify_element = bool(notify_element) + row.updated_at = datetime.now(UTC) + await session.commit() + await session.refresh(row) + return _to_dict(row) + + async def delete_subscription(self, task_id: str, user_id: str) -> bool: + async with self._sf() as session: + result = await session.execute( + delete(ScheduledTaskSubscriptionRow).where( + ScheduledTaskSubscriptionRow.task_id == task_id, + ScheduledTaskSubscriptionRow.user_id == user_id, + ) + ) + await session.commit() + return (result.rowcount or 0) > 0 + + async def list_notify_subscribers(self, task_id: str) -> list[dict[str, Any]]: + async with self._sf() as session: + result = await session.execute( + select(ScheduledTaskSubscriptionRow).where( + ScheduledTaskSubscriptionRow.task_id == task_id, + ScheduledTaskSubscriptionRow.notify_element.is_(True), + ) + ) + return [_to_dict(row) for row in result.scalars()] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py index 7edbcfb2..bf52df88 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py @@ -197,16 +197,70 @@ class ScheduledTaskService: return thread async def list_tasks(self, user_id: str) -> list[dict[str, Any]]: - return await self.store.list_tasks(user_id) + owned = await self.store.list_tasks(user_id) + for task in owned: + task["role"] = "owner" + subscribed = await self.store.list_subscribed_tasks(user_id) + owned_ids = {task["task_id"] for task in owned} + for task in subscribed: + if task["task_id"] in owned_ids: + continue + task["role"] = "subscriber" + owned.append(task) + return owned + + async def list_published(self, *, limit: int = 50, offset: int = 0) -> list[dict[str, Any]]: + """Public discovery list. Strips author prompt details to a short summary.""" + tasks = await self.store.list_published_tasks(limit=limit, offset=offset) + result: list[dict[str, Any]] = [] + for task in tasks: + prompt = (task.get("prompt") or "").strip() + summary = prompt if len(prompt) <= 200 else prompt[:200].rstrip() + "…" + result.append( + { + "task_id": task["task_id"], + "name": task.get("name"), + "prompt_summary": summary, + "schedule_text": task.get("schedule_text"), + "timezone": task.get("timezone"), + "published_at": task.get("published_at"), + "owner_user_id": task.get("user_id"), + } + ) + return result + + async def _accessible_task(self, user_id: str, task_id: str) -> dict[str, Any] | None: + task = await self.store.get_task_any(task_id) + if task is None: + return None + if task["user_id"] == user_id: + task["role"] = "owner" + return task + sub = await self.store.get_subscription(task_id, user_id) + if sub is None: + return None + task["role"] = "subscriber" + task["_subscription"] = sub + return task async def list_runs(self, user_id: str, task_id: str, *, limit: int = 50) -> list[dict[str, Any]]: - task = await self.store.get_task(task_id, user_id) - if task is None: + accessible = await self._accessible_task(user_id, task_id) + if accessible is None: return [] - return await self.store.list_runs(task_id, user_id, limit=limit) + return await self.store.list_runs(task_id, accessible["user_id"], limit=limit) async def get_run(self, user_id: str, run_id: str) -> dict[str, Any] | None: - return await self.store.get_run(run_id, user_id) + run = await self.store.get_run(run_id, user_id) + if run is not None: + return run + # Subscriber visibility: look up the run unfiltered, then verify the + # caller has a subscription on the parent task. + candidate = await self.store.get_run_any(run_id) + if candidate is None: + return None + if await self.store.get_subscription(candidate["task_id"], user_id) is None: + return None + return candidate async def delete_run(self, user_id: str, run_id: str) -> bool: deleted = await self.store.delete_run(run_id, user_id) @@ -655,15 +709,62 @@ class ScheduledTaskService: async def _deliver_element_result(self, task: dict[str, Any], result_text: str) -> None: if self._element_client is None: return - profile = await self.store.get_delivery_profile(task["user_id"]) + body = f"【定时任务】{task['name']}\n\n{result_text}" + # 1) Author + await self._deliver_to_user(task["user_id"], task, body) + # 2) Subscribers who opted into notifications + try: + subscribers = await self.store.list_notify_subscribers(task["task_id"]) + except Exception: + logger.exception("Failed to list scheduled task subscribers for %s", task["task_id"]) + subscribers = [] + for sub in subscribers: + sub_user_id = sub.get("user_id") + if not sub_user_id or sub_user_id == task["user_id"]: + continue + await self._deliver_to_user(sub_user_id, task, body) + + async def _deliver_to_user(self, user_id: str, task: dict[str, Any], body: str) -> None: + if self._element_client is None: + return + try: + profile = await self.store.get_delivery_profile(user_id) + except Exception: + logger.exception("Failed to load delivery profile for user %s", user_id) + return element_user_id = profile.get("element_user_id") if not isinstance(element_user_id, str) or not element_user_id.strip(): return existing_room_id: str | None = profile.get("element_room_id") or None - body = f"【定时任务】{task['name']}\n\n{result_text}" try: room_id = await self._element_client.send_direct_message(element_user_id.strip(), body, room_id=existing_room_id) if room_id and room_id != existing_room_id: - await self.store.update_delivery_profile(task["user_id"], {"element_room_id": room_id}) + await self.store.update_delivery_profile(user_id, {"element_room_id": room_id}) except Exception: logger.exception("Failed to deliver scheduled task %s to Element user %s", task["task_id"], element_user_id) + + # --- Publish / subscribe --- + + async def publish_task(self, user_id: str, task_id: str) -> dict[str, Any] | None: + return await self.store.set_task_published(task_id, user_id, published=True) + + async def unpublish_task(self, user_id: str, task_id: str) -> dict[str, Any] | None: + return await self.store.set_task_published(task_id, user_id, published=False) + + async def subscribe_task( + self, user_id: str, task_id: str, *, notify_element: bool + ) -> dict[str, Any] | None: + task = await self.store.get_task_any(task_id) + if task is None: + return None + if task["user_id"] == user_id: + # Authors don't need to subscribe to their own tasks. + return None + existing = await self.store.get_subscription(task_id, user_id) + if existing is None and not task.get("published"): + # First-time subscription requires the task to currently be published. + return None + return await self.store.upsert_subscription(task_id, user_id, notify_element=notify_element) + + async def unsubscribe_task(self, user_id: str, task_id: str) -> bool: + return await self.store.delete_subscription(task_id, user_id) diff --git a/offline-backend-20260512/backend/tests/test_scheduled_tasks_publish.py b/offline-backend-20260512/backend/tests/test_scheduled_tasks_publish.py new file mode 100644 index 00000000..d9860577 --- /dev/null +++ b/offline-backend-20260512/backend/tests/test_scheduled_tasks_publish.py @@ -0,0 +1,220 @@ +"""Tests for scheduled task publish/subscribe + mirrored Element delivery.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any + +import pytest + +from deerflow.persistence.scheduled_tasks import MemoryScheduledTaskStore +from deerflow.runtime.scheduler.service import ScheduledTaskService + + +@pytest.fixture +def store() -> MemoryScheduledTaskStore: + return MemoryScheduledTaskStore() + + +def _seed_task( + store: MemoryScheduledTaskStore, + *, + task_id: str, + user_id: str, + name: str = "demo", + prompt: str = "Hello world", + published: bool = False, +) -> dict[str, Any]: + task = { + "task_id": task_id, + "user_id": user_id, + "scheduler_thread_id": f"thread-{user_id}", + "name": name, + "prompt": prompt, + "schedule_text": "0 9 * * *", + "cron_expr": "0 9 * * *", + "timezone": "UTC", + "enabled": True, + "next_run_at": datetime.now(UTC), + "execution_context": {}, + "published": published, + "published_at": datetime.now(UTC) if published else None, + } + store._tasks[task_id] = task # noqa: SLF001 + return task + + +def _build_service(store: MemoryScheduledTaskStore) -> ScheduledTaskService: + # Build with no FastAPI app — the methods we exercise here don't touch app.state. + service = ScheduledTaskService(app=None, store=store, poll_interval_seconds=60.0) + service._element_client = None # noqa: SLF001 — disable real delivery + return service + + +@pytest.mark.asyncio +async def test_publish_unpublish_only_owner(store): + task = _seed_task(store, task_id="t1", user_id="alice") + service = _build_service(store) + + assert await service.publish_task("bob", task["task_id"]) is None, "non-owner cannot publish" + + published = await service.publish_task("alice", task["task_id"]) + assert published is not None + assert published["published"] is True + assert published["published_at"] is not None + + listed = await service.list_published(limit=10, offset=0) + assert len(listed) == 1 + assert listed[0]["task_id"] == "t1" + assert listed[0]["prompt_summary"] == "Hello world" + assert "prompt" not in listed[0], "raw prompt must not leak through summary endpoint" + + unpublished = await service.unpublish_task("alice", task["task_id"]) + assert unpublished["published"] is False + assert (await service.list_published()) == [] + + +@pytest.mark.asyncio +async def test_subscribe_requires_published_for_first_time(store): + task = _seed_task(store, task_id="t1", user_id="alice", published=False) + service = _build_service(store) + + # Cannot subscribe to a private task. + assert await service.subscribe_task("bob", task["task_id"], notify_element=False) is None + + # Author publishes; bob subscribes. + await service.publish_task("alice", task["task_id"]) + sub = await service.subscribe_task("bob", task["task_id"], notify_element=True) + assert sub is not None and sub["notify_element"] is True + + # Author cannot subscribe to their own task. + assert await service.subscribe_task("alice", task["task_id"], notify_element=True) is None + + +@pytest.mark.asyncio +async def test_unpublish_keeps_existing_subscribers(store): + task = _seed_task(store, task_id="t1", user_id="alice", published=True) + service = _build_service(store) + + sub = await service.subscribe_task("bob", task["task_id"], notify_element=False) + assert sub is not None + + # Author un-publishes — bob keeps acquired access. + await service.unpublish_task("alice", task["task_id"]) + assert await store.get_subscription(task["task_id"], "bob") is not None + + # New user cannot subscribe after un-publish. + assert await service.subscribe_task("carol", task["task_id"], notify_element=False) is None + + # Existing subscriber can still toggle their notification preference. + updated = await service.subscribe_task("bob", task["task_id"], notify_element=True) + assert updated is not None and updated["notify_element"] is True + + +@pytest.mark.asyncio +async def test_list_tasks_includes_subscribed_with_role(store): + _seed_task(store, task_id="t1", user_id="alice", published=True) + _seed_task(store, task_id="t2", user_id="bob") + service = _build_service(store) + + await service.subscribe_task("bob", "t1", notify_element=False) + + bobs = await service.list_tasks("bob") + by_id = {t["task_id"]: t for t in bobs} + assert by_id["t2"]["role"] == "owner" + assert by_id["t1"]["role"] == "subscriber" + assert by_id["t1"]["_subscription"]["notify_element"] is False + + +@pytest.mark.asyncio +async def test_subscriber_can_read_runs_and_get_run(store): + _seed_task(store, task_id="t1", user_id="alice", published=True) + service = _build_service(store) + await service.subscribe_task("bob", "t1", notify_element=False) + + # Seed a run owned by alice. + scheduled_for = datetime.now(UTC) + run = await store.create_run_once( + { + "id": "run-1", + "task_id": "t1", + "user_id": "alice", + "scheduler_thread_id": "thread-alice", + "scheduled_for": scheduled_for, + "status": "succeeded", + "started_at": scheduled_for, + } + ) + assert run is not None + + # Bob (subscriber) sees runs on the task. + bob_runs = await service.list_runs("bob", "t1", limit=10) + assert [r["id"] for r in bob_runs] == ["run-1"] + + # Bob can fetch the run by id. + fetched = await service.get_run("bob", "run-1") + assert fetched is not None and fetched["id"] == "run-1" + + # Carol (no subscription) cannot. + assert await service.list_runs("carol", "t1", limit=10) == [] + assert await service.get_run("carol", "run-1") is None + + +@pytest.mark.asyncio +async def test_unsubscribe_removes_visibility(store): + _seed_task(store, task_id="t1", user_id="alice", published=True) + service = _build_service(store) + + await service.subscribe_task("bob", "t1", notify_element=False) + assert await service.unsubscribe_task("bob", "t1") is True + bob_tasks = await service.list_tasks("bob") + assert all(t["task_id"] != "t1" for t in bob_tasks) + + +@pytest.mark.asyncio +async def test_mirror_element_delivery_to_subscribers(store): + _seed_task(store, task_id="t1", user_id="alice", published=True) + service = _build_service(store) + + # Author and two subscribers configure delivery profiles. + await store.update_delivery_profile("alice", {"element_user_id": "@alice:example"}) + await store.update_delivery_profile("bob", {"element_user_id": "@bob:example"}) + await store.update_delivery_profile("carol", {"element_user_id": "@carol:example"}) + + await service.subscribe_task("bob", "t1", notify_element=True) + await service.subscribe_task("carol", "t1", notify_element=False) # opted out + + sent: list[tuple[str, str]] = [] + + class _FakeElement: + async def send_direct_message(self, user_id: str, body: str, room_id: str | None = None) -> str: + sent.append((user_id, body)) + return f"!room-{user_id}" + + service._element_client = _FakeElement() # noqa: SLF001 + + task = await store.get_task_any("t1") + assert task is not None + await service._deliver_element_result(task, "result body") # noqa: SLF001 + + recipients = sorted(user for user, _ in sent) + assert recipients == ["@alice:example", "@bob:example"], "only author + opted-in subscriber receive" + + # All recipients get the same body content (we sent identical messages). + bodies = {body for _, body in sent} + assert bodies == {"【定时任务】demo\n\nresult body"} + + # Carol is not in the recipient list. + assert all(uid != "@carol:example" for uid, _ in sent) + + +@pytest.mark.asyncio +async def test_published_summary_truncates_long_prompt(store): + long_prompt = "abcdef" * 100 # 600 chars + _seed_task(store, task_id="t1", user_id="alice", prompt=long_prompt, published=True) + service = _build_service(store) + + listed = await service.list_published() + summary = listed[0]["prompt_summary"] + assert len(summary) <= 201 # 200 chars + ellipsis + assert summary.endswith("…")