diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 222239389..ab346e9a8 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -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|`, 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 ``-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 ``-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. diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index 72be91337..fd50727ad 100644 --- a/backend/app/channels/manager.py +++ b/backend/app/channels/manager.py @@ -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 diff --git a/backend/app/channels/service.py b/backend/app/channels/service.py index f46724beb..bab30d9cc 100644 --- a/backend/app/channels/service.py +++ b/backend/app/channels/service.py @@ -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: diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py index 080c9391f..08f0d0246 100644 --- a/backend/app/gateway/app.py +++ b/backend/app/gateway/app.py @@ -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 diff --git a/backend/app/gateway/deps.py b/backend/app/gateway/deps.py index a836acdd4..120f18872 100644 --- a/backend/app/gateway/deps.py +++ b/backend/app/gateway/deps.py @@ -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. diff --git a/backend/app/gateway/routers/__init__.py b/backend/app/gateway/routers/__init__.py index a948c522b..27afb4341 100644 --- a/backend/app/gateway/routers/__init__.py +++ b/backend/app/gateway/routers/__init__.py @@ -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", diff --git a/backend/app/gateway/routers/scheduled_tasks.py b/backend/app/gateway/routers/scheduled_tasks.py deleted file mode 100644 index 35cd74bf1..000000000 --- a/backend/app/gateway/routers/scheduled_tasks.py +++ /dev/null @@ -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) diff --git a/backend/app/scheduler/__init__.py b/backend/app/scheduler/__init__.py index a3c331940..2bec3007a 100644 --- a/backend/app/scheduler/__init__.py +++ b/backend/app/scheduler/__init__.py @@ -1,3 +1 @@ -from .service import ScheduledTaskService - -__all__ = ["ScheduledTaskService"] +"""Clock-driven entry point of the schedule context (the poller).""" diff --git a/backend/app/scheduler/service.py b/backend/app/scheduler/service.py deleted file mode 100644 index 58df7e5a1..000000000 --- a/backend/app/scheduler/service.py +++ /dev/null @@ -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 diff --git a/backend/docs/GITHUB_AGENTS.md b/backend/docs/GITHUB_AGENTS.md index d46004cec..f32977048 100644 --- a/backend/docs/GITHUB_AGENTS.md +++ b/backend/docs/GITHUB_AGENTS.md @@ -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 diff --git a/backend/packages/harness/deerflow/config/reload_boundary.py b/backend/packages/harness/deerflow/config/reload_boundary.py index 6f6f87d62..52e4a04f6 100644 --- a/backend/packages/harness/deerflow/config/reload_boundary.py +++ b/backend/packages/harness/deerflow/config/reload_boundary.py @@ -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 " diff --git a/backend/packages/harness/deerflow/domain/schedule/model/task.py b/backend/packages/harness/deerflow/domain/schedule/model/task.py index e526b70b9..bf5b6e848 100644 --- a/backend/packages/harness/deerflow/domain/schedule/model/task.py +++ b/backend/packages/harness/deerflow/domain/schedule/model/task.py @@ -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 diff --git a/backend/packages/harness/deerflow/domain/schedule/service.py b/backend/packages/harness/deerflow/domain/schedule/service.py index 2cc0d7593..7f31d37be 100644 --- a/backend/packages/harness/deerflow/domain/schedule/service.py +++ b/backend/packages/harness/deerflow/domain/schedule/service.py @@ -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 diff --git a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/__init__.py b/backend/packages/harness/deerflow/persistence/scheduled_task_runs/__init__.py index 8c014cf90..6669b4054 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/__init__.py +++ b/backend/packages/harness/deerflow/persistence/scheduled_task_runs/__init__.py @@ -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"] diff --git a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py b/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py deleted file mode 100644 index afedb4b4f..000000000 --- a/backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py +++ /dev/null @@ -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) diff --git a/backend/packages/harness/deerflow/persistence/scheduled_tasks/__init__.py b/backend/packages/harness/deerflow/persistence/scheduled_tasks/__init__.py index 18b6ae7fe..8cd1e94f6 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_tasks/__init__.py +++ b/backend/packages/harness/deerflow/persistence/scheduled_tasks/__init__.py @@ -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"] diff --git a/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py b/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py deleted file mode 100644 index 7f05f12bd..000000000 --- a/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py +++ /dev/null @@ -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) diff --git a/backend/packages/harness/deerflow/scheduler/__init__.py b/backend/packages/harness/deerflow/scheduler/__init__.py deleted file mode 100644 index b2262f3e8..000000000 --- a/backend/packages/harness/deerflow/scheduler/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from .schedules import next_run_at, normalize_cron_expression, validate_timezone - -__all__ = ["next_run_at", "normalize_cron_expression", "validate_timezone"] diff --git a/backend/packages/harness/deerflow/scheduler/schedules.py b/backend/packages/harness/deerflow/scheduler/schedules.py deleted file mode 100644 index 8509c3523..000000000 --- a/backend/packages/harness/deerflow/scheduler/schedules.py +++ /dev/null @@ -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}") diff --git a/backend/tests/schedule_fakes.py b/backend/tests/schedule_fakes.py index 3fe575b1a..4e7fae5f3 100644 --- a/backend/tests/schedule_fakes.py +++ b/backend/tests/schedule_fakes.py @@ -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. """ diff --git a/backend/tests/test_schedule_dispatch_race.py b/backend/tests/test_schedule_dispatch_race.py new file mode 100644 index 000000000..d502b3978 --- /dev/null +++ b/backend/tests/test_schedule_dispatch_race.py @@ -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() diff --git a/backend/tests/test_schedule_fakes.py b/backend/tests/test_schedule_fakes.py index 68c4b1972..abed843ff 100644 --- a/backend/tests/test_schedule_fakes.py +++ b/backend/tests/test_schedule_fakes.py @@ -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. """ diff --git a/backend/tests/test_schedule_router.py b/backend/tests/test_schedule_router.py index 85315862b..2a02eb7ac 100644 --- a/backend/tests/test_schedule_router.py +++ b/backend/tests/test_schedule_router.py @@ -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 diff --git a/backend/tests/test_scheduled_task_claims.py b/backend/tests/test_scheduled_task_claims.py deleted file mode 100644 index 83db44628..000000000 --- a/backend/tests/test_scheduled_task_claims.py +++ /dev/null @@ -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() diff --git a/backend/tests/test_scheduled_task_dispatch_race.py b/backend/tests/test_scheduled_task_dispatch_race.py deleted file mode 100644 index 6dcc01dc0..000000000 --- a/backend/tests/test_scheduled_task_dispatch_race.py +++ /dev/null @@ -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() diff --git a/backend/tests/test_scheduled_task_repository.py b/backend/tests/test_scheduled_task_repository.py deleted file mode 100644 index a9101f440..000000000 --- a/backend/tests/test_scheduled_task_repository.py +++ /dev/null @@ -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() diff --git a/backend/tests/test_scheduled_task_router.py b/backend/tests/test_scheduled_task_router.py deleted file mode 100644 index 4c28161c1..000000000 --- a/backend/tests/test_scheduled_task_router.py +++ /dev/null @@ -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 diff --git a/backend/tests/test_scheduled_task_router_behavior.py b/backend/tests/test_scheduled_task_router_behavior.py deleted file mode 100644 index 70e516752..000000000 --- a/backend/tests/test_scheduled_task_router_behavior.py +++ /dev/null @@ -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 diff --git a/backend/tests/test_scheduled_task_schedules.py b/backend/tests/test_scheduled_task_schedules.py deleted file mode 100644 index fcaac8676..000000000 --- a/backend/tests/test_scheduled_task_schedules.py +++ /dev/null @@ -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) diff --git a/backend/tests/test_scheduled_task_service.py b/backend/tests/test_scheduled_task_service.py deleted file mode 100644 index 07ee0b8a9..000000000 --- a/backend/tests/test_scheduled_task_service.py +++ /dev/null @@ -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 diff --git a/backend/tests/test_scheduled_task_skip_writeback.py b/backend/tests/test_scheduled_task_skip_writeback.py deleted file mode 100644 index d3745700d..000000000 --- a/backend/tests/test_scheduled_task_skip_writeback.py +++ /dev/null @@ -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()