github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-08-01 19:06:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(schedule): delete the superseded pre-hexagonal implementation

Browse Source 
Remove the legacy stack the hexagonal slice replaced, now that nothing
assembles it: app/scheduler/service.py (the old orchestration),
app/gateway/routers/scheduled_tasks.py (the old dict-returning router,
no longer mounted), the deerflow/scheduler package (its cron/timezone
rules live in ScheduleSpec), the dict-returning repositories in
persistence/scheduled_task*/sql.py, and the deps.py providers and
app.state wiring that served them. The ORM rows and the
uq_scheduled_task_run_active partial unique index stay -- the table
definitions live with the shared engine/alembic infrastructure and the
schedule adapters are their only readers and writers.

The legacy test suites go with the code they pinned; every scenario has
a counterpart in the new suites. The one suite that guarded semantics
rather than the old implementation -- the real-database dispatch-race
TOCTOU tests -- is migrated to the new stack as
test_schedule_dispatch_race.py, driving ScheduleService over the real
SQL adapters with the same barrier, natural-timing, and index-semantics
cases.

Docs and comments that named the old classes as the current wiring
(backend/AGENTS.md, reload_boundary.py, channel/service comments) now
name the composition-root wiring instead.
This commit is contained in:
rayhpeng 2026-07-29 19:56:51 +08:00
parent a6388bc967
commit 345b046a1c
31 changed files with 262 additions and 3387 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
backend/AGENTS.md
View File

