mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-05 04:18:47 +00:00
9d0b09558e
First step of the scheduled-task context's hexagonal migration: the inner-ring model, with zero infrastructure dependencies. - ScheduleSpec / SchedulePolicy value objects - ScheduledTask aggregate root - ScheduledRun aggregate - 9 domain errors, 5 enums Rules are migrated verbatim from their current homes, each method's docstring citing the source line: timezone/cron/next-run calculation from deerflow/scheduler/schedules.py, context-mode and re-arm rules from routers/scheduled_tasks.py, and the four status-derivation rules from app/scheduler/service.py. Two things previously held by convention are now enforced by construction. Validation and normalization live in __post_init__, so building a ScheduleSpec field-by-field cannot bypass them. The skipped tombstone is a separate factory, so it can never be written as the transient queued row that would collide with uq_scheduled_task_run_active. The domain does not serialize itself: mapping the stored schedule_spec JSON in and out stays with the adapter layer, keeping Mapping[str, Any] out of every domain signature. Production code still runs through app/scheduler/service.py -- this commit adds no call sites and changes no behavior.
98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
|
|
from deerflow.domain.schedule.model.enums import RunStatus
|
|
|
|
ACTIVE_RUN_STATUSES: tuple[RunStatus, ...] = (RunStatus.QUEUED, RunStatus.RUNNING)
|
|
"""The statuses that occupy a task's single active-run slot.
|
|
|
|
Must stay in lockstep with the predicate of the partial unique index
|
|
``uq_scheduled_task_run_active`` (``status IN ('queued','running')``, declared
|
|
in ``persistence/scheduled_task_runs/model.py``). Drift here silently
|
|
decouples the overlap fast path from its atomic arbiter — the consistency
|
|
assertion lives in a separate test module rather than the domain tests, which
|
|
stay dependency-free.
|
|
"""
|
|
|
|
TERMINAL_RUN_STATUSES: frozenset[RunStatus] = frozenset(
|
|
{
|
|
RunStatus.SUCCESS,
|
|
RunStatus.FAILED,
|
|
RunStatus.SKIPPED,
|
|
RunStatus.INTERRUPTED,
|
|
}
|
|
)
|
|
"""Statuses a run row can no longer leave.
|
|
|
|
Used by the repository's ``protect_terminal`` compare-and-set: a fast-failing
|
|
run can reach the completion hook before the launch path's own write lands.
|
|
"""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ScheduledRun:
|
|
"""One execution record of a scheduled task — the history row.
|
|
|
|
A separate aggregate from ``ScheduledTask``: the two are written in
|
|
independent transactions and reference each other only by ``task_id``.
|
|
"""
|
|
|
|
record_id: str
|
|
task_id: str
|
|
thread_id: str
|
|
scheduled_for: datetime
|
|
trigger: str
|
|
status: RunStatus
|
|
run_id: str | None = None
|
|
error: str | None = None
|
|
started_at: datetime | None = None
|
|
finished_at: datetime | None = None
|
|
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
|
|
|
|
@classmethod
|
|
def queued(cls, *, task_id: str, thread_id: str, scheduled_for: datetime, trigger: str) -> ScheduledRun:
|
|
"""The active row of a normal dispatch.
|
|
|
|
Inserting this is what the unique index arbitrates: the loser of a
|
|
concurrent insert surfaces as ``ActiveRunConflictError``.
|
|
|
|
The ``task-run-{hex}`` id shape is depended on by existing rows and by
|
|
the run metadata that links a Gateway run back to this record — do not
|
|
change it.
|
|
"""
|
|
return cls(
|
|
record_id=f"task-run-{uuid.uuid4().hex}",
|
|
task_id=task_id,
|
|
thread_id=thread_id,
|
|
scheduled_for=scheduled_for,
|
|
trigger=trigger,
|
|
status=RunStatus.QUEUED,
|
|
)
|
|
|
|
@classmethod
|
|
def skipped_tombstone(cls, *, task_id: str, thread_id: str, scheduled_for: datetime, trigger: str) -> ScheduledRun:
|
|
"""A dropped occurrence, created directly as terminal ``SKIPPED``.
|
|
|
|
Deliberately a second factory rather than ``queued()`` followed by a
|
|
status change: ``QUEUED`` falls inside ``uq_scheduled_task_run_active``'s
|
|
predicate and would collide with the pre-existing run that still holds
|
|
the task's single active slot. ``SKIPPED`` is outside the predicate and
|
|
can never conflict (service.py:249-256).
|
|
"""
|
|
return cls(
|
|
record_id=f"task-run-{uuid.uuid4().hex}",
|
|
task_id=task_id,
|
|
thread_id=thread_id,
|
|
scheduled_for=scheduled_for,
|
|
trigger=trigger,
|
|
status=RunStatus.SKIPPED,
|
|
)
|
|
|
|
@property
|
|
def is_active(self) -> bool:
|
|
"""Whether this row occupies the task's single active-run slot."""
|
|
return self.status in ACTIVE_RUN_STATUSES
|