feat: 定时任务运行详情执行过程视图 + 管理员下载删除 + MySQL 迁移修复

定时任务: - 运行详情页新增「执行过程」标签页,按 agent_run_id 拉取该次运行的 agent 消息(GET /scheduled-tasks/runs/{id}/messages),运行中每 4s 自动刷新,可观察卡住的任务停在哪一步 - 详情页「返回」按钮回到来源列表(普通/管理员)并重新选中原任务 - 管理员可下载、删除任意用户的执行记录(DELETE /runs/{id} 加管理员兜底) MySQL 兼容修复: - 合并迁移 20260518_06 汇合 ai-writing 与 thread-shares 两个 head - 20260517_02 改用 SQLAlchemy Core 构造,规避保留字 key 的 1064 报错 - list_messages_by_run 各事件存储补 user_id 形参以保持接口一致测试: 新增 test_scheduled_task_admin_run_delete.py Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-18 21:13:58 +08:00
1 parent 449703a6f0
commit e3cde9be25
@@ -227,7 +227,7 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA
 | **Runs** (`/api/runs`) | `POST /stream` - stateless run + SSE; `POST /wait` - stateless run + block; `GET /{rid}/messages` - paginated messages by run_id `{data, has_more}` (cursor: `after_seq`/`before_seq`); `GET /{rid}/feedback` - list feedback by run_id |
 | **Thread Shares** (`/api/thread-shares`) | Per-conversation sharing with two modes (`thread_shares.mode`). **`import`**: `POST /` `{thread_id, mode}` - create (or reuse) a revocable share code; `GET /{code}/preview` + `POST /{code}/import` - copy that conversation into the caller's account (fresh `thread_id` via `app/gateway/thread_copy.py` + new `threads_meta` row + `user-data` copy; dedup on `metadata.shared_origin`). **`view`**: a public, read-only static page — see Public Shares below. Both modes: `GET /` - list my codes; `DELETE /{code}` - revoke. |
 | **Public Shares** (`/api/public/shares`) | **Unauthenticated** (the `/api/public/` prefix is whitelisted in `auth_middleware._PUBLIC_PATH_PREFIXES`). `GET /{code}` - read-only conversation snapshot (title + serialized messages + artifacts) for a `view`-mode share; messages are read straight from the latest checkpoint. `GET /{code}/files/{path}` - serves an **image** referenced by that conversation (images only — non-image paths 404; SVG forced to download). Missing/revoked/non-`view` codes all return 404. |
-| **Scheduled Tasks** (`/api/scheduled-tasks`) | Per-user cron-like agent tasks. `GET/POST /` - list/create owned tasks; `PATCH/DELETE /{id}` - update/delete (owner); `POST /{id}/{pause,resume,trigger}`; `GET /{id}/runs`, `GET/DELETE /runs/{rid}`, run-file preview/edit/download; publish/subscribe under `/{id}/{publish,unpublish,subscribe}` and `GET /published`. **Admin (cross-user):** `GET /admin/all` - every task; `PATCH/DELETE /admin/{id}` - edit/delete any task; `POST /admin/{id}/trigger`; `GET /admin/{id}/runs`. Admin routes require `system_role == "admin"`. |
+| **Scheduled Tasks** (`/api/scheduled-tasks`) | Per-user cron-like agent tasks. `GET/POST /` - list/create owned tasks; `PATCH/DELETE /{id}` - update/delete (owner); `POST /{id}/{pause,resume,trigger}`; `GET /{id}/runs`, `GET /runs/{rid}`, run-file preview/edit/download; `GET /runs/{rid}/messages` - agent messages for the run's underlying agent run (used to inspect a stuck/running run; authorized via `_load_run`); `DELETE /runs/{rid}` - owner-scoped, with an admin fallback so admins can delete any user's run; publish/subscribe under `/{id}/{publish,unpublish,subscribe}` and `GET /published`. **Admin (cross-user):** `GET /admin/all` - every task; `PATCH/DELETE /admin/{id}` - edit/delete any task; `POST /admin/{id}/trigger`; `GET /admin/{id}/runs`. Admin routes require `system_role == "admin"`. |

 Proxied through nginx: `/api/langgraph/*` → LangGraph, all other `/api/*` → Gateway.

