mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
refactor(schedule): turn the write use cases into commands
Add domain/schedule/commands.py with one frozen dataclass per HTTP-driven write use case -- CreateScheduledTask, UpdateScheduledTask, PauseTask, ResumeTask, DeleteTask, TriggerTask -- and make the service methods their handlers, keeping the naming chain aligned across all three spellings (command / handler / <Command>Request). The clock stays an explicit now= handler parameter: it is a rule input, not part of the client's intent. UpdateScheduledTask expresses partial updates with an UNSET sentinel, so absence is unambiguous; the wire keeps its historical None-means-omitted convention and the request model's to_command owns the translation. The former ContextChange moves from service.py into commands.py unchanged. The request models are renamed to <Command>Request and own transformation ① (to_command): identity is injected server-side and pinned by a test to never appear on the wire models. Clock- and callback-driven writes (run_once, dispatch_task, handle_run_completion, reconcile_on_startup) deliberately stay plain methods -- those drivers have no wire shape to translate, their inputs are already domain vocabulary. The context package now also exports the commands and the service, which completes its public API and retires the stale 'service not landed' note.
This commit is contained in:
rayhpeng
parent
f88c8e61bc
commit
fa0d709f29
@ -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")
|
||||
|
||||
@ -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)
|
||||
|
||||
|
||||
|
||||
@ -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",
|
||||
]
|
||||
|
||||
130
backend/packages/harness/deerflow/domain/schedule/commands.py
Normal file
130
backend/packages/harness/deerflow/domain/schedule/commands.py
Normal file
@ -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
|
||||
@ -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]:
|
||||
|
||||
@ -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):
|
||||
|
||||
@ -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)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user