mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
c38d291505
Replaces the pre-hexagonal scheduled-task implementation with a slice built to the layering spec: a pure domain (two aggregates, two state machines, the policy value object), output ports it declares itself, SQL/launcher/thread adapters implementing them under `app/adapters/`, and a composition root that is the one place any of them is instantiated. The old implementation mixed all of that into `app/scheduler/service.py` and a router that reached straight into repositories, so the rules that matter -- overlap policy, lease handling, which write owns which timestamp -- were only reachable through a live database. They are now unit-assertable on in-memory fakes, with the contract suite running each port against both the fake and real sqlite, and the concurrency invariants pinned by dedicated race tests. Two bugs the old shape hid are fixed on the way: a completion hook that replayed a stale snapshot and rolled back the launch write, and a corrupt stored row surfacing to the client as a 4xx. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
59 lines
2.5 KiB
Python
59 lines
2.5 KiB
Python
import re
|
|
|
|
from sqlalchemy import Index
|
|
|
|
from deerflow.config.app_config import AppConfig
|
|
from deerflow.domain.schedule.model import ACTIVE_RUN_STATUSES
|
|
from deerflow.persistence.models import ScheduledTaskRow, ScheduledTaskRunRow
|
|
|
|
|
|
def test_app_config_exposes_scheduler_section():
|
|
config = AppConfig.model_validate(
|
|
{
|
|
"models": [],
|
|
"sandbox": {"use": "local"},
|
|
}
|
|
)
|
|
assert config.scheduler.enabled is False
|
|
assert config.scheduler.poll_interval_seconds == 5
|
|
assert config.scheduler.lease_seconds == 120
|
|
|
|
|
|
def test_scheduled_task_models_registered():
|
|
assert ScheduledTaskRow.__tablename__ == "scheduled_tasks"
|
|
assert ScheduledTaskRunRow.__tablename__ == "scheduled_task_runs"
|
|
|
|
|
|
def _active_run_index() -> Index:
|
|
return next(arg for arg in ScheduledTaskRunRow.__table_args__ if isinstance(arg, Index) and arg.name == "uq_scheduled_task_run_active")
|
|
|
|
|
|
def test_active_run_index_arbitrates_one_active_run_per_task():
|
|
"""The index is the atomic arbiter of the overlap rule, so its shape is a
|
|
contract, not a detail: unique, keyed on `task_id` alone."""
|
|
index = _active_run_index()
|
|
assert index.unique is True
|
|
assert [column.name for column in index.expressions] == ["task_id"]
|
|
|
|
|
|
def test_active_run_index_predicate_matches_the_domain_constant():
|
|
"""`ACTIVE_RUN_STATUSES` and this predicate must stay in lockstep.
|
|
|
|
The domain's fast path (`has_active`) and the index disagree the moment
|
|
they drift, which silently decouples the overlap check from its arbiter.
|
|
The domain tests cannot assert this -- they are deliberately
|
|
dependency-free and cannot import an ORM model -- so the assertion the
|
|
`ACTIVE_RUN_STATUSES` docstring promises lives here.
|
|
|
|
Both dialect predicates are checked: `create_all` renders the SQLite one
|
|
and production renders the Postgres one, so a drift in either is real.
|
|
"""
|
|
index = _active_run_index()
|
|
expected = {str(status) for status in ACTIVE_RUN_STATUSES}
|
|
assert expected == {"queued", "running"}, "domain constant changed -- update the ORM predicates below"
|
|
|
|
predicates = {key: str(value) for key, value in index.dialect_kwargs.items() if key.endswith("_where")}
|
|
assert set(predicates) == {"sqlite_where", "postgresql_where"}, "a dialect lost its partial-index predicate"
|
|
for dialect, predicate in predicates.items():
|
|
assert set(re.findall(r"'([^']+)'", predicate)) == expected, f"{dialect} predicate drifted from ACTIVE_RUN_STATUSES"
|