@@ -14,7 +14,7 @@ from fastapi import APIRouter, HTTPException, Query, Request
 from fastapi.responses import FileResponse, Response, StreamingResponse
 from pydantic import BaseModel, Field

-from app.gateway.deps import get_current_user, get_local_provider, get_optional_user_from_request, get_tag_store
+from app.gateway.deps import get_current_user, get_local_provider, get_optional_user_from_request, get_run_event_store, get_run_store, get_tag_store
 from deerflow.runtime.scheduler import DuplicateTaskNameError, get_scheduled_task_service

 router = APIRouter(prefix="/api/scheduled-tasks", tags=["scheduled-tasks"])
@@ -234,6 +234,42 @@ async def get_scheduled_task_run(run_id: str, request: Request) -> dict:
    return run


+@router.get("/runs/{run_id}/messages")
+async def get_scheduled_task_run_messages(run_id: str, request: Request, limit: int = 500) -> dict:
+    """Agent messages produced by a scheduled run's underlying agent run.
+
+    Lets the run-detail page show what the agent is doing — useful when a run
+    is stuck in ``running``. Authorization reuses :func:`_load_run` (owner /
+    subscriber / admin); the event-store query then passes ``user_id=None``
+    because access has already been checked.
+    """
+    service = get_scheduled_task_service()
+    user_id = await _user_id(request)
+    run = await _load_run(service, request, run_id, user_id)
+    if run is None:
+        raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
+    agent_run_id = run.get("agent_run_id")
+    if not agent_run_id:
+        return {"messages": [], "has_more": False}
+    # Each scheduled run executes on its own throwaway runtime thread — NOT the
+    # run row's ``scheduler_thread_id`` — so resolve the real thread from the
+    # run store before querying its message events.
+    run_store = get_run_store(request)
+    agent_run = await run_store.get(agent_run_id, user_id=None)
+    if agent_run is None:
+        return {"messages": [], "has_more": False}
+    capped = max(1, min(limit, 1000))
+    event_store = get_run_event_store(request)
+    rows = await event_store.list_messages_by_run(
+        agent_run["thread_id"],
+        agent_run_id,
+        limit=capped + 1,
+        user_id=None,
+    )
+    has_more = len(rows) > capped
+    return {"messages": rows[:capped] if has_more else rows, "has_more": has_more}
+
+
@router.delete("/runs/{run_id}")
 async def delete_scheduled_task_run(run_id: str, request: Request) -> dict:
    service = get_scheduled_task_service()
@@ -241,7 +277,15 @@ async def delete_scheduled_task_run(run_id: str, request: Request) -> dict:
    try:
        deleted = await service.delete_run(user_id, run_id)
    except PermissionError as exc:
-        raise HTTPException(status_code=403, detail=str(exc))
+        # Owner-only for regular users; admins may still delete any run.
+        if await _is_admin(request):
+            deleted = await service.admin_delete_run(run_id)
+        else:
+            raise HTTPException(status_code=403, detail=str(exc))
+    # A run the caller does not own returns deleted=False — fall back to an
+    # admin-scoped delete so admins can clean up other users' runs.
+    if not deleted and await _is_admin(request):
+        deleted = await service.admin_delete_run(run_id)
    if not deleted:
        raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
    return {"success": True}
@@ -76,29 +76,39 @@ def upgrade() -> None:
            sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
        )

