diff --git a/backend/app/gateway/routers/schedule/models.py b/backend/app/gateway/routers/schedule/models.py index 0d21312c1..e2cb833a7 100644 --- a/backend/app/gateway/routers/schedule/models.py +++ b/backend/app/gateway/routers/schedule/models.py @@ -29,6 +29,7 @@ from typing import Annotated, Any from pydantic import BaseModel, Field, PlainSerializer +from deerflow.domain.schedule.commands import UNSET, ContextChange, CreateScheduledTask, UpdateScheduledTask from deerflow.domain.schedule.model import ScheduledRun, ScheduledTask, ScheduleSpec, ScheduleType UtcTimestamp = Annotated[datetime, PlainSerializer(lambda value: value.isoformat(), return_type=str)] @@ -60,7 +61,13 @@ def _spec_to_wire(spec: ScheduleSpec) -> dict[str, str]: return {"run_at": spec.run_at.isoformat() if spec.run_at else ""} -class ScheduledTaskCreateRequest(BaseModel): +class CreateScheduledTaskRequest(BaseModel): + """Body of ``POST /scheduled-tasks``. + + No identity field: ``user_id`` is resolved server-side and injected + through ``to_command``, never read off the wire. + """ + thread_id: str | None = None context_mode: str = "fresh_thread_per_run" title: str = Field(min_length=1) @@ -69,6 +76,17 @@ class ScheduledTaskCreateRequest(BaseModel): schedule_spec: dict[str, Any] timezone: str + def to_command(self, user_id: str) -> CreateScheduledTask: + """Wire shape -> command (transformation ① of the chain).""" + return CreateScheduledTask( + user_id=user_id, + title=self.title, + prompt=self.prompt, + schedule=self.to_schedule(), + context_mode=self.context_mode, + thread_id=self.thread_id, + ) + def to_schedule(self) -> ScheduleSpec: """Parse the submitted triple into the value object. @@ -86,7 +104,16 @@ class ScheduledTaskCreateRequest(BaseModel): ) -class ScheduledTaskUpdateRequest(BaseModel): +class UpdateScheduledTaskRequest(BaseModel): + """Body of ``PATCH /scheduled-tasks/{task_id}``. + + On the wire, ``None`` means "not supplied" -- an explicit ``null`` has + always meant that on this endpoint, and unbinding a thread is expressed + by switching ``context_mode``, not by nulling ``thread_id``. The + ``None``-to-``UNSET`` translation in ``to_command`` is where that wire + convention meets the command's unambiguous three-state fields. + """ + context_mode: str | None = None thread_id: str | None = None title: str | None = Field(default=None, min_length=1) @@ -94,16 +121,44 @@ class ScheduledTaskUpdateRequest(BaseModel): schedule_spec: dict[str, Any] | None = None timezone: str | None = None + def changes_schedule(self) -> bool: + return self.schedule_spec is not None or self.timezone is not None + + def changes_context(self) -> bool: + return self.context_mode is not None or self.thread_id is not None + + def to_command(self, task_id: str, user_id: str, current: ScheduledTask | None) -> UpdateScheduledTask: + """Wire shape -> command (transformation ① of the chain). + + A schedule or context change needs the parts the client omitted, so + the router reads the current task once and passes it in -- this model + stays IO-free and only translates. ``current`` may be ``None`` when + neither composite field is being changed. + """ + schedule = UNSET + if current is not None and self.changes_schedule(): + schedule = self.to_schedule(current.schedule) + context = UNSET + if current is not None and self.changes_context(): + context = ContextChange( + context_mode=self.context_mode if self.context_mode is not None else str(current.context_mode), + thread_id=self.thread_id if self.thread_id is not None else current.thread_id, + ) + return UpdateScheduledTask( + task_id=task_id, + user_id=user_id, + title=self.title if self.title is not None else UNSET, + prompt=self.prompt if self.prompt is not None else UNSET, + schedule=schedule, + context=context, + ) + 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") diff --git a/backend/app/gateway/routers/schedule/router.py b/backend/app/gateway/routers/schedule/router.py index 9b0a05bfb..5443712c1 100644 --- a/backend/app/gateway/routers/schedule/router.py +++ b/backend/app/gateway/routers/schedule/router.py @@ -23,13 +23,14 @@ from fastapi import APIRouter, HTTPException, Query, Request from app.gateway.authz import require_permission from app.gateway.deps import ScheduleServiceDep, get_optional_user_from_request from app.gateway.routers.schedule.models import ( + CreateScheduledTaskRequest, DeleteResponse, ScheduledRunResponse, - ScheduledTaskCreateRequest, ScheduledTaskResponse, - ScheduledTaskUpdateRequest, TriggerResponse, + UpdateScheduledTaskRequest, ) +from deerflow.domain.schedule.commands import DeleteTask, PauseTask, ResumeTask, TriggerTask from deerflow.domain.schedule.exceptions import ( InvalidContextModeError, InvalidScheduleError, @@ -39,7 +40,6 @@ from deerflow.domain.schedule.exceptions import ( ThreadNotFoundError, ) from deerflow.domain.schedule.model import DispatchOutcome -from deerflow.domain.schedule.service import ContextChange router = APIRouter(prefix="/api", tags=["scheduled-tasks"]) @@ -97,17 +97,9 @@ async def list_scheduled_tasks(request: Request, service: ScheduleServiceDep): @router.post("/scheduled-tasks", response_model=ScheduledTaskResponse) @require_permission("threads", "write") @_map_domain_errors -async def create_scheduled_task(request: Request, body: ScheduledTaskCreateRequest, service: ScheduleServiceDep): +async def create_scheduled_task(request: Request, body: CreateScheduledTaskRequest, service: ScheduleServiceDep): user_id = await _require_user_id(request) - task = await service.create_task( - user_id=user_id, - title=body.title, - prompt=body.prompt, - schedule=body.to_schedule(), - context_mode=body.context_mode, - thread_id=body.thread_id, - now=datetime.now(UTC), - ) + task = await service.create_scheduled_task(body.to_command(user_id), now=datetime.now(UTC)) return ScheduledTaskResponse.from_domain(task) @@ -125,43 +117,16 @@ async def get_scheduled_task(task_id: str, request: Request, service: ScheduleSe async def update_scheduled_task( task_id: str, request: Request, - body: ScheduledTaskUpdateRequest, + body: UpdateScheduledTaskRequest, service: ScheduleServiceDep, ): user_id = await _require_user_id(request) - # `exclude_none` rather than `exclude_unset`, matching the pre-migration - # router: an explicit `null` has always meant "not supplied" on this - # endpoint, and unbinding a thread is expressed by switching - # `context_mode`, not by nulling `thread_id`. - supplied = body.model_dump(exclude_none=True) - - changes_schedule = "schedule_spec" in supplied or "timezone" in supplied - changes_context = "context_mode" in supplied or "thread_id" in supplied - - # Both a schedule and a context change need the parts the client omitted, - # so the current task is read once and supplies the defaults. That one - # extra fetch is what lets the service receive whole value objects instead - # 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 = 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: - context = ContextChange( - context_mode=supplied.get("context_mode", str(current.context_mode)), - thread_id=supplied.get("thread_id", current.thread_id), - ) - - task = await service.update_task( - task_id, - user_id=user_id, - now=datetime.now(UTC), - title=supplied.get("title"), - prompt=supplied.get("prompt"), - schedule=schedule, - context=context, - ) + # A schedule or context change needs the parts the client omitted, so the + # current task is read once and handed to `to_command`. That one extra + # fetch is what lets the command carry whole value objects instead of a + # patch of loose fields. + current = await service.get_task(task_id, user_id=user_id) if body.changes_schedule() or body.changes_context() else None + task = await service.update_scheduled_task(body.to_command(task_id, user_id, current), now=datetime.now(UTC)) return ScheduledTaskResponse.from_domain(task) @@ -170,7 +135,8 @@ async def update_scheduled_task( @_map_domain_errors async def pause_scheduled_task(task_id: str, request: Request, service: ScheduleServiceDep): user_id = await _require_user_id(request) - return ScheduledTaskResponse.from_domain(await service.pause_task(task_id, user_id=user_id)) + # No body, so no request model: the command is built right here (spec §2.1 ①). + return ScheduledTaskResponse.from_domain(await service.pause_task(PauseTask(task_id=task_id, user_id=user_id))) @router.post("/scheduled-tasks/{task_id}/resume", response_model=ScheduledTaskResponse) @@ -178,7 +144,7 @@ async def pause_scheduled_task(task_id: str, request: Request, service: Schedule @_map_domain_errors async def resume_scheduled_task(task_id: str, request: Request, service: ScheduleServiceDep): user_id = await _require_user_id(request) - return ScheduledTaskResponse.from_domain(await service.resume_task(task_id, user_id=user_id)) + return ScheduledTaskResponse.from_domain(await service.resume_task(ResumeTask(task_id=task_id, user_id=user_id))) @router.post("/scheduled-tasks/{task_id}/trigger", response_model=TriggerResponse) @@ -186,7 +152,7 @@ async def resume_scheduled_task(task_id: str, request: Request, service: Schedul @_map_domain_errors async def trigger_scheduled_task(task_id: str, request: Request, service: ScheduleServiceDep): user_id = await _require_user_id(request) - result = await service.trigger_task(task_id, user_id=user_id, now=datetime.now(UTC)) + result = await service.trigger_task(TriggerTask(task_id=task_id, user_id=user_id), now=datetime.now(UTC)) if result.outcome is DispatchOutcome.CONFLICT: raise HTTPException(status_code=409, detail=result.error or "Scheduled task trigger conflicted with an active run") if result.outcome is DispatchOutcome.FAILED: @@ -201,7 +167,7 @@ async def trigger_scheduled_task(task_id: str, request: Request, service: Schedu @_map_domain_errors async def delete_scheduled_task(task_id: str, request: Request, service: ScheduleServiceDep): user_id = await _require_user_id(request) - await service.delete_task(task_id, user_id=user_id) + await service.delete_task(DeleteTask(task_id=task_id, user_id=user_id)) return DeleteResponse(id=task_id, deleted=True) diff --git a/backend/packages/harness/deerflow/domain/schedule/__init__.py b/backend/packages/harness/deerflow/domain/schedule/__init__.py index fbeb8d7fc..c0bcd05fd 100644 --- a/backend/packages/harness/deerflow/domain/schedule/__init__.py +++ b/backend/packages/harness/deerflow/domain/schedule/__init__.py @@ -1,12 +1,21 @@ """Schedule bounded context: standing instructions to run a prompt on time. -Public API of the context. Import domain objects from here; import ports from -`deerflow.domain.schedule.ports` -- they are contracts consumed by adapters and -tests, not everyday call-site symbols. - -`ScheduleService` is not exported yet: the application service has not landed. +Public API of the context. Import domain objects, commands, errors, and the +service from here; import ports from `deerflow.domain.schedule.ports` -- they +are contracts consumed by adapters and tests, not everyday call-site symbols. """ +from deerflow.domain.schedule.commands import ( + UNSET, + ContextChange, + CreateScheduledTask, + DeleteTask, + PauseTask, + ResumeTask, + TriggerTask, + UnsetType, + UpdateScheduledTask, +) from deerflow.domain.schedule.exceptions import ( ActiveRunConflictError, InvalidContextModeError, @@ -33,20 +42,29 @@ from deerflow.domain.schedule.model import ( TaskStatus, TriggerKind, ) +from deerflow.domain.schedule.service import DispatchResult, ScheduleService __all__ = [ "ACTIVE_RUN_STATUSES", "TERMINAL_RUN_STATUSES", "TERMINAL_TASK_STATUSES", + "UNSET", "ActiveRunConflictError", + "ContextChange", "ContextMode", + "CreateScheduledTask", + "DeleteTask", "DispatchOutcome", + "DispatchResult", "InvalidContextModeError", "InvalidScheduleError", "LaunchFailedError", + "PauseTask", + "ResumeTask", "RunStatus", "ScheduleError", "SchedulePolicy", + "ScheduleService", "ScheduleSpec", "ScheduleType", "ScheduledRun", @@ -57,4 +75,7 @@ __all__ = [ "ThreadBusyError", "ThreadNotFoundError", "TriggerKind", + "TriggerTask", + "UnsetType", + "UpdateScheduledTask", ] diff --git a/backend/packages/harness/deerflow/domain/schedule/commands.py b/backend/packages/harness/deerflow/domain/schedule/commands.py new file mode 100644 index 000000000..75cf4f09b --- /dev/null +++ b/backend/packages/harness/deerflow/domain/schedule/commands.py @@ -0,0 +1,130 @@ +"""Commands of the schedule context. + +One frozen dataclass per HTTP-driven state-changing use case -- the named +carrier of "the information required to perform an operation on the domain" +(AWS hexagonal guidance). Commands are dumb data on purpose: business +validation stays on the aggregate (``ScheduledTask``'s factory and +transitions), and structural validation stays on the primary adapter's api +model, so error attribution (a malformed schedule reported before an unknown +thread) is owned by the handler's construction order. + +Two groups of use cases are deliberately NOT commands: + +- **Queries** (``list_tasks``, ``get_task``, ``list_task_runs``, ...) keep + plain parameters -- a command expresses an intent to change state, and + wrapping reads would be pure boilerplate. +- **Clock- and callback-driven writes** (``run_once``, ``dispatch_task``, + ``handle_run_completion``, ``reconcile_on_startup``). Those drivers have no + wire shape to translate (spec §5.3): the poller hands the service a claimed + aggregate and a clock reading, and the completion hook hands it an already + domain-typed ``RunOutcome``. A command would re-wrap domain vocabulary in + more domain vocabulary. + +``now`` is not a command field: it is the server's clock reading, passed +explicitly to the handler (``now=``) like every other rule input, not part of +the client's intent. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Final, final + +if TYPE_CHECKING: + from deerflow.domain.schedule.model import ContextMode, ScheduleSpec + + +@final +class UnsetType: + """The type of ``UNSET`` -- "the client did not supply this field". + + Partial updates need three states (absent, null, value). ``None`` cannot + carry two of them, so absence gets its own singleton; a field that is + ``UNSET`` is left untouched by the handler. + """ + + _instance: UnsetType | None = None + + def __new__(cls) -> UnsetType: + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + + def __repr__(self) -> str: + return "UNSET" + + def __bool__(self) -> bool: + return False + + +UNSET: Final[UnsetType] = UnsetType() + + +@dataclass(frozen=True) +class ContextChange: + """A requested change of execution context. + + The mode and the thread always move together -- ``with_context`` takes + both, and clearing the thread is what switching to a fresh-thread mode + means. Packaging them keeps ``None`` unambiguous everywhere else: the one + field for which ``None`` is a meaningful value travels inside this object. + """ + + context_mode: str | ContextMode + thread_id: str | None = None + + +@dataclass(frozen=True) +class CreateScheduledTask: + """Register a new standing instruction to run a prompt on time.""" + + user_id: str + title: str + prompt: str + schedule: ScheduleSpec + context_mode: str | ContextMode + thread_id: str | None = None + + +@dataclass(frozen=True) +class UpdateScheduledTask: + """Partially update a task; ``UNSET`` means "not supplied".""" + + task_id: str + user_id: str + title: str | UnsetType = UNSET + prompt: str | UnsetType = UNSET + schedule: ScheduleSpec | UnsetType = UNSET + context: ContextChange | UnsetType = UNSET + + +@dataclass(frozen=True) +class PauseTask: + """Stop claiming this task until it is resumed.""" + + task_id: str + user_id: str + + +@dataclass(frozen=True) +class ResumeTask: + """Re-admit this task to claiming.""" + + task_id: str + user_id: str + + +@dataclass(frozen=True) +class DeleteTask: + """Remove the task; its execution history rows go with it.""" + + task_id: str + user_id: str + + +@dataclass(frozen=True) +class TriggerTask: + """Dispatch a task on demand, even while it is paused.""" + + task_id: str + user_id: str diff --git a/backend/packages/harness/deerflow/domain/schedule/service.py b/backend/packages/harness/deerflow/domain/schedule/service.py index 846dc6070..2cc0d7593 100644 --- a/backend/packages/harness/deerflow/domain/schedule/service.py +++ b/backend/packages/harness/deerflow/domain/schedule/service.py @@ -17,6 +17,15 @@ from __future__ import annotations from dataclasses import dataclass, replace from datetime import datetime +from deerflow.domain.schedule.commands import ( + UNSET, + CreateScheduledTask, + DeleteTask, + PauseTask, + ResumeTask, + TriggerTask, + UpdateScheduledTask, +) from deerflow.domain.schedule.exceptions import ( ActiveRunConflictError, LaunchFailedError, @@ -31,7 +40,6 @@ from deerflow.domain.schedule.model import ( ScheduledRun, ScheduledTask, SchedulePolicy, - ScheduleSpec, ScheduleType, TriggerKind, ) @@ -50,21 +58,6 @@ _ACTIVE_RUN_CONFLICT_ERROR = "task already has an active run" _SKIP_ACTIVE_RUN_ERROR = "skipped: a previous run of this task is still active" -@dataclass(frozen=True) -class ContextChange: - """A requested change of execution context. - - The mode and the thread always move together -- `with_context` takes - both, and clearing the thread is what switching to a fresh-thread mode - means. Packaging them removes the one place where `None` was ambiguous - ("unbind the thread" versus "leave it alone") and lets every other - update field use plain `None` for "not supplied". - """ - - context_mode: str | ContextMode - thread_id: str | None = None - - @dataclass(frozen=True) class DispatchResult: """What one dispatch attempt produced. @@ -134,17 +127,7 @@ class ScheduleService: # ------------------------------------------------------------------ writes - async def create_task( - self, - *, - user_id: str, - title: str, - prompt: str, - schedule: ScheduleSpec, - context_mode: str | ContextMode, - thread_id: str | None, - now: datetime, - ) -> ScheduledTask: + async def create_scheduled_task(self, cmd: CreateScheduledTask, *, now: datetime) -> ScheduledTask: """Register a new standing instruction. The aggregate is built first so a malformed schedule or context mode is @@ -157,79 +140,68 @@ class ScheduleService: cannot use. """ task = ScheduledTask.create( - user_id=user_id, - title=title, - prompt=prompt, - schedule=schedule, - context_mode=context_mode, - thread_id=thread_id, + user_id=cmd.user_id, + title=cmd.title, + prompt=cmd.prompt, + schedule=cmd.schedule, + context_mode=cmd.context_mode, + thread_id=cmd.thread_id, now=now, policy=self._policy, ) await self._require_thread(task) return await self._tasks.add(task) - async def update_task( - self, - task_id: str, - *, - user_id: str, - now: datetime, - title: str | None = None, - prompt: str | None = None, - schedule: ScheduleSpec | None = None, - context: ContextChange | None = None, - ) -> ScheduledTask: - """Partially update a task; `None` means "not supplied". + async def update_scheduled_task(self, cmd: UpdateScheduledTask, *, now: datetime) -> ScheduledTask: + """Partially update a task; ``UNSET`` means "not supplied". - No sentinel is needed because nothing here has `None` as a meaningful - value -- the one field that does, `thread_id`, travels inside - `ContextChange` where it is unambiguous. + The one field for which ``None`` is a meaningful value, `thread_id`, + travels inside `ContextChange` where it is unambiguous. Context and schedule are applied through the aggregate's own transitions, so the re-arm rule and the running-task gate cannot be bypassed by patching fields directly. """ - task = await self.get_task(task_id, user_id=user_id) + task = await self.get_task(cmd.task_id, user_id=cmd.user_id) task.ensure_mutable() - if context is not None: - task = task.with_context(context.context_mode, context.thread_id) + if cmd.context is not UNSET: + task = task.with_context(cmd.context.context_mode, cmd.context.thread_id) await self._require_thread(task) - if schedule is not None: - task = task.with_schedule(schedule, now=now, policy=self._policy) - if title is not None: - task = replace(task, title=title) - if prompt is not None: - task = replace(task, prompt=prompt) + if cmd.schedule is not UNSET: + task = task.with_schedule(cmd.schedule, now=now, policy=self._policy) + if cmd.title is not UNSET: + task = replace(task, title=cmd.title) + if cmd.prompt is not UNSET: + task = replace(task, prompt=cmd.prompt) return await self._save(task) - async def pause_task(self, task_id: str, *, user_id: str) -> ScheduledTask: + async def pause_task(self, cmd: PauseTask) -> ScheduledTask: """Stop claiming this task until it is resumed. Refused while the task is being dispatched -- see `ensure_mutable`. """ - task = await self.get_task(task_id, user_id=user_id) + task = await self.get_task(cmd.task_id, user_id=cmd.user_id) return await self._save(task.paused()) - async def resume_task(self, task_id: str, *, user_id: str) -> ScheduledTask: + async def resume_task(self, cmd: ResumeTask) -> ScheduledTask: """Re-admit this task to claiming. Same gate as `pause_task`.""" - task = await self.get_task(task_id, user_id=user_id) + task = await self.get_task(cmd.task_id, user_id=cmd.user_id) return await self._save(task.resumed()) - async def delete_task(self, task_id: str, *, user_id: str) -> None: + async def delete_task(self, cmd: DeleteTask) -> None: """Deleting is deliberately not gated on the task being idle: the pre-migration router applied that gate to update/pause/resume only.""" - if not await self._tasks.delete(task_id, user_id=user_id): + if not await self._tasks.delete(cmd.task_id, user_id=cmd.user_id): raise TaskNotFoundError("Scheduled task not found") # ------------------------------------------------------------------ dispatch - async def trigger_task(self, task_id: str, *, user_id: str, now: datetime) -> DispatchResult: + async def trigger_task(self, cmd: TriggerTask, *, now: datetime) -> DispatchResult: """Dispatch a task on demand. Unlike the scheduled path this is allowed while the task is paused, and leaves it paused.""" - task = await self.get_task(task_id, user_id=user_id) + task = await self.get_task(cmd.task_id, user_id=cmd.user_id) return await self.dispatch_task(task, now=now, trigger=TriggerKind.MANUAL) async def run_once(self, *, now: datetime) -> list[DispatchResult]: diff --git a/backend/tests/test_schedule_router.py b/backend/tests/test_schedule_router.py index 9449b9daa..85315862b 100644 --- a/backend/tests/test_schedule_router.py +++ b/backend/tests/test_schedule_router.py @@ -30,8 +30,9 @@ from schedule_fakes import ( ) from app.gateway.routers.schedule import router as router_module +from deerflow.domain.schedule.commands import UNSET from deerflow.domain.schedule.exceptions import LaunchFailedError, ThreadBusyError -from deerflow.domain.schedule.model import RunStatus, ScheduledRun, SchedulePolicy, TaskStatus, TriggerKind +from deerflow.domain.schedule.model import RunStatus, ScheduledRun, ScheduledTask, SchedulePolicy, ScheduleSpec, TaskStatus, TriggerKind from deerflow.domain.schedule.service import ScheduleService USER = "user-1" @@ -99,11 +100,11 @@ def _create_body(**overrides): schedule_spec={"cron": "0 9 * * *"}, timezone="UTC", ) - return router_module.ScheduledTaskCreateRequest(**{**defaults, **overrides}) + return router_module.CreateScheduledTaskRequest(**{**defaults, **overrides}) def _update_body(**fields): - return router_module.ScheduledTaskUpdateRequest(**fields) + return router_module.UpdateScheduledTaskRequest(**fields) async def _create(service, **overrides): @@ -255,6 +256,50 @@ class TestRead: assert [task.title for task in listed] == ["Bound"] +class TestToCommand: + """Transformation ① of the chain: wire shape -> command, owned by the + request models. Identity is injected as a parameter, never read off the + wire; the wire's `None` = "not supplied" convention becomes the command's + unambiguous `UNSET` here and nowhere else.""" + + def test_identity_is_injected_not_read_off_the_wire(self): + assert "user_id" not in router_module.CreateScheduledTaskRequest.model_fields + assert "user_id" not in router_module.UpdateScheduledTaskRequest.model_fields + cmd = _create_body().to_command(USER) + assert cmd.user_id == USER + + def test_create_carries_the_parsed_value_object(self): + cmd = _create_body().to_command(USER) + assert cmd.schedule.cron == "0 9 * * *" + assert cmd.title == "Daily summary" + + def test_update_translates_wire_none_to_unset(self): + cmd = _update_body(title="Renamed").to_command("task-1", USER, None) + assert cmd.title == "Renamed" + assert cmd.prompt is UNSET + assert cmd.schedule is UNSET + assert cmd.context is UNSET + + def test_update_completes_composite_fields_from_the_current_task(self): + current = ScheduledTask.create( + user_id=USER, + title="t", + prompt="p", + schedule=ScheduleSpec.cron_schedule("0 9 * * *", "Asia/Shanghai"), + context_mode="fresh_thread_per_run", + thread_id=None, + now=datetime(2026, 7, 27, tzinfo=UTC), + policy=SchedulePolicy(), + ) + cmd = _update_body(schedule_spec={"cron": "30 2 * * *"}).to_command(current.task_id, USER, current) + assert cmd.schedule.cron == "30 2 * * *" + assert cmd.schedule.timezone == "Asia/Shanghai", "the omitted zone comes from the current value object" + + cmd = _update_body(thread_id=THREAD).to_command(current.task_id, USER, current) + assert str(cmd.context.context_mode) == "fresh_thread_per_run", "the omitted mode comes from the current task" + assert cmd.context.thread_id == THREAD + + class TestUpdate: @pytest.mark.asyncio async def test_a_title_is_updated_in_place(self, service, as_user): diff --git a/backend/tests/test_schedule_service.py b/backend/tests/test_schedule_service.py index 60241d15b..e6344a94f 100644 --- a/backend/tests/test_schedule_service.py +++ b/backend/tests/test_schedule_service.py @@ -23,6 +23,17 @@ from schedule_fakes import ( InMemoryScheduledTaskRepository, ) +from deerflow.domain.schedule.commands import ( + UNSET, + ContextChange, + CreateScheduledTask, + DeleteTask, + PauseTask, + ResumeTask, + TriggerTask, + UnsetType, + UpdateScheduledTask, +) from deerflow.domain.schedule.exceptions import ( InvalidScheduleError, LaunchFailedError, @@ -41,7 +52,7 @@ from deerflow.domain.schedule.model import ( TriggerKind, ) from deerflow.domain.schedule.ports import RunOutcome -from deerflow.domain.schedule.service import ContextChange, ScheduleService +from deerflow.domain.schedule.service import ScheduleService pytestmark = pytest.mark.asyncio @@ -120,13 +131,15 @@ def make_service( async def create_cron_task(service: ScheduleService, *, schedule: ScheduleSpec = CRON): - return await service.create_task( - user_id="user-1", - title="Daily summary", - prompt="summarize", - schedule=schedule, - context_mode=ContextMode.FRESH_THREAD_PER_RUN, - thread_id=None, + return await service.create_scheduled_task( + CreateScheduledTask( + user_id="user-1", + title="Daily summary", + prompt="summarize", + schedule=schedule, + context_mode=ContextMode.FRESH_THREAD_PER_RUN, + thread_id=None, + ), now=NOW, ) @@ -200,18 +213,18 @@ class TestFullLifecycle: assert await runs.count_active() == 0, "the active slot is free again" # -- pause / resume --------------------------------------------------- - paused = await service.pause_task(task.task_id, user_id="user-1") + paused = await service.pause_task(PauseTask(task_id=task.task_id, user_id="user-1")) assert paused.status is TaskStatus.PAUSED assert await service.run_once(now=overlap_at + timedelta(days=2)) == [], "a paused task is not claimed" - resumed = await service.resume_task(task.task_id, user_id="user-1") + resumed = await service.resume_task(ResumeTask(task_id=task.task_id, user_id="user-1")) assert resumed.status is TaskStatus.ENABLED # -- history and delete ---------------------------------------------- history = await service.list_task_runs(task.task_id, user_id="user-1") assert sorted(run.status for run in history) == [RunStatus.SKIPPED, RunStatus.SUCCESS] - await service.delete_task(task.task_id, user_id="user-1") + await service.delete_task(DeleteTask(task_id=task.task_id, user_id="user-1")) with pytest.raises(TaskNotFoundError): await service.get_task(task.task_id, user_id="user-1") @@ -244,7 +257,7 @@ class TestDispatchOutcomes: await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) before = len(runs.all_runs()) - result = await service.trigger_task(task.task_id, user_id="user-1", now=NOW) + result = await service.trigger_task(TriggerTask(task_id=task.task_id, user_id="user-1"), now=NOW) assert result.outcome is DispatchOutcome.CONFLICT assert result.record_id is None @@ -354,13 +367,15 @@ class TestConflictCollapse: # mints a new uuid per dispatch, which would make the two runs differ # for a reason that has nothing to do with the collapse. service = make_service(runs=run_repo, threads=FakeThreadLookup({"thread-1": "user-1"})) - task = await service.create_task( - user_id="user-1", - title="t", - prompt="p", - schedule=CRON, - context_mode=ContextMode.REUSE_THREAD, - thread_id="thread-1", + task = await service.create_scheduled_task( + CreateScheduledTask( + user_id="user-1", + title="t", + prompt="p", + schedule=CRON, + context_mode=ContextMode.REUSE_THREAD, + thread_id="thread-1", + ), now=NOW, ) await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) @@ -392,13 +407,15 @@ class TestOnceTaskDispatch: """Declaring it complete at launch would stick if the run failed or the process died before the hook could correct it.""" service = make_service() - task = await service.create_task( - user_id="user-1", - title="one shot", - prompt="go", - schedule=once_spec(), - context_mode=ContextMode.FRESH_THREAD_PER_RUN, - thread_id=None, + task = await service.create_scheduled_task( + CreateScheduledTask( + user_id="user-1", + title="one shot", + prompt="go", + schedule=once_spec(), + context_mode=ContextMode.FRESH_THREAD_PER_RUN, + thread_id=None, + ), now=NOW, ) @@ -412,13 +429,15 @@ class TestOnceTaskDispatch: that never happened.""" runs = InMemoryScheduledRunRepository() service = make_service(runs=runs) - task = await service.create_task( - user_id="user-1", - title="one shot", - prompt="go", - schedule=once_spec(), - context_mode=ContextMode.FRESH_THREAD_PER_RUN, - thread_id=None, + task = await service.create_scheduled_task( + CreateScheduledTask( + user_id="user-1", + title="one shot", + prompt="go", + schedule=once_spec(), + context_mode=ContextMode.FRESH_THREAD_PER_RUN, + thread_id=None, + ), now=NOW, ) await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) @@ -497,13 +516,15 @@ class TestRunCompletion: async def _launched_once_task(self): runs = InMemoryScheduledRunRepository() service = make_service(runs=runs) - task = await service.create_task( - user_id="user-1", - title="one shot", - prompt="go", - schedule=once_spec(), - context_mode=ContextMode.FRESH_THREAD_PER_RUN, - thread_id=None, + task = await service.create_scheduled_task( + CreateScheduledTask( + user_id="user-1", + title="one shot", + prompt="go", + schedule=once_spec(), + context_mode=ContextMode.FRESH_THREAD_PER_RUN, + thread_id=None, + ), now=NOW, ) await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) @@ -570,18 +591,20 @@ class TestRunCompletion: """ runs = InMemoryScheduledRunRepository() service = make_service(runs=runs) - task = await service.create_task( - user_id="user-1", - title="one shot", - prompt="go", - schedule=once_spec(), - context_mode=ContextMode.FRESH_THREAD_PER_RUN, - thread_id=None, + task = await service.create_scheduled_task( + CreateScheduledTask( + user_id="user-1", + title="one shot", + prompt="go", + schedule=once_spec(), + context_mode=ContextMode.FRESH_THREAD_PER_RUN, + thread_id=None, + ), now=NOW, ) await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) record = runs.all_runs()[0] - await service.delete_task(task.task_id, user_id="user-1") + await service.delete_task(DeleteTask(task_id=task.task_id, user_id="user-1")) await service.handle_run_completion( RunOutcome( @@ -642,13 +665,15 @@ class TestReconcileOnStartup: runs = InMemoryScheduledRunRepository() tasks = InMemoryScheduledTaskRepository() service = make_service(tasks=tasks, runs=runs) - task = await service.create_task( - user_id="user-1", - title="one shot", - prompt="go", - schedule=once_spec(), - context_mode=ContextMode.FRESH_THREAD_PER_RUN, - thread_id=None, + task = await service.create_scheduled_task( + CreateScheduledTask( + user_id="user-1", + title="one shot", + prompt="go", + schedule=once_spec(), + context_mode=ContextMode.FRESH_THREAD_PER_RUN, + thread_id=None, + ), now=NOW, ) await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) @@ -670,25 +695,29 @@ class TestTaskManagement: async def test_reuse_thread_requires_an_accessible_thread(self): service = make_service(threads=FakeThreadLookup({"thread-1": "user-1"})) - created = await service.create_task( - user_id="user-1", - title="t", - prompt="p", - schedule=CRON, - context_mode=ContextMode.REUSE_THREAD, - thread_id="thread-1", - now=NOW, - ) - assert created.thread_id == "thread-1" - - with pytest.raises(ThreadNotFoundError): - await service.create_task( - user_id="user-2", + created = await service.create_scheduled_task( + CreateScheduledTask( + user_id="user-1", title="t", prompt="p", schedule=CRON, context_mode=ContextMode.REUSE_THREAD, thread_id="thread-1", + ), + now=NOW, + ) + assert created.thread_id == "thread-1" + + with pytest.raises(ThreadNotFoundError): + await service.create_scheduled_task( + CreateScheduledTask( + user_id="user-2", + title="t", + prompt="p", + schedule=CRON, + context_mode=ContextMode.REUSE_THREAD, + thread_id="thread-1", + ), now=NOW, ) @@ -697,32 +726,60 @@ class TestTaskManagement: service = make_service(tasks=tasks) with pytest.raises(InvalidScheduleError): - await service.create_task( - user_id="user-1", - title="t", - prompt="p", - schedule=once_spec(after_seconds=10), # inside min_once_delay - context_mode=ContextMode.FRESH_THREAD_PER_RUN, - thread_id=None, + await service.create_scheduled_task( + CreateScheduledTask( + user_id="user-1", + title="t", + prompt="p", + schedule=once_spec(after_seconds=10), # inside min_once_delay + context_mode=ContextMode.FRESH_THREAD_PER_RUN, + thread_id=None, + ), now=NOW, ) assert await service.list_tasks("user-1") == [] + def test_commands_are_dumb_data(self): + # A command carries intent without validating it: business rules stay + # on the aggregate, so error attribution (a malformed schedule before + # an unknown thread) is owned by the handler's construction order, + # not by the command's own constructor. + cmd = CreateScheduledTask(user_id="u", title="", prompt="", schedule=CRON, context_mode="not-a-mode", thread_id=None) + assert cmd.context_mode == "not-a-mode" + + def test_unset_is_a_singleton_distinct_from_none(self): + # Three states, not two: an update field is UNSET (leave it alone), + # None can stay a meaningful value elsewhere, and UNSET is falsy so + # it cannot masquerade as a supplied value. + assert UnsetType() is UNSET + cmd = UpdateScheduledTask(task_id="t", user_id="u") + assert cmd.title is UNSET + assert cmd.title is not None + assert not UNSET + async def test_update_leaves_omitted_fields_alone(self): service = make_service() task = await create_cron_task(service) - updated = await service.update_task(task.task_id, user_id="user-1", now=NOW, title="renamed") + updated = await service.update_scheduled_task(UpdateScheduledTask(task_id=task.task_id, user_id="user-1", title="renamed"), now=NOW) assert updated.title == "renamed" assert updated.prompt == task.prompt assert updated.schedule == task.schedule + async def test_update_with_everything_unset_changes_nothing(self): + service = make_service() + task = await create_cron_task(service) + + updated = await service.update_scheduled_task(UpdateScheduledTask(task_id=task.task_id, user_id="user-1"), now=NOW) + + assert updated == task + async def test_update_can_change_the_prompt(self): service = make_service() task = await create_cron_task(service) - updated = await service.update_task(task.task_id, user_id="user-1", now=NOW, prompt="new instructions") + updated = await service.update_scheduled_task(UpdateScheduledTask(task_id=task.task_id, user_id="user-1", prompt="new instructions"), now=NOW) assert updated.prompt == "new instructions" assert updated.title == task.title @@ -741,9 +798,9 @@ class TestTaskManagement: task = await create_cron_task(service) with pytest.raises(TaskNotFoundError): - await service.update_task(task.task_id, user_id="user-1", now=NOW, title="x") + await service.update_scheduled_task(UpdateScheduledTask(task_id=task.task_id, user_id="user-1", title="x"), now=NOW) with pytest.raises(TaskNotFoundError): - await service.pause_task(task.task_id, user_id="user-1") + await service.pause_task(PauseTask(task_id=task.task_id, user_id="user-1")) async def test_context_is_changed_through_the_packaged_value(self): """context_mode and thread_id move together, so they are supplied @@ -752,20 +809,16 @@ class TestTaskManagement: service = make_service(threads=FakeThreadLookup({"thread-1": "user-1"})) task = await create_cron_task(service) - bound = await service.update_task( - task.task_id, - user_id="user-1", + bound = await service.update_scheduled_task( + UpdateScheduledTask(task_id=task.task_id, user_id="user-1", context=ContextChange(ContextMode.REUSE_THREAD, "thread-1")), now=NOW, - context=ContextChange(ContextMode.REUSE_THREAD, "thread-1"), ) assert bound.context_mode is ContextMode.REUSE_THREAD assert bound.thread_id == "thread-1" - unbound = await service.update_task( - task.task_id, - user_id="user-1", + unbound = await service.update_scheduled_task( + UpdateScheduledTask(task_id=task.task_id, user_id="user-1", context=ContextChange(ContextMode.FRESH_THREAD_PER_RUN)), now=NOW, - context=ContextChange(ContextMode.FRESH_THREAD_PER_RUN), ) assert unbound.thread_id is None, "switching to a fresh thread clears the binding" @@ -774,11 +827,9 @@ class TestTaskManagement: task = await create_cron_task(service) with pytest.raises(ThreadNotFoundError): - await service.update_task( - task.task_id, - user_id="user-1", + await service.update_scheduled_task( + UpdateScheduledTask(task_id=task.task_id, user_id="user-1", context=ContextChange(ContextMode.REUSE_THREAD, "thread-1")), now=NOW, - context=ContextChange(ContextMode.REUSE_THREAD, "thread-1"), ) async def test_rescheduling_a_terminal_task_re_arms_it(self): @@ -787,11 +838,9 @@ class TestTaskManagement: task = await create_cron_task(service) tasks.seed(replace(task, status=TaskStatus.FAILED)) - updated = await service.update_task( - task.task_id, - user_id="user-1", + updated = await service.update_scheduled_task( + UpdateScheduledTask(task_id=task.task_id, user_id="user-1", schedule=ScheduleSpec.cron_schedule("0 10 * * *", "UTC")), now=NOW, - schedule=ScheduleSpec.cron_schedule("0 10 * * *", "UTC"), ) assert updated.status is TaskStatus.ENABLED, "otherwise it would never be claimed again" @@ -803,9 +852,9 @@ class TestTaskManagement: tasks.seed(replace(task, status=TaskStatus.RUNNING)) with pytest.raises(TaskNotMutableError): - await service.update_task(task.task_id, user_id="user-1", now=NOW, title="nope") + await service.update_scheduled_task(UpdateScheduledTask(task_id=task.task_id, user_id="user-1", title="nope"), now=NOW) with pytest.raises(TaskNotMutableError): - await service.pause_task(task.task_id, user_id="user-1") + await service.pause_task(PauseTask(task_id=task.task_id, user_id="user-1")) async def test_a_running_task_can_still_be_deleted(self): """The pre-migration router gated update/pause/resume on this, but not @@ -815,37 +864,37 @@ class TestTaskManagement: task = await create_cron_task(service) tasks.seed(replace(task, status=TaskStatus.RUNNING)) - await service.delete_task(task.task_id, user_id="user-1") + await service.delete_task(DeleteTask(task_id=task.task_id, user_id="user-1")) @pytest.mark.parametrize("call", ["get", "update", "pause", "resume", "delete", "runs"]) async def test_another_users_task_is_reported_as_missing(self, call): service = make_service() task = await create_cron_task(service) - kwargs = {"user_id": "intruder"} - with pytest.raises(TaskNotFoundError): if call == "get": - await service.get_task(task.task_id, **kwargs) + await service.get_task(task.task_id, user_id="intruder") elif call == "update": - await service.update_task(task.task_id, now=NOW, title="x", **kwargs) + await service.update_scheduled_task(UpdateScheduledTask(task_id=task.task_id, user_id="intruder", title="x"), now=NOW) elif call == "pause": - await service.pause_task(task.task_id, **kwargs) + await service.pause_task(PauseTask(task_id=task.task_id, user_id="intruder")) elif call == "resume": - await service.resume_task(task.task_id, **kwargs) + await service.resume_task(ResumeTask(task_id=task.task_id, user_id="intruder")) elif call == "delete": - await service.delete_task(task.task_id, **kwargs) + await service.delete_task(DeleteTask(task_id=task.task_id, user_id="intruder")) else: - await service.list_task_runs(task.task_id, **kwargs) + await service.list_task_runs(task.task_id, user_id="intruder") async def test_tasks_are_listed_per_user_and_per_thread(self): service = make_service(threads=FakeThreadLookup({"thread-1": "user-1"})) - bound = await service.create_task( - user_id="user-1", - title="bound", - prompt="p", - schedule=CRON, - context_mode=ContextMode.REUSE_THREAD, - thread_id="thread-1", + bound = await service.create_scheduled_task( + CreateScheduledTask( + user_id="user-1", + title="bound", + prompt="p", + schedule=CRON, + context_mode=ContextMode.REUSE_THREAD, + thread_id="thread-1", + ), now=NOW, ) await create_cron_task(service)