diff --git a/backend/app/adapters/schedule/scheduled_task_repository.py b/backend/app/adapters/schedule/scheduled_task_repository.py index 1d65e89ca..a7ff6a1fd 100644 --- a/backend/app/adapters/schedule/scheduled_task_repository.py +++ b/backend/app/adapters/schedule/scheduled_task_repository.py @@ -5,9 +5,10 @@ This context owns the `scheduled_tasks` table and writes its own queries, so SQL/ORM vocabulary stops at this file: methods exchange domain objects and normalize SQLite's tz-naive reads. -Sibling of `scheduled_run_repository.py`. The stored `schedule_spec` JSON -column is translated by `spec_column.py`, which belongs to this side of the -boundary only -- the HTTP shape has its own translation next to the router. +Sibling of `scheduled_run_repository.py`. Translating the row is this class's +own private business, `_to_domain` / `_apply` and the two `schedule_spec` +helpers beside them -- the HTTP shape has its own translation next to the +router, and neither side imports the other. **The queries are migrated unchanged from the legacy repository.** The claim statement's `FOR UPDATE SKIP LOCKED` and the `protect_terminal` conditional @@ -25,12 +26,13 @@ from datetime import UTC, datetime, timedelta from sqlalchemy import and_, or_, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker -from app.adapters.schedule.spec_column import column_to_spec, spec_to_column from deerflow.domain.schedule.model import ( TERMINAL_TASK_STATUSES, ContextMode, InvalidScheduleError, ScheduledTask, + ScheduleSpec, + ScheduleType, TaskStatus, ) from deerflow.domain.schedule.ports import ScheduledTaskRepository @@ -63,6 +65,36 @@ class SqlScheduledTaskRepository(ScheduledTaskRepository): # ------------------------------------------------------------ conversion + @staticmethod + def _spec_from_row(row: ScheduledTaskRow) -> ScheduleSpec: + """The stored triple -> the value object. + + All this side owns is which two keys this table's JSON uses; the rule + for turning them into a schedule belongs to the domain, which is why + the router's own mapping can state the same thing without either + importing the other. + """ + stored = row.schedule_spec or {} + return ScheduleSpec.from_primitives( + row.schedule_type, + cron=stored.get("cron"), + run_at=stored.get("run_at"), + timezone=row.timezone, + ) + + @staticmethod + def _spec_to_column(spec: ScheduleSpec) -> dict[str, str]: + """The value object -> the stored JSON. + + Emits the normalized value rather than echoing the caller's bytes: a + trailing-Z input is stored as "+00:00". Both forms parse back, so this + is deliberate -- preferable to carrying the raw dict on the value + object just to preserve the exact input spelling. + """ + if spec.schedule_type is ScheduleType.CRON: + return {"cron": spec.cron or ""} + return {"run_at": spec.run_at.isoformat() if spec.run_at else ""} + @staticmethod def _to_domain(row: ScheduledTaskRow) -> ScheduledTask: """ORM row -> aggregate. @@ -78,7 +110,7 @@ class SqlScheduledTaskRepository(ScheduledTaskRepository): user_id=row.user_id, title=row.title, prompt=row.prompt, - schedule=column_to_spec(row.schedule_type, row.schedule_spec, row.timezone), + schedule=SqlScheduledTaskRepository._spec_from_row(row), context_mode=ContextMode(row.context_mode), thread_id=row.thread_id, assistant_id=row.assistant_id, @@ -110,7 +142,7 @@ class SqlScheduledTaskRepository(ScheduledTaskRepository): row.title = task.title row.prompt = task.prompt row.schedule_type = str(task.schedule.schedule_type) - row.schedule_spec = spec_to_column(task.schedule) + row.schedule_spec = SqlScheduledTaskRepository._spec_to_column(task.schedule) row.timezone = task.schedule.timezone row.context_mode = str(task.context_mode) row.thread_id = task.thread_id diff --git a/backend/app/adapters/schedule/spec_column.py b/backend/app/adapters/schedule/spec_column.py deleted file mode 100644 index d0fc32c74..000000000 --- a/backend/app/adapters/schedule/spec_column.py +++ /dev/null @@ -1,78 +0,0 @@ -"""Boundary mapping (not a port implementation) -- the schedule_spec JSON column. - -Unlike its siblings in this package, this module implements no port: it is the -translation `scheduled_task_repository` needs between the stored column and the -value object, so the domain never grows a `Mapping[str, Any]` in its -signatures. - -Its counterpart on the other side of the application is -`app/gateway/routers/schedule/spec_wire.py`, which does the same job for the -HTTP request/response body. The two are near-identical today and are still kept -apart on purpose: a primary adapter must not import a secondary one, and the -two shapes are only equal by coincidence -- the day the API grows a field the -column does not have, they diverge without either side having to be untangled -first. The function names differ (`column_to_spec` here, `wire_to_spec` there) -so an import from the wrong side is visible rather than silently working. - -The duplication is bounded because the split inside each is deliberate: -**structural** checks (is the key present? is it a str?) belong to the -boundary, **value** rules (5-field cron, resolvable timezone, run_at present) -belong to `ScheduleSpec.__post_init__`. Only the structural half is repeated, -and `tests/test_schedule_spec_parity.py` feeds both the same malformed inputs -so a drift between them fails a test rather than reaching production. -""" - -from __future__ import annotations - -from collections.abc import Mapping -from datetime import datetime -from typing import Any - -from deerflow.domain.schedule.model import InvalidScheduleError, ScheduleSpec, ScheduleType - - -def column_to_spec(schedule_type: str, spec: Mapping[str, Any] | None, timezone: str) -> ScheduleSpec: - """Parse the stored/submitted triple into the value object. - - Raises: - InvalidScheduleError: unknown schedule type, or the type's required key - is missing or not a string. Raising a *domain* error from an - adapter is intentional -- domain errors are the vocabulary the - outer ring uses to say "this violates a domain rule", and the - router maps this one family uniformly. - """ - try: - kind = ScheduleType(schedule_type) - except ValueError as exc: - raise InvalidScheduleError(f"Unsupported schedule_type: {schedule_type}") from exc - - fields = spec or {} - if kind is ScheduleType.CRON: - raw_cron = fields.get("cron") - if not isinstance(raw_cron, str): - raise InvalidScheduleError("cron schedule requires schedule_spec.cron") - return ScheduleSpec.cron_schedule(raw_cron, timezone) - - raw_run_at = fields.get("run_at") - if not isinstance(raw_run_at, str): - raise InvalidScheduleError("once schedule requires run_at") - try: - run_at = datetime.fromisoformat(raw_run_at) - except ValueError as exc: - raise InvalidScheduleError(f"once schedule has an unparseable run_at: {raw_run_at!r}") from exc - return ScheduleSpec.once_at(run_at, timezone) - - -def spec_to_column(spec: ScheduleSpec) -> dict[str, str]: - """Rebuild the persisted/wire JSON shape. - - Note this normalizes the stored string rather than echoing the caller's - bytes: the frontend submits an already-UTC-aware ISO value - (`zonedLocalToUtcIso`), so a trailing-Z input round-trips out as "+00:00". - Both forms parse on either side, so the normalization is deliberate -- - preferable to carrying the raw dict on the value object just to preserve - the exact input spelling. - """ - if spec.schedule_type is ScheduleType.CRON: - return {"cron": spec.cron or ""} - return {"run_at": spec.run_at.isoformat() if spec.run_at else ""} diff --git a/backend/app/gateway/routers/schedule/models.py b/backend/app/gateway/routers/schedule/models.py index 51da2a13d..0d21312c1 100644 --- a/backend/app/gateway/routers/schedule/models.py +++ b/backend/app/gateway/routers/schedule/models.py @@ -29,8 +29,7 @@ from typing import Annotated, Any from pydantic import BaseModel, Field, PlainSerializer -from app.gateway.routers.schedule.spec_wire import spec_to_wire -from deerflow.domain.schedule.model import ScheduledRun, ScheduledTask +from deerflow.domain.schedule.model import ScheduledRun, ScheduledTask, ScheduleSpec, ScheduleType UtcTimestamp = Annotated[datetime, PlainSerializer(lambda value: value.isoformat(), return_type=str)] @@ -42,6 +41,25 @@ def _utc(value: datetime | None) -> datetime | None: return value.astimezone(UTC) if value.tzinfo is not None else value.replace(tzinfo=UTC) +def _spec_to_wire(spec: ScheduleSpec) -> dict[str, str]: + """The value object -> the `schedule_spec` body field. + + Emits the normalized value rather than echoing the caller's bytes: the + frontend submits an already-UTC-aware ISO value (`zonedLocalToUtcIso`), so + a trailing-Z input comes back as "+00:00". Both forms parse on either side, + so the normalization is deliberate. + + All this side owns is which two keys the body uses. Its counterpart is + `SqlScheduledTaskRepository._spec_to_column`; the two are near-identical + today by coincidence, and are kept apart because a primary adapter must not + import a secondary one and because the day the API grows a field the column + does not have, they diverge without either having to be untangled. + """ + if spec.schedule_type is ScheduleType.CRON: + return {"cron": spec.cron or ""} + return {"run_at": spec.run_at.isoformat() if spec.run_at else ""} + + class ScheduledTaskCreateRequest(BaseModel): thread_id: str | None = None context_mode: str = "fresh_thread_per_run" @@ -51,6 +69,22 @@ class ScheduledTaskCreateRequest(BaseModel): schedule_spec: dict[str, Any] timezone: str + def to_schedule(self) -> ScheduleSpec: + """Parse the submitted triple into the value object. + + Raises `InvalidScheduleError` -- a *domain* error out of a primary + adapter, on purpose: it is the vocabulary the outer ring uses to say + "this violates a domain rule", and the router maps that one family onto + 422. Structural problems (key missing, wrong type) and value problems + (5-field cron, resolvable timezone) both arrive as it. + """ + return ScheduleSpec.from_primitives( + self.schedule_type, + cron=self.schedule_spec.get("cron"), + run_at=self.schedule_spec.get("run_at"), + timezone=self.timezone, + ) + class ScheduledTaskUpdateRequest(BaseModel): context_mode: str | None = None @@ -60,6 +94,30 @@ class ScheduledTaskUpdateRequest(BaseModel): schedule_spec: dict[str, Any] | None = None timezone: str | None = None + def to_schedule(self, current: ScheduleSpec) -> ScheduleSpec: + """Build the replacement spec, taking what was omitted from `current`. + + The schedule *type* is not patchable; only its spec and its zone are. + Omitted parts are read straight off the current value object rather + than round-tripped through the wire shape and back. + + `None` means "not supplied" on this endpoint -- an explicit `null` has + always meant that here, and unbinding is expressed by switching + `context_mode`, not by nulling a field. + """ + if self.schedule_spec is not None: + cron = self.schedule_spec.get("cron") + run_at = self.schedule_spec.get("run_at") + else: + cron = current.cron + run_at = current.run_at.isoformat() if current.run_at else None + return ScheduleSpec.from_primitives( + str(current.schedule_type), + cron=cron, + run_at=run_at, + timezone=self.timezone if self.timezone is not None else current.timezone, + ) + class ScheduledTaskResponse(BaseModel): """One scheduled task as the client sees it. @@ -94,7 +152,7 @@ class ScheduledTaskResponse(BaseModel): title=task.title, prompt=task.prompt, schedule_type=str(task.schedule.schedule_type), - schedule_spec=spec_to_wire(task.schedule), + schedule_spec=_spec_to_wire(task.schedule), timezone=task.schedule.timezone, status=str(task.status), next_run_at=_utc(task.next_run_at), diff --git a/backend/app/gateway/routers/schedule/router.py b/backend/app/gateway/routers/schedule/router.py index badf99401..6476712df 100644 --- a/backend/app/gateway/routers/schedule/router.py +++ b/backend/app/gateway/routers/schedule/router.py @@ -30,7 +30,6 @@ from app.gateway.routers.schedule.models import ( ScheduledTaskUpdateRequest, TriggerResponse, ) -from app.gateway.routers.schedule.spec_wire import spec_to_wire, wire_to_spec from deerflow.domain.schedule.model import ( DispatchOutcome, InvalidContextModeError, @@ -104,7 +103,7 @@ async def create_scheduled_task(request: Request, body: ScheduledTaskCreateReque user_id=user_id, title=body.title, prompt=body.prompt, - schedule=wire_to_spec(body.schedule_type, body.schedule_spec, body.timezone), + schedule=body.to_schedule(), context_mode=body.context_mode, thread_id=body.thread_id, now=datetime.now(UTC), @@ -145,14 +144,7 @@ async def update_scheduled_task( # of a patch of loose fields. current = await service.get_task(task_id, user_id=user_id) if changes_schedule or changes_context else None - schedule = None - if current is not None and changes_schedule: - schedule = wire_to_spec( - # The schedule *type* is not patchable; only its spec and zone are. - str(current.schedule.schedule_type), - supplied.get("schedule_spec", spec_to_wire(current.schedule)), - supplied.get("timezone", current.schedule.timezone), - ) + schedule = body.to_schedule(current.schedule) if current is not None and changes_schedule else None context = None if current is not None and changes_context: diff --git a/backend/app/gateway/routers/schedule/spec_wire.py b/backend/app/gateway/routers/schedule/spec_wire.py deleted file mode 100644 index ca891b483..000000000 --- a/backend/app/gateway/routers/schedule/spec_wire.py +++ /dev/null @@ -1,68 +0,0 @@ -"""Boundary mapping -- the schedule_spec HTTP body field. - -The primary adapter's own translation between the request/response shape and -`ScheduleSpec`. Its counterpart is `app/adapters/schedule/spec_column.py`, which -does the same job for the stored JSON column; see that module for why the two -are kept apart rather than shared. - -The split is deliberate: **structural** checks (is the key present? is it a -str?) belong here, **value** rules (5-field cron, resolvable timezone, run_at -present) belong to `ScheduleSpec.__post_init__`. That is why this module is -thin -- most of what could go wrong is caught one layer in, and reported with -the same domain error, so the router maps one family onto 422. -""" - -from __future__ import annotations - -from collections.abc import Mapping -from datetime import datetime -from typing import Any - -from deerflow.domain.schedule.model import InvalidScheduleError, ScheduleSpec, ScheduleType - - -def wire_to_spec(schedule_type: str, spec: Mapping[str, Any] | None, timezone: str) -> ScheduleSpec: - """Parse the submitted triple into the value object. - - Raises: - InvalidScheduleError: unknown schedule type, or the type's required key - is missing or not a string. Raising a *domain* error from a primary - adapter is intentional -- domain errors are the vocabulary the outer - ring uses to say "this violates a domain rule", and the router maps - that one family uniformly onto 422. - """ - try: - kind = ScheduleType(schedule_type) - except ValueError as exc: - raise InvalidScheduleError(f"Unsupported schedule_type: {schedule_type}") from exc - - fields = spec or {} - if kind is ScheduleType.CRON: - raw_cron = fields.get("cron") - if not isinstance(raw_cron, str): - raise InvalidScheduleError("cron schedule requires schedule_spec.cron") - return ScheduleSpec.cron_schedule(raw_cron, timezone) - - raw_run_at = fields.get("run_at") - if not isinstance(raw_run_at, str): - raise InvalidScheduleError("once schedule requires run_at") - try: - run_at = datetime.fromisoformat(raw_run_at) - except ValueError as exc: - raise InvalidScheduleError(f"once schedule has an unparseable run_at: {raw_run_at!r}") from exc - return ScheduleSpec.once_at(run_at, timezone) - - -def spec_to_wire(spec: ScheduleSpec) -> dict[str, str]: - """Rebuild the response body shape. - - Note this emits the normalized value rather than echoing the caller's - bytes: the frontend submits an already-UTC-aware ISO value - (`zonedLocalToUtcIso`), so a trailing-Z input comes back as "+00:00". Both - forms parse on either side, so the normalization is deliberate -- - preferable to carrying the raw dict on the value object just to preserve - the exact input spelling. - """ - if spec.schedule_type is ScheduleType.CRON: - return {"cron": spec.cron or ""} - return {"run_at": spec.run_at.isoformat() if spec.run_at else ""} diff --git a/backend/docs/SCHEDULE_DESIGN_zh.md b/backend/docs/SCHEDULE_DESIGN_zh.md index 1a3918a7f..2541cdd60 100644 --- a/backend/docs/SCHEDULE_DESIGN_zh.md +++ b/backend/docs/SCHEDULE_DESIGN_zh.md @@ -55,11 +55,11 @@ flowchart LR SV --> M end subgraph PRIM["✅ 主适配器 · app"] - R["gateway/routers/schedule/
router · models · spec_wire"] + R["gateway/routers/schedule/
router · models"] PL["scheduler/poller.py
轮询时钟"] end subgraph SEC["✅ 从适配器 · app"] - AD["adapters/schedule/
两个仓储 · run_launcher · thread_lookup
spec_column · run_outcome_mapping"] + AD["adapters/schedule/
两个仓储 · run_launcher · thread_lookup
run_outcome_mapping"] end subgraph CR["✅ 组合根"] CO["composition.py
build_domain_services()"] @@ -239,7 +239,8 @@ America/New_York 的 "0 9 * * *" {"cron": "0 9 * * *"} ←──→ ScheduleSpec(CRON, "Asia/Shanghai", cron="0 9 * * *") （JSON 列 / HTTP 字段） （值对象） ↑ - app/adapters/schedule/spec_mapping.py + ScheduleSpec.from_primitives()（领域，一份） + + 每个适配器自己那几行「本格式用哪两个键」 ``` 这不是洁癖，是一条可执行的判据：**一旦领域方法的签名里出现 `Mapping[str, Any]`，就说明领域在处理持久化/传输格式了**。解析这件事天然可以切成两半——结构校验（键在不在？值是不是字符串？）属于边界，值校验（cron 是不是 5 段、时区认不认识、`run_at` 有没有）属于 `__post_init__`。切开之后领域完全不需要看见 dict，签名全部强类型。 @@ -522,7 +523,7 @@ await service.update_task( 1. `model/enums.py` 的 `ScheduleType` 加成员 2. `model/spec.py`：加承载参数的字段（如 `interval_seconds`）、在 `__post_init__` 加校验、在 `next_after` 加一个分支 -3. 适配器的 `spec_mapping`（见 §4.5）：进出两个方向各加一个分支 +3. `ScheduleSpec.from_primitives`（见 §4.5）加一个分支；两个适配器各自的 `_spec_to_column` / `_spec_to_wire` 也各加一个 4. **逐个检查 `task.py` 里四个 `status_after_*`**——它们目前都在问"是不是 ONCE"，新类型会落进 else 分支。确认那是你要的语义（大概率是：interval 与 cron 同属周期性） 5. 域测试：新类型在 §5.4 四张表里各补一行 @@ -627,8 +628,7 @@ await service.update_task( | 文件 | 内容 | |---|---| | [`gateway/routers/schedule/router.py`](../app/gateway/routers/schedule/router.py) | 10 个 HTTP 端点，只做协议转换 + 领域错误→状态码 | -| [`gateway/routers/schedule/models.py`](../app/gateway/routers/schedule/models.py) | 请求/响应模型；响应是白名单，不是 ORM 转储 | -| [`gateway/routers/schedule/spec_wire.py`](../app/gateway/routers/schedule/spec_wire.py) | HTTP body ↔ `ScheduleSpec` | +| [`gateway/routers/schedule/models.py`](../app/gateway/routers/schedule/models.py) | 请求/响应模型；响应是白名单，不是 ORM 转储；`schedule_spec` 的进出转换是模型自己的方法 | | [`app/scheduler/poller.py`](../app/scheduler/poller.py) | 轮询时钟 + 启动恢复 | **从适配器** @@ -639,7 +639,6 @@ await service.update_task( | [`adapters/schedule/scheduled_run_repository.py`](../app/adapters/schedule/scheduled_run_repository.py) | 自有持久化；`IntegrityError → ActiveRunConflictError` 的翻译点 | | [`adapters/schedule/run_launcher.py`](../app/adapters/schedule/run_launcher.py) | 防腐层；`ConflictError` / `HTTPException(409)` → `ThreadBusyError` | | [`adapters/schedule/thread_lookup.py`](../app/adapters/schedule/thread_lookup.py) | 防腐层；`check_access(require_existing=True)` | -| [`adapters/schedule/spec_column.py`](../app/adapters/schedule/spec_column.py) | JSON 列 ↔ `ScheduleSpec`（与 `spec_wire` 成对，见其 docstring） | | [`adapters/schedule/run_outcome_mapping.py`](../app/adapters/schedule/run_outcome_mapping.py) | `RunRecord → RunOutcome \| None`，承接旧完成钩子的内联过滤 | **组合根** diff --git a/backend/packages/harness/deerflow/domain/schedule/model/spec.py b/backend/packages/harness/deerflow/domain/schedule/model/spec.py index 84223af04..e3189abed 100644 --- a/backend/packages/harness/deerflow/domain/schedule/model/spec.py +++ b/backend/packages/harness/deerflow/domain/schedule/model/spec.py @@ -96,6 +96,50 @@ class ScheduleSpec: """Readability sugar — all validation lives in __post_init__.""" return cls(ScheduleType.ONCE, timezone, run_at=run_at) + @classmethod + def from_primitives(cls, schedule_type: str, *, cron: str | None, run_at: str | None, timezone: str) -> ScheduleSpec: + """Build from the loose strings both boundaries arrive as. + + The HTTP body and the stored JSON column both carry a schedule type + plus a mapping, so the rule for turning that into a value object lives + here instead of being written out once per adapter. Each adapter keeps + only what is genuinely its own — which keys its own format uses — and + the errors below stay one family the router maps uniformly. + + Four strings rather than a `Mapping[str, Any]` on purpose: a mapping in + this signature would mean the domain is handling a transport or storage + format, which is the thing this class exists to avoid. + + `cron` and `run_at` are annotated as the contract expects them but + checked rather than trusted — both callers read from data a client can + influence, so a non-string must be rejected here instead of reaching + `fromisoformat` or the cron parser. The field the given type does not + use is ignored: a boundary that carries both keys is not an error. + + Raises: + InvalidScheduleError: unknown schedule type, the type's field + missing or not a string, or an unparseable `run_at`. Value + rules — 5-field cron, resolvable timezone — are left to + __post_init__ and surface as the same error. + """ + try: + kind = ScheduleType(schedule_type) + except ValueError as exc: + raise InvalidScheduleError(f"Unsupported schedule_type: {schedule_type}") from exc + + if kind is ScheduleType.CRON: + if not isinstance(cron, str): + raise InvalidScheduleError("cron schedule requires schedule_spec.cron") + return cls.cron_schedule(cron, timezone) + + if not isinstance(run_at, str): + raise InvalidScheduleError("once schedule requires run_at") + try: + parsed = datetime.fromisoformat(run_at) + except ValueError as exc: + raise InvalidScheduleError(f"once schedule has an unparseable run_at: {run_at!r}") from exc + return cls.once_at(parsed, timezone) + def next_after(self, now: datetime) -> datetime | None: """Next fire time in UTC, or None when there is no future occurrence. diff --git a/backend/tests/test_schedule_domain.py b/backend/tests/test_schedule_domain.py index d7a1b74cf..2a86bf894 100644 --- a/backend/tests/test_schedule_domain.py +++ b/backend/tests/test_schedule_domain.py @@ -112,6 +112,75 @@ class TestScheduleSpecInvariants: assert spec.run_at == datetime(2026, 8, 1, 1, 0, tzinfo=UTC) +# ---------------------------------------------------------------- A2. from_primitives + + +class TestFromPrimitives: + """The one construction path that starts from untrusted strings. + + Both boundaries -- the HTTP body and the stored JSON column -- arrive as a + schedule type plus a loose mapping, so the rule for turning that into a + value object lives here rather than being written out once per adapter. + Each adapter is left with the part that is genuinely its own: which keys + its format uses. + + Values are type-checked rather than assumed. The signature says `str | + None` because that is the contract, but both callers read from data a + client can influence, so a non-string has to be rejected here rather than + reaching `datetime.fromisoformat` or the cron parser. + """ + + def test_builds_a_cron_schedule(self): + spec = ScheduleSpec.from_primitives("cron", cron="0 9 * * *", run_at=None, timezone="Asia/Shanghai") + assert spec.schedule_type is ScheduleType.CRON + assert spec.cron == "0 9 * * *" + assert spec.timezone == "Asia/Shanghai" + + def test_builds_a_once_schedule_from_an_iso_string(self): + spec = ScheduleSpec.from_primitives("once", cron=None, run_at="2026-08-01T09:00:00", timezone="Asia/Shanghai") + assert spec.schedule_type is ScheduleType.ONCE + assert spec.run_at == datetime(2026, 8, 1, 1, 0, tzinfo=UTC) + + def test_a_trailing_z_run_at_is_the_same_instant(self): + """The shape the frontend submits (`zonedLocalToUtcIso`).""" + spec = ScheduleSpec.from_primitives("once", cron=None, run_at="2026-08-01T01:00:00Z", timezone="Asia/Shanghai") + assert spec.run_at == datetime(2026, 8, 1, 1, 0, tzinfo=UTC) + + def test_unknown_schedule_type_is_rejected(self): + with pytest.raises(InvalidScheduleError, match="Unsupported schedule_type"): + ScheduleSpec.from_primitives("teleport", cron="0 9 * * *", run_at=None, timezone="UTC") + + @pytest.mark.parametrize("cron", [None, 5, ""]) + def test_cron_without_a_usable_expression_is_rejected(self, cron): + with pytest.raises(InvalidScheduleError, match="requires schedule_spec"): + ScheduleSpec.from_primitives("cron", cron=cron, run_at=None, timezone="UTC") + + @pytest.mark.parametrize("run_at", [None, 5]) + def test_once_without_a_string_run_at_is_rejected(self, run_at): + with pytest.raises(InvalidScheduleError, match="requires run_at"): + ScheduleSpec.from_primitives("once", cron=None, run_at=run_at, timezone="UTC") + + def test_an_unparseable_run_at_is_rejected(self): + with pytest.raises(InvalidScheduleError, match="unparseable run_at"): + ScheduleSpec.from_primitives("once", cron=None, run_at="next tuesday", timezone="UTC") + + def test_the_irrelevant_field_is_ignored(self): + """A boundary that carries both keys must not be rejected for it.""" + spec = ScheduleSpec.from_primitives("cron", cron="0 9 * * *", run_at="2026-08-01T09:00:00", timezone="UTC") + assert spec.run_at is None + + def test_value_rules_still_come_from_post_init(self): + """Not re-implemented here -- this is why each adapter stays thin.""" + with pytest.raises(InvalidScheduleError, match="exactly 5 fields"): + ScheduleSpec.from_primitives("cron", cron="0 9 * *", run_at=None, timezone="UTC") + with pytest.raises(InvalidScheduleError, match="Unknown timezone"): + ScheduleSpec.from_primitives("cron", cron="0 9 * * *", run_at=None, timezone="Mars/Olympus_Mons") + + def test_whitespace_in_a_cron_expression_is_normalized(self): + spec = ScheduleSpec.from_primitives("cron", cron=" 0 9 * * * ", run_at=None, timezone="UTC") + assert spec.cron == "0 9 * * *" + + # ---------------------------------------------------------------- B. next_after diff --git a/backend/tests/test_schedule_router.py b/backend/tests/test_schedule_router.py index 36611c86a..7120a7c94 100644 --- a/backend/tests/test_schedule_router.py +++ b/backend/tests/test_schedule_router.py @@ -302,6 +302,30 @@ class TestUpdate: assert updated.schedule_spec == created.schedule_spec assert updated.timezone == "Asia/Shanghai" + @pytest.mark.asyncio + async def test_a_timezone_change_on_a_once_task_keeps_the_same_instant(self, service, as_user): + """The `once` half of the fallback above. + + The omitted `run_at` is read straight off the current value object, and + the stored instant is already offset-aware, so re-zoning the schedule + must relabel it without moving it. + """ + created = await _create( + service, + schedule_type="once", + schedule_spec={"run_at": "2026-08-01T09:00:00+00:00"}, + timezone="UTC", + ) + updated = await _call( + router_module.update_scheduled_task, + task_id=created.id, + body=_update_body(timezone="Asia/Shanghai"), + service=service, + ) + assert updated.schedule_spec == created.schedule_spec + assert updated.timezone == "Asia/Shanghai" + assert updated.next_run_at == created.next_run_at + @pytest.mark.asyncio async def test_a_running_task_cannot_be_updated(self, service, tasks, as_user): """Red line: the mutability gate lives in the aggregate now, and the diff --git a/backend/tests/test_schedule_spec_parity.py b/backend/tests/test_schedule_spec_parity.py deleted file mode 100644 index bcc9b269f..000000000 --- a/backend/tests/test_schedule_spec_parity.py +++ /dev/null @@ -1,162 +0,0 @@ -"""Boundary tests for the two schedule_spec mappings, run against both. - -`schedule_spec` crosses two boundaries -- an HTTP body field and a JSON column --- and each side owns its own translation, because a primary adapter must not -import a secondary one. The shapes are equal today only by coincidence. - -Coincidence is exactly what needs a test. Every case here runs against both -implementations, so the day one side is changed without the other, this file -fails instead of production accepting a spec the repository cannot read back. -That is the same N-cases-x-2-implementations shape `test_schedule_fakes.py` -uses for the repository ports. - -The split each side enforces is the point: **structural** problems (key -missing, wrong type, unknown schedule type) are caught at the boundary, -**value** problems (5-field cron, resolvable timezone) are left to -`ScheduleSpec.__post_init__` -- and both surface as the same domain error, so -the router maps one family. -""" - -from __future__ import annotations - -from datetime import UTC, datetime - -import pytest - -from app.adapters.schedule.spec_column import column_to_spec, spec_to_column -from app.gateway.routers.schedule.spec_wire import spec_to_wire, wire_to_spec -from deerflow.domain.schedule.model import InvalidScheduleError, ScheduleSpec, ScheduleType - - -@pytest.fixture( - params=[ - pytest.param((column_to_spec, spec_to_column), id="column"), - pytest.param((wire_to_spec, spec_to_wire), id="wire"), - ] -) -def mapping(request): - """The (parse, emit) pair under test, once per boundary.""" - return request.param - - -@pytest.fixture -def parse(mapping): - return mapping[0] - - -@pytest.fixture -def emit(mapping): - return mapping[1] - - -class TestStructuralChecks: - def test_unknown_schedule_type_is_rejected(self, parse): - """The one rule the domain cannot state: ScheduleSpec only accepts the - enum, so a bad string has to be caught at the boundary.""" - with pytest.raises(InvalidScheduleError, match="Unsupported schedule_type"): - parse("teleport", {"cron": "0 9 * * *"}, "UTC") - - @pytest.mark.parametrize("spec", [{}, None, {"cron": 5}, {"run_at": "..."}]) - def test_cron_without_a_string_cron_is_rejected(self, parse, spec): - with pytest.raises(InvalidScheduleError, match="requires schedule_spec"): - parse("cron", spec, "UTC") - - @pytest.mark.parametrize("spec", [{}, None, {"run_at": 5}, {"cron": "0 9 * * *"}]) - def test_once_without_a_string_run_at_is_rejected(self, parse, spec): - with pytest.raises(InvalidScheduleError, match="requires run_at"): - parse("once", spec, "UTC") - - def test_an_unparseable_run_at_is_rejected(self, parse): - with pytest.raises(InvalidScheduleError, match="unparseable run_at"): - parse("once", {"run_at": "next tuesday"}, "UTC") - - -class TestValueChecksStayInTheDomain: - """These are not re-implemented on either side -- they arrive from - __post_init__, which is why each boundary can stay thin and why the - duplication between them is bounded.""" - - def test_a_bad_cron_expression_still_raises(self, parse): - with pytest.raises(InvalidScheduleError, match="exactly 5 fields"): - parse("cron", {"cron": "0 9 * *"}, "UTC") - - def test_a_bad_timezone_still_raises(self, parse): - with pytest.raises(InvalidScheduleError, match="Unknown timezone"): - parse("cron", {"cron": "0 9 * * *"}, "Mars/Olympus_Mons") - - -class TestRoundTrip: - def test_cron_round_trips_normalized(self, parse, emit): - spec = parse("cron", {"cron": " 0 9 * * * "}, "Asia/Shanghai") - assert spec.schedule_type is ScheduleType.CRON - assert emit(spec) == {"cron": "0 9 * * *"} - assert parse("cron", emit(spec), "Asia/Shanghai") == spec - - def test_once_round_trips_to_the_same_instant(self, parse, emit): - spec = parse("once", {"run_at": "2026-08-01T09:00:00"}, "Asia/Shanghai") - assert spec.run_at == datetime(2026, 8, 1, 1, 0, tzinfo=UTC) - assert parse("once", emit(spec), "Asia/Shanghai") == spec - - def test_a_trailing_z_input_parses_and_re_emits_as_an_offset(self, parse, emit): - """The shape the frontend actually submits (`zonedLocalToUtcIso`). - Both spellings parse on either side, so normalizing is deliberate.""" - spec = parse("once", {"run_at": "2026-08-01T01:00:00Z"}, "Asia/Shanghai") - emitted = emit(spec)["run_at"] - assert emitted.endswith("+00:00") - assert parse("once", {"run_at": emitted}, "Asia/Shanghai") == spec - - def test_emitted_output_is_idempotent(self, parse, emit): - spec = ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0, tzinfo=UTC), "UTC") - once = emit(spec) - assert emit(parse("once", once, "UTC")) == once - - -class TestCrossBoundaryParity: - """The two sides must agree, not merely each be self-consistent. - - The cases above run against both and would catch a behavioural drift; - these compare the outputs directly, which is what catches a *silent* one -- - a side that starts emitting a different but still-self-consistent shape. - """ - - @pytest.mark.parametrize( - ("schedule_type", "spec", "timezone"), - [ - ("cron", {"cron": "0 9 * * *"}, "UTC"), - ("cron", {"cron": " 30 2 * * 1 "}, "Asia/Shanghai"), - ("once", {"run_at": "2026-08-01T09:00:00"}, "Asia/Shanghai"), - ("once", {"run_at": "2026-08-01T01:00:00Z"}, "UTC"), - ], - ) - def test_both_sides_parse_to_the_same_value_object(self, schedule_type, spec, timezone): - assert column_to_spec(schedule_type, spec, timezone) == wire_to_spec(schedule_type, spec, timezone) - - @pytest.mark.parametrize( - "spec", - [ - ScheduleSpec.cron_schedule("0 9 * * *", "UTC"), - ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0, tzinfo=UTC), "Asia/Shanghai"), - ], - ) - def test_both_sides_emit_the_same_shape(self, spec): - assert spec_to_column(spec) == spec_to_wire(spec) - - @pytest.mark.parametrize( - ("schedule_type", "spec"), - [ - ("teleport", {"cron": "0 9 * * *"}), - ("cron", {}), - ("cron", {"cron": 5}), - ("once", {}), - ("once", {"run_at": 5}), - ("once", {"run_at": "next tuesday"}), - ], - ) - def test_both_sides_reject_the_same_inputs_with_the_same_message(self, schedule_type, spec): - """Same message, not just same type: the router turns this text into a - 422 detail, so a divergence here is user-visible.""" - with pytest.raises(InvalidScheduleError) as from_column: - column_to_spec(schedule_type, spec, "UTC") - with pytest.raises(InvalidScheduleError) as from_wire: - wire_to_spec(schedule_type, spec, "UTC") - assert str(from_column.value) == str(from_wire.value)