-    # 预置默认文章类型（幂等，跳过已存在的 key）
+    # 预置默认文章类型（幂等，跳过已存在的 key）。
+    # 用 SQLAlchemy Core 构造而非裸 SQL：``key`` 是 MySQL 保留字，裸
+    # ``SELECT key`` 会触发 1064 语法错误。Core 编译器会按方言自动给保留字
+    # 加引号（MySQL 反引号 / Postgres 双引号），保持跨后端可移植。
+    article_types = sa.table(
+        "ai_writing_article_types",
+        sa.column("id", sa.String),
+        sa.column("key", sa.String),
+        sa.column("label", sa.String),
+        sa.column("description", sa.Text),
+        sa.column("prompt_hint", sa.Text),
+        sa.column("default_word_count", sa.Integer),
+        sa.column("sort_order", sa.Integer),
+        sa.column("created_at", sa.DateTime),
+        sa.column("updated_at", sa.DateTime),
+    )
    conn = op.get_bind()
-    existing_keys = {row[0] for row in conn.execute(sa.text("SELECT key FROM ai_writing_article_types")).fetchall()}
+    existing_keys = {row[0] for row in conn.execute(sa.select(article_types.c.key)).fetchall()}
    now = datetime.now(UTC)
    for t in _DEFAULT_TYPES:
        if t["key"] not in existing_keys:
            conn.execute(
-                sa.text(
-                    "INSERT INTO ai_writing_article_types "
-                    "(id, key, label, description, prompt_hint, default_word_count, sort_order, created_at, updated_at) "
-                    "VALUES (:id, :key, :label, :description, :prompt_hint, :default_word_count, :sort_order, :created_at, :updated_at)"
-                ),
-                {
-                    "id": str(uuid.uuid4()),
-                    "key": t["key"],
-                    "label": t["label"],
-                    "description": t["description"],
-                    "prompt_hint": t["prompt_hint"],
-                    "default_word_count": t["default_word_count"],
-                    "sort_order": t["sort_order"],
-                    "created_at": now,
-                    "updated_at": now,
-                },
+                article_types.insert().values(
+                    id=str(uuid.uuid4()),
+                    key=t["key"],
+                    label=t["label"],
+                    description=t["description"],
+                    prompt_hint=t["prompt_hint"],
+                    default_word_count=t["default_word_count"],
+                    sort_order=t["sort_order"],
+                    created_at=now,
+                    updated_at=now,
+                )
            )


@@ -0,0 +1,28 @@
+"""Merge the ai-writing and thread-shares migration heads.
+
+Two independent migration branches diverged from ``20260515_01``: the AI
+writing feature (``20260517_01..03``) and the thread-shares / scheduled-task
+work (``20260518_01..05``). Merging the two branches into one repo left Alembic
+with multiple heads, so ``upgrade head`` failed. This is a no-op merge revision
+that joins both lineages — it changes no schema.
+
+Revision ID: 20260518_06
+Revises: 20260517_03, 20260518_05
+Create Date: 2026-05-18
+"""
+
+from __future__ import annotations
+
+
+revision = "20260518_06"
+down_revision = ("20260517_03", "20260518_05")
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass
@@ -85,6 +85,11 @@ class ScheduledTaskStore(ABC):
    async def delete_run(self, run_id: str, user_id: str) -> bool:
        raise NotImplementedError

+    @abstractmethod
+    async def delete_run_any(self, run_id: str) -> bool:
+        """Delete a run by id without restricting by owner. Admin-only callers."""
+        raise NotImplementedError
+
    @abstractmethod
    async def get_delivery_profile(self, user_id: str) -> dict[str, Any]:
        raise NotImplementedError
@@ -121,11 +121,21 @@ class MemoryScheduledTaskStore(ScheduledTaskStore):
        run = self._runs.get(run_id)
        if run is None or run["user_id"] != user_id:
            return False
-        self._runs.pop(run_id, None)
+        self._drop_run(run)
+        return True
+
+    async def delete_run_any(self, run_id: str) -> bool:
+        run = self._runs.get(run_id)
+        if run is None:
+            return False
+        self._drop_run(run)
+        return True
+
+    def _drop_run(self, run: dict[str, Any]) -> None:
+        self._runs.pop(run["id"], None)
        scheduled_for = run.get("scheduled_for")
        if hasattr(scheduled_for, "isoformat"):
            self._runs_by_key.discard((run["task_id"], scheduled_for.isoformat()))