@ -17,7 +17,7 @@ DeerFlow is a LangGraph-based AI super agent system with a full-stack architectu
- Gateway streams `write_file` and `str_replace` argument deltas in bounded batches when clients also subscribe to `values`; messages-only consumers retain the original per-chunk contract, while `values` preserves the complete tool call.
- With `stream_subgraphs`, subgraph frames keep their namespace in the SSE event name (`values|<ns>`, LangGraph Platform style) instead of impersonating root frames — a delegated subagent inherits the parent checkpoint namespace, so publishing its `values` snapshot as bare `values` replaces the whole thread view in SDK clients (#4399). Root-only consumers (file-tool chunk batcher, subagent event persistence, LLM error-fallback detection) ignore namespaced frames. The web frontend does not request subgraph streaming; subtask progress rides root-namespace `task_*` custom events.
- Scheduled-task executions must reuse that same Gateway run lifecycle. The scheduler may decide *when* work runs, but it must dispatch through the existing run path rather than introducing a parallel execution stack.
- Scheduled-task dispatch enforces "at most one active run per task when `overlap_policy=skip`" at the DB layer via the partial unique index `uq_scheduled_task_run_active` (`scheduled_task_runs.task_id WHERE status IN ('queued','running')`). `ScheduledTaskService.dispatch_task`'s `has_active_runs` check is a non-atomic fast path (its own session, separated from the `create()` insert by `await` points), so two concurrent dispatches — a manual `POST /scheduled-tasks/{id}/trigger` racing the poller, a double-click, or a client retry — can both pass it; the index is the atomic arbiter, and the losing `create` surfaces as `ActiveScheduledRunConflict` (translated from `IntegrityError` in the repository) and collapses to the same outcome as the fast path (manual → 409 conflict, scheduled → a `"skipped"` tombstone). The scheduled-skip tombstone is created directly as terminal `"skipped"` (not a transient `"queued"`) so it never occupies the active slot the pre-existing run still holds. Sibling of the `runs` table's `uq_runs_thread_active` (PR #4003), which keys on `thread_id` and so does not cover the default `fresh_thread_per_run` context where every dispatch gets a new thread. Index is status-only, not `overlap_policy`-conditional (the policy is fixed to `"skip"` in the MVP).
- Scheduled-task dispatch enforces "at most one active run per task when `overlap_policy=skip`" at the DB layer via the partial unique index `uq_scheduled_task_run_active` (`scheduled_task_runs.task_id WHERE status IN ('queued','running')`). `ScheduleService.dispatch_task`'s `has_active` check is a non-atomic fast path (its own session, separated from the queued-record insert by `await` points), so two concurrent dispatches — a manual `POST /scheduled-tasks/{id}/trigger` racing the poller, a double-click, or a client retry — can both pass it; the index is the atomic arbiter, and the losing insert surfaces as `ActiveRunConflictError` (translated from `IntegrityError` in the `SqlScheduledRunRepository` adapter) and collapses to the same outcome as the fast path (manual → 409 conflict, scheduled → a `"skipped"` tombstone). The scheduled-skip tombstone is created directly as terminal `"skipped"` (not a transient `"queued"`) so it never occupies the active slot the pre-existing run still holds. Sibling of the `runs` table's `uq_runs_thread_active` (PR #4003), which keys on `thread_id` and so does not cover the default `fresh_thread_per_run` context where every dispatch gets a new thread. Index is status-only, not `overlap_policy`-conditional (the policy is fixed to `"skip"` in the MVP).
- The dispatch path and the run-completion hook write to the same `scheduled_tasks` row concurrently, so **the two writes own disjoint field sets and neither may write through the whole aggregate**. A run that fails fast reaches its completion hook before the dispatch path's own bookkeeping write lands, so the hook is holding a snapshot taken *before* it. The launch write (`record_launch`) owns the schedule — `next_run_at`, `last_run_at`, `last_run_id`, `last_thread_id`, `run_count`, the claim — and the completion write (`record_completion`) owns only the verdict: the terminal status and `last_error`. Expressing the completion as a read-modify-write through the aggregate replays the stale snapshot and rolls the launch write back, restoring an elapsed `next_run_at` and leaving the task in `running` with no live claim — a shape neither `claim_due` branch nor `cancel_stuck_once_tasks` can reach, i.e. a cron task that is permanently unschedulable with no recovery path. `record_launch`'s `protect_terminal` flag closes the mirror-image race in the other direction (the hook's terminal status must survive a launch write landing after it). Pinned by the `TestRecordCompletion` contract cases (both the in-memory double and real sqlite) and by `test_a_cron_task_survives_a_launch_write_landing_mid_completion`.
**Project Structure**:
@ -743,7 +743,7 @@ Lets a caller pass per-request, short-lived end-user credentials (e.g. an ERP to
- **Declare**: a skill lists the secrets it needs in `SKILL.md` frontmatter — `required-secrets:` as a string list or `{name, optional}` mappings. `name` is both the lookup key and the env var name exposed to scripts. Parsed by `skills/parser.py::parse_required_secrets` into `Skill.required_secrets` (`SecretRequirement`); malformed entries are dropped with a warning.
- **Carry**: the caller sends values out-of-band in the run request's `context.secrets` mapping (never a message). `runtime/secret_context.py` owns the contract (`SECRETS_CONTEXT_KEY`, `extract_request_secrets`). The existing `context` passthrough carries it to `runtime.context` without mirroring into `configurable`. `build_run_config` still sets `configurable.thread_id` on the context path — the checkpointer requires it. MCP tool interceptors can read the same live carrier from `langgraph.config.get_config()["context"]["secrets"]`; see `docs/MCP_SERVER.md`.
- **Admission and redaction ownership**: `services.py::start_run()` validates both legacy request mappings, `metadata.auth_token` and `config.metadata.auth_token`, before any run or thread persistence. `runtime/secret_context.py::redact_config_secrets()` also removes nested config metadata secrets from observable and persisted config copies; historical `RunResponse.kwargs` applies the same redaction non-mutatively, leaving stored `RunRecord` data unchanged. Keep callers on `config.context.secrets` rather than adding another credential carrier. Scheduled task definitions have no durable credential carrier: `ScheduledTaskService` supplies only `scheduled_task_id`, `scheduled_task_run_id`, and `scheduled_trigger` as run metadata.
- **Admission and redaction ownership**: `services.py::start_run()` validates both legacy request mappings, `metadata.auth_token` and `config.metadata.auth_token`, before any run or thread persistence. `runtime/secret_context.py::redact_config_secrets()` also removes nested config metadata secrets from observable and persisted config copies; historical `RunResponse.kwargs` applies the same redaction non-mutatively, leaving stored `RunRecord` data unchanged. Keep callers on `config.context.secrets` rather than adding another credential carrier. Scheduled task definitions have no durable credential carrier: the schedule dispatch path supplies only `scheduled_task_id`, `scheduled_task_run_id`, and `scheduled_trigger` as run metadata.
- **Bind (point A+)**: `SkillActivationMiddleware._resolve_secret_bindings` recomputes the injection set (`runtime.context[__active_skill_secrets]`) on every model call from two unioned sources, then REPLACES the key. (1) *Slash*: the run's most recent `/skill` activation, persisted as a source on the run context (only the activated skill's **canonical container path**, never its declared secrets) so the whole tool loop after the activation call keeps the binding; a new activation replaces it. Slash reads the genuine user text via `get_original_user_content_text`; `InputSanitizationMiddleware` preserves it (`ORIGINAL_USER_CONTENT_KEY`), so activation fires even after sanitization. (2) *In-context* (autonomous invocation): skills the model actually loaded in this thread — `ThreadState.skill_context` entries. **Both sources resolve the live registry skill by normalized container path on every call** (`_resolve_registry_skill`) and bind only that skill's own declared secrets — enabled + allowlist checked for both; the `secrets-autonomous: false` opt-out (malformed values fail closed to `false`) additionally gates the in-context path but exempts explicit slash. Resolving by registry — not by trusting the source's stored data — is what makes a caller-forged `__slash_skill_secret_source` harmless (`runtime.context` is caller-mergeable; the gateway also strips caller `__`-keys in `build_run_config`), #3938. Authorization is three-gated regardless of activation style: skill **enabled** by the operator × values **supplied per-request** by the caller (`context.secrets`) × names **declared** in frontmatter (∩ semantics). Because the set is recomputed per call, a skill evicted from `skill_context` (capacity) or a caller that stops supplying a value loses injection on the next call. The injected value always comes from the caller's request, never the host environment (scrubbed first — see below), so a declared name that also exists in the host env is safe: the caller's value wins and the host value is dropped (the #3861 per-user-key-overrides-shared-key case). Missing required secrets are logged once per binding change, not injected; binding changes are recorded as a `middleware:skill_secrets` journal event (skill and secret names only, never values).
- **Inject**: `bash_tool` reads the injection set and passes it as `execute_command(env=...)`. Scope is the activation turn/run only — a run without `/skill` activation injects nothing.
- **AIO image requirement**: on `AioSandbox` the env path uses the `bash.exec` API (`POST /v1/bash/exec`), which upstream all-in-one-sandbox only ships since `1.9.3` — older images (including a `latest` tag frozen on the `1.0.0.x` line) 404 the whole `/v1/bash/*` namespace. `AioSandbox` detects the 404, remembers the capability gap on the instance, and fails fast with an actionable upgrade error instead of letting the model retry raw 404s; there is deliberately **no** fallback through the legacy shell path because none keeps the secret values out of the command string (#3921). Regression tests: `tests/test_aio_sandbox.py::TestBashExecUnsupportedFailFast`.
@ -832,7 +832,7 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_
**GitHub event-driven agents** (webhook-driven IM channel):
- Custom agents declare a `github:` block in their `config.yaml` to bind to repos and event triggers; the webhook route is fail-closed by default (mounted only when `GITHUB_WEBHOOK_SECRET` is set) and exempt from auth/CSRF because authenticity is enforced by HMAC.
- Outbound is **log-only** by design: each agent posts its own reply mid-run via the `gh` CLI from its sandbox, so the manager uses `fire_and_forget=True` and `runs.create()` returns once pending.
- **Follow-up buffering while busy** (issue #4121): because outbound is log-only, the pre-existing `THREAD_BUSY_MESSAGE` reply on a `ConflictError` was invisible to the commenter — a comment posted while a run was already active looked like it had been silently ignored. When `ChannelRunPolicy.buffer_followups_on_busy=True` (GitHub's default), a `ConflictError` on `runs.create()` now also appends the triggering message to a per-thread, in-memory buffer (`ChannelManager._followup_buffers`) — deduped by GitHub delivery id, capped at `FOLLOWUP_BUFFER_MAX_PER_THREAD` (20, oldest dropped with a WARNING log on overflow). The first successful `runs.create()` on a thread now captures its `run_id` and spawns a background watcher that subscribes to that run's `StreamBridge` stream; once it observes `END_SENTINEL`, the watcher drains up to `FOLLOWUP_DRAIN_BATCH_SIZE` (10) buffered entries into one `<followups-while-busy>`-wrapped input and fires a follow-up `runs.create()` — itself watched the same way, so a backlog deeper than one batch chains into further drain cycles instead of growing one unbounded input. If that follow-up `runs.create()` itself hits `ConflictError` (e.g. a manual Web UI turn or a scheduled run raced onto the same thread), the batch is requeued rather than lost, and is retried whenever this manager next successfully creates and watches a run on that thread. Reactions/acknowledgment (e.g. GitHub's `eyes`/`confused` reaction API) on buffered comments are intentionally **out of scope** for this mechanism and left to a follow-up — comments are coalesced silently. **Plumbing**: the watcher needs the Gateway's `StreamBridge` singleton, which `ChannelManager` did not previously have access to; it is threaded from `app.py`'s lifespan (where `app.state.stream_bridge` is already set by `langgraph_runtime`) through `start_channel_service(get_stream_bridge=...)` → `ChannelService.__init__` → `ChannelManager.__init__`, as a zero-arg closure mirroring the existing `launch_run=lambda **kwargs: launch_scheduled_thread_run(app=app, **kwargs)` pattern used for `ScheduledTaskService` in the same lifespan function. A `ChannelManager` constructed without it (e.g. directly in a test) still buffers safely — it just has no watcher to auto-drain. **Scope limitation**: the buffer and watcher state are per-process, in-memory. Under `GATEWAY_WORKERS>1` or multi-pod, a follow-up comment routed to a different worker process than the one running the busy thread's agent will not see that buffer. This is a known, deliberately deferred limitation with the same shape as the cross-pod gap described for issue #4120 (a shared buffer store or IM-leader election would be needed to close it) — single-process/single-pod deployments, the safe default, see no correctness issue from this, only the documented per-process scope.
- **Follow-up buffering while busy** (issue #4121): because outbound is log-only, the pre-existing `THREAD_BUSY_MESSAGE` reply on a `ConflictError` was invisible to the commenter — a comment posted while a run was already active looked like it had been silently ignored. When `ChannelRunPolicy.buffer_followups_on_busy=True` (GitHub's default), a `ConflictError` on `runs.create()` now also appends the triggering message to a per-thread, in-memory buffer (`ChannelManager._followup_buffers`) — deduped by GitHub delivery id, capped at `FOLLOWUP_BUFFER_MAX_PER_THREAD` (20, oldest dropped with a WARNING log on overflow). The first successful `runs.create()` on a thread now captures its `run_id` and spawns a background watcher that subscribes to that run's `StreamBridge` stream; once it observes `END_SENTINEL`, the watcher drains up to `FOLLOWUP_DRAIN_BATCH_SIZE` (10) buffered entries into one `<followups-while-busy>`-wrapped input and fires a follow-up `runs.create()` — itself watched the same way, so a backlog deeper than one batch chains into further drain cycles instead of growing one unbounded input. If that follow-up `runs.create()` itself hits `ConflictError` (e.g. a manual Web UI turn or a scheduled run raced onto the same thread), the batch is requeued rather than lost, and is retried whenever this manager next successfully creates and watches a run on that thread. Reactions/acknowledgment (e.g. GitHub's `eyes`/`confused` reaction API) on buffered comments are intentionally **out of scope** for this mechanism and left to a follow-up — comments are coalesced silently. **Plumbing**: the watcher needs the Gateway's `StreamBridge` singleton, which `ChannelManager` did not previously have access to; it is threaded from `app.py`'s lifespan (where `app.state.stream_bridge` is already set by `langgraph_runtime`) through `start_channel_service(get_stream_bridge=...)` → `ChannelService.__init__` → `ChannelManager.__init__`, as a zero-arg closure mirroring the existing `launch_run=lambda **kwargs: launch_scheduled_thread_run(app=app, **kwargs)` pattern used for the schedule composition root in the same lifespan function. A `ChannelManager` constructed without it (e.g. directly in a test) still buffers safely — it just has no watcher to auto-drain. **Scope limitation**: the buffer and watcher state are per-process, in-memory. Under `GATEWAY_WORKERS>1` or multi-pod, a follow-up comment routed to a different worker process than the one running the busy thread's agent will not see that buffer. This is a known, deliberately deferred limitation with the same shape as the cross-pod gap described for issue #4120 (a shared buffer store or IM-leader election would be needed to close it) — single-process/single-pod deployments, the safe default, see no correctness issue from this, only the documented per-process scope.
- See [backend/docs/GITHUB_AGENTS.md](docs/GITHUB_AGENTS.md) for the architecture diagrams: webhook → fan-out → `InboundMessage` dispatch, `preferred_thread_id = UUID5(repo, number, agent_name)` thread determinism, mention-handle precedence chain, GH token lifecycle via `GH_TOKEN`/`GITHUB_TOKEN` per-call `extra_env`, and the narrow `ConflictError` (HTTP 409) thread-create race recovery.

2
backend/app/channels/manager.py
View File

@ -930,7 +930,7 @@ class ChannelManager:
self._require_bound_identity = require_bound_identity
# Zero-arg accessor for the FastAPI app's StreamBridge singleton,
# threaded in from app.py's lifespan via start_channel_service() ->
# ChannelService.__init__ (mirrors how ScheduledTaskService gets a
# ChannelService.__init__ (mirrors how the schedule composition root gets a
# launch_run closure over `app` in the same lifespan function). None
# when not wired (e.g. a ChannelManager constructed directly in
# tests) — follow-up buffering still works, but no watcher is

2
backend/app/channels/service.py
View File

@ -441,7 +441,7 @@ async def start_channel_service(
``ChannelRunPolicy.buffer_followups_on_busy`` (currently GitHub) can watch
a run's completion and auto-drain buffered follow-ups. ``app.py``'s
lifespan passes a closure over ``app.state.stream_bridge`` here, the same
pattern it already uses for ``ScheduledTaskService``'s ``launch_run``.
pattern it already uses for the schedule composition root's ``launch_run``.
"""
global _channel_service
if _channel_service is not None:

2
backend/app/gateway/app.py
View File

@ -287,7 +287,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
try:
from app.channels.service import start_channel_service
# Closure over `app` (mirrors ScheduledTaskService's `launch_run`
# Closure over `app` (mirrors the schedule composition root's `launch_run`
# below) rather than resolving `app.state.stream_bridge` here
# directly: `stream_bridge` is a STARTUP_ONLY_FIELDS singleton set
# once, above, by `langgraph_runtime(app, startup_config)`, so

33
backend/app/gateway/deps.py
View File

@ -414,18 +414,6 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen
app.state.thread_store = make_thread_store(sf, app.state.store)
if sf is not None:
from deerflow.persistence.scheduled_task_runs import (
ScheduledTaskRunRepository,
)
from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
app.state.scheduled_task_repo = ScheduledTaskRepository(sf)
app.state.scheduled_task_run_repo = ScheduledTaskRunRepository(sf)
else:
app.state.scheduled_task_repo = None
app.state.scheduled_task_run_repo = None
# Hexagonal slices: every adapter is instantiated by the composition
# root and nowhere else. It is a pure function of the infrastructure
# built above, so what it decides — including "no SQL backend means no
@ -576,27 +564,6 @@ def get_thread_store(request: Request) -> ThreadMetaStore:
return val
def get_scheduled_task_repo(request: Request):
val = getattr(request.app.state, "scheduled_task_repo", None)
if val is None:
raise HTTPException(status_code=503, detail="Scheduled task repo not available")
return val
def get_scheduled_task_run_repo(request: Request):
val = getattr(request.app.state, "scheduled_task_run_repo", None)
if val is None:
raise HTTPException(status_code=503, detail="Scheduled task run repo not available")
return val
def get_scheduled_task_service(request: Request):
val = getattr(request.app.state, "scheduled_task_service", None)
if val is None:
raise HTTPException(status_code=503, detail="Scheduled task service not available")
return val
def get_schedule_service(request: Request) -> ScheduleService:
"""The schedule context's input port.

2
backend/app/gateway/routers/__init__.py
View File

@ -5,7 +5,6 @@ from . import (
input_polish,
mcp,
models,
scheduled_tasks,
skills,
suggestions,
thread_runs,
@ -20,7 +19,6 @@ __all__ = [
"input_polish",
"mcp",
"models",
"scheduled_tasks",
"skills",
"suggestions",
"threads",

310
backend/app/gateway/routers/scheduled_tasks.py
View File

@ -1,310 +0,0 @@
from __future__ import annotations
import uuid
from datetime import UTC, datetime
from typing import Any
from fastapi import APIRouter, HTTPException, Query, Request
from pydantic import BaseModel, Field
from app.gateway.authz import require_permission
from app.gateway.deps import (
get_config,
get_optional_user_from_request,
get_scheduled_task_repo,
get_scheduled_task_run_repo,
get_scheduled_task_service,
get_thread_store,
)
from deerflow.scheduler.schedules import (
next_run_at as compute_next_run_at,
)
from deerflow.scheduler.schedules import (
normalize_cron_expression,
validate_timezone,
)
router = APIRouter(prefix="/api", tags=["scheduled-tasks"])
def _ensure_task_mutable(task: dict[str, Any]) -> None:
if task.get("status") == "running":
raise HTTPException(
status_code=409,
detail="Scheduled task is currently running; retry after the active execution finishes",
)
class ScheduledTaskCreateRequest(BaseModel):
thread_id: str | None = None
context_mode: str = "fresh_thread_per_run"
title: str = Field(min_length=1)
prompt: str = Field(min_length=1)
schedule_type: str
schedule_spec: dict[str, Any]
timezone: str
class ScheduledTaskUpdateRequest(BaseModel):
context_mode: str | None = None
thread_id: str | None = None
title: str | None = Field(default=None, min_length=1)
prompt: str | None = Field(default=None, min_length=1)
schedule_spec: dict[str, Any] | None = None
timezone: str | None = None
@router.get("/scheduled-tasks")
@require_permission("threads", "read")
async def list_scheduled_tasks(request: Request):
repo = get_scheduled_task_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
return []
return await repo.list_by_user(str(user.id))
@router.post("/scheduled-tasks")
@require_permission("threads", "write")
async def create_scheduled_task(request: Request, body: ScheduledTaskCreateRequest):
config = get_config()
repo = get_scheduled_task_repo(request)
thread_store = get_thread_store(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
if body.context_mode not in {"fresh_thread_per_run", "reuse_thread"}:
raise HTTPException(status_code=422, detail="Unsupported context_mode")
if body.context_mode == "reuse_thread":
if not body.thread_id:
raise HTTPException(status_code=422, detail="reuse_thread requires thread_id")
if not await thread_store.check_access(body.thread_id, str(user.id), require_existing=True):
raise HTTPException(status_code=404, detail="Thread not found")
if body.schedule_type not in {"once", "cron"}:
raise HTTPException(status_code=422, detail="Unsupported schedule_type")
schedule_spec = dict(body.schedule_spec)
try:
validate_timezone(body.timezone)
if body.schedule_type == "cron":
raw_cron = schedule_spec.get("cron")
if not isinstance(raw_cron, str):
raise HTTPException(status_code=422, detail="cron schedule requires schedule_spec.cron")
schedule_spec["cron"] = normalize_cron_expression(raw_cron)
next_run_at = compute_next_run_at(
body.schedule_type,
schedule_spec,
body.timezone,
now=datetime.now(UTC),
)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
if body.schedule_type == "once" and next_run_at is None:
raise HTTPException(status_code=422, detail="once schedule must be in the future")
if body.schedule_type == "once" and next_run_at is not None and (next_run_at - datetime.now(UTC)).total_seconds() < config.scheduler.min_once_delay_seconds:
raise HTTPException(
status_code=422,
detail=(f"once schedule must be at least {config.scheduler.min_once_delay_seconds} seconds in the future"),
)
return await repo.create(
task_id=f"task-{uuid.uuid4().hex}",
user_id=str(user.id),
thread_id=body.thread_id,
context_mode=body.context_mode,
assistant_id="lead_agent",
title=body.title,
prompt=body.prompt,
schedule_type=body.schedule_type,
schedule_spec=schedule_spec,
timezone=body.timezone,
next_run_at=next_run_at,
)
@router.get("/scheduled-tasks/{task_id}")
@require_permission("threads", "read")
async def get_scheduled_task(task_id: str, request: Request):
repo = get_scheduled_task_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
task = await repo.get(task_id, user_id=str(user.id))
if task is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
return task
@router.patch("/scheduled-tasks/{task_id}")
@require_permission("threads", "write")
async def update_scheduled_task(task_id: str, request: Request, body: ScheduledTaskUpdateRequest):
config = get_config()
repo = get_scheduled_task_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
existing = await repo.get(task_id, user_id=str(user.id))
if existing is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
_ensure_task_mutable(existing)
updates = body.model_dump(exclude_none=True)
if "context_mode" in updates:
if updates["context_mode"] not in {"fresh_thread_per_run", "reuse_thread"}:
raise HTTPException(status_code=422, detail="Unsupported context_mode")
effective_context_mode = str(updates.get("context_mode", existing["context_mode"]))
effective_thread_id = updates.get("thread_id", existing.get("thread_id"))
if effective_context_mode == "reuse_thread":
if not effective_thread_id:
raise HTTPException(status_code=422, detail="reuse_thread requires thread_id")
thread_store = get_thread_store(request)
if not await thread_store.check_access(str(effective_thread_id), str(user.id), require_existing=True):
raise HTTPException(status_code=404, detail="Thread not found")
elif effective_context_mode == "fresh_thread_per_run":
updates["thread_id"] = None
if "timezone" in updates:
try:
validate_timezone(str(updates["timezone"]))
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
if "schedule_spec" in updates or "timezone" in updates:
schedule_spec = dict(existing["schedule_spec"])
if "schedule_spec" in updates and isinstance(updates["schedule_spec"], dict):
schedule_spec = dict(updates["schedule_spec"])
timezone = str(updates.get("timezone", existing["timezone"]))
try:
if existing["schedule_type"] == "cron":
raw_cron = schedule_spec.get("cron")
if not isinstance(raw_cron, str):
raise HTTPException(
status_code=422,
detail="cron schedule requires schedule_spec.cron",
)
schedule_spec["cron"] = normalize_cron_expression(raw_cron)
next_run_at = compute_next_run_at(
existing["schedule_type"],
schedule_spec,
timezone,
now=datetime.now(UTC),
)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
if existing["schedule_type"] == "once" and next_run_at is None:
raise HTTPException(status_code=422, detail="once schedule must be in the future")
if existing["schedule_type"] == "once" and next_run_at is not None and (next_run_at - datetime.now(UTC)).total_seconds() < config.scheduler.min_once_delay_seconds:
raise HTTPException(
status_code=422,
detail=(f"once schedule must be at least {config.scheduler.min_once_delay_seconds} seconds in the future"),
)
updates["schedule_spec"] = schedule_spec
updates["next_run_at"] = next_run_at
# A terminal task (completed/failed/cancelled) whose schedule was just
# pushed into the future must be re-armed: claim_due_tasks only admits
# "enabled" rows, so leaving the terminal status would return 200 with
# a next_run_at that silently never fires.
if next_run_at is not None and existing["status"] in {"completed", "failed", "cancelled"}:
updates["status"] = "enabled"
updated = await repo.update(
task_id,
user_id=str(user.id),
updates=updates,
)
return updated
@router.post("/scheduled-tasks/{task_id}/pause")
@require_permission("threads", "write")
async def pause_scheduled_task(task_id: str, request: Request):
repo = get_scheduled_task_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
existing = await repo.get(task_id, user_id=str(user.id))
if existing is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
_ensure_task_mutable(existing)
updated = await repo.update(task_id, user_id=str(user.id), updates={"status": "paused"})
if updated is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
return updated
@router.post("/scheduled-tasks/{task_id}/resume")
@require_permission("threads", "write")
async def resume_scheduled_task(task_id: str, request: Request):
repo = get_scheduled_task_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
existing = await repo.get(task_id, user_id=str(user.id))
if existing is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
_ensure_task_mutable(existing)
updated = await repo.update(task_id, user_id=str(user.id), updates={"status": "enabled"})
if updated is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
return updated
@router.post("/scheduled-tasks/{task_id}/trigger")
@require_permission("threads", "write")
async def trigger_scheduled_task(task_id: str, request: Request):
repo = get_scheduled_task_repo(request)
service = get_scheduled_task_service(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
task = await repo.get(task_id, user_id=str(user.id))
if task is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
result = await service.dispatch_task(task, now=datetime.now(UTC), trigger="manual")
if result["outcome"] == "conflict":
raise HTTPException(status_code=409, detail=result["error"] or "Scheduled task trigger conflicted with an active run")
if result["outcome"] == "failed":
raise HTTPException(status_code=502, detail=result["error"] or "Scheduled task trigger failed")
return {"id": task_id, "triggered": True}
@router.delete("/scheduled-tasks/{task_id}")
@require_permission("threads", "write")
async def delete_scheduled_task(task_id: str, request: Request):
repo = get_scheduled_task_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
deleted = await repo.delete(task_id, user_id=str(user.id))
if not deleted:
raise HTTPException(status_code=404, detail="Scheduled task not found")
return {"id": task_id, "deleted": deleted}
@router.get("/scheduled-tasks/{task_id}/runs")
@require_permission("threads", "read")
async def list_scheduled_task_runs(
task_id: str,
request: Request,
limit: int = Query(default=50, ge=1, le=200),
offset: int = Query(default=0, ge=0),
):
task_repo = get_scheduled_task_repo(request)
run_repo = get_scheduled_task_run_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
task = await task_repo.get(task_id, user_id=str(user.id))
if task is None:
raise HTTPException(status_code=404, detail="Scheduled task not found")
return await run_repo.list_by_task(task_id, limit=limit, offset=offset)
@router.get("/threads/{thread_id}/scheduled-tasks")
@require_permission("threads", "read", owner_check=True)
async def list_thread_scheduled_tasks(thread_id: str, request: Request):
repo = get_scheduled_task_repo(request)
user = await get_optional_user_from_request(request)
if user is None:
raise HTTPException(status_code=401, detail="Authentication required")
return await repo.list_by_user_and_thread(str(user.id), thread_id)

4
backend/app/scheduler/__init__.py
View File

@ -1,3 +1 @@
from .service import ScheduledTaskService
__all__ = ["ScheduledTaskService"]
"""Clock-driven entry point of the schedule context (the poller)."""

428
backend/app/scheduler/service.py
View File

@ -1,428 +0,0 @@
from __future__ import annotations
import asyncio
import logging
import socket
import uuid
from datetime import UTC, datetime
from typing import Any, Literal
from fastapi import HTTPException
from deerflow.persistence.scheduled_task_runs import ActiveScheduledRunConflict
from deerflow.runtime import ConflictError, RunRecord
from deerflow.scheduler.schedules import next_run_at
logger = logging.getLogger(__name__)
# Shared so the has_active_runs fast path and the unique-index race path return
# byte-identical outcomes for the same "task already has an active run" condition.
_ACTIVE_RUN_CONFLICT_ERROR = "task already has an active run"
_SKIP_ACTIVE_RUN_ERROR = "skipped: a previous run of this task is still active"
def _as_datetime(value: Any) -> datetime | None:
"""Coerce a timestamp read back off a task dict into a real datetime.
``ScheduledTaskRepository._row_to_dict`` runs every timestamp column
through ``coerce_iso``, so what a caller reads is an ISO *string* while the
column it goes back into is a ``DateTime``. Only the skip path performs
that round trip -- every other write passes ``now`` straight through -- so
the coercion lives here rather than being pushed into the repository, which
would loosen the type of an otherwise strict boundary.
"""
if value is None or isinstance(value, datetime):
return value
try:
parsed = datetime.fromisoformat(str(value))
except (TypeError, ValueError):
# Display-only bookkeeping: an unparseable stored timestamp must not
# take down the poll cycle along with every task queued behind it.
logger.warning("Dropping unparseable timestamp %r while recording a skipped occurrence", value)
return None
return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC)
class ScheduledTaskService:
def __init__(
self,
*,
task_repo,
task_run_repo,
launch_run,
poll_interval_seconds: int,
lease_seconds: int,
max_concurrent_runs: int,
) -> None:
self._task_repo = task_repo
self._task_run_repo = task_run_repo
self._launch_run = launch_run
self._poll_interval_seconds = poll_interval_seconds
self._lease_seconds = lease_seconds
self._max_concurrent_runs = max_concurrent_runs
self._lease_owner = f"{socket.gethostname()}:{uuid.uuid4().hex}"
self._task: asyncio.Task | None = None
self._stop = asyncio.Event()
async def run_once(self, *, now: datetime) -> None:
# ``max_concurrent_runs`` is a global cap on active scheduled runs, not
# just a per-poll claim batch: long runs accumulate across poll cycles,
# so each cycle only claims into the remaining budget.
active = await self._task_run_repo.count_active_runs()
budget = self._max_concurrent_runs - active
if budget <= 0:
return
claimed = await self._task_repo.claim_due_tasks(
now=now,
lease_owner=self._lease_owner,
lease_seconds=self._lease_seconds,
limit=budget,
)
for task in claimed:
await self.dispatch_task(task, now=now, trigger="scheduled")
@staticmethod
def _is_overlap_conflict(exc: Exception) -> bool:
if isinstance(exc, ConflictError):
return True
return isinstance(exc, HTTPException) and exc.status_code == 409
@staticmethod
def _task_status_for_failure(task: dict[str, Any], *, trigger: str) -> str:
if trigger == "manual":
# A failed manual trigger must not consume the task's scheduled
# future: a `once` task with run_at still ahead would otherwise be
# flipped to "failed" and never claimed again.
return task.get("status") or "enabled"
if task["schedule_type"] == "once":
return "failed"
return "enabled"
@staticmethod
def _task_status_for_skip(task: dict[str, Any]) -> str:
if task["schedule_type"] == "once":
# The single occurrence was lost to an overlapping run; "completed"
# would claim an execution that never happened.
return "failed"
return "enabled"
async def dispatch_task(
self,
task: dict[str, Any],
*,
now: datetime,
trigger: str,
) -> dict[str, Any]:
execution_thread_id = task.get("thread_id")
if task.get("context_mode") == "fresh_thread_per_run" or not execution_thread_id:
execution_thread_id = str(uuid.uuid4())
# "skip" must hold for fresh-thread runs too, where every run gets a new
# thread and the same-thread multitask ConflictError below can never
# fire. Checked before creating this dispatch's own run row so the row
# does not count itself as the active run. A manual trigger against an
# active run is rejected outright (409 at the router) instead of being
# recorded as a skipped occurrence — nothing was scheduled to happen.
#
# This has_active_runs check is a non-atomic fast path: it runs in its
# own session and is separated from the create() below by await points,
# so two concurrent dispatches (double-click / client retry / a manual
# trigger racing the poller) can both observe no active run. The DB is
# the atomic arbiter — the partial unique index uq_scheduled_task_run_active
# rejects the second active insert, surfaced as ActiveScheduledRunConflict
# and collapsed to the SAME outcome as this fast path just below.
overlap_skip = task.get("overlap_policy", "skip") == "skip"
if overlap_skip and await self._task_run_repo.has_active_runs(task["id"]):
if trigger == "manual":
return self._active_run_conflict_result(execution_thread_id)
return await self._record_scheduled_skip(task, thread_id=execution_thread_id, now=now, trigger=trigger)
task_run_id = f"task-run-{uuid.uuid4().hex}"
try:
await self._task_run_repo.create(
run_record_id=task_run_id,
task_id=task["id"],
thread_id=execution_thread_id,
scheduled_for=now,
trigger=trigger,
status="queued",
)
except ActiveScheduledRunConflict:
# Lost the create race for the task's single active slot: a
# concurrent dispatch passed the same fast-path check and inserted
# its active row first. Identical outcome to the fast path above.
if trigger == "manual":
return self._active_run_conflict_result(execution_thread_id)
return await self._record_scheduled_skip(task, thread_id=execution_thread_id, now=now, trigger=trigger)
try:
result = await self._launch_run(
thread_id=execution_thread_id,
assistant_id=task.get("assistant_id"),
prompt=task["prompt"],
owner_user_id=task.get("user_id"),
metadata={
"scheduled_task_id": task["id"],
"scheduled_task_run_id": task_run_id,
"scheduled_trigger": trigger,
},
)
next_at = next_run_at(
task["schedule_type"],
task["schedule_spec"],
task["timezone"],
now=now,
)
if task["schedule_type"] == "once":
# Stay "running" until handle_run_completion sees the real
# terminal outcome; declaring "completed" at launch would stick
# if the run fails or the process dies (startup reconciliation
# is cancel_stuck_once_tasks).
task_status = "running"
elif trigger == "manual" and task.get("status") == "paused":
task_status = "paused"
else:
task_status = "enabled"
await self._task_run_repo.update_status(
task_run_id,
status="running",
run_id=result["run_id"],
started_at=now,
# A fast-failing run can reach handle_run_completion before this
# write resumes; never clobber its terminal status.
protect_terminal=True,
)
await self._task_repo.update_after_launch(
task["id"],
status=task_status,
next_run_at=next_at,
last_run_at=now,
last_run_id=result["run_id"],
last_thread_id=result["thread_id"],
last_error=None,
increment_run_count=True,
# Same race as the run-row write above: a fast-failing run's
# completion hook may have already finalized a `once` task.
protect_terminal=True,
)
return {
"outcome": "launched",
"task_run_id": task_run_id,
"run_id": result["run_id"],
"thread_id": result["thread_id"],
"error": None,
}
except Exception as exc:
next_at = next_run_at(
task["schedule_type"],
task["schedule_spec"],
task["timezone"],
now=now,
)
if self._is_overlap_conflict(exc) and trigger == "scheduled" and task.get("overlap_policy", "skip") == "skip":
return await self._finalize_skip(task, task_run_id=task_run_id, thread_id=execution_thread_id, now=now, error=str(exc))
task_status = self._task_status_for_failure(task, trigger=trigger)
await self._task_run_repo.update_status(
task_run_id,
status="failed",
error=str(exc),
started_at=now,
finished_at=now,
)
await self._task_repo.update_after_launch(
task["id"],
status=task_status,
next_run_at=next_at,
last_run_at=now,
last_run_id=None,
last_thread_id=execution_thread_id,
last_error=str(exc),
increment_run_count=False,
)
return {
"outcome": "conflict" if self._is_overlap_conflict(exc) else "failed",
"task_run_id": task_run_id,
"run_id": None,
"thread_id": execution_thread_id,
"error": str(exc),
}
def _active_run_conflict_result(self, thread_id: str) -> dict[str, Any]:
"""Manual-trigger response when the task already has an active run.
Nothing was scheduled to happen, so no run-history row is recorded; the
router maps this to a 409.
"""
return {
"outcome": "conflict",
"task_run_id": None,
"run_id": None,
"thread_id": thread_id,
"error": _ACTIVE_RUN_CONFLICT_ERROR,
}
async def _record_scheduled_skip(
self,
task: dict[str, Any],
*,
thread_id: str,
now: datetime,
trigger: str,
) -> dict[str, Any]:
"""Record a skipped occurrence for a scheduled dispatch that overlapped an active run.
The tombstone is created directly as terminal ``"skipped"`` rather than
the transient ``"queued"`` the launch path uses: a queued row is active
and would itself trip ``uq_scheduled_task_run_active`` against the
pre-existing run that is still holding the task's single active slot.
``"skipped"`` is outside the index predicate, so it never conflicts.
"""
task_run_id = f"task-run-{uuid.uuid4().hex}"
await self._task_run_repo.create(
run_record_id=task_run_id,
task_id=task["id"],
thread_id=thread_id,
scheduled_for=now,
trigger=trigger,
status="skipped",
)
return await self._finalize_skip(task, task_run_id=task_run_id, thread_id=thread_id, now=now, error=_SKIP_ACTIVE_RUN_ERROR)
async def _finalize_skip(
self,
task: dict[str, Any],
*,
task_run_id: str,
thread_id: str,
now: datetime,
error: str,
) -> dict[str, Any]:
next_at = next_run_at(
task["schedule_type"],
task["schedule_spec"],
task["timezone"],
now=now,
)
await self._task_run_repo.update_status(
task_run_id,
status="skipped",
error=error,
started_at=now,
finished_at=now,
)
await self._task_repo.update_after_launch(
task["id"],
status=self._task_status_for_skip(task),
next_run_at=next_at,
# A skip is not an execution, so the launch bookkeeping is carried
# over unchanged. `update_after_launch` assigns unconditionally, so
# "unchanged" has to be spelled out by passing the current values
# back -- and `last_run_at` comes back as an ISO string.
last_run_at=_as_datetime(task.get("last_run_at")),
last_run_id=task.get("last_run_id"),
last_thread_id=task.get("last_thread_id"),
last_error=error if task["schedule_type"] == "once" else None,
increment_run_count=False,
)
return {
"outcome": "skipped",
"task_run_id": task_run_id,
"run_id": None,
"thread_id": thread_id,
"error": error,
}
async def handle_run_completion(self, record: RunRecord) -> None:
metadata = record.metadata or {}
task_id = metadata.get("scheduled_task_id")
task_run_id = metadata.get("scheduled_task_run_id")
user_id = record.user_id
if not isinstance(task_id, str) or not isinstance(task_run_id, str) or not user_id:
return
terminal_status: Literal["success", "failed", "interrupted"] | None
if record.status.value == "success":
terminal_status = "success"
error = None
elif record.status.value == "interrupted":
# Distinct from "failed": an interrupt (user cancel, same-thread
# takeover) carries no error and is not an execution failure.
terminal_status = "interrupted"
error = record.error or "run was interrupted before completion"
elif record.status.value in {"error", "timeout"}:
terminal_status = "failed"
error = record.error
else:
terminal_status = None
error = record.error
if terminal_status is None:
return
await self._task_run_repo.update_status(
task_run_id,
status=terminal_status,
run_id=record.run_id,
error=error,
finished_at=datetime.now(UTC),
)
task = await self._task_repo.get(task_id, user_id=user_id)
if task is None:
return
updates: dict[str, Any] = {"last_error": error}
if task["schedule_type"] == "once":
# The single occurrence is consumed either way (the run did launch,
# so re-arming risks duplicate side effects), but an interrupt ends
# as "cancelled", not "failed".
if terminal_status == "success":
updates["status"] = "completed"
elif terminal_status == "interrupted":
updates["status"] = "cancelled"
else:
updates["status"] = "failed"
await self._task_repo.update(task_id, user_id=user_id, updates=updates)
async def start(self) -> None:
if self._task is not None:
return
restart_error = "interrupted: gateway restarted before the run reached a terminal state"
try:
stale = await self._task_run_repo.mark_stale_active_runs(error=restart_error)
if stale:
logger.warning("Marked %d stale scheduled task run(s) as interrupted after restart", stale)
except Exception:
logger.exception("Failed to sweep stale scheduled task runs at startup")
try:
# The run rows above are only half the story: a launched `once`
# task is parked in "running" until the (now dead) completion hook
# would have finalized it, so reconcile the parent rows too.
stuck = await self._task_repo.cancel_stuck_once_tasks(error=restart_error)
if stuck:
logger.warning("Cancelled %d stuck once task(s) after restart", stuck)
except Exception:
logger.exception("Failed to reconcile stuck once tasks at startup")
self._stop.clear()
self._task = asyncio.create_task(self._run_loop())
async def stop(self) -> None:
if self._task is None:
return
self._stop.set()
await self._task
self._task = None
async def _run_loop(self) -> None:
while not self._stop.is_set():
try:
await self.run_once(now=datetime.now(UTC))
except Exception:
# A transient DB error (e.g. SQLite "database is locked") must
# not kill the poller task for the rest of the process life.
logger.exception("Scheduled task poll failed; retrying next interval")
try:
await asyncio.wait_for(
self._stop.wait(),
timeout=self._poll_interval_seconds,
)
except TimeoutError:
continue

2
backend/docs/GITHUB_AGENTS.md
View File

@ -259,7 +259,7 @@ Key properties:
- **Drain-conflict edge case**: if the drain's own `runs.create()` also hits `ConflictError` — e.g. a manual Web UI turn or a scheduled run raced onto the same thread — the popped batch is requeued (not lost, not retried in a tight loop). It is picked up again the next time this manager successfully creates *and watches* a run on that thread, which is guaranteed to eventually happen once whatever is occupying the thread finishes and any subsequent trigger succeeds.
- **Reactions are out of scope for this slice.** GitHub's reaction API (`eyes`/`confused` acknowledgment) has no existing integration in this codebase and needs its own design pass; buffered comments are coalesced silently, with no per-comment acknowledgment.
- **Config**: gated behind `ChannelRunPolicy.buffer_followups_on_busy` (default `False`); GitHub's own registration in `app/gateway/github/run_policy.py` opts in, since it is exactly the fire_and_forget + log-only-send channel this was designed for. Any other channel that adopts `fire_and_forget=True` in the future keeps the old silent-drop-with-log behavior unless it opts in too.
- **Plumbing**: the watcher subscribes to the Gateway's `StreamBridge` singleton (`app.state.stream_bridge`), which `ChannelManager` previously had no way to reach — every other consumer gets it via `get_stream_bridge(request)`, which needs an HTTP `Request` the bus-consumer loop doesn't have. It is threaded as a zero-arg closure from `app.py`'s lifespan (`get_stream_bridge=lambda: getattr(app.state, "stream_bridge", None)`) through `start_channel_service()` → `ChannelService.__init__` → `ChannelManager.__init__`, mirroring the existing `launch_run=lambda **kwargs: launch_scheduled_thread_run(app=app, **kwargs)` closure already used for `ScheduledTaskService` in the same lifespan function.
- **Plumbing**: the watcher subscribes to the Gateway's `StreamBridge` singleton (`app.state.stream_bridge`), which `ChannelManager` previously had no way to reach — every other consumer gets it via `get_stream_bridge(request)`, which needs an HTTP `Request` the bus-consumer loop doesn't have. It is threaded as a zero-arg closure from `app.py`'s lifespan (`get_stream_bridge=lambda: getattr(app.state, "stream_bridge", None)`) through `start_channel_service()` → `ChannelService.__init__` → `ChannelManager.__init__`, mirroring the existing `launch_run=lambda **kwargs: launch_scheduled_thread_run(app=app, **kwargs)` closure already used for the schedule composition root in the same lifespan function.
- **Single-process scope (known limitation)**: the buffer and watcher tasks live in one `ChannelManager` instance's process memory. Under `GATEWAY_WORKERS>1` or a multi-pod deployment, a follow-up comment that a load balancer or webhook fan-out routes to a *different* worker than the one that created the busy run will not see that worker's buffer. This mirrors the cross-pod gap already documented for issue #4120 (IM leader election or a shared buffer store would close it) and is deliberately deferred — single-process/single-pod deployments, the safe default, are unaffected.
## Outbound is Log-only

4
backend/packages/harness/deerflow/config/reload_boundary.py
View File

@ -66,8 +66,8 @@ STARTUP_ONLY_FIELDS: dict[str, str] = {
"start_channel_service() wires the connection repository and channel workers once at startup, and the channel-connections router caches the merged provider config on app.state; channel_connections.* edits need a restart."
),
"scheduler": (
"ScheduledTaskService is constructed and started once during Gateway lifespan startup; enabled, poll_interval_seconds, lease_seconds, "
"and max_concurrent_runs are captured into the service instance and the background poller task is not rebuilt on config.yaml edits."
"ScheduleService and SchedulePoller are constructed and started once during Gateway lifespan startup; enabled, poll_interval_seconds, lease_seconds, "
"and max_concurrent_runs are captured into the assembled service/poller and the background poller task is not rebuilt on config.yaml edits."
),
"run_ownership": (
"RunOwnershipConfig is captured once into RunManager at langgraph_runtime() startup; the lease heartbeat background task is created and "

2
backend/packages/harness/deerflow/domain/schedule/model/task.py
View File

@ -22,7 +22,7 @@ same-valued ones:
there would hand back a next_run_at that silently never fires.
- The repository's `protect_terminal` compare-and-set refuses to overwrite
them, because a fast-failing run's completion hook can land before the launch
path's own write (scheduled_tasks/sql.py:12).
path's own write.
The two coincide because terminal *is* the definition of re-armable. The name
states what the statuses are; what each rule does with them belongs to that

6
backend/packages/harness/deerflow/domain/schedule/service.py
View File

@ -7,9 +7,9 @@ to the aggregate -- and knows nothing about HTTP, SQL, or the run runtime.
primary adapter's job.
The dispatch path deliberately mirrors the structure of the pre-migration
`app/scheduler/service.py`, including its ordering and its comments, because
its concurrency and idempotency semantics are load-bearing and are pinned by
tests that must keep passing unchanged.
`app/scheduler/service.py` (since deleted), including its ordering and its
comments, because its concurrency and idempotency semantics are load-bearing
and are pinned by tests that must keep passing unchanged.
"""
from __future__ import annotations

14
backend/packages/harness/deerflow/persistence/scheduled_task_runs/__init__.py
View File

@ -1,4 +1,12 @@
from .model import ScheduledTaskRunRow
from .sql import ActiveScheduledRunConflict, ScheduledTaskRunRepository
"""ORM row of the scheduled_task_runs table.
__all__ = ["ActiveScheduledRunConflict", "ScheduledTaskRunRow", "ScheduledTaskRunRepository"]
The table definition -- including the partial unique index
``uq_scheduled_task_run_active`` that arbitrates the single active slot --
lives here with the shared engine/alembic infrastructure; its only reader and
writer is the schedule context's secondary adapter
(``app/adapters/schedule/scheduled_run_repository.py``).
"""
from .model import ScheduledTaskRunRow
__all__ = ["ScheduledTaskRunRow"]

171
backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py
View File

@ -1,171 +0,0 @@
from __future__ import annotations
from datetime import UTC, datetime
from typing import Any
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.persistence.scheduled_task_runs.model import ScheduledTaskRunRow
from deerflow.utils.time import coerce_iso
TERMINAL_RUN_STATUSES: frozenset[str] = frozenset({"success", "failed", "skipped", "interrupted"})
ACTIVE_RUN_STATUSES: tuple[str, ...] = ("queued", "running")
class ActiveScheduledRunConflict(Exception):
"""A concurrent dispatch already holds the task's single active-run slot.
Raised by :meth:`ScheduledTaskRunRepository.create` when inserting an
active (queued/running) run row would violate the partial unique index
``uq_scheduled_task_run_active`` (at most one active run per ``task_id``).
This is the atomic counterpart to the non-atomic ``has_active_runs`` check
in ``ScheduledTaskService.dispatch_task``: two dispatches can both pass that
check, but only one can insert the active row — the loser lands here.
Translating the SQLAlchemy ``IntegrityError`` into a domain exception at
the repository boundary keeps the service layer free of ``sqlalchemy.exc``
coupling (mirrors ``deerflow.runtime.ConflictError`` for the runs table).
"""
def __init__(self, task_id: str) -> None:
self.task_id = task_id
super().__init__(f"scheduled task {task_id!r} already has an active run")
class ScheduledTaskRunRepository:
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
self._sf = session_factory
@staticmethod
def _row_to_dict(row: ScheduledTaskRunRow) -> dict[str, Any]:
data = row.to_dict()
for key in ("scheduled_for", "started_at", "finished_at", "created_at"):
if data.get(key) is not None:
data[key] = coerce_iso(data[key])
return data
async def create(
self,
*,
run_record_id: str,
task_id: str,
thread_id: str,
scheduled_for: datetime,
trigger: str,
status: str,
) -> dict[str, Any]:
row = ScheduledTaskRunRow(
id=run_record_id,
task_id=task_id,
thread_id=thread_id,
scheduled_for=scheduled_for,
trigger=trigger,
status=status,
created_at=datetime.now(UTC),
)
async with self._sf() as session:
session.add(row)
try:
await session.commit()
except IntegrityError:
await session.rollback()
# Only active-status inserts can trip the partial unique index
# ``uq_scheduled_task_run_active``; a terminal-status row (e.g.
# a "skipped" tombstone) is outside its predicate and cannot
# conflict, so any IntegrityError there is a genuine fault and
# is re-raised untranslated.
if status in ACTIVE_RUN_STATUSES:
raise ActiveScheduledRunConflict(task_id) from None
raise
await session.refresh(row)
return self._row_to_dict(row)
async def list_by_task(self, task_id: str, *, limit: int = 50, offset: int = 0) -> list[dict[str, Any]]:
stmt = (
select(ScheduledTaskRunRow)
.where(ScheduledTaskRunRow.task_id == task_id)
.order_by(
ScheduledTaskRunRow.created_at.desc(),
ScheduledTaskRunRow.id.desc(),
)
.limit(limit)
.offset(offset)
)
async with self._sf() as session:
result = await session.execute(stmt)
return [self._row_to_dict(row) for row in result.scalars()]
async def count_active_runs(self) -> int:
"""Global count of queued/running rows, used to bound cross-task concurrency."""
stmt = select(func.count()).select_from(ScheduledTaskRunRow).where(ScheduledTaskRunRow.status.in_(ACTIVE_RUN_STATUSES))
async with self._sf() as session:
result = await session.execute(stmt)
return int(result.scalar() or 0)
async def update_status(
self,
run_record_id: str,
*,
status: str,
run_id: str | None = None,
error: str | None = None,
started_at: datetime | None = None,
finished_at: datetime | None = None,
protect_terminal: bool = False,
) -> None:
async with self._sf() as session:
row = await session.get(ScheduledTaskRunRow, run_record_id)
if row is None:
return
if protect_terminal and row.status in TERMINAL_RUN_STATUSES:
# The launch-path "running" write lost the race against the
# completion hook; keep the terminal status/error and only
# backfill bookkeeping the completion write could not know.
if row.run_id is None and run_id is not None:
row.run_id = run_id
if row.started_at is None and started_at is not None:
row.started_at = started_at
await session.commit()
return
row.status = status
row.run_id = run_id
row.error = error
if started_at is not None:
row.started_at = started_at
if finished_at is not None:
row.finished_at = finished_at
await session.commit()
async def has_active_runs(self, task_id: str) -> bool:
stmt = (
select(ScheduledTaskRunRow.id)
.where(
ScheduledTaskRunRow.task_id == task_id,
ScheduledTaskRunRow.status.in_(ACTIVE_RUN_STATUSES),
)
.limit(1)
)
async with self._sf() as session:
result = await session.execute(stmt)
return result.scalars().first() is not None
async def mark_stale_active_runs(self, *, error: str) -> int:
"""Fail-fast bookkeeping for runs orphaned by a process crash.
Agent runs execute in-process, so any ``queued``/``running`` row found
at scheduler startup belongs to a run whose process is gone. Only valid
under the MVP's single-scheduler-instance assumption.
"""
stmt = select(ScheduledTaskRunRow).where(ScheduledTaskRunRow.status.in_(ACTIVE_RUN_STATUSES))
now = datetime.now(UTC)
async with self._sf() as session:
result = await session.execute(stmt)
rows = list(result.scalars())
for row in rows:
row.status = "interrupted"
row.error = error
row.finished_at = now
await session.commit()
return len(rows)

12
backend/packages/harness/deerflow/persistence/scheduled_tasks/__init__.py
View File

@ -1,4 +1,10 @@
from .model import ScheduledTaskRow
from .sql import ScheduledTaskRepository
"""ORM row of the scheduled_tasks table.
__all__ = ["ScheduledTaskRow", "ScheduledTaskRepository"]
The table definition lives here with the shared engine/alembic
infrastructure; its only reader and writer is the schedule context's
secondary adapter (``app/adapters/schedule/scheduled_task_repository.py``).
"""
from .model import ScheduledTaskRow
__all__ = ["ScheduledTaskRow"]

232
backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py
View File

@ -1,232 +0,0 @@
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from typing import Any
from sqlalchemy import and_, or_, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow
from deerflow.utils.time import coerce_iso
TERMINAL_TASK_STATUSES: frozenset[str] = frozenset({"completed", "failed", "cancelled"})
class ScheduledTaskRepository:
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
self._sf = session_factory
@staticmethod
def _row_to_dict(row: ScheduledTaskRow) -> dict[str, Any]:
data = row.to_dict()
for key in (
"created_at",
"updated_at",
"next_run_at",
"last_run_at",
"lease_expires_at",
):
if data.get(key) is not None:
data[key] = coerce_iso(data[key])
return data
async def create(
self,
*,
task_id: str,
user_id: str,
thread_id: str | None,
context_mode: str,
assistant_id: str | None,
title: str,
prompt: str,
schedule_type: str,
schedule_spec: dict[str, Any],
timezone: str,
next_run_at: datetime | None,
) -> dict[str, Any]:
now = datetime.now(UTC)
row = ScheduledTaskRow(
id=task_id,
user_id=user_id,
thread_id=thread_id,
context_mode=context_mode,
assistant_id=assistant_id,
title=title,
prompt=prompt,
schedule_type=schedule_type,
schedule_spec=schedule_spec,
timezone=timezone,
next_run_at=next_run_at,
created_at=now,
updated_at=now,
)
async with self._sf() as session:
session.add(row)
await session.commit()
await session.refresh(row)
return self._row_to_dict(row)
async def get(self, task_id: str, *, user_id: str) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(ScheduledTaskRow, task_id)
if row is None or row.user_id != user_id:
return None
return self._row_to_dict(row)
async def list_by_user(self, user_id: str) -> list[dict[str, Any]]:
stmt = select(ScheduledTaskRow).where(ScheduledTaskRow.user_id == user_id).order_by(ScheduledTaskRow.created_at.desc(), ScheduledTaskRow.id.desc())
async with self._sf() as session:
result = await session.execute(stmt)
return [self._row_to_dict(row) for row in result.scalars()]
async def update(
self,
task_id: str,
*,
user_id: str,
updates: dict[str, Any],
) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(ScheduledTaskRow, task_id)
if row is None or row.user_id != user_id:
return None
for key, value in updates.items():
if hasattr(row, key):
setattr(row, key, value)
row.updated_at = datetime.now(UTC)
await session.commit()
await session.refresh(row)
return self._row_to_dict(row)
async def delete(self, task_id: str, *, user_id: str) -> bool:
async with self._sf() as session:
row = await session.get(ScheduledTaskRow, task_id)
if row is None or row.user_id != user_id:
return False
await session.delete(row)
await session.commit()
return True
async def claim_due_tasks(
self,
*,
now: datetime,
lease_owner: str,
lease_seconds: int,
limit: int,
) -> list[dict[str, Any]]:
lease_expires_at = now + timedelta(seconds=lease_seconds)
stmt = (
select(ScheduledTaskRow)
.where(
ScheduledTaskRow.next_run_at.is_not(None),
ScheduledTaskRow.next_run_at <= now,
or_(
and_(
ScheduledTaskRow.status == "enabled",
or_(
ScheduledTaskRow.lease_expires_at.is_(None),
ScheduledTaskRow.lease_expires_at < now,
),
),
# A task stuck in "running" with an expired lease means the
# claiming process died between claim and dispatch; it must
# stay reclaimable or the task is dead forever.
and_(
ScheduledTaskRow.status == "running",
ScheduledTaskRow.lease_expires_at.is_not(None),
ScheduledTaskRow.lease_expires_at < now,
),
),
)
.order_by(ScheduledTaskRow.next_run_at.asc(), ScheduledTaskRow.id.asc())
.limit(limit)
.with_for_update(skip_locked=True)
)
async with self._sf() as session:
result = await session.execute(stmt)
rows = list(result.scalars())
for row in rows:
row.lease_owner = lease_owner
row.lease_expires_at = lease_expires_at
row.status = "running"
row.updated_at = datetime.now(UTC)
await session.commit()
return [self._row_to_dict(row) for row in rows]
async def update_after_launch(
self,
task_id: str,
*,
status: str,
next_run_at: datetime | None,
last_run_at: datetime | None,
last_run_id: str | None,
last_thread_id: str | None,
last_error: str | None,
increment_run_count: bool,
protect_terminal: bool = False,
) -> None:
async with self._sf() as session:
row = await session.get(ScheduledTaskRow, task_id)
if row is None:
return
if protect_terminal and row.status in TERMINAL_TASK_STATUSES:
# A fast-failing run can reach handle_run_completion (which
# finalizes a `once` task) before this launch-path write
# commits; keep the hook's status/error and only record the
# launch bookkeeping.
pass
else:
row.status = status
row.last_error = last_error
row.next_run_at = next_run_at
row.last_run_at = last_run_at
row.last_run_id = last_run_id
row.last_thread_id = last_thread_id
if increment_run_count:
row.run_count += 1
row.lease_owner = None
row.lease_expires_at = None
row.updated_at = datetime.now(UTC)
await session.commit()
async def list_by_user_and_thread(self, user_id: str, thread_id: str) -> list[dict[str, Any]]:
stmt = (
select(ScheduledTaskRow)
.where(
ScheduledTaskRow.user_id == user_id,
ScheduledTaskRow.thread_id == thread_id,
)
.order_by(ScheduledTaskRow.created_at.desc(), ScheduledTaskRow.id.desc())
)
async with self._sf() as session:
result = await session.execute(stmt)
return [self._row_to_dict(row) for row in result.scalars()]
async def cancel_stuck_once_tasks(self, *, error: str) -> int:
"""Reconcile ``once`` tasks orphaned in ``running`` by a process crash.
A launched ``once`` task stays ``running`` until the in-process
completion hook moves it to a terminal status; its lease was cleared at
launch, so the claim query's expired-lease reclaim branch never sees
it. After a crash the hook is gone and the task would be stuck forever.
Tasks still holding a lease are left alone — they were claimed but not
launched, and expired-lease reclaim recovers them safely.
"""
stmt = select(ScheduledTaskRow).where(
ScheduledTaskRow.schedule_type == "once",
ScheduledTaskRow.status == "running",
ScheduledTaskRow.lease_expires_at.is_(None),
)
async with self._sf() as session:
result = await session.execute(stmt)
rows = list(result.scalars())
now = datetime.now(UTC)
for row in rows:
row.status = "cancelled"
row.last_error = error
row.updated_at = now
await session.commit()
return len(rows)

3
backend/packages/harness/deerflow/scheduler/__init__.py
View File

@ -1,3 +0,0 @@
from .schedules import next_run_at, normalize_cron_expression, validate_timezone
__all__ = ["next_run_at", "normalize_cron_expression", "validate_timezone"]

55
backend/packages/harness/deerflow/scheduler/schedules.py
View File

@ -1,55 +0,0 @@
from __future__ import annotations
from datetime import UTC, datetime
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from croniter import croniter
def validate_timezone(timezone_name: str) -> str:
try:
ZoneInfo(timezone_name)
except ZoneInfoNotFoundError as exc:
raise ValueError(f"Unknown timezone: {timezone_name}") from exc
return timezone_name
def normalize_cron_expression(expr: str) -> str:
parts = [part for part in expr.split() if part]
if len(parts) != 5:
raise ValueError("Cron expression must contain exactly 5 fields")
return " ".join(parts)
def next_run_at(
schedule_type: str,
schedule_spec: dict[str, object],
timezone_name: str,
*,
now: datetime,
) -> datetime | None:
validate_timezone(timezone_name)
if now.tzinfo is None:
now = now.replace(tzinfo=UTC)
if schedule_type == "once":
run_at_raw = schedule_spec.get("run_at")
if not isinstance(run_at_raw, str):
raise ValueError("once schedule requires run_at")
run_at = datetime.fromisoformat(run_at_raw)
if run_at.tzinfo is None:
# A naive run_at means "wall-clock time in the task's declared
# timezone", matching how cron schedules interpret it.
run_at = run_at.replace(tzinfo=ZoneInfo(timezone_name))
return run_at if run_at > now else None
if schedule_type == "cron":
cron_expr = normalize_cron_expression(str(schedule_spec.get("cron", "")))
zone = ZoneInfo(timezone_name)
local_now = now.astimezone(zone)
next_local = croniter(cron_expr, local_now).get_next(datetime)
if next_local.tzinfo is None:
next_local = next_local.replace(tzinfo=zone)
return next_local.astimezone(UTC)
raise ValueError(f"Unsupported schedule_type: {schedule_type}")

2
backend/tests/schedule_fakes.py
View File

@ -8,7 +8,7 @@ relying on the tests directory happening to be on ``sys.path``.
single-threaded semantics of each port -- which rows `claim_due` selects, that
`add` refuses a second active run -- and provide no atomicity whatsoever. A
green run here says nothing about two dispatchers racing; that is covered
against a real database in ``test_scheduled_task_dispatch_race.py``. Do not
against a real database in ``test_schedule_dispatch_race.py``. Do not
read a passing contract suite as licence to run more than one scheduler.
"""

223
backend/tests/test_schedule_dispatch_race.py Normal file
View File

@ -0,0 +1,223 @@
"""Concurrency regression tests for the scheduled-task dispatch TOCTOU.
``ScheduleService.dispatch_task`` guards "at most one active run per task
when overlap_policy=skip" with a non-atomic ``has_active`` fast path followed
by a separate queued-record insert. Two concurrent dispatches (double-click,
client retry, or a manual trigger racing the poller) can both pass the check
and both launch. The database is the atomic arbiter via the partial unique
index ``uq_scheduled_task_run_active`` (``task_id WHERE status IN
('queued','running')``); the losing insert is translated to
``ActiveRunConflictError`` and collapsed to the same outcome as the fast
path.
These tests drive the REAL ``SqlScheduledRunRepository`` +
``SqlScheduledTaskRepository`` + ``ScheduleService`` against a real
file-backed sqlite database (so the index is actually enforced), with a fake
launcher that only records launches. The contract suite deliberately does not
own this: its doubles provide no atomicity, so a green contract run says
nothing about two dispatchers racing -- this file is where that is proven.
"""
from __future__ import annotations
import asyncio
from datetime import UTC, datetime
import pytest
from schedule_fakes import FakeThreadLookup
from app.adapters.schedule.scheduled_run_repository import SqlScheduledRunRepository
from app.adapters.schedule.scheduled_task_repository import SqlScheduledTaskRepository
from deerflow.config.database_config import DatabaseConfig
from deerflow.domain.schedule.exceptions import ActiveRunConflictError
from deerflow.domain.schedule.model import DispatchOutcome, RunStatus, ScheduledRun, ScheduledTask, SchedulePolicy, ScheduleSpec, TriggerKind
from deerflow.domain.schedule.ports import LaunchedRun
from deerflow.domain.schedule.service import ScheduleService
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
pytestmark = pytest.mark.asyncio
NOW = datetime(2026, 7, 27, 12, 0, tzinfo=UTC)
class _BarrierRunRepo(SqlScheduledRunRepository):
"""Real repository that only releases both dispatchers past ``has_active``
once both have read it, so their ``add()`` calls genuinely race for the
task's single active slot -- a deterministic reproduction of the
check-then-insert TOCTOU."""
def __init__(self, session_factory, barrier: asyncio.Barrier | None) -> None:
super().__init__(session_factory)
self._barrier = barrier
async def has_active(self, task_id: str) -> bool:
result = await super().has_active(task_id)
if self._barrier is not None:
await self._barrier.wait()
return result
class _RecordingLauncher:
"""Launch double that yields first, so a truly-concurrent sibling can
interleave before the launch is recorded."""
def __init__(self) -> None:
self.calls: list[dict] = []
async def launch(self, *, thread_id, assistant_id, prompt, owner_user_id, metadata) -> LaunchedRun:
await asyncio.sleep(0)
self.calls.append({"thread_id": thread_id, "metadata": metadata})
return LaunchedRun(run_id=f"run-{len(self.calls)}", thread_id=thread_id)
def _make_service(tasks, runs, launcher) -> ScheduleService:
return ScheduleService(
tasks=tasks,
runs=runs,
launcher=launcher,
threads=FakeThreadLookup(),
policy=SchedulePolicy(min_once_delay_seconds=60, max_concurrent_runs=10, lease_seconds=120),
)
async def _seed_task(tasks: SqlScheduledTaskRepository, title: str) -> ScheduledTask:
# fresh_thread_per_run: every dispatch gets a NEW thread_id, so #4003's
# per-thread uq_runs_thread_active can never fire for two dispatches of the
# same task -- this is precisely the gap the per-task index closes.
return await tasks.add(
ScheduledTask.create(
user_id="user-1",
title=title,
prompt="do the thing",
schedule=ScheduleSpec.cron_schedule("*/5 * * * *", "UTC"),
context_mode="fresh_thread_per_run",
thread_id=None,
now=NOW,
policy=SchedulePolicy(),
)
)
async def _active_run_count(runs: SqlScheduledRunRepository, task_id: str) -> int:
rows = await runs.list_by_task(task_id, limit=100)
return sum(1 for row in rows if row.is_active)
async def test_two_concurrent_manual_dispatches_launch_exactly_once(tmp_path):
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
try:
sf = get_session_factory()
assert sf is not None
tasks = SqlScheduledTaskRepository(sf)
runs = _BarrierRunRepo(sf, asyncio.Barrier(2))
launcher = _RecordingLauncher()
service = _make_service(tasks, runs, launcher)
task = await _seed_task(tasks, "task-race-manual")
results = await asyncio.gather(
service.dispatch_task(task, now=NOW, trigger=TriggerKind.MANUAL),
service.dispatch_task(task, now=NOW, trigger=TriggerKind.MANUAL),
)
outcomes = sorted((result.outcome for result in results), key=str)
# Exactly one wins the active slot; the loser is a 409-style conflict.
assert outcomes == [DispatchOutcome.CONFLICT, DispatchOutcome.LAUNCHED], outcomes
assert len(launcher.calls) == 1, launcher.calls
assert await _active_run_count(runs, task.task_id) == 1
# The manual loser records no run-history row (nothing was scheduled).
conflict = next(r for r in results if r.outcome is DispatchOutcome.CONFLICT)
assert conflict.record_id is None
finally:
await close_engine()
async def test_scheduled_and_manual_dispatch_launch_exactly_once(tmp_path):
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
try:
sf = get_session_factory()
assert sf is not None
tasks = SqlScheduledTaskRepository(sf)
runs = _BarrierRunRepo(sf, asyncio.Barrier(2))
launcher = _RecordingLauncher()
service = _make_service(tasks, runs, launcher)
task = await _seed_task(tasks, "task-race-mixed")
results = await asyncio.gather(
service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED),
service.dispatch_task(task, now=NOW, trigger=TriggerKind.MANUAL),
)
outcomes = [result.outcome for result in results]
# Whichever won launched; the loser is conflict (manual) or skipped
# (scheduled). Which one wins is timing-dependent, but exactly one runs.
assert outcomes.count(DispatchOutcome.LAUNCHED) == 1, outcomes
assert set(outcomes) <= {DispatchOutcome.LAUNCHED, DispatchOutcome.CONFLICT, DispatchOutcome.SKIPPED}, outcomes
assert len(launcher.calls) == 1, launcher.calls
assert await _active_run_count(runs, task.task_id) == 1
finally:
await close_engine()
async def test_natural_timing_concurrent_dispatch_launches_exactly_once(tmp_path):
# No barrier: exercise the fix under the same natural interleaving that
# reproduced the bug (5/5 both-launch before the index). The fix must hold
# whether the second dispatch is caught by the has_active fast path or by
# the index-violation path.
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
try:
sf = get_session_factory()
assert sf is not None
tasks = SqlScheduledTaskRepository(sf)
runs = SqlScheduledRunRepository(sf)
for i in range(5):
launcher = _RecordingLauncher()
service = _make_service(tasks, runs, launcher)
task = await _seed_task(tasks, f"task-natural-{i}")
results = await asyncio.gather(
service.dispatch_task(task, now=NOW, trigger=TriggerKind.MANUAL),
service.dispatch_task(task, now=NOW, trigger=TriggerKind.MANUAL),
)
outcomes = [result.outcome for result in results]
assert outcomes.count(DispatchOutcome.LAUNCHED) == 1, (i, outcomes)
assert len(launcher.calls) == 1, (i, launcher.calls)
assert await _active_run_count(runs, task.task_id) == 1, i
finally:
await close_engine()
async def test_partial_unique_index_enforces_one_active_run_per_task(tmp_path):
# Focused repository-level test of the index semantics + the typed conflict.
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
try:
sf = get_session_factory()
assert sf is not None
runs = SqlScheduledRunRepository(sf)
now = datetime(2026, 7, 2, 1, 0, tzinfo=UTC)
first = ScheduledRun.queued(task_id="t1", thread_id="th1", scheduled_for=now, trigger=TriggerKind.SCHEDULED)
await runs.add(first)
# queued -> running is a same-row UPDATE: keeps the one active slot, no
# violation (this is the normal launch transition).
await runs.update_status(first.record_id, status=RunStatus.RUNNING, run_id="run-1", started_at=now)
assert await runs.has_active("t1") is True
# A second active insert for the same task is a domain conflict.
with pytest.raises(ActiveRunConflictError):
await runs.add(ScheduledRun.queued(task_id="t1", thread_id="th2", scheduled_for=now, trigger=TriggerKind.MANUAL))
# Terminal-status rows for the same task are outside the index predicate.
await runs.add(ScheduledRun.skipped_tombstone(task_id="t1", thread_id="th3", scheduled_for=now, trigger=TriggerKind.SCHEDULED))
# A different task's active row is independent.
await runs.add(ScheduledRun.queued(task_id="t2", thread_id="th4", scheduled_for=now, trigger=TriggerKind.SCHEDULED))
# Finishing the active run frees the slot; a fresh active row is allowed.
await runs.update_status(first.record_id, status=RunStatus.SUCCESS, run_id="run-1", finished_at=now)
assert await runs.has_active("t1") is False
await runs.add(ScheduledRun.queued(task_id="t1", thread_id="th5", scheduled_for=now, trigger=TriggerKind.SCHEDULED))
assert await runs.has_active("t1") is True
finally:
await close_engine()

2
backend/tests/test_schedule_fakes.py
View File

@ -10,7 +10,7 @@ selects, that `add` refuses a second active record, what `protect_terminal`
preserves. What it deliberately does **not** own is atomicity -- the doubles
provide none, and a green run here says nothing about two dispatchers racing.
That is covered against a real database in
``test_scheduled_task_dispatch_race.py``. Do not read a passing contract suite
``test_schedule_dispatch_race.py``. Do not read a passing contract suite
as licence to run more than one scheduler.
"""

7
backend/tests/test_schedule_router.py
View File

@ -10,10 +10,9 @@ being raised.
own suite) while keeping `_map_domain_errors` in the call path, which is the
layer under test here.
The legacy `test_scheduled_task_router_behavior.py` remains the reference for
what this endpoint must do; every scenario it pins has a counterpart below,
plus the response-shape and error-mapping cases the old dict-returning router
could not express.
Every scenario the deleted legacy suite (`test_scheduled_task_router_behavior.py`)
pinned has a counterpart below, plus the response-shape and error-mapping
cases the old dict-returning router could not express.
"""
from __future__ import annotations

152
backend/tests/test_scheduled_task_claims.py
View File

@ -1,152 +0,0 @@
from datetime import UTC, datetime, timedelta
import pytest
from deerflow.config.database_config import DatabaseConfig
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
@pytest.mark.asyncio
async def test_claim_due_tasks_claims_only_due_rows(tmp_path):
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
sf = get_session_factory()
assert sf is not None
repo = ScheduledTaskRepository(sf)
due = datetime.now(UTC) - timedelta(minutes=1)
future = datetime.now(UTC) + timedelta(hours=1)
await repo.create(
task_id="due-1",
user_id="user-1",
thread_id="thread-1",
context_mode="reuse_thread",
assistant_id="lead_agent",
title="Due",
prompt="Prompt",
schedule_type="cron",
schedule_spec={"cron": "0 9 * * *"},
timezone="UTC",
next_run_at=due,
)
await repo.create(
task_id="future-1",
user_id="user-1",
thread_id="thread-1",
context_mode="reuse_thread",
assistant_id="lead_agent",
title="Future",
prompt="Prompt",
schedule_type="cron",
schedule_spec={"cron": "0 9 * * *"},
timezone="UTC",
next_run_at=future,
)
claimed = await repo.claim_due_tasks(
now=datetime.now(UTC),
lease_owner="worker-1",
lease_seconds=120,
limit=10,
)
assert [task["id"] for task in claimed] == ["due-1"]
await close_engine()
@pytest.mark.asyncio
async def test_claim_reclaims_task_stuck_in_running_with_expired_lease(tmp_path):
"""A task whose claiming process died mid-dispatch must stay reclaimable.
Regression for the lease dead-end bug: claim flips status to ``running``,
and the old claim query only selected ``status == 'enabled'``, so a crash
between claim and dispatch left the task permanently un-triggerable.
"""
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
sf = get_session_factory()
assert sf is not None
repo = ScheduledTaskRepository(sf)
now = datetime.now(UTC)
due = now - timedelta(minutes=5)
await repo.create(
task_id="stuck-1",
user_id="user-1",
thread_id="thread-1",
context_mode="reuse_thread",
assistant_id="lead_agent",
title="Stuck",
prompt="Prompt",
schedule_type="cron",
schedule_spec={"cron": "0 9 * * *"},
timezone="UTC",
next_run_at=due,
)
first_claim = await repo.claim_due_tasks(
now=now,
lease_owner="dead-worker",
lease_seconds=60,
limit=10,
)
assert first_claim[0]["id"] == "stuck-1"
assert first_claim[0]["status"] == "running"
# Simulate the claiming process dying: lease expires, status stays "running".
expired_now = now + timedelta(seconds=120)
reclaimed = await repo.claim_due_tasks(
now=expired_now,
lease_owner="new-worker",
lease_seconds=60,
limit=10,
)
assert [task["id"] for task in reclaimed] == ["stuck-1"]
assert reclaimed[0]["lease_owner"] == "new-worker"
await close_engine()
@pytest.mark.asyncio
async def test_claim_skips_task_with_active_lease(tmp_path):
"""A task whose lease has not expired must not be reclaimed."""
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
sf = get_session_factory()
assert sf is not None
repo = ScheduledTaskRepository(sf)
now = datetime.now(UTC)
due = now - timedelta(minutes=5)
await repo.create(
task_id="active-1",
user_id="user-1",
thread_id="thread-1",
context_mode="reuse_thread",
assistant_id="lead_agent",
title="Active",
prompt="Prompt",
schedule_type="cron",
schedule_spec={"cron": "0 9 * * *"},
timezone="UTC",
next_run_at=due,
)
await repo.claim_due_tasks(
now=now,
lease_owner="worker-1",
lease_seconds=300,
limit=10,
)
# Lease still valid — second claim within the same process must not re-grab it.
reclaimed = await repo.claim_due_tasks(
now=now + timedelta(seconds=10),
lease_owner="worker-2",
lease_seconds=300,
limit=10,
)
assert reclaimed == []
await close_engine()

219
backend/tests/test_scheduled_task_dispatch_race.py
View File

@ -1,219 +0,0 @@
"""Concurrency regression tests for the scheduled-task dispatch TOCTOU.
``ScheduledTaskService.dispatch_task`` guards the "at most one active run per
task when overlap_policy=skip" invariant with a non-atomic
``has_active_runs`` check followed by a separate ``create(status="queued")``
insert. Two concurrent dispatches (double-click, client retry, or a manual
trigger racing the poller) can both pass the check and both launch. The fix
makes the database the atomic arbiter via the partial unique index
``uq_scheduled_task_run_active`` (``task_id WHERE status IN
('queued','running')``); the losing insert is translated to the typed
``ActiveScheduledRunConflict`` and collapsed to the same outcome as the
fast-path check.
These tests drive the REAL ``ScheduledTaskRunRepository`` + ``ScheduledTaskService``
against a real file-backed ``sqlite+aiosqlite`` DB (so the index is actually
enforced), with a fake ``launch_run`` that only records launches.
"""
from __future__ import annotations
import asyncio
from datetime import UTC, datetime
import pytest
from app.scheduler.service import ScheduledTaskService
from deerflow.config.database_config import DatabaseConfig
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
from deerflow.persistence.scheduled_task_runs import ActiveScheduledRunConflict, ScheduledTaskRunRepository
from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
pytestmark = pytest.mark.asyncio
_ACTIVE_STATUSES = {"queued", "running"}
class _BarrierRunRepo(ScheduledTaskRunRepository):
"""Real repository that only releases both dispatchers past
``has_active_runs`` once both have read it, so their ``create()`` calls
genuinely race for the task's single active slot — a deterministic
reproduction of the check-then-insert TOCTOU."""
def __init__(self, session_factory, barrier: asyncio.Barrier | None) -> None:
super().__init__(session_factory)
self._barrier = barrier
async def has_active_runs(self, task_id: str) -> bool:
result = await super().has_active_runs(task_id)
if self._barrier is not None:
await self._barrier.wait()
return result
def _make_service(task_repo, run_repo, launched: list) -> ScheduledTaskService:
async def fake_launch(**kwargs):
# Yield so a truly-concurrent sibling can interleave, then record.
await asyncio.sleep(0)
launched.append(kwargs)
return {"run_id": f"run-{len(launched)}", "thread_id": kwargs["thread_id"]}
return ScheduledTaskService(
task_repo=task_repo,
task_run_repo=run_repo,
launch_run=fake_launch,
poll_interval_seconds=5,
lease_seconds=120,
max_concurrent_runs=10,
)
async def _seed_task(task_repo: ScheduledTaskRepository, task_id: str) -> dict:
# fresh_thread_per_run: every dispatch gets a NEW thread_id, so #4003's
# per-thread uq_runs_thread_active can never fire for two dispatches of the
# same task — this is precisely the gap the per-task index closes.
await task_repo.create(
task_id=task_id,
user_id="user-1",
thread_id=None,
context_mode="fresh_thread_per_run",
assistant_id="lead_agent",
title=task_id,
prompt="do the thing",
schedule_type="cron",
schedule_spec={"cron": "*/5 * * * *"},
timezone="UTC",
next_run_at=None,
)
task = await task_repo.get(task_id, user_id="user-1")
assert task is not None
assert task["overlap_policy"] == "skip"
return task
async def _active_run_count(run_repo: ScheduledTaskRunRepository, task_id: str) -> int:
rows = await run_repo.list_by_task(task_id, limit=100)
return sum(1 for row in rows if row["status"] in _ACTIVE_STATUSES)
async def test_two_concurrent_manual_dispatches_launch_exactly_once(tmp_path):
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
try:
sf = get_session_factory()
assert sf is not None
task_repo = ScheduledTaskRepository(sf)
run_repo = _BarrierRunRepo(sf, asyncio.Barrier(2))
launched: list = []
service = _make_service(task_repo, run_repo, launched)
task = await _seed_task(task_repo, "task-race-manual")
now = datetime.now(UTC)
results = await asyncio.gather(
service.dispatch_task(dict(task), now=now, trigger="manual"),
service.dispatch_task(dict(task), now=now, trigger="manual"),
)
outcomes = sorted(result["outcome"] for result in results)
# Exactly one wins the active slot; the loser is a 409-style conflict.
assert outcomes == ["conflict", "launched"], outcomes
assert len(launched) == 1, launched
assert await _active_run_count(run_repo, "task-race-manual") == 1
# The manual loser records no run-history row (nothing was scheduled).
conflict = next(r for r in results if r["outcome"] == "conflict")
assert conflict["task_run_id"] is None
finally:
await close_engine()
async def test_scheduled_and_manual_dispatch_launch_exactly_once(tmp_path):
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
try:
sf = get_session_factory()
assert sf is not None
task_repo = ScheduledTaskRepository(sf)
run_repo = _BarrierRunRepo(sf, asyncio.Barrier(2))
launched: list = []
service = _make_service(task_repo, run_repo, launched)
task = await _seed_task(task_repo, "task-race-mixed")
now = datetime.now(UTC)
results = await asyncio.gather(
service.dispatch_task(dict(task), now=now, trigger="scheduled"),
service.dispatch_task(dict(task), now=now, trigger="manual"),
)
outcomes = sorted(result["outcome"] for result in results)
# Whichever won launched; the loser is conflict (manual) or skipped
# (scheduled). Which one wins is timing-dependent, but exactly one runs.
assert outcomes.count("launched") == 1, outcomes
assert set(outcomes) <= {"launched", "conflict", "skipped"}, outcomes
assert len(launched) == 1, launched
assert await _active_run_count(run_repo, "task-race-mixed") == 1
finally:
await close_engine()
async def test_natural_timing_concurrent_dispatch_launches_exactly_once(tmp_path):
# No barrier: exercise the fix under the same natural interleaving that
# reproduced the bug (5/5 both-launch on main). The fix must hold whether
# the second dispatch is caught by the has_active_runs fast path or by the
# index-violation path.
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
try:
sf = get_session_factory()
assert sf is not None
task_repo = ScheduledTaskRepository(sf)
run_repo = ScheduledTaskRunRepository(sf)
for i in range(5):
launched: list = []
service = _make_service(task_repo, run_repo, launched)
task_id = f"task-natural-{i}"
task = await _seed_task(task_repo, task_id)
now = datetime.now(UTC)
results = await asyncio.gather(
service.dispatch_task(dict(task), now=now, trigger="manual"),
service.dispatch_task(dict(task), now=now, trigger="manual"),
)
outcomes = sorted(result["outcome"] for result in results)
assert outcomes.count("launched") == 1, (i, outcomes)
assert len(launched) == 1, (i, launched)
assert await _active_run_count(run_repo, task_id) == 1, i
finally:
await close_engine()
async def test_partial_unique_index_enforces_one_active_run_per_task(tmp_path):
# Focused repository-level test of the index semantics + the typed conflict.
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
try:
sf = get_session_factory()
assert sf is not None
run_repo = ScheduledTaskRunRepository(sf)
now = datetime(2026, 7, 2, 1, 0, tzinfo=UTC)
await run_repo.create(run_record_id="r1", task_id="t1", thread_id="th1", scheduled_for=now, trigger="scheduled", status="queued")
# queued -> running is a same-row UPDATE: keeps the one active slot, no
# violation (this is the normal launch transition).
await run_repo.update_status("r1", status="running", run_id="run-1", started_at=now)
assert await run_repo.has_active_runs("t1") is True
# A second active insert for the same task is a domain conflict.
with pytest.raises(ActiveScheduledRunConflict):
await run_repo.create(run_record_id="r2", task_id="t1", thread_id="th2", scheduled_for=now, trigger="manual", status="queued")
# Terminal-status rows for the same task are outside the index predicate.
await run_repo.create(run_record_id="r3", task_id="t1", thread_id="th3", scheduled_for=now, trigger="scheduled", status="skipped")
# A different task's active row is independent.
await run_repo.create(run_record_id="r4", task_id="t2", thread_id="th4", scheduled_for=now, trigger="scheduled", status="queued")
# Finishing the active run frees the slot; a fresh active row is allowed.
await run_repo.update_status("r1", status="success", run_id="run-1", finished_at=now)
assert await run_repo.has_active_runs("t1") is False
await run_repo.create(run_record_id="r5", task_id="t1", thread_id="th5", scheduled_for=now, trigger="scheduled", status="queued")
assert await run_repo.has_active_runs("t1") is True
finally:
await close_engine()

324
backend/tests/test_scheduled_task_repository.py
View File

@ -1,324 +0,0 @@
from datetime import UTC, datetime
import pytest
from deerflow.config.database_config import DatabaseConfig
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
from deerflow.persistence.scheduled_task_runs import ScheduledTaskRunRepository
from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
@pytest.mark.asyncio
async def test_scheduled_task_repository_create_and_list(tmp_path):
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
sf = get_session_factory()
assert sf is not None
repo = ScheduledTaskRepository(sf)
created = 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 this thread",
schedule_type="cron",
schedule_spec={"cron": "0 9 * * *"},
timezone="Asia/Shanghai",
next_run_at=datetime(2026, 7, 2, 1, 0, tzinfo=UTC),
)
assert created["id"] == "task-1"
listed = await repo.list_by_user("user-1")
assert [task["id"] for task in listed] == ["task-1"]
await close_engine()
@pytest.mark.asyncio
async def test_scheduled_task_run_repository_records_history(tmp_path):
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
sf = get_session_factory()
assert sf is not None
repo = ScheduledTaskRunRepository(sf)
row = await repo.create(
run_record_id="task-run-1",
task_id="task-1",
thread_id="thread-1",
scheduled_for=datetime(2026, 7, 2, 1, 0, tzinfo=UTC),
trigger="manual",
status="queued",
)
assert row["id"] == "task-run-1"
history = await repo.list_by_task("task-1")
assert [entry["id"] for entry in history] == ["task-run-1"]
await close_engine()
@pytest.mark.asyncio
async def test_mark_stale_active_runs_fails_orphaned_runs(tmp_path):
"""Runs stuck in queued/running after a process crash are swept to interrupted."""
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
sf = get_session_factory()
assert sf is not None
repo = ScheduledTaskRunRepository(sf)
# The queued and running rows live on different tasks: mark_stale_active_runs
# is a global sweep (no task filter), and the uq_scheduled_task_run_active
# partial unique index forbids two active rows on one task_id, so the pair
# that proves both active statuses get swept must be spread across tasks.
await repo.create(
run_record_id="task-run-queued",
task_id="task-1",
thread_id="thread-1",
scheduled_for=datetime(2026, 7, 2, 1, 0, tzinfo=UTC),
trigger="scheduled",
status="queued",
)
await repo.create(
run_record_id="task-run-running",
task_id="task-2",
thread_id="thread-2",
scheduled_for=datetime(2026, 7, 2, 1, 0, tzinfo=UTC),
trigger="scheduled",
status="running",
)
# A terminal row on task-1 (outside the index predicate) coexists with the
# active queued row and must be left untouched by the sweep.
await repo.create(
run_record_id="task-run-success",
task_id="task-1",
thread_id="thread-1",
scheduled_for=datetime(2026, 7, 2, 1, 0, tzinfo=UTC),
trigger="scheduled",
status="success",
)
swept = await repo.mark_stale_active_runs(error="interrupted: gateway restarted")
assert swept == 2
by_id = {entry["id"]: entry for entry in await repo.list_by_task("task-1")}
by_id.update({entry["id"]: entry for entry in await repo.list_by_task("task-2")})
assert by_id["task-run-queued"]["status"] == "interrupted"
assert by_id["task-run-running"]["status"] == "interrupted"
assert by_id["task-run-success"]["status"] == "success"
await close_engine()
@pytest.mark.asyncio
async def test_update_status_protect_terminal_keeps_completion_result(tmp_path):
"""The launch-path "running" write must not clobber a terminal status
already committed by the completion hook (launch/completion race)."""
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
sf = get_session_factory()
assert sf is not None
repo = ScheduledTaskRunRepository(sf)
await repo.create(
run_record_id="task-run-race",
task_id="task-1",
thread_id="thread-1",
scheduled_for=datetime(2026, 7, 2, 1, 0, tzinfo=UTC),
trigger="scheduled",
status="queued",
)
# Completion hook wins the race and commits the terminal state first.
await repo.update_status("task-run-race", status="failed", run_id="run-1", error="boom", finished_at=datetime(2026, 7, 2, 1, 1, tzinfo=UTC))
# Late launch-path write: keeps terminal status/error, backfills started_at.
await repo.update_status("task-run-race", status="running", run_id="run-1", started_at=datetime(2026, 7, 2, 1, 0, tzinfo=UTC), protect_terminal=True)
entry = (await repo.list_by_task("task-1"))[0]
assert entry["status"] == "failed"
assert entry["error"] == "boom"
assert entry["started_at"] is not None
await close_engine()
@pytest.mark.asyncio
async def test_has_active_runs_sees_only_queued_and_running(tmp_path):
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
sf = get_session_factory()
assert sf is not None
repo = ScheduledTaskRunRepository(sf)
assert await repo.has_active_runs("task-1") is False
await repo.create(
run_record_id="task-run-active",
task_id="task-1",
thread_id="thread-1",
scheduled_for=datetime(2026, 7, 2, 1, 0, tzinfo=UTC),
trigger="scheduled",
status="running",
)
assert await repo.has_active_runs("task-1") is True
await repo.update_status("task-run-active", status="success", run_id="run-1")
assert await repo.has_active_runs("task-1") is False
await close_engine()
@pytest.mark.asyncio
async def test_cancel_stuck_once_tasks_reconciles_orphaned_running(tmp_path):
"""Launched (lease cleared) once tasks stuck in running are cancelled at
startup; leased ones are left for expired-lease reclaim."""
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
sf = get_session_factory()
assert sf is not None
repo = ScheduledTaskRepository(sf)
for task_id, schedule_type, status in (
("task-once-stuck", "once", "running"),
("task-once-done", "once", "completed"),
("task-cron-running", "cron", "running"),
):
await repo.create(
task_id=task_id,
user_id="user-1",
thread_id=None,
context_mode="fresh_thread_per_run",
assistant_id="lead_agent",
title=task_id,
prompt="p",
schedule_type=schedule_type,
schedule_spec={"run_at": "2026-07-02T01:00:00+00:00"} if schedule_type == "once" else {"cron": "0 9 * * *"},
timezone="UTC",
next_run_at=None,
)
await repo.update(task_id, user_id="user-1", updates={"status": status})
# A claimed-but-not-launched once task still holds its lease: keep it.
await repo.create(
task_id="task-once-leased",
user_id="user-1",
thread_id=None,
context_mode="fresh_thread_per_run",
assistant_id="lead_agent",
title="task-once-leased",
prompt="p",
schedule_type="once",
schedule_spec={"run_at": "2026-07-02T01:00:00+00:00"},
timezone="UTC",
next_run_at=datetime(2026, 7, 2, 1, 0, tzinfo=UTC),
)
await repo.update("task-once-leased", user_id="user-1", updates={"status": "running", "lease_expires_at": datetime(2026, 7, 2, 1, 2, tzinfo=UTC)})
cancelled = await repo.cancel_stuck_once_tasks(error="interrupted: gateway restarted")
assert cancelled == 1
by_id = {t["id"]: t for t in await repo.list_by_user("user-1")}
assert by_id["task-once-stuck"]["status"] == "cancelled"
assert by_id["task-once-stuck"]["last_error"] == "interrupted: gateway restarted"
assert by_id["task-once-done"]["status"] == "completed"
assert by_id["task-cron-running"]["status"] == "running"
assert by_id["task-once-leased"]["status"] == "running"
await close_engine()
@pytest.mark.asyncio
async def test_update_after_launch_protect_terminal_keeps_hook_result(tmp_path):
"""The launch-path bookkeeping write must not clobber a terminal task
status committed first by the completion hook (fast-failing run)."""
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
sf = get_session_factory()
assert sf is not None
repo = ScheduledTaskRepository(sf)
await repo.create(
task_id="task-race",
user_id="user-1",
thread_id=None,
context_mode="fresh_thread_per_run",
assistant_id="lead_agent",
title="task-race",
prompt="p",
schedule_type="once",
schedule_spec={"run_at": "2026-07-02T01:00:00+00:00"},
timezone="UTC",
next_run_at=datetime(2026, 7, 2, 1, 0, tzinfo=UTC),
)
# Completion hook wins the race: task finalized as failed.
await repo.update("task-race", user_id="user-1", updates={"status": "failed", "last_error": "boom"})
# Late launch-path write with protection keeps the hook's outcome.
await repo.update_after_launch(
"task-race",
status="running",
next_run_at=None,
last_run_at=datetime(2026, 7, 2, 1, 0, tzinfo=UTC),
last_run_id="run-1",
last_thread_id="thread-1",
last_error=None,
increment_run_count=True,
protect_terminal=True,
)
task = await repo.get("task-race", user_id="user-1")
assert task is not None
assert task["status"] == "failed"
assert task["last_error"] == "boom"
# Launch bookkeeping still recorded.
assert task["last_run_id"] == "run-1"
assert task["run_count"] == 1
await close_engine()
@pytest.mark.asyncio
async def test_list_by_task_paginates(tmp_path):
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
sf = get_session_factory()
assert sf is not None
repo = ScheduledTaskRunRepository(sf)
for i in range(5):
await repo.create(
run_record_id=f"task-run-{i}",
task_id="task-1",
thread_id="thread-1",
scheduled_for=datetime(2026, 7, 2, 1, i, tzinfo=UTC),
trigger="scheduled",
status="success",
)
assert await repo.count_active_runs() == 0
page1 = await repo.list_by_task("task-1", limit=2)
page2 = await repo.list_by_task("task-1", limit=2, offset=2)
assert len(page1) == 2
assert len(page2) == 2
assert {e["id"] for e in page1}.isdisjoint({e["id"] for e in page2})
await close_engine()
@pytest.mark.asyncio
async def test_list_by_user_and_thread_filters_in_sql(tmp_path):
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
sf = get_session_factory()
assert sf is not None
repo = ScheduledTaskRepository(sf)
for task_id, thread_id in (("task-a", "thread-1"), ("task-b", "thread-2"), ("task-c", "thread-1")):
await repo.create(
task_id=task_id,
user_id="user-1",
thread_id=thread_id,
context_mode="reuse_thread",
assistant_id="lead_agent",
title=task_id,
prompt="p",
schedule_type="cron",
schedule_spec={"cron": "0 9 * * *"},
timezone="UTC",
next_run_at=None,
)
listed = await repo.list_by_user_and_thread("user-1", "thread-1")
assert sorted(t["id"] for t in listed) == ["task-a", "task-c"]
assert await repo.list_by_user_and_thread("user-2", "thread-1") == []
await close_engine()

38
backend/tests/test_scheduled_task_router.py
View File

@ -1,38 +0,0 @@
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.gateway.routers import scheduled_tasks
def test_router_registers_list_endpoint():
app = FastAPI()
app.include_router(scheduled_tasks.router)
client = TestClient(app)
response = client.get("/api/scheduled-tasks")
assert response.status_code != 404
def test_router_registers_trigger_route():
app = FastAPI()
app.include_router(scheduled_tasks.router)
client = TestClient(app)
response = client.post("/api/scheduled-tasks/task-1/trigger")
assert response.status_code != 404
def test_router_registers_create_route():
app = FastAPI()
app.include_router(scheduled_tasks.router)
client = TestClient(app)
response = client.post(
"/api/scheduled-tasks",
json={
"thread_id": "thread-1",
"title": "Daily summary",
"prompt": "Summarize thread",
"schedule_type": "cron",
"schedule_spec": {"cron": "0 9 * * *"},
"timezone": "UTC",
},
)
assert response.status_code != 404

652
backend/tests/test_scheduled_task_router_behavior.py
View File

@ -1,652 +0,0 @@
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

49
backend/tests/test_scheduled_task_schedules.py
View File

@ -1,49 +0,0 @@
from datetime import UTC, datetime
import pytest
from deerflow.scheduler.schedules import (
next_run_at,
normalize_cron_expression,
validate_timezone,
)
def test_validate_timezone_accepts_iana_name():
assert validate_timezone("Asia/Shanghai") == "Asia/Shanghai"
def test_validate_timezone_rejects_unknown_name():
with pytest.raises(ValueError):
validate_timezone("Mars/Base")
def test_normalize_cron_accepts_five_fields():
assert normalize_cron_expression("0 9 * * 1") == "0 9 * * 1"
def test_normalize_cron_rejects_seconds_field():
with pytest.raises(ValueError):
normalize_cron_expression("0 0 9 * * 1")
def test_next_run_at_for_once_returns_none_after_fire_time():
now = datetime(2026, 7, 2, 2, 0, tzinfo=UTC)
result = next_run_at(
"once",
{"run_at": "2026-07-02T01:00:00+00:00"},
"UTC",
now=now,
)
assert result is None
def test_next_run_at_for_cron_uses_timezone():
now = datetime(2026, 7, 1, 0, 30, tzinfo=UTC)
result = next_run_at(
"cron",
{"cron": "0 9 * * *"},
"Asia/Shanghai",
now=now,
)
assert result == datetime(2026, 7, 1, 1, 0, tzinfo=UTC)

543
backend/tests/test_scheduled_task_service.py
View File

@ -1,543 +0,0 @@
from datetime import UTC, datetime, timedelta
import pytest
from app.scheduler.service import ScheduledTaskService
from deerflow.runtime import ConflictError, RunStatus
from deerflow.runtime.runs.manager import RunRecord
from deerflow.runtime.runs.schemas import DisconnectMode
class DummyTaskRepo:
def __init__(self, rows):
self.rows = rows
self.claimed = False
self.updated = None
self.cancelled_stuck_once = None
async def cancel_stuck_once_tasks(self, *, error):
self.cancelled_stuck_once = error
return 0
async def claim_due_tasks(self, **_kwargs):
if self.claimed:
return []
self.claimed = True
return self.rows
async def update_after_launch(self, *args, **kwargs):
self.updated = (args, kwargs)
async def get(self, task_id: str, *, user_id: str):
row = next((item for item in self.rows if item["id"] == task_id and item["user_id"] == user_id), None)
return dict(row) if row is not None else None
async def update(self, task_id: str, *, user_id: str, updates):
row = next((item for item in self.rows if item["id"] == task_id and item["user_id"] == user_id), None)
if row is None:
return None
row.update(updates)
return dict(row)
class DummyRunRepo:
def __init__(self, *, active=False, active_count=0):
self.created = None
self.updated = []
self.active = active
self.active_count = active_count
self.stale_marked = None
async def count_active_runs(self):
return self.active_count
async def create(self, **kwargs):
self.created = kwargs
return {"id": kwargs["run_record_id"]}
async def update_status(self, run_record_id, **kwargs):
self.updated.append((run_record_id, kwargs))
async def has_active_runs(self, task_id):
return self.active
async def mark_stale_active_runs(self, *, error):
self.stale_marked = error
return 0
@pytest.mark.asyncio
async def test_service_claims_and_dispatches_due_task():
async def fake_launch(**kwargs):
assert kwargs["owner_user_id"] == "user-1"
assert kwargs["metadata"]["scheduled_task_id"] == "task-1"
assert kwargs["metadata"]["scheduled_trigger"] == "scheduled"
return {"run_id": "run-1", "thread_id": kwargs["thread_id"]}
task_repo = DummyTaskRepo(
[
{
"id": "task-1",
"user_id": "user-1",
"thread_id": "thread-1",
"context_mode": "reuse_thread",
"assistant_id": "lead_agent",
"prompt": "Summarize thread",
"schedule_type": "once",
"schedule_spec": {"run_at": "2026-07-02T01:00:00+00:00"},
"timezone": "UTC",
}
]
)
run_repo = DummyRunRepo()
service = ScheduledTaskService(
task_repo=task_repo,
task_run_repo=run_repo,
launch_run=fake_launch,
poll_interval_seconds=5,
lease_seconds=120,
max_concurrent_runs=3,
)
await service.run_once(now=datetime.now(UTC) + timedelta(days=1))
assert run_repo.created["task_id"] == "task-1"
assert run_repo.updated[0][1]["status"] == "running"
assert run_repo.updated[0][1]["protect_terminal"] is True
# `once` terminal status is owned by handle_run_completion, not the launch.
assert task_repo.updated[1]["status"] == "running"
@pytest.mark.asyncio
async def test_manual_trigger_keeps_paused_cron_task_paused():
async def fake_launch(**kwargs):
return {"run_id": "run-2", "thread_id": kwargs["thread_id"]}
task_repo = DummyTaskRepo(
[
{
"id": "task-2",
"user_id": "user-1",
"thread_id": "thread-1",
"context_mode": "reuse_thread",
"assistant_id": "lead_agent",
"prompt": "Summarize thread",
"schedule_type": "cron",
"schedule_spec": {"cron": "0 9 * * *"},
"timezone": "UTC",
"status": "paused",
}
]
)
run_repo = DummyRunRepo()
service = ScheduledTaskService(
task_repo=task_repo,
task_run_repo=run_repo,
launch_run=fake_launch,
poll_interval_seconds=5,
lease_seconds=120,
max_concurrent_runs=3,
)
await service.dispatch_task(
task_repo.rows[0],
now=datetime.now(UTC),
trigger="manual",
)
assert task_repo.updated[1]["status"] == "paused"
@pytest.mark.asyncio
async def test_fresh_thread_per_run_creates_new_execution_thread():
async def fake_launch(**kwargs):
assert kwargs["thread_id"] != "thread-template"
return {"run_id": "run-3", "thread_id": kwargs["thread_id"]}
task_repo = DummyTaskRepo(
[
{
"id": "task-3",
"user_id": "user-1",
"thread_id": "thread-template",
"context_mode": "fresh_thread_per_run",
"assistant_id": "lead_agent",
"prompt": "Summarize thread",
"schedule_type": "cron",
"schedule_spec": {"cron": "0 9 * * *"},
"timezone": "UTC",
"status": "enabled",
}
]
)
run_repo = DummyRunRepo()
service = ScheduledTaskService(
task_repo=task_repo,
task_run_repo=run_repo,
launch_run=fake_launch,
poll_interval_seconds=5,
lease_seconds=120,
max_concurrent_runs=3,
)
await service.dispatch_task(
task_repo.rows[0],
now=datetime.now(UTC),
trigger="scheduled",
)
assert run_repo.created["thread_id"] != "thread-template"
assert task_repo.updated[1]["last_thread_id"] == run_repo.created["thread_id"]
@pytest.mark.asyncio
async def test_scheduled_overlap_conflict_is_recorded_as_skip():
async def fake_launch(**_kwargs):
raise ConflictError("Thread thread-1 already has an active run")
task_repo = DummyTaskRepo(
[
{
"id": "task-4",
"user_id": "user-1",
"thread_id": "thread-1",
"context_mode": "reuse_thread",
"assistant_id": "lead_agent",
"prompt": "Summarize thread",
"schedule_type": "cron",
"schedule_spec": {"cron": "0 9 * * *"},
"timezone": "UTC",
"status": "running",
"overlap_policy": "skip",
"last_run_id": "run-old",
"last_thread_id": "thread-1",
"last_run_at": "2026-07-01T00:00:00+00:00",
}
]
)
run_repo = DummyRunRepo()
service = ScheduledTaskService(
task_repo=task_repo,
task_run_repo=run_repo,
launch_run=fake_launch,
poll_interval_seconds=5,
lease_seconds=120,
max_concurrent_runs=3,
)
result = await service.dispatch_task(
task_repo.rows[0],
now=datetime.now(UTC),
trigger="scheduled",
)
assert result["outcome"] == "skipped"
assert run_repo.updated[-1][1]["status"] == "skipped"
assert task_repo.updated[1]["status"] == "enabled"
@pytest.mark.asyncio
async def test_manual_overlap_conflict_returns_conflict():
async def fake_launch(**_kwargs):
raise ConflictError("Thread thread-1 already has an active run")
task_repo = DummyTaskRepo(
[
{
"id": "task-5",
"user_id": "user-1",
"thread_id": "thread-1",
"context_mode": "reuse_thread",
"assistant_id": "lead_agent",
"prompt": "Summarize thread",
"schedule_type": "cron",
"schedule_spec": {"cron": "0 9 * * *"},
"timezone": "UTC",
"status": "enabled",
"overlap_policy": "skip",
}
]
)
run_repo = DummyRunRepo()
service = ScheduledTaskService(
task_repo=task_repo,
task_run_repo=run_repo,
launch_run=fake_launch,
poll_interval_seconds=5,
lease_seconds=120,
max_concurrent_runs=3,
)
result = await service.dispatch_task(
task_repo.rows[0],
now=datetime.now(UTC),
trigger="manual",
)
assert result["outcome"] == "conflict"
assert run_repo.updated[-1][1]["status"] == "failed"
@pytest.mark.asyncio
async def test_handle_run_completion_persists_success():
task_repo = DummyTaskRepo(
[
{
"id": "task-6",
"user_id": "user-1",
"thread_id": None,
"context_mode": "fresh_thread_per_run",
"assistant_id": "lead_agent",
"prompt": "Summarize thread",
"schedule_type": "cron",
"schedule_spec": {"cron": "0 9 * * *"},
"timezone": "UTC",
"status": "enabled",
}
]
)
run_repo = DummyRunRepo()
service = ScheduledTaskService(
task_repo=task_repo,
task_run_repo=run_repo,
launch_run=lambda **_kwargs: None,
poll_interval_seconds=5,
lease_seconds=120,
max_concurrent_runs=3,
)
record = RunRecord(
run_id="run-6",
thread_id="thread-6",
assistant_id="lead_agent",
status=RunStatus.success,
on_disconnect=DisconnectMode.continue_,
metadata={
"scheduled_task_id": "task-6",
"scheduled_task_run_id": "task-run-6",
},
user_id="user-1",
)
await service.handle_run_completion(record)
assert run_repo.updated[-1][0] == "task-run-6"
assert run_repo.updated[-1][1]["status"] == "success"
assert task_repo.rows[0]["last_error"] is None
def _make_service(task_repo, run_repo):
return ScheduledTaskService(
task_repo=task_repo,
task_run_repo=run_repo,
launch_run=lambda **_kwargs: None,
poll_interval_seconds=5,
lease_seconds=120,
max_concurrent_runs=3,
)
def _once_task_row(task_id="task-once", status="running"):
return {
"id": task_id,
"user_id": "user-1",
"thread_id": None,
"context_mode": "fresh_thread_per_run",
"assistant_id": "lead_agent",
"prompt": "Summarize thread",
"schedule_type": "once",
"schedule_spec": {"run_at": "2026-07-02T01:00:00+00:00"},
"timezone": "UTC",
"status": status,
}
def _completion_record(status, *, task_id="task-once", error=None):
return RunRecord(
run_id="run-x",
thread_id="thread-x",
assistant_id="lead_agent",
status=status,
on_disconnect=DisconnectMode.continue_,
metadata={
"scheduled_task_id": task_id,
"scheduled_task_run_id": "task-run-x",
},
user_id="user-1",
error=error,
)
@pytest.mark.asyncio
async def test_once_task_completes_only_via_completion_hook():
task_repo = DummyTaskRepo([_once_task_row()])
run_repo = DummyRunRepo()
service = _make_service(task_repo, run_repo)
await service.handle_run_completion(_completion_record(RunStatus.success))
assert run_repo.updated[-1][1]["status"] == "success"
assert task_repo.rows[0]["status"] == "completed"
@pytest.mark.asyncio
async def test_once_task_failed_run_marks_task_failed():
task_repo = DummyTaskRepo([_once_task_row()])
run_repo = DummyRunRepo()
service = _make_service(task_repo, run_repo)
await service.handle_run_completion(_completion_record(RunStatus.error, error="boom"))
assert run_repo.updated[-1][1]["status"] == "failed"
assert run_repo.updated[-1][1]["error"] == "boom"
assert task_repo.rows[0]["status"] == "failed"
assert task_repo.rows[0]["last_error"] == "boom"
@pytest.mark.asyncio
async def test_interrupted_run_is_distinct_and_cancels_once_task():
task_repo = DummyTaskRepo([_once_task_row()])
run_repo = DummyRunRepo()
service = _make_service(task_repo, run_repo)
await service.handle_run_completion(_completion_record(RunStatus.interrupted))
run_update = run_repo.updated[-1][1]
assert run_update["status"] == "interrupted"
assert run_update["error"] == "run was interrupted before completion"
assert task_repo.rows[0]["status"] == "cancelled"
@pytest.mark.asyncio
async def test_interrupted_cron_run_keeps_task_enabled():
row = _once_task_row(task_id="task-cron")
row.update({"schedule_type": "cron", "schedule_spec": {"cron": "0 9 * * *"}, "status": "enabled"})
task_repo = DummyTaskRepo([row])
run_repo = DummyRunRepo()
service = _make_service(task_repo, run_repo)
await service.handle_run_completion(_completion_record(RunStatus.interrupted, task_id="task-cron"))
assert run_repo.updated[-1][1]["status"] == "interrupted"
assert task_repo.rows[0]["status"] == "enabled"
@pytest.mark.asyncio
async def test_skip_policy_applies_to_fresh_thread_runs():
launched = []
async def fake_launch(**kwargs):
launched.append(kwargs)
return {"run_id": "run-9", "thread_id": kwargs["thread_id"]}
row = _once_task_row(task_id="task-9")
row.update({"schedule_type": "cron", "schedule_spec": {"cron": "* * * * *"}, "status": "running", "overlap_policy": "skip"})
task_repo = DummyTaskRepo([row])
run_repo = DummyRunRepo(active=True)
service = ScheduledTaskService(
task_repo=task_repo,
task_run_repo=run_repo,
launch_run=fake_launch,
poll_interval_seconds=5,
lease_seconds=120,
max_concurrent_runs=3,
)
result = await service.dispatch_task(row, now=datetime.now(UTC), trigger="scheduled")
assert result["outcome"] == "skipped"
assert launched == []
# The skip tombstone is created directly as terminal "skipped" (not the
# transient "queued" the launch path uses): a queued row is active and would
# itself trip the uq_scheduled_task_run_active partial unique index against
# the pre-existing run still holding the task's single active slot.
assert run_repo.created["status"] == "skipped"
assert run_repo.updated[-1][1]["status"] == "skipped"
assert task_repo.updated[1]["status"] == "enabled"
assert task_repo.updated[1]["increment_run_count"] is False
@pytest.mark.asyncio
async def test_startup_sweep_reconciles_stale_runs_and_stuck_once_tasks():
task_repo = DummyTaskRepo([])
run_repo = DummyRunRepo()
service = _make_service(task_repo, run_repo)
await service.start()
await service.stop()
assert run_repo.stale_marked is not None
assert task_repo.cancelled_stuck_once == run_repo.stale_marked
@pytest.mark.asyncio
async def test_manual_trigger_with_active_run_returns_conflict_without_launching():
launched = []
async def fake_launch(**kwargs):
launched.append(kwargs)
return {"run_id": "run-x", "thread_id": kwargs["thread_id"]}
row = _once_task_row(task_id="task-manual-busy")
row.update({"schedule_type": "cron", "schedule_spec": {"cron": "* * * * *"}, "status": "enabled", "overlap_policy": "skip"})
task_repo = DummyTaskRepo([row])
run_repo = DummyRunRepo(active=True)
service = ScheduledTaskService(
task_repo=task_repo,
task_run_repo=run_repo,
launch_run=fake_launch,
poll_interval_seconds=5,
lease_seconds=120,
max_concurrent_runs=3,
)
result = await service.dispatch_task(row, now=datetime.now(UTC), trigger="manual")
assert result["outcome"] == "conflict"
assert launched == []
# Nothing was scheduled to happen, so no run-history row is recorded.
assert run_repo.created is None
assert result["task_run_id"] is None
@pytest.mark.asyncio
async def test_run_once_claims_only_into_remaining_global_budget():
claim_limits = []
class BudgetTaskRepo(DummyTaskRepo):
async def claim_due_tasks(self, **kwargs):
claim_limits.append(kwargs["limit"])
return []
task_repo = BudgetTaskRepo([])
run_repo = DummyRunRepo(active_count=2)
service = _make_service(task_repo, run_repo)
await service.run_once(now=datetime.now(UTC))
assert claim_limits == [1]
run_repo.active_count = 3
await service.run_once(now=datetime.now(UTC))
# Budget exhausted: no claim at all this cycle.
assert claim_limits == [1]
@pytest.mark.asyncio
async def test_launch_bookkeeping_passes_protect_terminal():
async def fake_launch(**kwargs):
return {"run_id": "run-pt", "thread_id": kwargs["thread_id"]}
task_repo = DummyTaskRepo([_once_task_row(task_id="task-pt", status="enabled")])
run_repo = DummyRunRepo()
service = ScheduledTaskService(
task_repo=task_repo,
task_run_repo=run_repo,
launch_run=fake_launch,
poll_interval_seconds=5,
lease_seconds=120,
max_concurrent_runs=3,
)
await service.dispatch_task(task_repo.rows[0], now=datetime.now(UTC), trigger="scheduled")
assert task_repo.updated[1]["protect_terminal"] is True

148
backend/tests/test_scheduled_task_skip_writeback.py
View File

@ -1,148 +0,0 @@
"""Regression test for the write-back on an overlap-skipped dispatch.
``ScheduledTaskService._finalize_skip`` preserves a task's launch bookkeeping
by reading the current values off the task dict and passing them straight back
into ``update_after_launch``. But repository dicts carry ISO *strings* for
timestamps (``ScheduledTaskRepository._row_to_dict`` runs every datetime column
through ``coerce_iso``), while ``scheduled_tasks.last_run_at`` is a
``DateTime`` column -- so the round trip fed a string into a datetime bind
parameter and the write raised.
It only reproduces once the task has actually launched at least once: before
that ``last_run_at`` is NULL, and NULL is accepted by the column. That is why
the existing dispatch tests, which all seed a fresh task, never hit it.
Uses the REAL repositories against a file-backed ``sqlite+aiosqlite`` DB,
because an in-memory fake stores whatever object it is handed and cannot
reject the type.
"""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
import pytest
from app.scheduler.service import ScheduledTaskService
from deerflow.config.database_config import DatabaseConfig
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
from deerflow.persistence.scheduled_task_runs import ScheduledTaskRunRepository
from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
pytestmark = pytest.mark.asyncio
def _make_service(task_repo, run_repo, launched: list) -> ScheduledTaskService:
async def fake_launch(**kwargs):
launched.append(kwargs)
return {"run_id": f"run-{len(launched)}", "thread_id": kwargs["thread_id"]}
return ScheduledTaskService(
task_repo=task_repo,
task_run_repo=run_repo,
launch_run=fake_launch,
poll_interval_seconds=5,
lease_seconds=120,
max_concurrent_runs=10,
)
async def _seed_task(task_repo: ScheduledTaskRepository, task_id: str, *, schedule_type: str = "cron") -> dict:
spec = {"cron": "*/5 * * * *"} if schedule_type == "cron" else {"run_at": "2099-01-01T00:00:00+00:00"}
await task_repo.create(
task_id=task_id,
user_id="user-1",
thread_id=None,
context_mode="fresh_thread_per_run",
assistant_id="lead_agent",
title=task_id,
prompt="do the thing",
schedule_type=schedule_type,
schedule_spec=spec,
timezone="UTC",
next_run_at=None,
)
task = await task_repo.get(task_id, user_id="user-1")
assert task is not None
return task
async def test_scheduled_skip_preserves_bookkeeping_of_a_previously_launched_task(tmp_path):
"""A cron task that already ran once, then overlaps, must record the skip
without disturbing (or failing to write) its launch bookkeeping."""
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
try:
sf = get_session_factory()
assert sf is not None
task_repo = ScheduledTaskRepository(sf)
run_repo = ScheduledTaskRunRepository(sf)
launched: list = []
service = _make_service(task_repo, run_repo, launched)
task = await _seed_task(task_repo, "task-skip-cron")
first_at = datetime(2026, 7, 27, 12, 0, tzinfo=UTC)
first = await service.dispatch_task(dict(task), now=first_at, trigger="scheduled")
assert first["outcome"] == "launched"
after_launch = await task_repo.get("task-skip-cron", user_id="user-1")
assert after_launch is not None
# The precondition the old dispatch tests never established.
assert after_launch["last_run_at"] is not None
# The fake launcher never terminates the run, so its row is still
# active and this second dispatch overlaps. `after_launch` is passed
# verbatim, exactly as the poller passes what claim_due_tasks returned.
second = await service.dispatch_task(after_launch, now=first_at + timedelta(minutes=5), trigger="scheduled")
assert second["outcome"] == "skipped"
assert len(launched) == 1, "the overlapping dispatch must not launch"
after_skip = await task_repo.get("task-skip-cron", user_id="user-1")
assert after_skip is not None
# A skip is not an execution: the launch bookkeeping is carried over
# untouched and the counter does not move.
assert after_skip["last_run_at"] == after_launch["last_run_at"]
assert after_skip["last_run_id"] == after_launch["last_run_id"]
assert after_skip["last_thread_id"] == after_launch["last_thread_id"]
assert after_skip["run_count"] == after_launch["run_count"] == 1
# A cron task stays claimable; only `once` burns its occurrence.
assert after_skip["status"] == "enabled"
assert after_skip["last_error"] is None
# The skip itself is recorded as a terminal tombstone.
rows = await run_repo.list_by_task("task-skip-cron", limit=10)
assert sorted(row["status"] for row in rows) == ["running", "skipped"]
finally:
await close_engine()
async def test_scheduled_skip_of_a_previously_launched_once_task_records_the_lost_occurrence(tmp_path):
"""Same write-back path, `once` branch: the single occurrence is lost, so
the task ends `failed` and carries the skip reason."""
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
try:
sf = get_session_factory()
assert sf is not None
task_repo = ScheduledTaskRepository(sf)
run_repo = ScheduledTaskRunRepository(sf)
launched: list = []
service = _make_service(task_repo, run_repo, launched)
task = await _seed_task(task_repo, "task-skip-once", schedule_type="once")
first_at = datetime(2026, 7, 27, 12, 0, tzinfo=UTC)
first = await service.dispatch_task(dict(task), now=first_at, trigger="scheduled")
assert first["outcome"] == "launched"
after_launch = await task_repo.get("task-skip-once", user_id="user-1")
assert after_launch is not None
assert after_launch["last_run_at"] is not None
second = await service.dispatch_task(after_launch, now=first_at + timedelta(minutes=5), trigger="scheduled")
assert second["outcome"] == "skipped"
after_skip = await task_repo.get("task-skip-once", user_id="user-1")
assert after_skip is not None
assert after_skip["last_run_at"] == after_launch["last_run_at"]
assert after_skip["status"] == "failed"
assert after_skip["last_error"] == "skipped: a previous run of this task is still active"
finally:
await close_engine()