feat(cron): share group task lists and allow admin deletion

astrbot/core/cron/manager.py

@@ -112,6 +112,10 @@ def __init__(self, db: BaseDatabase) -> None:
         # The scheduler may start early via _schedule_job; track DB sync separately.
         self._db_synced = False
         self._running_tasks: dict[str, set[asyncio.Task]] = {}
+        # Ref-counted deletion markers. A run must not register after deletion
+        # has started, and nested/concurrent deletes must not clear the marker
+        # before the last deletion finishes.
+        self._deleting_jobs: dict[str, int] = {}
 
     async def start(self, ctx: "Context") -> None:
         self.ctx: Context = ctx  # star context
@@ -227,20 +231,31 @@ async def update_job(self, job_id: str, **kwargs) -> CronJob | None:
     async def delete_job(self, job_id: str) -> None:
         """Remove a schedule and drain its active runs before deleting its row.
 
+        Runs that have already started are cancelled and drained. Runs that
+        start after deletion begins return without executing.
+
         Args:
             job_id: Job to delete, including any currently executing run.
         """
-        self._remove_scheduled(job_id)
-        self._basic_handlers.pop(job_id, None)
-        tasks = [
-            task
-            for task in self._running_tasks.get(job_id, set())
-            if task is not asyncio.current_task() and not task.done()
-        ]
-        for task in tasks:
-            task.cancel()
-        await asyncio.gather(*tasks, return_exceptions=True)
-        await self.db.delete_cron_job(job_id)
+        self._deleting_jobs[job_id] = self._deleting_jobs.get(job_id, 0) + 1
+        try:
+            self._remove_scheduled(job_id)
+            self._basic_handlers.pop(job_id, None)
+            tasks = [
+                task
+                for task in self._running_tasks.get(job_id, set())
+                if task is not asyncio.current_task() and not task.done()
+            ]
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            await self.db.delete_cron_job(job_id)
+        finally:
+            remaining = self._deleting_jobs.get(job_id, 0) - 1
+            if remaining > 0:
+                self._deleting_jobs[job_id] = remaining
+            else:
+                self._deleting_jobs.pop(job_id, None)
 
     async def list_jobs(self, job_type: str | None = None) -> list[CronJob]:
         return await self.db.list_cron_jobs(job_type)
@@ -361,12 +376,16 @@ async def _run_job(
         ignore_enabled: bool = False,
         delete_run_once: bool = True,
     ) -> None:
-        job = await self.db.get_cron_job(job_id)
-        if not job or (not job.enabled and not ignore_enabled):
-            return
         task = asyncio.current_task()
+        if task is None or job_id in self._deleting_jobs:
+            return
+        # Register before the first await so delete_job always snapshots a run
+        # that has passed the deletion-marker check.
         self._running_tasks.setdefault(job_id, set()).add(task)
         try:
+            job = await self.db.get_cron_job(job_id)
+            if not job or (not job.enabled and not ignore_enabled):
+                return
             start_time = datetime.now(timezone.utc)
             await self.db.update_cron_job(
                 job_id, status="running", last_run_at=start_time, last_error=None
@@ -400,9 +419,11 @@ async def _run_job(
                     # One-shot runs remove themselves, excluding their own task.
                     await self.delete_job(job_id)
         finally:
-            self._running_tasks[job_id].discard(task)
-            if not self._running_tasks[job_id]:
-                del self._running_tasks[job_id]
+            running = self._running_tasks.get(job_id)
+            if running is not None:
+                running.discard(task)
+                if not running:
+                    del self._running_tasks[job_id]
 
     async def _run_basic_job(self, job: CronJob) -> None:
         handler = self._basic_handlers.get(job.job_id)

astrbot/core/tools/cron_tools.py

@@ -59,7 +59,10 @@ class FutureTaskTool(FunctionTool[AstrAgentContext]):
         "Use action='create' to schedule a recurring cron task or one-time run_at task. "
         "Use action='edit' to update an existing task. "
         "Use action='list' to inspect existing tasks. "
-        "Use action='delete' to remove a task by job_id."
+        "Use action='delete' to remove a task by job_id. "
+        "Group members can list all tasks in the current group, including their creators. "
+        "Only task creators can edit tasks. Group members can delete only their own tasks, "
+        "while AstrBot admins can also delete other members' tasks in the current group."
     )
     parameters: dict = Field(
         default_factory=lambda: {
@@ -191,8 +194,9 @@ async def call(
             )
             return f"Scheduled future task {job.job_id} ({job.name}) {suffix}."
 
-        current_umo = context.context.event.unified_msg_origin
-        current_sender_id = str(context.context.event.get_sender_id())
+        event = context.context.event
+        current_umo = event.unified_msg_origin
+        current_sender_id = str(event.get_sender_id())
         if action == "edit":
             job_id = kwargs.get("job_id")
             if not job_id:
@@ -218,8 +222,7 @@ async def call(
                     )
                 if (
                     same_session
-                    and context.context.event.get_message_type()
-                    == MessageType.GROUP_MESSAGE
+                    and event.get_message_type() == MessageType.GROUP_MESSAGE
                 ):
                     return (
                         f"error: cron job {job_id} was created by another member of "
@@ -296,8 +299,15 @@ async def call(
             job = await cron_mgr.db.get_cron_job(str(job_id))
             if not job:
                 return f"error: cron job {job_id} not found."
-            if not _job_belongs_to_current_sender(job, current_umo, current_sender_id):
-                same_session = _extract_job_session(job) == current_umo
+            same_session = _extract_job_session(job) == current_umo
+            can_delete = _job_belongs_to_current_sender(
+                job, current_umo, current_sender_id
+            ) or (
+                same_session
+                and event.get_message_type() == MessageType.GROUP_MESSAGE
+                and event.is_admin()
+            )
+            if not can_delete:
                 if same_session and not _extract_job_sender(job):
                     # Dashboard / legacy rows have a session but no member as
                     # their creator, so blaming another member would be wrong.
@@ -308,13 +318,13 @@ async def call(
                     )
                 if (
                     same_session
-                    and context.context.event.get_message_type()
-                    == MessageType.GROUP_MESSAGE
+                    and event.get_message_type() == MessageType.GROUP_MESSAGE
                 ):
                     return (
                         f"error: cron job {job_id} was created by another member of "
                         "this group chat, so you cannot delete it. Only the member who "
-                        "created it can delete it; tell the user to ask that member."
+                        "created it or an AstrBot admin can delete it; tell the user "
+                        "to ask that member."
                     )
                 return (
                     f"error: cron job {job_id} was not created by you, so you cannot "
@@ -325,26 +335,29 @@ async def call(
 
         if action == "list":
             all_jobs = await cron_mgr.list_jobs()
+            is_group = event.get_message_type() == MessageType.GROUP_MESSAGE
             jobs = [
                 job
                 for job in all_jobs
                 if _job_belongs_to_current_sender(job, current_umo, current_sender_id)
+                or (is_group and _extract_job_session(job) == current_umo)
             ]
-            # Tasks in this session that were created by somebody else stay
-            # out of the result. Saying so stops an agent from reading "No cron
-            # jobs found." as "the task no longer exists".
+            # Private chats retain sender isolation. Explain hidden tasks so
+            # the agent does not mistake a filtered list for a missing task.
             hidden_note = ""
-            for job in all_jobs:
-                if _extract_job_session(job) != current_umo:
-                    continue
-                if _job_belongs_to_current_sender(job, current_umo, current_sender_id):
-                    continue
-                hidden_note = (
-                    "\n\nNote: tasks in this chat that were not created by you "
-                    "are not listed here, and can only be edited or deleted by "
-                    "whoever created them."
-                )
-                break
+            if not is_group:
+                for job in all_jobs:
+                    if _extract_job_session(job) != current_umo:
+                        continue
+                    if _job_belongs_to_current_sender(
+                        job, current_umo, current_sender_id
+                    ):
+                        continue
+                    hidden_note = (
+                        "\n\nNote: tasks in this chat that were not created by you "
+                        "are not listed here. Only their creators can edit or delete them."
+                    )
+                    break
             if not jobs:
                 return "No cron jobs found." + hidden_note
             tz_name = str(
@@ -374,9 +387,14 @@ async def call(
                     next_run = (
                         next_run.astimezone(tzinfo) if tzinfo else next_run.astimezone()
                     )
-                lines.append(
-                    f"{j.job_id} | {j.name} | {j.job_type} | run_once={getattr(j, 'run_once', False)} | enabled={j.enabled} | next={next_run}"
+                line = (
+                    f"{j.job_id} | {j.name} | {j.job_type} | "
+                    f"run_once={getattr(j, 'run_once', False)} | enabled={j.enabled} | "
+                    f"next={next_run}"
                 )
+                if is_group:
+                    line += f" | creator={_extract_job_sender(j) or 'unknown'}"
+                lines.append(line)
             return "\n".join(lines) + hidden_note
 
         return "error: action must be one of create, edit, delete, or list."

docs/en/use/proactive-agent.md

@@ -41,7 +41,7 @@ You can also ask it to:
 - “Move my break reminder to 3 PM tomorrow.”
 - “Delete the news task I just created.”
 
-Chat-based listing, editing, and deletion are limited to tasks created by **the current sender in the current conversation**. Group members cannot edit or delete each other's tasks. Tasks created in the WebUI do not belong to a chat sender; administrators should manage them in the WebUI.
+All group members can list **every task in the current group conversation**, including its creator. Regular members can only edit and delete tasks they created in that group. AstrBot admins can also delete other members' tasks in the current group, but editing remains limited to the creator. Tasks created in the WebUI without a recorded chat creator also appear in the corresponding group's list and can be deleted by AstrBot admins. These permissions stay within the current conversation and do not allow access across groups. In private chats, listing, editing, and deletion remain limited to tasks created by the current sender in the current conversation.
 
 ## Create and manage tasks in the WebUI
 
@@ -119,4 +119,4 @@ Create a one-off task a few minutes ahead that sends a short test reminder. Conf
 | Task ran but no message arrived | Check the delivery target, platform support, connection, and send permissions. No target means no automatic delivery. |
 | Search or file operations fail | Test those tools' configuration, dependencies, and permissions separately. Scheduling only wakes the Agent. |
 | A one-off task disappeared | Scheduled execution deletes it even on failure. Inspect logs for the outcome. |
-| Chat cannot find an existing task | Query as the original sender in the original conversation. Use the WebUI for tasks created by others or by the dashboard. |
+| Chat cannot find an existing task | Query in the task's original conversation. All group members can list every task in their current group; private chats require the original sender. Otherwise use the WebUI. |

docs/zh/use/proactive-agent.md

@@ -41,7 +41,7 @@
 - “把我的休息提醒改到明天下午三点。”
 - “删除我刚才设置的新闻任务。”
 
-聊天中的任务查询、修改和删除限定为**当前会话、当前发送者创建的任务**。群成员不能修改或删除别人创建的任务；从 WebUI 创建的任务也不能直接在聊天中当作自己的任务管理，管理员应使用 WebUI。
+群聊中的所有成员都可以查询**本群会话的全部任务**，列表会显示创建者。普通成员只能修改和删除自己在本群创建的任务；AstrBot 管理员还可以删除本群其他成员创建的任务，但编辑仍仅限创建者。从 WebUI 创建且未记录创建者的任务也会显示在对应群聊的列表中，并可由 AstrBot 管理员删除。以上权限均限定在当前会话，不能跨群管理；私聊中的查询、修改和删除仍限定为当前发送者在当前会话创建的任务。
 
 ## 通过 WebUI 创建和管理
 
@@ -119,4 +119,4 @@ AstrBot 内置 `send_message_to_user` 工具，支持文本、图片、语音、
 | 执行了但没有消息 | 检查投递地、支持的平台、平台连接和发送权限；未设置投递地不会自动回传结果。 |
 | 搜索或文件操作失败 | 单独验证相应工具的配置、依赖和权限；主动型能力只负责定时唤醒。 |
 | 一次性任务不见了 | 定时触发后无论成功失败都会删除，到日志中检查执行结果。 |
-| 聊天里找不到某个已有任务 | 确认在原会话由同一个发送者查询；其他成员或 WebUI 创建的任务需要到 WebUI 管理。 |
+| 聊天里找不到某个已有任务 | 确认在任务所属的原会话查询。所有群成员均可查看本群全部任务；私聊中需由原创建者查询。其他情况请到 WebUI 管理。 |

tests/unit/test_cron_manager.py

@@ -1,5 +1,6 @@
 """Tests for CronJobManager."""
 
+import asyncio
 import json
 from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
@@ -73,6 +74,7 @@ def test_init(self, mock_db):
         assert manager._basic_handlers == {}
         assert manager._started is False
         assert manager._db_synced is False
+        assert manager._deleting_jobs == {}
 
 
 class TestCronJobManagerStart:
@@ -315,6 +317,113 @@ async def test_delete_job(self, cron_manager, mock_db):
         mock_db.delete_cron_job.assert_called_once_with("test-job-id")
         assert "test-job-id" not in cron_manager._basic_handlers
 
+    @pytest.mark.asyncio
+    async def test_delete_job_cancels_run_blocked_in_row_lookup(
+        self, cron_manager, mock_db, sample_cron_job
+    ):
+        """A run that fetched the row must not escape deletion before registering."""
+        lookup_started = asyncio.Event()
+        allow_lookup = asyncio.Event()
+
+        async def delayed_get(job_id):
+            lookup_started.set()
+            await allow_lookup.wait()
+            return sample_cron_job
+
+        mock_db.get_cron_job.side_effect = delayed_get
+        handler = MagicMock()
+        cron_manager._basic_handlers[sample_cron_job.job_id] = handler
+
+        run_task = asyncio.create_task(cron_manager._run_job(sample_cron_job.job_id))
+        await asyncio.wait_for(lookup_started.wait(), 2)
+        await asyncio.wait_for(cron_manager.delete_job(sample_cron_job.job_id), 2)
+
+        assert run_task.cancelled()
+        allow_lookup.set()
+        handler.assert_not_called()
+        mock_db.delete_cron_job.assert_awaited_once_with(sample_cron_job.job_id)
+        assert cron_manager._running_tasks == {}
+        assert cron_manager._deleting_jobs == {}
+
+    @pytest.mark.asyncio
+    async def test_delete_job_blocks_run_starting_after_deletion_begins(
+        self, cron_manager, mock_db, sample_cron_job
+    ):
+        """A run must not start once the job deletion marker is set."""
+        delete_started = asyncio.Event()
+        allow_delete = asyncio.Event()
+
+        async def delayed_delete(job_id):
+            delete_started.set()
+            await allow_delete.wait()
+
+        mock_db.delete_cron_job.side_effect = delayed_delete
+        mock_db.get_cron_job.return_value = sample_cron_job
+        handler = MagicMock()
+        cron_manager._basic_handlers[sample_cron_job.job_id] = handler
+
+        delete_task = asyncio.create_task(
+            cron_manager.delete_job(sample_cron_job.job_id)
+        )
+        await asyncio.wait_for(delete_started.wait(), 2)
+
+        await asyncio.wait_for(cron_manager._run_job(sample_cron_job.job_id), 2)
+
+        allow_delete.set()
+        await asyncio.wait_for(delete_task, 2)
+
+        handler.assert_not_called()
+        mock_db.get_cron_job.assert_not_awaited()
+        assert cron_manager._deleting_jobs == {}
+
+    @pytest.mark.asyncio
+    async def test_concurrent_deletes_keep_marker_until_last_finishes(
+        self, cron_manager, mock_db, sample_cron_job
+    ):
+        """One deletion finishing must not unblock runs while another is pending."""
+        first_delete_started = asyncio.Event()
+        second_delete_started = asyncio.Event()
+        allow_first_delete = asyncio.Event()
+        allow_second_delete = asyncio.Event()
+        delete_calls = 0
+
+        async def delayed_delete(job_id):
+            nonlocal delete_calls
+            delete_calls += 1
+            if delete_calls == 1:
+                first_delete_started.set()
+                await allow_first_delete.wait()
+            else:
+                second_delete_started.set()
+                await allow_second_delete.wait()
+
+        mock_db.delete_cron_job.side_effect = delayed_delete
+        mock_db.get_cron_job.return_value = sample_cron_job
+        handler = MagicMock()
+        cron_manager._basic_handlers[sample_cron_job.job_id] = handler
+
+        first_delete = asyncio.create_task(
+            cron_manager.delete_job(sample_cron_job.job_id)
+        )
+        second_delete = asyncio.create_task(
+            cron_manager.delete_job(sample_cron_job.job_id)
+        )
+        await asyncio.wait_for(first_delete_started.wait(), 2)
+        await asyncio.wait_for(second_delete_started.wait(), 2)
+
+        allow_first_delete.set()
+        await asyncio.wait_for(first_delete, 2)
+        assert cron_manager._deleting_jobs == {sample_cron_job.job_id: 1}
+
+        await asyncio.wait_for(cron_manager._run_job(sample_cron_job.job_id), 2)
+
+        allow_second_delete.set()
+        await asyncio.wait_for(second_delete, 2)
+
+        handler.assert_not_called()
+        mock_db.get_cron_job.assert_not_awaited()
+        assert cron_manager._deleting_jobs == {}
+
 
 class TestListJobs:
     """Tests for list_jobs method."""