-        return True

    async def get_delivery_profile(self, user_id: str) -> dict[str, Any]:
        return dict(self._profiles.get(user_id) or {"user_id": user_id, "element_user_id": None})
@@ -234,6 +234,15 @@ class ScheduledTaskRepository(ScheduledTaskStore):
            await session.commit()
            return True

+    async def delete_run_any(self, run_id: str) -> bool:
+        async with self._sf() as session:
+            row = await session.get(ScheduledTaskRunRow, run_id)
+            if row is None:
+                return False
+            await session.delete(row)
+            await session.commit()
+            return True
+
    async def get_delivery_profile(self, user_id: str) -> dict[str, Any]:
        async with self._sf() as session:
            row = await session.get(ScheduledTaskDeliveryProfileRow, user_id)
@@ -286,7 +295,14 @@ class ScheduledTaskRepository(ScheduledTaskStore):
            result = await session.execute(
                select(ScheduledTaskRow)
                .where(ScheduledTaskRow.published.is_(True))
-                .order_by(ScheduledTaskRow.published_at.desc().nullslast(), ScheduledTaskRow.created_at.desc())
+                # ``NULLS LAST`` is unsupported by MySQL — sort by an
+                # ``IS NULL`` flag instead (False<True ⇒ non-null rows first),
+                # which is portable across mysql/sqlite/postgres.
+                .order_by(
+                    ScheduledTaskRow.published_at.is_(None),
+                    ScheduledTaskRow.published_at.desc(),
+                    ScheduledTaskRow.created_at.desc(),
+                )
                .offset(max(0, offset))
                .limit(max(1, limit))
            )
@@ -87,6 +87,7 @@ class RunEventStore(abc.ABC):
        limit: int = 50,
        before_seq: int | None = None,
        after_seq: int | None = None,
+        user_id: str | None = None,
    ) -> list[dict]:
        """Return displayable messages (category=message) for a specific run, ordered by seq ascending.

@@ -152,7 +152,9 @@ class JsonlRunEventStore(RunEventStore):
            events = [e for e in events if e.get("event_type") in event_types]
        return events[:limit]

-    async def list_messages_by_run(self, thread_id, run_id, *, limit=50, before_seq=None, after_seq=None):
+    async def list_messages_by_run(self, thread_id, run_id, *, limit=50, before_seq=None, after_seq=None, user_id=None):
+        # JSONL events carry no per-event owner; user_id is accepted for
+        # interface parity with DbRunEventStore but ignored.
        events = self._read_run_events(thread_id, run_id)
        filtered = [e for e in events if e.get("category") == "message"]
        if before_seq is not None:
@@ -97,7 +97,9 @@ class MemoryRunEventStore(RunEventStore):
            filtered = [e for e in filtered if e["event_type"] in event_types]
        return filtered[:limit]

-    async def list_messages_by_run(self, thread_id, run_id, *, limit=50, before_seq=None, after_seq=None):
+    async def list_messages_by_run(self, thread_id, run_id, *, limit=50, before_seq=None, after_seq=None, user_id=None):
+        # The in-memory store keeps no per-event owner, so user_id is accepted
+        # for interface parity with DbRunEventStore but ignored.
        all_events = self._events.get(thread_id, [])
        filtered = [e for e in all_events if e["run_id"] == run_id and e["category"] == "message"]
        if before_seq is not None:
@@ -358,15 +358,20 @@ class ScheduledTaskService:
            "kind": item.get("kind"),
        }

+    @staticmethod
+    def _remove_run_archive(run_id: str) -> None:
+        """Best-effort cleanup of a run's on-disk result archive."""
+        archive_dir = get_paths().base_dir / "scheduled-task-runs" / run_id
+        if archive_dir.exists():
+            try:
+                shutil.rmtree(archive_dir)
+            except Exception:
+                logger.warning("Failed to delete scheduled task run archive %s", archive_dir, exc_info=True)
+
    async def delete_run(self, user_id: str, run_id: str) -> bool:
        deleted = await self.store.delete_run(run_id, user_id)
        if deleted:
