吉鑫升andClaude Opus 4.7
ba2bc46042
- 前端：新会话选 Agent 后直接跳转并自动发送，智能体页输入框宽度与首页对齐 - 前端：澄清卡片支持多选+多问题合并提交，消除单题逐一发送的割裂感 - 前端：定时任务弹窗 Agent 下拉支持长描述截断，避免撑破布局 - 后端：ask_clarification 支持同回合多次调用以提多个独立问题 - 后端：LLM 错误处理与定时任务失败提示统一改为中文 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1201 lines
51 KiB
Python
1201 lines
51 KiB
Python
"""Scheduled task orchestration service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import mimetypes
|
|
import re
|
|
import shutil
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from croniter import croniter
|
|
from langgraph.checkpoint.base import empty_checkpoint
|
|
|
|
from deerflow.config.agents_config import AGENT_ID_PATTERN
|
|
from deerflow.config.paths import get_paths
|
|
from deerflow.persistence.scheduled_tasks import ScheduledTaskStore
|
|
from deerflow.runtime import serialize_channel_values
|
|
from deerflow.runtime.scheduler.element import ElementDeliveryClient
|
|
from deerflow.runtime.user_context import reset_current_user, set_current_user
|
|
from deerflow.utils.time import now_iso
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# A single execution attempt may not exceed this wall-clock budget. When it
|
|
# does, the run is treated as stuck (the agent loop hung, a tool blocked, etc.)
|
|
# and retried. The watchdog uses the same window to recover runs orphaned by a
|
|
# crash or restart — those stay "running" in the DB with no in-process owner.
|
|
_RUN_ATTEMPT_TIMEOUT_SECONDS = 30 * 60 # 30 minutes
|
|
# Retries allowed after the first attempt before the run is marked "failed".
|
|
_MAX_RUN_RETRIES = 3
|
|
|
|
|
|
class DuplicateTaskNameError(Exception):
|
|
"""A user already owns a scheduled task with the requested name.
|
|
|
|
Task names are unique per user (``uq_scheduled_tasks_user_name``). The
|
|
service raises this instead of letting the raw DB ``IntegrityError`` bubble
|
|
up so the API layer can return a friendly 409 rather than a 500.
|
|
"""
|
|
|
|
def __init__(self, name: str) -> None:
|
|
self.name = name
|
|
super().__init__(f"A scheduled task named {name!r} already exists")
|
|
|
|
|
|
@dataclass
|
|
class _RuntimeUser:
|
|
id: str
|
|
|
|
|
|
def _coerce_datetime(value: datetime | str | None) -> datetime | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, datetime):
|
|
dt = value
|
|
else:
|
|
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=UTC)
|
|
return dt.astimezone(UTC)
|
|
|
|
|
|
def _normalize_timezone(timezone: str | None) -> str:
|
|
tz = (timezone or "UTC").strip() or "UTC"
|
|
try:
|
|
ZoneInfo(tz)
|
|
return tz
|
|
except Exception:
|
|
return "UTC"
|
|
|
|
|
|
def _schedule_to_cron(schedule: str) -> str:
|
|
value = schedule.strip()
|
|
if re.fullmatch(r"(\S+\s+){4}\S+", value):
|
|
return value
|
|
|
|
lower = value.lower()
|
|
hour_min = re.search(r"(?:(?:at|@)\s*)?(\d{1,2})(?::(\d{2}))?\s*(am|pm)?", lower)
|
|
cn_hour_min = re.search(r"(\d{1,2})\s*[点時时](?:(\d{1,2})\s*分?)?", value)
|
|
|
|
hour = 9
|
|
minute = 0
|
|
ampm = None
|
|
if hour_min:
|
|
hour = int(hour_min.group(1))
|
|
minute = int(hour_min.group(2) or 0)
|
|
ampm = hour_min.group(3)
|
|
elif cn_hour_min:
|
|
hour = int(cn_hour_min.group(1))
|
|
minute = int(cn_hour_min.group(2) or 0)
|
|
|
|
if ampm == "pm" and hour < 12:
|
|
hour += 12
|
|
if ampm == "am" and hour == 12:
|
|
hour = 0
|
|
hour = max(0, min(23, hour))
|
|
minute = max(0, min(59, minute))
|
|
|
|
if any(token in lower for token in ("weekday", "weekdays", "workday")) or "工作日" in value:
|
|
return f"{minute} {hour} * * 1-5"
|
|
|
|
weekdays = {
|
|
"monday": 1,
|
|
"tuesday": 2,
|
|
"wednesday": 3,
|
|
"thursday": 4,
|
|
"friday": 5,
|
|
"saturday": 6,
|
|
"sunday": 0,
|
|
"周一": 1,
|
|
"星期一": 1,
|
|
"周二": 2,
|
|
"星期二": 2,
|
|
"周三": 3,
|
|
"星期三": 3,
|
|
"周四": 4,
|
|
"星期四": 4,
|
|
"周五": 5,
|
|
"星期五": 5,
|
|
"周六": 6,
|
|
"星期六": 6,
|
|
"周日": 0,
|
|
"周天": 0,
|
|
"星期日": 0,
|
|
"星期天": 0,
|
|
}
|
|
for token, weekday in weekdays.items():
|
|
if token in lower or token in value:
|
|
return f"{minute} {hour} * * {weekday}"
|
|
|
|
every_hours = re.search(r"every\s+(\d+)\s*h", lower)
|
|
cn_every_hours = re.search(r"每隔\s*(\d+)\s*小?时", value)
|
|
if every_hours or cn_every_hours:
|
|
hours = int((every_hours or cn_every_hours).group(1))
|
|
hours = max(1, min(23, hours))
|
|
return f"0 */{hours} * * *"
|
|
|
|
return f"{minute} {hour} * * *"
|
|
|
|
|
|
def _next_run_at(cron_expr: str, timezone: str, base: datetime | None = None) -> datetime:
|
|
tz = ZoneInfo(timezone)
|
|
base_local = (base or datetime.now(UTC)).astimezone(tz)
|
|
next_local = croniter(cron_expr, base_local).get_next(datetime)
|
|
if next_local.tzinfo is None:
|
|
next_local = next_local.replace(tzinfo=tz)
|
|
return next_local.astimezone(UTC)
|
|
|
|
|
|
def _looks_like_markdown(content: str) -> bool:
|
|
markdown_patterns = (
|
|
r"^#{1,6}\s+",
|
|
r"^\s*[-*+]\s+",
|
|
r"^\s*\d+\.\s+",
|
|
r"```",
|
|
r"\|.+\|",
|
|
r"\[[^\]]+\]\([^)]+\)",
|
|
)
|
|
return any(re.search(pattern, content, flags=re.MULTILINE) for pattern in markdown_patterns)
|
|
|
|
|
|
def _build_result(
|
|
task: dict[str, Any],
|
|
*,
|
|
scheduled_for: datetime,
|
|
content: str | None,
|
|
files: list[dict[str, Any]] | None = None,
|
|
) -> dict[str, Any]:
|
|
result_content = content or f"Scheduled task completed: {task['name']}"
|
|
task_tz = ZoneInfo(task.get("timezone") or "UTC")
|
|
local_time = scheduled_for.astimezone(task_tz)
|
|
return {
|
|
"title": f"{task['name']} - {local_time.strftime('%Y-%m-%d %H:%M')}",
|
|
"content": result_content,
|
|
"format": "markdown" if _looks_like_markdown(result_content) else "text",
|
|
"files": files or [],
|
|
"created_at": datetime.now(UTC).isoformat(),
|
|
}
|
|
|
|
|
|
class ScheduledTaskService:
|
|
def __init__(self, app: Any, store: ScheduledTaskStore, *, poll_interval_seconds: float = 30.0) -> None:
|
|
self.app = app
|
|
self.store = store
|
|
self.poll_interval_seconds = poll_interval_seconds
|
|
self._loop_task: asyncio.Task | None = None
|
|
self._stopping = asyncio.Event()
|
|
self._running_task_ids: set[str] = set()
|
|
# All scheduled jobs deliver into one per-user scheduler thread. The
|
|
# current RunManager rejects concurrent runs on the same thread, so
|
|
# serialize starts here instead of letting reports fail with conflicts.
|
|
self._agent_run_lock = asyncio.Lock()
|
|
self._element_client = ElementDeliveryClient.from_env()
|
|
|
|
async def start(self) -> None:
|
|
if self._loop_task is None or self._loop_task.done():
|
|
self._stopping.clear()
|
|
self._loop_task = asyncio.create_task(self._run_loop(), name="scheduled-task-loop")
|
|
|
|
async def stop(self) -> None:
|
|
self._stopping.set()
|
|
if self._loop_task is not None:
|
|
self._loop_task.cancel()
|
|
try:
|
|
await self._loop_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
async def ensure_scheduler_thread(self, user_id: str) -> dict[str, Any]:
|
|
thread = await self.store.get_or_create_scheduler_thread(user_id, lambda: str(uuid.uuid4()))
|
|
await self._ensure_thread_meta(user_id, thread["thread_id"])
|
|
return thread
|
|
|
|
async def list_tasks(self, user_id: str) -> list[dict[str, Any]]:
|
|
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]]:
|
|
accessible = await self._accessible_task(user_id, task_id)
|
|
if accessible is None:
|
|
return []
|
|
runs = await self.store.list_runs(task_id, accessible["user_id"], limit=limit)
|
|
can_edit = accessible.get("role") == "owner"
|
|
for run in runs:
|
|
run["can_edit"] = can_edit
|
|
return runs
|
|
|
|
async def get_run(self, user_id: str, run_id: str) -> dict[str, Any] | None:
|
|
run = await self.store.get_run(run_id, user_id)
|
|
if run is not None:
|
|
run["can_edit"] = True
|
|
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
|
|
candidate["can_edit"] = False
|
|
return candidate
|
|
|
|
async def update_run_file(
|
|
self,
|
|
user_id: str,
|
|
run_id: str,
|
|
file_index: int,
|
|
new_content: str,
|
|
) -> dict[str, Any]:
|
|
"""Overwrite an archived run file in place. Owner-only.
|
|
|
|
Returns updated file metadata. Raises:
|
|
- PermissionError if the caller is not the run owner (subscribers blocked)
|
|
- FileNotFoundError if the run or file index does not exist
|
|
- ValueError if the file is not editable (binary / not previewable)
|
|
"""
|
|
run = await self.store.get_run(run_id, user_id)
|
|
if run is None:
|
|
candidate = await self.store.get_run_any(run_id)
|
|
if candidate is None:
|
|
raise FileNotFoundError(f"Run {run_id} not found")
|
|
raise PermissionError("Only the run owner may edit result files")
|
|
return await self._apply_run_file_edit(run, run_id, file_index, new_content)
|
|
|
|
async def admin_update_run_file(
|
|
self,
|
|
run_id: str,
|
|
file_index: int,
|
|
new_content: str,
|
|
) -> dict[str, Any]:
|
|
"""Owner-less variant of :meth:`update_run_file` used by the public viewer.
|
|
|
|
The public scheduled-task page is unauthenticated, so it cannot supply a
|
|
``user_id``. Mutations from that page route here instead. The disk-write
|
|
and result-row update logic is identical to the owner path.
|
|
"""
|
|
run = await self.store.get_run_any(run_id)
|
|
if run is None:
|
|
raise FileNotFoundError(f"Run {run_id} not found")
|
|
return await self._apply_run_file_edit(run, run_id, file_index, new_content)
|
|
|
|
async def admin_update_run_body(self, run_id: str, new_content: str) -> dict[str, Any]:
|
|
"""Overwrite the inline Markdown body (``result.content``) of a run.
|
|
|
|
Used by the public scheduled-task viewer so anonymous editors can fix
|
|
typos in the "正文" tab. Unlike :meth:`admin_update_run_file`, no disk
|
|
artifact backs the body — it lives entirely in the run row.
|
|
"""
|
|
run = await self.store.get_run_any(run_id)
|
|
if run is None:
|
|
raise FileNotFoundError(f"Run {run_id} not found")
|
|
result = run.get("result") if isinstance(run.get("result"), dict) else {}
|
|
new_result = {**result, "content": new_content}
|
|
await self.store.update_run(run_id, {"result": new_result})
|
|
return {"content": new_content}
|
|
|
|
async def _apply_run_file_edit(
|
|
self,
|
|
run: dict[str, Any],
|
|
run_id: str,
|
|
file_index: int,
|
|
new_content: str,
|
|
) -> dict[str, Any]:
|
|
result = run.get("result") or {}
|
|
files = result.get("files") if isinstance(result, dict) else None
|
|
if not isinstance(files, list) or file_index < 0 or file_index >= len(files):
|
|
raise FileNotFoundError(f"File index {file_index} not found in run {run_id}")
|
|
item = files[file_index]
|
|
if not isinstance(item, dict) or not isinstance(item.get("path"), str):
|
|
raise FileNotFoundError(f"File index {file_index} has no archive path")
|
|
|
|
archive_path = Path(item["path"])
|
|
if not archive_path.exists() or not archive_path.is_file():
|
|
raise FileNotFoundError(f"Archived file {archive_path} is no longer available")
|
|
|
|
# Constrain edits to text-previewable files. We mirror the same rule the
|
|
# router uses for previews so binary files cannot be silently corrupted.
|
|
suffix = archive_path.suffix.lower()
|
|
mime_type = item.get("mime_type")
|
|
is_text = (
|
|
(isinstance(mime_type, str) and (mime_type.startswith("text/") or mime_type in {"application/json", "application/xml"}))
|
|
or suffix in {".md", ".markdown", ".txt", ".json", ".csv", ".xml", ".yaml", ".yml", ".log"}
|
|
)
|
|
if not is_text:
|
|
raise ValueError("Only text-previewable files (markdown, txt, json, etc.) can be edited")
|
|
|
|
# Make sure we stay inside the per-run archive root before touching disk.
|
|
archive_root = (get_paths().base_dir / "scheduled-task-runs" / run_id).resolve()
|
|
try:
|
|
archive_path.resolve().relative_to(archive_root)
|
|
except ValueError as exc:
|
|
raise PermissionError("Refusing to edit file outside the run archive") from exc
|
|
|
|
archive_path.write_text(new_content, encoding="utf-8", newline="")
|
|
new_size = archive_path.stat().st_size
|
|
|
|
# Reflect the new size in the run row so subsequent list/preview calls
|
|
# report the latest figure. Other fields stay untouched.
|
|
files[file_index] = {**item, "size": new_size}
|
|
new_result = {**result, "files": files}
|
|
await self.store.update_run(run_id, {"result": new_result})
|
|
|
|
return {
|
|
"index": file_index,
|
|
"name": item.get("name") or archive_path.name,
|
|
"size": new_size,
|
|
"mime_type": mime_type,
|
|
"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:
|
|
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:
|
|
raise PermissionError("Only the run owner may delete scheduled task runs")
|
|
return deleted
|
|
|
|
async def get_delivery_profile(self, user_id: str) -> dict[str, Any]:
|
|
return await self.store.get_delivery_profile(user_id)
|
|
|
|
async def update_delivery_profile(self, user_id: str, data: dict[str, Any]) -> dict[str, Any]:
|
|
return await self.store.update_delivery_profile(user_id, data)
|
|
|
|
async def _assert_name_available(self, user_id: str, name: str, *, exclude_task_id: str | None = None) -> None:
|
|
"""Raise :class:`DuplicateTaskNameError` if ``user_id`` already owns ``name``.
|
|
|
|
Enforces ``uq_scheduled_tasks_user_name`` proactively so callers get a
|
|
clean conflict error instead of a raw DB ``IntegrityError``. The check
|
|
compares the trimmed name; ``exclude_task_id`` skips the task being
|
|
renamed so an unchanged name on update is not flagged as a collision.
|
|
"""
|
|
for existing in await self.store.list_tasks(user_id):
|
|
if exclude_task_id is not None and existing.get("task_id") == exclude_task_id:
|
|
continue
|
|
if (existing.get("name") or "").strip() == name:
|
|
raise DuplicateTaskNameError(name)
|
|
|
|
async def create_task(
|
|
self,
|
|
*,
|
|
user_id: str,
|
|
name: str,
|
|
prompt: str,
|
|
schedule: str,
|
|
timezone: str | None = None,
|
|
description: str | None = None,
|
|
source_thread_id: str | None = None,
|
|
source_agent_name: str | None = None,
|
|
execution_agent_name: str | None = None,
|
|
execution_context: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
tz = _normalize_timezone(timezone)
|
|
cron_expr = _schedule_to_cron(schedule)
|
|
next_run_at = _next_run_at(cron_expr, tz)
|
|
final_name = name.strip() or "Scheduled task"
|
|
await self._assert_name_available(user_id, final_name)
|
|
scheduler_thread = await self.ensure_scheduler_thread(user_id)
|
|
context = dict(execution_context or {})
|
|
if execution_agent_name:
|
|
context["agent_name"] = execution_agent_name
|
|
normalized_description = description.strip() if isinstance(description, str) else None
|
|
task = await self.store.create_task(
|
|
{
|
|
"task_id": str(uuid.uuid4()),
|
|
"user_id": user_id,
|
|
"scheduler_thread_id": scheduler_thread["thread_id"],
|
|
"source_thread_id": source_thread_id,
|
|
"source_agent_name": source_agent_name,
|
|
"execution_agent_name": execution_agent_name,
|
|
"name": final_name,
|
|
"description": normalized_description or None,
|
|
"prompt": prompt.strip(),
|
|
"schedule_text": schedule.strip(),
|
|
"cron_expr": cron_expr,
|
|
"timezone": tz,
|
|
"enabled": True,
|
|
"next_run_at": next_run_at,
|
|
"execution_context": context,
|
|
}
|
|
)
|
|
return task
|
|
|
|
@staticmethod
|
|
def _build_task_update(data: dict[str, Any], existing: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""Translate an API update payload into a storage update dict.
|
|
|
|
Shared by owner-scoped :meth:`update_task` and admin :meth:`admin_update_task`
|
|
so both apply identical schedule/timezone normalization.
|
|
"""
|
|
update_data: dict[str, Any] = {}
|
|
for key in ("name", "prompt", "enabled", "execution_agent_name"):
|
|
if key in data and data[key] is not None:
|
|
update_data[key] = data[key]
|
|
# Description is nullable — accept ``None`` (clear) and empty strings
|
|
# (also clear) explicitly, so the field can be removed via PATCH.
|
|
if "description" in data:
|
|
raw = data["description"]
|
|
if isinstance(raw, str):
|
|
trimmed = raw.strip()
|
|
update_data["description"] = trimmed or None
|
|
else:
|
|
update_data["description"] = None
|
|
# Names are matched against the per-user uniqueness constraint, so keep
|
|
# the stored value trimmed to match create_task's normalization.
|
|
if isinstance(update_data.get("name"), str):
|
|
update_data["name"] = update_data["name"].strip() or "Scheduled task"
|
|
timezone = data.get("timezone")
|
|
if timezone is not None:
|
|
update_data["timezone"] = _normalize_timezone(timezone)
|
|
if data.get("schedule") is not None:
|
|
schedule = str(data["schedule"])
|
|
cron_expr = _schedule_to_cron(schedule)
|
|
tz = update_data.get("timezone") or (existing or {}).get("timezone") or "UTC"
|
|
update_data["schedule_text"] = schedule
|
|
update_data["cron_expr"] = cron_expr
|
|
update_data["next_run_at"] = _next_run_at(cron_expr, tz)
|
|
if data.get("execution_context") is not None:
|
|
update_data["execution_context"] = data["execution_context"]
|
|
# Re-enabling a task must also schedule it forward from now, otherwise a
|
|
# paused task with a stale/empty next_run_at would never fire again.
|
|
if update_data.get("enabled") is True and "next_run_at" not in update_data:
|
|
cron = (existing or {}).get("cron_expr")
|
|
if cron:
|
|
tz = update_data.get("timezone") or (existing or {}).get("timezone") or "UTC"
|
|
update_data["next_run_at"] = _next_run_at(cron, tz)
|
|
return update_data
|
|
|
|
async def update_task(self, user_id: str, task_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
|
|
existing = await self.store.get_task(task_id, user_id)
|
|
if existing is None:
|
|
return None
|
|
update_data = self._build_task_update(data, existing)
|
|
if "name" in update_data:
|
|
await self._assert_name_available(user_id, update_data["name"], exclude_task_id=task_id)
|
|
return await self.store.update_task(task_id, user_id, update_data)
|
|
|
|
async def delete_task(self, user_id: str, task_id: str) -> bool:
|
|
return await self.store.delete_task(task_id, user_id)
|
|
|
|
# --- Admin (cross-user management) ---
|
|
|
|
async def admin_list_tasks(self) -> list[dict[str, Any]]:
|
|
"""List every scheduled task across all users. Admin-only callers."""
|
|
tasks = await self.store.list_all_tasks()
|
|
for task in tasks:
|
|
task["role"] = "owner"
|
|
return tasks
|
|
|
|
async def admin_update_task(self, task_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
|
|
"""Update any task regardless of owner. Admin-only callers."""
|
|
existing = await self.store.get_task_any(task_id)
|
|
if existing is None:
|
|
return None
|
|
update_data = self._build_task_update(data, existing)
|
|
if "name" in update_data:
|
|
await self._assert_name_available(existing["user_id"], update_data["name"], exclude_task_id=task_id)
|
|
return await self.store.update_task_any(task_id, update_data)
|
|
|
|
async def admin_delete_task(self, task_id: str) -> bool:
|
|
"""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)
|
|
for run in runs:
|
|
run["can_edit"] = True
|
|
return runs
|
|
|
|
async def admin_get_run(self, run_id: str) -> dict[str, Any] | None:
|
|
"""Fetch a single run regardless of owner. Admin-only callers."""
|
|
run = await self.store.get_run_any(run_id)
|
|
if run is not None:
|
|
run["can_edit"] = True
|
|
return run
|
|
|
|
async def admin_trigger_task(self, task_id: str) -> dict[str, Any] | None:
|
|
"""Trigger an immediate run of any task regardless of owner."""
|
|
task = await self.store.get_task_any(task_id)
|
|
if task is None:
|
|
return None
|
|
asyncio.create_task(self._execute_tracked(task, scheduled_for=datetime.now(UTC), manual=True))
|
|
return task
|
|
|
|
async def pause_task(self, user_id: str, task_id: str) -> dict[str, Any] | None:
|
|
return await self.store.update_task(task_id, user_id, {"enabled": False})
|
|
|
|
async def resume_task(self, user_id: str, task_id: str) -> dict[str, Any] | None:
|
|
existing = await self.store.get_task(task_id, user_id)
|
|
if existing is None:
|
|
return None
|
|
next_run_at = _next_run_at(existing["cron_expr"], existing.get("timezone") or "UTC")
|
|
return await self.store.update_task(task_id, user_id, {"enabled": True, "next_run_at": next_run_at})
|
|
|
|
async def trigger_task(self, user_id: str, task_id: str) -> dict[str, Any] | None:
|
|
task = await self.store.get_task(task_id, user_id)
|
|
if task is None:
|
|
return None
|
|
asyncio.create_task(self._execute_tracked(task, scheduled_for=datetime.now(UTC), manual=True))
|
|
return task
|
|
|
|
async def _run_loop(self) -> None:
|
|
while not self._stopping.is_set():
|
|
# Recover hung / orphaned runs first so a single stuck task cannot
|
|
# silently wedge the scheduler indefinitely.
|
|
try:
|
|
await self._reconcile_stuck_runs()
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
logger.exception("Scheduled task watchdog scan failed")
|
|
try:
|
|
due = await self.store.due_tasks(datetime.now(UTC), limit=20)
|
|
for task in due:
|
|
task_id = task["task_id"]
|
|
if task_id in self._running_task_ids:
|
|
continue
|
|
self._running_task_ids.add(task_id)
|
|
asyncio.create_task(self._execute_and_untrack(task))
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
logger.exception("Scheduled task scan failed")
|
|
try:
|
|
await asyncio.wait_for(self._stopping.wait(), timeout=self.poll_interval_seconds)
|
|
except TimeoutError:
|
|
pass
|
|
|
|
async def _execute_and_untrack(self, task: dict[str, Any]) -> None:
|
|
try:
|
|
scheduled_for = _coerce_datetime(task.get("next_run_at")) or datetime.now(UTC)
|
|
await self._execute_task(task, scheduled_for=scheduled_for, manual=False)
|
|
finally:
|
|
self._running_task_ids.discard(task["task_id"])
|
|
|
|
async def _execute_tracked(self, task: dict[str, Any], *, scheduled_for: datetime, manual: bool) -> None:
|
|
"""Run a task while marking it busy so the watchdog won't double-drive it."""
|
|
self._running_task_ids.add(task["task_id"])
|
|
try:
|
|
await self._execute_task(task, scheduled_for=scheduled_for, manual=manual)
|
|
finally:
|
|
self._running_task_ids.discard(task["task_id"])
|
|
|
|
async def _reconcile_stuck_runs(self) -> None:
|
|
"""Recover runs left in the ``running`` state with no in-process owner.
|
|
|
|
A run becomes stuck when the agent loop hangs past the per-attempt
|
|
timeout, or when the process crashes / restarts mid-run (the original
|
|
asyncio task is gone but the DB row still says ``running``). For each
|
|
such run we retry up to ``_MAX_RUN_RETRIES`` times, then mark it
|
|
``failed`` so it no longer shows as perpetually executing.
|
|
"""
|
|
cutoff = datetime.now(UTC) - timedelta(seconds=_RUN_ATTEMPT_TIMEOUT_SECONDS)
|
|
stuck = await self.store.list_stuck_runs(cutoff, limit=20)
|
|
for run in stuck:
|
|
task_id = run["task_id"]
|
|
# A run still owned by a live in-process execution is handled by
|
|
# that execution's own timeout/retry loop — don't touch it here.
|
|
if task_id in self._running_task_ids:
|
|
continue
|
|
self._running_task_ids.add(task_id)
|
|
asyncio.create_task(self._recover_and_untrack(run))
|
|
|
|
async def _recover_and_untrack(self, run: dict[str, Any]) -> None:
|
|
try:
|
|
await self._recover_stuck_run(run)
|
|
except Exception:
|
|
logger.exception("Failed to recover stuck scheduled task run %s", run.get("id"))
|
|
finally:
|
|
self._running_task_ids.discard(run["task_id"])
|
|
|
|
async def _recover_stuck_run(self, run: dict[str, Any]) -> None:
|
|
run_id = run["id"]
|
|
retry_count = int(run.get("retry_count") or 0)
|
|
task = await self.store.get_task_any(run["task_id"])
|
|
scheduled_for = _coerce_datetime(run.get("scheduled_for")) or datetime.now(UTC)
|
|
if task is None:
|
|
# Parent task was deleted — just close out the orphaned run.
|
|
await self.store.update_run(
|
|
run_id,
|
|
{"status": "failed", "finished_at": datetime.now(UTC), "error": "执行超时：所属任务已被删除"},
|
|
)
|
|
return
|
|
if retry_count >= _MAX_RUN_RETRIES:
|
|
await self._fail_run(
|
|
task,
|
|
run_id,
|
|
scheduled_for,
|
|
error=f"执行超时：超过 {_RUN_ATTEMPT_TIMEOUT_SECONDS // 60} 分钟未完成，重试 {_MAX_RUN_RETRIES} 次后仍失败",
|
|
retry_count=retry_count,
|
|
)
|
|
return
|
|
logger.warning("Recovering stuck scheduled task run %s (retry budget %s/%s)", run_id, retry_count, _MAX_RUN_RETRIES)
|
|
await self._execute_task(task, scheduled_for=scheduled_for, manual=False, run_row=run)
|
|
|
|
async def _fail_run(
|
|
self,
|
|
task: dict[str, Any],
|
|
run_id: str,
|
|
scheduled_for: datetime,
|
|
*,
|
|
error: str,
|
|
retry_count: int = 0,
|
|
) -> None:
|
|
"""Mark a run failed and reflect the outcome on the parent task."""
|
|
result = _build_result(task, scheduled_for=scheduled_for, content=f"定时任务执行失败：{error}")
|
|
await self.store.update_run(
|
|
run_id,
|
|
{"status": "failed", "finished_at": datetime.now(UTC), "error": error, "result": result, "retry_count": retry_count},
|
|
)
|
|
await self.store.update_task(task["task_id"], task["user_id"], {"last_run_at": datetime.now(UTC), "last_status": "failed"})
|
|
|
|
async def _execute_task(
|
|
self,
|
|
task: dict[str, Any],
|
|
*,
|
|
scheduled_for: datetime,
|
|
manual: bool,
|
|
run_row: dict[str, Any] | None = None,
|
|
) -> None:
|
|
# Each run uses its own throwaway thread so message history cannot accumulate
|
|
# across runs and blow past the model's input window. The per-user
|
|
# task["scheduler_thread_id"] is no longer used to drive the LLM call —
|
|
# we keep it on the task row only for backwards compatibility.
|
|
is_recovery = run_row is not None
|
|
if run_row is None:
|
|
run_row = await self.store.create_run_once(
|
|
{
|
|
"id": str(uuid.uuid4()),
|
|
"task_id": task["task_id"],
|
|
"user_id": task["user_id"],
|
|
"scheduler_thread_id": str(uuid.uuid4()),
|
|
"scheduled_for": scheduled_for,
|
|
"status": "running",
|
|
"started_at": datetime.now(UTC),
|
|
}
|
|
)
|
|
if run_row is None:
|
|
return
|
|
else:
|
|
# Recovery path: re-claim an existing (orphaned) run row.
|
|
await self.store.update_run(run_row["id"], {"status": "running", "started_at": datetime.now(UTC)})
|
|
|
|
# Advancing the schedule only happens once, when the run is first
|
|
# created — recovery retries reuse the original scheduled_for.
|
|
if not manual and not is_recovery:
|
|
next_run_at = _next_run_at(task["cron_expr"], task.get("timezone") or "UTC", base=scheduled_for)
|
|
await self.store.update_task(task["task_id"], task["user_id"], {"next_run_at": next_run_at, "last_status": "running"})
|
|
|
|
run_id = run_row["id"]
|
|
retry_count = int(run_row.get("retry_count") or 0)
|
|
|
|
while True:
|
|
runtime_thread_id = str(uuid.uuid4())
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
self._run_agent_attempt(task, runtime_thread_id=runtime_thread_id, scheduled_for=scheduled_for, run_id=run_id),
|
|
timeout=_RUN_ATTEMPT_TIMEOUT_SECONDS,
|
|
)
|
|
except TimeoutError:
|
|
# The attempt overran the 30-minute cap. Retry a bounded number
|
|
# of times, then give up and mark the run failed.
|
|
if retry_count >= _MAX_RUN_RETRIES:
|
|
await self._fail_run(
|
|
task,
|
|
run_id,
|
|
scheduled_for,
|
|
error=f"执行超时：单次执行超过 {_RUN_ATTEMPT_TIMEOUT_SECONDS // 60} 分钟，重试 {_MAX_RUN_RETRIES} 次后仍失败",
|
|
retry_count=retry_count,
|
|
)
|
|
return
|
|
retry_count += 1
|
|
logger.warning("Scheduled task %s timed out; retry %s/%s", task["task_id"], retry_count, _MAX_RUN_RETRIES)
|
|
await self.store.update_run(
|
|
run_id,
|
|
{
|
|
"retry_count": retry_count,
|
|
"status": "running",
|
|
"started_at": datetime.now(UTC),
|
|
"error": f"执行超时，正在重试（第 {retry_count}/{_MAX_RUN_RETRIES} 次）",
|
|
},
|
|
)
|
|
continue
|
|
except Exception as exc:
|
|
logger.exception("Scheduled task %s failed", task["task_id"])
|
|
await self._fail_run(task, run_id, scheduled_for, error=str(exc), retry_count=retry_count)
|
|
return
|
|
|
|
await self.store.update_run(
|
|
run_id,
|
|
{"status": "succeeded", "finished_at": datetime.now(UTC), "result": result, "retry_count": retry_count, "error": None},
|
|
)
|
|
await self.store.update_task(task["task_id"], task["user_id"], {"last_run_at": datetime.now(UTC), "last_status": "succeeded"})
|
|
await self._deliver_element_result(task, result["content"])
|
|
return
|
|
|
|
async def _run_agent_attempt(
|
|
self,
|
|
task: dict[str, Any],
|
|
*,
|
|
runtime_thread_id: str,
|
|
scheduled_for: datetime,
|
|
run_id: str,
|
|
) -> dict[str, Any]:
|
|
"""Run the agent once and build the result payload.
|
|
|
|
Wrapped in :func:`asyncio.wait_for` by the caller so a hung agent loop
|
|
cannot hold the shared run lock — and the whole scheduler — forever.
|
|
"""
|
|
await self._ensure_thread_meta(task["user_id"], runtime_thread_id)
|
|
async with self._agent_run_lock:
|
|
output_snapshot = self._snapshot_result_files(runtime_thread_id, task["user_id"])
|
|
record = await self._start_agent_run(task, scheduled_for=scheduled_for, runtime_thread_id=runtime_thread_id)
|
|
await self.store.update_run(run_id, {"agent_run_id": record.run_id})
|
|
if record.task is not None:
|
|
await record.task
|
|
result_text = await self._extract_latest_ai_text(runtime_thread_id)
|
|
result_files = await self._extract_result_files(
|
|
runtime_thread_id,
|
|
task["user_id"],
|
|
run_id=run_id,
|
|
output_snapshot=output_snapshot,
|
|
)
|
|
return _build_result(task, scheduled_for=scheduled_for, content=result_text, files=result_files)
|
|
|
|
async def _ensure_thread_meta(self, user_id: str, thread_id: str) -> None:
|
|
token = set_current_user(_RuntimeUser(id=user_id))
|
|
try:
|
|
thread_store = self.app.state.thread_store
|
|
existing = await thread_store.get(thread_id)
|
|
if existing is None:
|
|
await thread_store.create(
|
|
thread_id,
|
|
assistant_id="lead_agent",
|
|
display_name="定时任务助手",
|
|
metadata={"thread_type": "scheduler", "system": True},
|
|
)
|
|
checkpointer = self.app.state.checkpointer
|
|
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
|
await checkpointer.aput(
|
|
config,
|
|
empty_checkpoint(),
|
|
{
|
|
"step": -1,
|
|
"source": "input",
|
|
"writes": None,
|
|
"parents": {},
|
|
"created_at": now_iso(),
|
|
"thread_type": "scheduler",
|
|
},
|
|
{},
|
|
)
|
|
finally:
|
|
reset_current_user(token)
|
|
|
|
async def _start_agent_run(self, task: dict[str, Any], *, scheduled_for: datetime, runtime_thread_id: str):
|
|
from app.gateway.routers.thread_runs import RunCreateRequest
|
|
from app.gateway.services import start_run
|
|
|
|
token = set_current_user(_RuntimeUser(id=task["user_id"]))
|
|
try:
|
|
request = SimpleNamespace(app=self.app)
|
|
execution_context = dict(task.get("execution_context") or {})
|
|
stored_agent_ref = task.get("execution_agent_name")
|
|
if stored_agent_ref:
|
|
resolved_agent_id = await self._resolve_agent_id(stored_agent_ref, task["user_id"])
|
|
if resolved_agent_id:
|
|
execution_context["agent_name"] = resolved_agent_id
|
|
else:
|
|
logger.warning(
|
|
"Scheduled task %s references unknown agent %r; falling back to default lead agent",
|
|
task.get("task_id"),
|
|
stored_agent_ref,
|
|
)
|
|
execution_context.update(
|
|
{
|
|
"is_scheduled_run": True,
|
|
"thread_id": runtime_thread_id,
|
|
"thinking_enabled": execution_context.get("thinking_enabled", True),
|
|
"is_plan_mode": execution_context.get("is_plan_mode", True),
|
|
}
|
|
)
|
|
content = (
|
|
f"【定时任务触发】{task['name']}\n"
|
|
f"计划时间：{scheduled_for.astimezone(ZoneInfo(task.get('timezone') or 'UTC')).isoformat()}\n\n"
|
|
f"{task['prompt']}"
|
|
)
|
|
body = RunCreateRequest(
|
|
assistant_id="lead_agent",
|
|
input={"messages": [{"type": "human", "content": content}]},
|
|
metadata={"scheduled_task_id": task["task_id"], "scheduled_task_name": task["name"]},
|
|
context=execution_context,
|
|
multitask_strategy="enqueue" if False else "reject",
|
|
)
|
|
return await start_run(body, runtime_thread_id, request) # type: ignore[arg-type]
|
|
finally:
|
|
reset_current_user(token)
|
|
|
|
async def _resolve_agent_id(self, stored_ref: str, user_id: str) -> str | None:
|
|
"""Resolve a scheduled task's ``execution_agent_name`` to a stable agent id.
|
|
|
|
Older tasks (created before the frontend was fixed) stored the agent's
|
|
Chinese **display name** here; the runtime validator only accepts ids
|
|
matching ``^[A-Za-z0-9_-]+$``, so those rows would 422 forever. We
|
|
gracefully recover by looking the stored value up first as an id, then
|
|
as a display name. Returns ``None`` when nothing visible matches — the
|
|
caller then falls back to the default lead agent rather than crashing.
|
|
"""
|
|
store = getattr(self.app.state, "agent_store", None)
|
|
if store is None:
|
|
return stored_ref if AGENT_ID_PATTERN.fullmatch(stored_ref) else None
|
|
if AGENT_ID_PATTERN.fullmatch(stored_ref):
|
|
try:
|
|
row = await store.get_visible(stored_ref, user_id)
|
|
except Exception:
|
|
row = None
|
|
if row is not None:
|
|
return row["id"]
|
|
try:
|
|
agents = await store.list_visible(user_id)
|
|
except Exception:
|
|
return None
|
|
for agent in agents:
|
|
if agent.get("name") == stored_ref:
|
|
return agent.get("id")
|
|
return None
|
|
|
|
async def _extract_latest_ai_text(self, thread_id: str) -> str | None:
|
|
checkpointer = self.app.state.checkpointer
|
|
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
|
checkpoint_tuple = await checkpointer.aget_tuple(config)
|
|
if checkpoint_tuple is None:
|
|
return None
|
|
checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {}
|
|
channel_values = checkpoint.get("channel_values", {})
|
|
values = serialize_channel_values(channel_values)
|
|
messages = values.get("messages") or []
|
|
for message in reversed(messages):
|
|
if not isinstance(message, dict):
|
|
continue
|
|
message_type = message.get("type") or message.get("role")
|
|
if message_type not in {"ai", "assistant"}:
|
|
continue
|
|
content = message.get("content")
|
|
if isinstance(content, str):
|
|
return content
|
|
if isinstance(content, list):
|
|
parts: list[str] = []
|
|
for part in content:
|
|
if isinstance(part, dict) and part.get("type") == "text" and isinstance(part.get("text"), str):
|
|
parts.append(part["text"])
|
|
if parts:
|
|
return "\n".join(parts)
|
|
return None
|
|
|
|
def _result_scan_dirs(self, thread_id: str, user_id: str) -> list[tuple[str, Path]]:
|
|
paths = get_paths()
|
|
return [
|
|
("/mnt/user-data/outputs", paths.sandbox_outputs_dir(thread_id, user_id=user_id)),
|
|
("/mnt/user-data/workspace", paths.sandbox_work_dir(thread_id, user_id=user_id)),
|
|
]
|
|
|
|
def _snapshot_result_files(self, thread_id: str, user_id: str) -> dict[str, tuple[int, int]]:
|
|
snapshot: dict[str, tuple[int, int]] = {}
|
|
for _, directory in self._result_scan_dirs(thread_id, user_id):
|
|
if not directory.exists() or not directory.is_dir():
|
|
continue
|
|
try:
|
|
for path in directory.rglob("*"):
|
|
if not path.is_file():
|
|
continue
|
|
stat = path.stat()
|
|
snapshot[str(path.resolve())] = (stat.st_mtime_ns, stat.st_size)
|
|
except Exception:
|
|
logger.debug("Failed to snapshot scheduled task result dir %s", directory, exc_info=True)
|
|
return snapshot
|
|
|
|
def _virtual_result_path(self, virtual_prefix: str, result_dir: Path, path: Path) -> str | None:
|
|
try:
|
|
relative = path.resolve().relative_to(result_dir.resolve())
|
|
except ValueError:
|
|
return None
|
|
return f"{virtual_prefix}/{relative.as_posix()}"
|
|
|
|
def _collect_changed_output_files(
|
|
self,
|
|
thread_id: str,
|
|
user_id: str,
|
|
output_snapshot: dict[str, tuple[int, int]],
|
|
) -> list[tuple[str, Path]]:
|
|
changed: list[tuple[str, Path]] = []
|
|
for virtual_prefix, directory in self._result_scan_dirs(thread_id, user_id):
|
|
if not directory.exists() or not directory.is_dir():
|
|
continue
|
|
try:
|
|
for path in sorted(directory.rglob("*"), key=lambda item: item.as_posix()):
|
|
if not path.is_file():
|
|
continue
|
|
resolved = path.resolve()
|
|
stat = path.stat()
|
|
previous = output_snapshot.get(str(resolved))
|
|
if previous == (stat.st_mtime_ns, stat.st_size):
|
|
continue
|
|
virtual_path = self._virtual_result_path(virtual_prefix, directory, resolved)
|
|
if virtual_path is not None:
|
|
changed.append((virtual_path, resolved))
|
|
except Exception:
|
|
logger.debug("Failed to collect changed scheduled task result files from %s", directory, exc_info=True)
|
|
return changed
|
|
|
|
def _is_in_result_scan_dirs(self, thread_id: str, user_id: str, path: Path) -> bool:
|
|
try:
|
|
resolved = path.resolve()
|
|
except Exception:
|
|
return False
|
|
for _, directory in self._result_scan_dirs(thread_id, user_id):
|
|
try:
|
|
resolved.relative_to(directory.resolve())
|
|
return True
|
|
except ValueError:
|
|
continue
|
|
return False
|
|
|
|
def _archive_result_file(self, run_id: str, path: Path) -> Path | None:
|
|
if not path.exists() or not path.is_file():
|
|
return None
|
|
|
|
archive_dir = get_paths().base_dir / "scheduled-task-runs" / run_id / "files"
|
|
archive_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
candidate = archive_dir / path.name
|
|
if candidate.exists():
|
|
stem = path.stem or "file"
|
|
suffix = path.suffix
|
|
index = 2
|
|
while candidate.exists():
|
|
candidate = archive_dir / f"{stem}-{index}{suffix}"
|
|
index += 1
|
|
|
|
shutil.copy2(path, candidate)
|
|
return candidate
|
|
|
|
def _build_file_result(self, *, virtual_path: str, source_path: Path, archive_path: Path) -> dict[str, Any]:
|
|
mime_type, _ = mimetypes.guess_type(archive_path.name)
|
|
suffix = archive_path.suffix.lower()
|
|
return {
|
|
"name": archive_path.name,
|
|
"path": str(archive_path),
|
|
"source_path": str(source_path),
|
|
"virtual_path": virtual_path,
|
|
"mime_type": mime_type or ("text/markdown" if suffix == ".md" else None),
|
|
"size": archive_path.stat().st_size if archive_path.exists() and archive_path.is_file() else None,
|
|
"kind": "markdown" if suffix == ".md" else "artifact",
|
|
}
|
|
|
|
def _is_new_or_changed_output_file(
|
|
self,
|
|
thread_id: str,
|
|
user_id: str,
|
|
path: Path,
|
|
output_snapshot: dict[str, tuple[int, int]],
|
|
) -> bool:
|
|
if not self._is_in_result_scan_dirs(thread_id, user_id, path):
|
|
return True
|
|
|
|
resolved = path.resolve()
|
|
if not resolved.exists() or not resolved.is_file():
|
|
return False
|
|
stat = resolved.stat()
|
|
return output_snapshot.get(str(resolved)) != (stat.st_mtime_ns, stat.st_size)
|
|
|
|
async def _extract_result_files(
|
|
self,
|
|
thread_id: str,
|
|
user_id: str,
|
|
*,
|
|
run_id: str,
|
|
output_snapshot: dict[str, tuple[int, int]] | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
checkpointer = self.app.state.checkpointer
|
|
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
|
checkpoint_tuple = await checkpointer.aget_tuple(config)
|
|
artifacts: list[Any] = []
|
|
if checkpoint_tuple is not None:
|
|
checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {}
|
|
channel_values = checkpoint.get("channel_values", {})
|
|
values = serialize_channel_values(channel_values)
|
|
raw_artifacts = values.get("artifacts") or []
|
|
if isinstance(raw_artifacts, list):
|
|
artifacts = raw_artifacts
|
|
|
|
result_files: list[dict[str, Any]] = []
|
|
seen: set[str] = set()
|
|
|
|
def add_file(virtual_path: str, source_path: Path) -> None:
|
|
resolved_source = source_path.resolve()
|
|
seen_key = str(resolved_source)
|
|
if seen_key in seen:
|
|
return
|
|
seen.add(seen_key)
|
|
try:
|
|
archive_path = self._archive_result_file(run_id, resolved_source)
|
|
except Exception:
|
|
logger.debug("Failed to archive scheduled task artifact %s", resolved_source, exc_info=True)
|
|
return
|
|
if archive_path is None:
|
|
return
|
|
result_files.append(self._build_file_result(virtual_path=virtual_path, source_path=resolved_source, archive_path=archive_path))
|
|
|
|
for artifact in artifacts:
|
|
if not isinstance(artifact, str) or not artifact:
|
|
continue
|
|
try:
|
|
actual_path = get_paths().resolve_virtual_path(thread_id, artifact, user_id=user_id)
|
|
except TypeError:
|
|
actual_path = get_paths().resolve_virtual_path(thread_id, artifact)
|
|
except Exception:
|
|
logger.debug("Failed to resolve scheduled task artifact %s", artifact, exc_info=True)
|
|
continue
|
|
path = Path(actual_path)
|
|
if self._is_new_or_changed_output_file(thread_id, user_id, path, output_snapshot or {}):
|
|
add_file(artifact, path)
|
|
|
|
for virtual_path, path in self._collect_changed_output_files(thread_id, user_id, output_snapshot or {}):
|
|
add_file(virtual_path, path)
|
|
|
|
return result_files
|
|
|
|
async def _deliver_element_result(self, task: dict[str, Any], result_text: str) -> None:
|
|
if self._element_client is None:
|
|
return
|
|
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
|
|
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(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)
|