mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-09 16:08:31 +00:00
4fc08b4f15
* feat: add scheduled tasks MVP
* fix: harden scheduled task execution semantics
* feat(scheduled-tasks): preset-driven schedule form with timezone and live preview
Replace the raw cron input with a preset Select (hourly/daily/weekly/monthly/custom)
plus structured inputs (time picker, weekday toggles, day-of-month), datetime-local
for one-time tasks, a timezone selector defaulting to the browser timezone, and a
live human-readable preview. Reuses one ScheduledTaskScheduleInput for create and
edit; backend contract unchanged; zero new deps (pure Intl + DST-safe offset helpers).
* feat(scheduled-tasks): full-page i18n + recipe templates + E2E locale pin
Localize the rest of the scheduled-tasks page (filters, detail pane, actions,
edit form, run list, enum values) via t.scheduledTasks.* in en/zh. Add four
built-in recipe templates (GitHub Trending, news digest, issue triage, weekly
report) exposed as a chip row that pre-fills title + prompt + schedule. Pin
Playwright locale to en-US so E2E selectors stay stable against i18n. No backend
change, no new deps.
* fix(scheduled-tasks): idempotent 0003 migration, update head constants, future-date once test
Merge with main surfaced three CI failures:
- 0003_scheduled_tasks create_table collided with legacy test seeds that
build from full metadata; guard with inspector.has_table so the revision
no-ops when the table already exists (0004/0005 are already idempotent via
_helpers.py).
- persistence bootstrap concurrency/regression tests pinned HEAD to main's
0002_runs_token_usage; bump to the new head 0005_scheduled_task_thread_nullable.
- once-task router test used a fixed past run_at and tripped the
must-be-in-the-future validation; use a future date.
* address review: ok-check, 502 for trigger failure, mock fields, migration filename, doc fences
- fetchThreadScheduledTasks now checks response.ok like the other fetchers.
- trigger endpoint returns 502 (not 409) when dispatch fails outright, so
clients can distinguish a real conflict from a server-side failure.
- E2E mock normalizes scheduled-task objects with context_mode/last_thread_id
and nullable thread_id, matching the backend contract the UI renders against.
- Rename 0002_scheduled_tasks.py -> 0003_scheduled_tasks.py to match its
revision id (file was renamed in spirit already; filename now follows).
- CONFIGURATION.md: close the Tool Groups yaml fence and drop the stray fence
after the Scheduler notes so the sections render correctly.
* fix(scheduled-tasks): harden lease, poller, config, and frontend UX after review
* fix(scheduled-tasks): harden run lifecycle, overlap skip, non_interactive gating, and DST conversion after review
- defer a once task's terminal status to the run-completion hook; the task
stays running until the real outcome, and a startup sweep cancels once
tasks orphaned by a crash (launch-time 'completed' could stick forever)
- record interrupted runs as a distinct 'interrupted' run status with a
readable message; an interrupted once task ends 'cancelled', not 'failed'
- enforce overlap_policy=skip for fresh_thread_per_run via an active-run
pre-check (same-thread ConflictError can never fire across fresh threads)
- protect terminal run statuses from the late launch-path 'running' write
- honor context.non_interactive only for internally-authenticated callers;
arbitrary clients can no longer strip ask_clarification
- fix DST-stale timezone offset in zonedLocalToUtcIso by re-deriving the
offset at the resolved instant (once tasks fired an hour late around
spring-forward and the create->edit round-trip diverged)
- drop dead ScheduledTaskRunRepository.update_by_run_id; share one Gateway
API error helper between channels and scheduled-tasks frontends
* fix(scheduled-tasks): close review round-3 gaps in guards, concurrency, and API ergonomics
- scrub internal-only context keys (non_interactive) from the assembled run
config for non-internal callers: gating body.context alone left the same
key smuggle-able through the free-form body.config copied verbatim by
build_run_config
- guard update_after_launch with protect_terminal so the launch bookkeeping
write cannot clobber a once task already finalized by a fast-failing run's
completion hook (parent-row sibling of the run-row guard)
- reject a manual trigger while the task has an active run (409) instead of
launching a duplicate concurrent run on fresh_thread_per_run
- re-arm a terminal once task to enabled when PATCH pushes run_at into the
future; previously the endpoint returned 200 with a next_run_at that could
never be claimed
- make max_concurrent_runs a real global cap: each poll claims only into the
remaining budget of active (queued/running) scheduled runs
- paginate GET /scheduled-tasks/{id}/runs (limit<=200, offset) and push the
thread filter of /threads/{id}/scheduled-tasks into SQL
- stamp context.user_id on scheduler-launched runs, matching IM channels, so
user-scoped guardrail providers see the owning user
---------
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
652 lines
22 KiB
Python
652 lines
22 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from app.gateway.routers import scheduled_tasks
|
|
|
|
|
|
class _Repo:
|
|
def __init__(self) -> None:
|
|
self.created = []
|
|
self.items = {}
|
|
|
|
async def list_by_user(self, user_id: str):
|
|
return [item for item in self.items.values() if item["user_id"] == user_id]
|
|
|
|
async def list_by_user_and_thread(self, user_id: str, thread_id: str):
|
|
return [item for item in self.items.values() if item["user_id"] == user_id and item["thread_id"] == thread_id]
|
|
|
|
async def create(self, **kwargs):
|
|
item = {
|
|
"id": kwargs["task_id"],
|
|
"user_id": kwargs["user_id"],
|
|
"thread_id": kwargs["thread_id"],
|
|
"context_mode": kwargs["context_mode"],
|
|
"title": kwargs["title"],
|
|
"prompt": kwargs["prompt"],
|
|
"schedule_type": kwargs["schedule_type"],
|
|
"schedule_spec": kwargs["schedule_spec"],
|
|
"timezone": kwargs["timezone"],
|
|
"status": "enabled",
|
|
"next_run_at": kwargs["next_run_at"],
|
|
}
|
|
self.items[item["id"]] = item
|
|
self.created.append(item)
|
|
return item
|
|
|
|
async def get(self, task_id: str, *, user_id: str):
|
|
item = self.items.get(task_id)
|
|
if item is None or item["user_id"] != user_id:
|
|
return None
|
|
return item
|
|
|
|
async def update(self, task_id: str, *, user_id: str, updates):
|
|
item = await self.get(task_id, user_id=user_id)
|
|
if item is None:
|
|
return None
|
|
item.update(updates)
|
|
return item
|
|
|
|
async def delete(self, task_id: str, *, user_id: str):
|
|
item = await self.get(task_id, user_id=user_id)
|
|
if item is None:
|
|
return False
|
|
self.items.pop(task_id, None)
|
|
return True
|
|
|
|
async def list_by_task(self, task_id: str):
|
|
return []
|
|
|
|
|
|
class _Service:
|
|
def __init__(self) -> None:
|
|
self.calls = []
|
|
self.result = {"outcome": "launched"}
|
|
|
|
async def dispatch_task(self, task, *, now, trigger):
|
|
self.calls.append((task, now, trigger))
|
|
return self.result
|
|
|
|
|
|
class _RunStore:
|
|
def __init__(self, runs):
|
|
self.runs = runs
|
|
|
|
async def get(self, run_id: str, *, user_id: str):
|
|
run = self.runs.get(run_id)
|
|
if run is None or run.get("user_id") != user_id:
|
|
return None
|
|
return run
|
|
|
|
|
|
class _Config:
|
|
def __init__(self, min_once_delay_seconds: int = 60) -> None:
|
|
self.scheduler = SimpleNamespace(min_once_delay_seconds=min_once_delay_seconds)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_scheduled_task_uses_repo():
|
|
repo = _Repo()
|
|
request = SimpleNamespace()
|
|
body = scheduled_tasks.ScheduledTaskCreateRequest(
|
|
thread_id="thread-1",
|
|
title="Daily summary",
|
|
prompt="Summarize thread",
|
|
schedule_type="once",
|
|
schedule_spec={"run_at": "2027-01-01T01:00:00+00:00"},
|
|
timezone="UTC",
|
|
)
|
|
|
|
user = SimpleNamespace(id="user-1")
|
|
thread_store = SimpleNamespace(check_access=AsyncMock(return_value=True))
|
|
config = _Config()
|
|
|
|
old_repo = scheduled_tasks.get_scheduled_task_repo
|
|
old_thread_store = scheduled_tasks.get_thread_store
|
|
old_config = scheduled_tasks.get_config
|
|
old_user = scheduled_tasks.get_optional_user_from_request
|
|
try:
|
|
scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
|
|
scheduled_tasks.get_thread_store = lambda _request: thread_store
|
|
scheduled_tasks.get_config = lambda: config
|
|
scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
|
|
|
|
created = await scheduled_tasks.create_scheduled_task.__wrapped__(
|
|
request=request,
|
|
body=body,
|
|
)
|
|
finally:
|
|
scheduled_tasks.get_scheduled_task_repo = old_repo
|
|
scheduled_tasks.get_thread_store = old_thread_store
|
|
scheduled_tasks.get_config = old_config
|
|
scheduled_tasks.get_optional_user_from_request = old_user
|
|
|
|
assert created["title"] == "Daily summary"
|
|
assert created["user_id"] == "user-1"
|
|
assert created["next_run_at"] == datetime(2027, 1, 1, 1, 0, tzinfo=UTC)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_fresh_thread_task_does_not_require_thread_id():
|
|
repo = _Repo()
|
|
request = SimpleNamespace()
|
|
body = scheduled_tasks.ScheduledTaskCreateRequest(
|
|
context_mode="fresh_thread_per_run",
|
|
thread_id=None,
|
|
title="Fresh task",
|
|
prompt="Run in fresh thread",
|
|
schedule_type="cron",
|
|
schedule_spec={"cron": "0 9 * * *"},
|
|
timezone="UTC",
|
|
)
|
|
|
|
user = SimpleNamespace(id="user-1")
|
|
thread_store = SimpleNamespace(check_access=AsyncMock(return_value=True))
|
|
config = _Config()
|
|
|
|
old_repo = scheduled_tasks.get_scheduled_task_repo
|
|
old_thread_store = scheduled_tasks.get_thread_store
|
|
old_config = scheduled_tasks.get_config
|
|
old_user = scheduled_tasks.get_optional_user_from_request
|
|
try:
|
|
scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
|
|
scheduled_tasks.get_thread_store = lambda _request: thread_store
|
|
scheduled_tasks.get_config = lambda: config
|
|
scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
|
|
|
|
created = await scheduled_tasks.create_scheduled_task.__wrapped__(
|
|
request=request,
|
|
body=body,
|
|
)
|
|
finally:
|
|
scheduled_tasks.get_scheduled_task_repo = old_repo
|
|
scheduled_tasks.get_thread_store = old_thread_store
|
|
scheduled_tasks.get_config = old_config
|
|
scheduled_tasks.get_optional_user_from_request = old_user
|
|
|
|
assert created["context_mode"] == "fresh_thread_per_run"
|
|
assert created["thread_id"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trigger_scheduled_task_dispatches_manual_run():
|
|
repo = _Repo()
|
|
service = _Service()
|
|
task = await repo.create(
|
|
task_id="task-1",
|
|
user_id="user-1",
|
|
thread_id="thread-1",
|
|
context_mode="reuse_thread",
|
|
assistant_id="lead_agent",
|
|
title="Daily summary",
|
|
prompt="Summarize thread",
|
|
schedule_type="cron",
|
|
schedule_spec={"cron": "0 9 * * *"},
|
|
timezone="UTC",
|
|
next_run_at=None,
|
|
)
|
|
request = SimpleNamespace()
|
|
user = SimpleNamespace(id="user-1")
|
|
|
|
old_repo = scheduled_tasks.get_scheduled_task_repo
|
|
old_service = scheduled_tasks.get_scheduled_task_service
|
|
old_user = scheduled_tasks.get_optional_user_from_request
|
|
try:
|
|
scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
|
|
scheduled_tasks.get_scheduled_task_service = lambda _request: service
|
|
scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
|
|
|
|
result = await scheduled_tasks.trigger_scheduled_task.__wrapped__(
|
|
task_id=task["id"],
|
|
request=request,
|
|
)
|
|
finally:
|
|
scheduled_tasks.get_scheduled_task_repo = old_repo
|
|
scheduled_tasks.get_scheduled_task_service = old_service
|
|
scheduled_tasks.get_optional_user_from_request = old_user
|
|
|
|
assert result == {"id": "task-1", "triggered": True}
|
|
assert len(service.calls) == 1
|
|
assert service.calls[0][2] == "manual"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trigger_scheduled_task_returns_conflict_when_dispatch_conflicts():
|
|
repo = _Repo()
|
|
service = _Service()
|
|
service.result = {"outcome": "conflict", "error": "Thread thread-1 already has an active run"}
|
|
task = await repo.create(
|
|
task_id="task-1",
|
|
user_id="user-1",
|
|
thread_id="thread-1",
|
|
context_mode="reuse_thread",
|
|
assistant_id="lead_agent",
|
|
title="Daily summary",
|
|
prompt="Summarize thread",
|
|
schedule_type="cron",
|
|
schedule_spec={"cron": "0 9 * * *"},
|
|
timezone="UTC",
|
|
next_run_at=None,
|
|
)
|
|
request = SimpleNamespace()
|
|
user = SimpleNamespace(id="user-1")
|
|
|
|
old_repo = scheduled_tasks.get_scheduled_task_repo
|
|
old_service = scheduled_tasks.get_scheduled_task_service
|
|
old_user = scheduled_tasks.get_optional_user_from_request
|
|
try:
|
|
scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
|
|
scheduled_tasks.get_scheduled_task_service = lambda _request: service
|
|
scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
await scheduled_tasks.trigger_scheduled_task.__wrapped__(
|
|
task_id=task["id"],
|
|
request=request,
|
|
)
|
|
finally:
|
|
scheduled_tasks.get_scheduled_task_repo = old_repo
|
|
scheduled_tasks.get_scheduled_task_service = old_service
|
|
scheduled_tasks.get_optional_user_from_request = old_user
|
|
|
|
assert "already has an active run" in str(exc_info.value)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_scheduled_task_writes_repo():
|
|
repo = _Repo()
|
|
task = await repo.create(
|
|
task_id="task-1",
|
|
user_id="user-1",
|
|
thread_id="thread-1",
|
|
context_mode="reuse_thread",
|
|
assistant_id="lead_agent",
|
|
title="Daily summary",
|
|
prompt="Summarize thread",
|
|
schedule_type="cron",
|
|
schedule_spec={"cron": "0 9 * * *"},
|
|
timezone="UTC",
|
|
next_run_at=None,
|
|
)
|
|
request = SimpleNamespace()
|
|
user = SimpleNamespace(id="user-1")
|
|
config = _Config()
|
|
thread_store = SimpleNamespace(check_access=AsyncMock(return_value=True))
|
|
|
|
old_repo = scheduled_tasks.get_scheduled_task_repo
|
|
old_thread_store = scheduled_tasks.get_thread_store
|
|
old_config = scheduled_tasks.get_config
|
|
old_user = scheduled_tasks.get_optional_user_from_request
|
|
try:
|
|
scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
|
|
scheduled_tasks.get_thread_store = lambda _request: thread_store
|
|
scheduled_tasks.get_config = lambda: config
|
|
scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
|
|
|
|
result = await scheduled_tasks.update_scheduled_task.__wrapped__(
|
|
task_id=task["id"],
|
|
request=request,
|
|
body=scheduled_tasks.ScheduledTaskUpdateRequest(title="Updated title"),
|
|
)
|
|
finally:
|
|
scheduled_tasks.get_scheduled_task_repo = old_repo
|
|
scheduled_tasks.get_thread_store = old_thread_store
|
|
scheduled_tasks.get_config = old_config
|
|
scheduled_tasks.get_optional_user_from_request = old_user
|
|
|
|
assert result["title"] == "Updated title"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_scheduled_task_deletes_repo_row():
|
|
repo = _Repo()
|
|
task = await repo.create(
|
|
task_id="task-1",
|
|
user_id="user-1",
|
|
thread_id="thread-1",
|
|
context_mode="reuse_thread",
|
|
assistant_id="lead_agent",
|
|
title="Daily summary",
|
|
prompt="Summarize thread",
|
|
schedule_type="cron",
|
|
schedule_spec={"cron": "0 9 * * *"},
|
|
timezone="UTC",
|
|
next_run_at=None,
|
|
)
|
|
request = SimpleNamespace()
|
|
user = SimpleNamespace(id="user-1")
|
|
|
|
old_repo = scheduled_tasks.get_scheduled_task_repo
|
|
old_user = scheduled_tasks.get_optional_user_from_request
|
|
try:
|
|
scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
|
|
scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
|
|
|
|
result = await scheduled_tasks.delete_scheduled_task.__wrapped__(
|
|
task_id=task["id"],
|
|
request=request,
|
|
)
|
|
finally:
|
|
scheduled_tasks.get_scheduled_task_repo = old_repo
|
|
scheduled_tasks.get_optional_user_from_request = old_user
|
|
|
|
assert result == {"id": "task-1", "deleted": True}
|
|
assert repo.items == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pause_and_resume_scheduled_task_update_status():
|
|
repo = _Repo()
|
|
task = await repo.create(
|
|
task_id="task-1",
|
|
user_id="user-1",
|
|
thread_id="thread-1",
|
|
context_mode="reuse_thread",
|
|
assistant_id="lead_agent",
|
|
title="Daily summary",
|
|
prompt="Summarize thread",
|
|
schedule_type="cron",
|
|
schedule_spec={"cron": "0 9 * * *"},
|
|
timezone="UTC",
|
|
next_run_at=None,
|
|
)
|
|
request = SimpleNamespace()
|
|
user = SimpleNamespace(id="user-1")
|
|
|
|
old_repo = scheduled_tasks.get_scheduled_task_repo
|
|
old_user = scheduled_tasks.get_optional_user_from_request
|
|
try:
|
|
scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
|
|
scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
|
|
|
|
paused = await scheduled_tasks.pause_scheduled_task.__wrapped__(
|
|
task_id=task["id"],
|
|
request=request,
|
|
)
|
|
paused_status = paused["status"]
|
|
resumed = await scheduled_tasks.resume_scheduled_task.__wrapped__(
|
|
task_id=task["id"],
|
|
request=request,
|
|
)
|
|
finally:
|
|
scheduled_tasks.get_scheduled_task_repo = old_repo
|
|
scheduled_tasks.get_optional_user_from_request = old_user
|
|
|
|
assert paused_status == "paused"
|
|
assert resumed["status"] == "enabled"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pause_rejects_running_task():
|
|
repo = _Repo()
|
|
task = await repo.create(
|
|
task_id="task-1",
|
|
user_id="user-1",
|
|
thread_id="thread-1",
|
|
context_mode="reuse_thread",
|
|
assistant_id="lead_agent",
|
|
title="Daily summary",
|
|
prompt="Summarize thread",
|
|
schedule_type="cron",
|
|
schedule_spec={"cron": "0 9 * * *"},
|
|
timezone="UTC",
|
|
next_run_at=None,
|
|
)
|
|
task["status"] = "running"
|
|
request = SimpleNamespace()
|
|
user = SimpleNamespace(id="user-1")
|
|
|
|
old_repo = scheduled_tasks.get_scheduled_task_repo
|
|
old_user = scheduled_tasks.get_optional_user_from_request
|
|
try:
|
|
scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
|
|
scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
await scheduled_tasks.pause_scheduled_task.__wrapped__(
|
|
task_id=task["id"],
|
|
request=request,
|
|
)
|
|
finally:
|
|
scheduled_tasks.get_scheduled_task_repo = old_repo
|
|
scheduled_tasks.get_optional_user_from_request = old_user
|
|
|
|
assert "currently running" in str(exc_info.value)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_rejects_running_task():
|
|
repo = _Repo()
|
|
task = await repo.create(
|
|
task_id="task-1",
|
|
user_id="user-1",
|
|
thread_id="thread-1",
|
|
context_mode="reuse_thread",
|
|
assistant_id="lead_agent",
|
|
title="Daily summary",
|
|
prompt="Summarize thread",
|
|
schedule_type="cron",
|
|
schedule_spec={"cron": "0 9 * * *"},
|
|
timezone="UTC",
|
|
next_run_at=None,
|
|
)
|
|
task["status"] = "running"
|
|
request = SimpleNamespace()
|
|
user = SimpleNamespace(id="user-1")
|
|
config = _Config()
|
|
thread_store = SimpleNamespace(check_access=AsyncMock(return_value=True))
|
|
|
|
old_repo = scheduled_tasks.get_scheduled_task_repo
|
|
old_thread_store = scheduled_tasks.get_thread_store
|
|
old_config = scheduled_tasks.get_config
|
|
old_user = scheduled_tasks.get_optional_user_from_request
|
|
try:
|
|
scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
|
|
scheduled_tasks.get_thread_store = lambda _request: thread_store
|
|
scheduled_tasks.get_config = lambda: config
|
|
scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
await scheduled_tasks.update_scheduled_task.__wrapped__(
|
|
task_id=task["id"],
|
|
request=request,
|
|
body=scheduled_tasks.ScheduledTaskUpdateRequest(title="Updated title"),
|
|
)
|
|
finally:
|
|
scheduled_tasks.get_scheduled_task_repo = old_repo
|
|
scheduled_tasks.get_thread_store = old_thread_store
|
|
scheduled_tasks.get_config = old_config
|
|
scheduled_tasks.get_optional_user_from_request = old_user
|
|
|
|
assert "currently running" in str(exc_info.value)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_thread_scheduled_tasks_filters_by_thread_id():
|
|
repo = _Repo()
|
|
await repo.create(
|
|
task_id="task-1",
|
|
user_id="user-1",
|
|
thread_id="thread-1",
|
|
context_mode="reuse_thread",
|
|
assistant_id="lead_agent",
|
|
title="Thread one task",
|
|
prompt="Prompt",
|
|
schedule_type="cron",
|
|
schedule_spec={"cron": "0 9 * * *"},
|
|
timezone="UTC",
|
|
next_run_at=None,
|
|
)
|
|
await repo.create(
|
|
task_id="task-2",
|
|
user_id="user-1",
|
|
thread_id="thread-2",
|
|
context_mode="reuse_thread",
|
|
assistant_id="lead_agent",
|
|
title="Thread two task",
|
|
prompt="Prompt",
|
|
schedule_type="cron",
|
|
schedule_spec={"cron": "0 9 * * *"},
|
|
timezone="UTC",
|
|
next_run_at=None,
|
|
)
|
|
|
|
request = SimpleNamespace()
|
|
user = SimpleNamespace(id="user-1")
|
|
|
|
old_repo = scheduled_tasks.get_scheduled_task_repo
|
|
old_user = scheduled_tasks.get_optional_user_from_request
|
|
try:
|
|
scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
|
|
scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
|
|
|
|
result = await scheduled_tasks.list_thread_scheduled_tasks.__wrapped__(
|
|
thread_id="thread-1",
|
|
request=request,
|
|
)
|
|
finally:
|
|
scheduled_tasks.get_scheduled_task_repo = old_repo
|
|
scheduled_tasks.get_optional_user_from_request = old_user
|
|
|
|
assert [task["id"] for task in result] == ["task-1"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_scheduled_task_runs_returns_persisted_rows_without_side_effects():
|
|
repo = _Repo()
|
|
task = await repo.create(
|
|
task_id="task-1",
|
|
user_id="user-1",
|
|
thread_id="thread-1",
|
|
context_mode="reuse_thread",
|
|
assistant_id="lead_agent",
|
|
title="Task",
|
|
prompt="Prompt",
|
|
schedule_type="cron",
|
|
schedule_spec={"cron": "0 9 * * *"},
|
|
timezone="UTC",
|
|
next_run_at=None,
|
|
)
|
|
run_repo = SimpleNamespace(
|
|
list_by_task=AsyncMock(
|
|
return_value=[
|
|
{
|
|
"id": "task-run-1",
|
|
"task_id": "task-1",
|
|
"thread_id": "thread-1",
|
|
"run_id": "run-1",
|
|
"status": "running",
|
|
"error": None,
|
|
}
|
|
]
|
|
),
|
|
)
|
|
request = SimpleNamespace()
|
|
user = SimpleNamespace(id="user-1")
|
|
|
|
old_task_repo = scheduled_tasks.get_scheduled_task_repo
|
|
old_run_repo = scheduled_tasks.get_scheduled_task_run_repo
|
|
old_user = scheduled_tasks.get_optional_user_from_request
|
|
try:
|
|
scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
|
|
scheduled_tasks.get_scheduled_task_run_repo = lambda _request: run_repo
|
|
scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
|
|
|
|
result = await scheduled_tasks.list_scheduled_task_runs.__wrapped__(
|
|
task_id=task["id"],
|
|
request=request,
|
|
)
|
|
finally:
|
|
scheduled_tasks.get_scheduled_task_repo = old_task_repo
|
|
scheduled_tasks.get_scheduled_task_run_repo = old_run_repo
|
|
scheduled_tasks.get_optional_user_from_request = old_user
|
|
|
|
assert result[0]["status"] == "running"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_once_task_enforces_minimum_delay():
|
|
repo = _Repo()
|
|
request = SimpleNamespace()
|
|
body = scheduled_tasks.ScheduledTaskCreateRequest(
|
|
thread_id="thread-1",
|
|
title="Soon task",
|
|
prompt="Run soon",
|
|
schedule_type="once",
|
|
schedule_spec={"run_at": (datetime.now(UTC) + timedelta(seconds=30)).isoformat()},
|
|
timezone="UTC",
|
|
)
|
|
user = SimpleNamespace(id="user-1")
|
|
thread_store = SimpleNamespace(check_access=AsyncMock(return_value=True))
|
|
config = _Config(min_once_delay_seconds=60)
|
|
|
|
old_repo = scheduled_tasks.get_scheduled_task_repo
|
|
old_thread_store = scheduled_tasks.get_thread_store
|
|
old_config = scheduled_tasks.get_config
|
|
old_user = scheduled_tasks.get_optional_user_from_request
|
|
try:
|
|
scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
|
|
scheduled_tasks.get_thread_store = lambda _request: thread_store
|
|
scheduled_tasks.get_config = lambda: config
|
|
scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
await scheduled_tasks.create_scheduled_task.__wrapped__(
|
|
request=request,
|
|
body=body,
|
|
)
|
|
finally:
|
|
scheduled_tasks.get_scheduled_task_repo = old_repo
|
|
scheduled_tasks.get_thread_store = old_thread_store
|
|
scheduled_tasks.get_config = old_config
|
|
scheduled_tasks.get_optional_user_from_request = old_user
|
|
|
|
assert "once schedule must be at least" in str(exc_info.value)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_terminal_once_task_with_future_run_at_rearms_it():
|
|
"""PATCHing a fresh future run_at onto a completed/failed/cancelled once
|
|
task must reset status to enabled — claim_due_tasks only admits enabled
|
|
rows, so keeping the terminal status returns a next_run_at that never fires."""
|
|
repo = _Repo()
|
|
task = await repo.create(
|
|
task_id="task-terminal",
|
|
user_id="user-1",
|
|
thread_id=None,
|
|
context_mode="fresh_thread_per_run",
|
|
assistant_id="lead_agent",
|
|
title="Once done",
|
|
prompt="p",
|
|
schedule_type="once",
|
|
schedule_spec={"run_at": "2026-07-01T00:00:00+00:00"},
|
|
timezone="UTC",
|
|
next_run_at=None,
|
|
)
|
|
task["status"] = "completed"
|
|
future_run_at = (datetime.now(UTC) + timedelta(hours=1)).isoformat()
|
|
request = SimpleNamespace()
|
|
user = SimpleNamespace(id="user-1")
|
|
|
|
old_repo = scheduled_tasks.get_scheduled_task_repo
|
|
old_config = scheduled_tasks.get_config
|
|
old_user = scheduled_tasks.get_optional_user_from_request
|
|
try:
|
|
scheduled_tasks.get_scheduled_task_repo = lambda _request: repo
|
|
scheduled_tasks.get_config = lambda: _Config()
|
|
scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user)
|
|
|
|
result = await scheduled_tasks.update_scheduled_task.__wrapped__(
|
|
task_id=task["id"],
|
|
request=request,
|
|
body=scheduled_tasks.ScheduledTaskUpdateRequest(schedule_spec={"run_at": future_run_at}),
|
|
)
|
|
finally:
|
|
scheduled_tasks.get_scheduled_task_repo = old_repo
|
|
scheduled_tasks.get_config = old_config
|
|
scheduled_tasks.get_optional_user_from_request = old_user
|
|
|
|
assert result["status"] == "enabled"
|
|
assert result["next_run_at"] is not None
|