-            archive_dir = get_paths().base_dir / "scheduled-task-runs" / run_id
-            if archive_dir.exists():
-                try:
-                    shutil.rmtree(archive_dir)
-                except Exception:
-                    logger.warning("Failed to delete scheduled task run archive %s", archive_dir, exc_info=True)
+            self._remove_run_archive(run_id)
            return True
        candidate = await self.store.get_run_any(run_id)
        if candidate is not None and await self.store.get_subscription(candidate["task_id"], user_id) is not None:
@@ -506,6 +511,13 @@ class ScheduledTaskService:
        """Delete any task regardless of owner. Admin-only callers."""
        return await self.store.delete_task_any(task_id)

+    async def admin_delete_run(self, run_id: str) -> bool:
+        """Delete a single run regardless of owner. Admin-only callers."""
+        deleted = await self.store.delete_run_any(run_id)
+        if deleted:
+            self._remove_run_archive(run_id)
+        return deleted
+
    async def admin_list_runs(self, task_id: str, *, limit: int = 50) -> list[dict[str, Any]]:
        """List runs for any task regardless of owner. Admin-only callers."""
        runs = await self.store.list_runs_any(task_id, limit=limit)
@@ -0,0 +1,77 @@
+"""Tests for admin (cross-user) deletion of scheduled-task runs.
+
+Regular users may only delete runs they own; admins may delete any run. This
+covers the service/store layer that the admin-aware ``DELETE /runs/{id}``
+route relies on.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import pytest
+
+from deerflow.persistence.scheduled_tasks import MemoryScheduledTaskStore
+from deerflow.runtime.scheduler.service import ScheduledTaskService
+
+
+@pytest.fixture
+def store() -> MemoryScheduledTaskStore:
+    return MemoryScheduledTaskStore()
+
+
+def _build_service(store: MemoryScheduledTaskStore) -> ScheduledTaskService:
+    service = ScheduledTaskService(app=None, store=store, poll_interval_seconds=60.0)
+    service._element_client = None  # noqa: SLF001
+    return service
+
+
+async def _seed_run(store: MemoryScheduledTaskStore, *, run_id: str, task_id: str, user_id: str) -> None:
+    await store.create_run_once(
+        {
+            "id": run_id,
+            "task_id": task_id,
+            "user_id": user_id,
+            "scheduler_thread_id": f"thread-{user_id}",
+            "scheduled_for": datetime.now(UTC),
+            "status": "succeeded",
+            "started_at": datetime.now(UTC),
+        }
+    )
+
+
+@pytest.mark.asyncio
+async def test_admin_delete_run_removes_any_owner_run(store):
+    await _seed_run(store, run_id="r1", task_id="t1", user_id="alice")
+    service = _build_service(store)
+
+    assert await service.admin_delete_run("r1") is True
+    assert await store.get_run_any("r1") is None
+
+
+@pytest.mark.asyncio
+async def test_admin_delete_run_missing_returns_false(store):
+    service = _build_service(store)
+    assert await service.admin_delete_run("nope") is False
+
+
+@pytest.mark.asyncio
+async def test_owner_delete_run_still_scoped_to_owner(store):
+    """A non-owner, non-subscriber deleting another user's run gets False."""
+    await _seed_run(store, run_id="r1", task_id="t1", user_id="alice")
+    service = _build_service(store)
+
+    assert await service.delete_run("bob", "r1") is False
+    assert await store.get_run_any("r1") is not None  # untouched
+
+    # The owner can still delete their own run.
+    assert await service.delete_run("alice", "r1") is True
+    assert await store.get_run_any("r1") is None
+
+
+@pytest.mark.asyncio
+async def test_store_delete_run_any_independent_of_owner(store):
+    await _seed_run(store, run_id="r1", task_id="t1", user_id="alice")
+
+    assert await store.delete_run_any("r1") is True
+    assert await store.delete_run_any("r1") is False