diff --git a/backend/packages/harness/deerflow/domain/schedule/__init__.py b/backend/packages/harness/deerflow/domain/schedule/__init__.py index 8c89fcd89..2619d8ade 100644 --- a/backend/packages/harness/deerflow/domain/schedule/__init__.py +++ b/backend/packages/harness/deerflow/domain/schedule/__init__.py @@ -14,6 +14,7 @@ from deerflow.domain.schedule.model import ( TERMINAL_TASK_STATUSES, ActiveRunConflictError, ContextMode, + DispatchOutcome, InvalidContextModeError, InvalidScheduleError, LaunchFailedError, @@ -39,6 +40,7 @@ __all__ = [ "TERMINAL_TASK_STATUSES", "ActiveRunConflictError", "ContextMode", + "DispatchOutcome", "InvalidContextModeError", "InvalidScheduleError", "LaunchFailedError", diff --git a/backend/packages/harness/deerflow/domain/schedule/model/__init__.py b/backend/packages/harness/deerflow/domain/schedule/model/__init__.py index fbc9edbcc..cb8a65bf3 100644 --- a/backend/packages/harness/deerflow/domain/schedule/model/__init__.py +++ b/backend/packages/harness/deerflow/domain/schedule/model/__init__.py @@ -21,6 +21,7 @@ breaks it. Keep it that way. from deerflow.domain.schedule.model.enums import ( ContextMode, + DispatchOutcome, RunStatus, ScheduleType, TaskStatus, @@ -48,6 +49,7 @@ __all__ = [ "TERMINAL_TASK_STATUSES", "ActiveRunConflictError", "ContextMode", + "DispatchOutcome", "InvalidContextModeError", "InvalidScheduleError", "LaunchFailedError", diff --git a/backend/packages/harness/deerflow/domain/schedule/model/enums.py b/backend/packages/harness/deerflow/domain/schedule/model/enums.py index f112fafed..0a9a81442 100644 --- a/backend/packages/harness/deerflow/domain/schedule/model/enums.py +++ b/backend/packages/harness/deerflow/domain/schedule/model/enums.py @@ -31,6 +31,21 @@ class RunStatus(StrEnum): SKIPPED = "skipped" +class DispatchOutcome(StrEnum): + """What one dispatch attempt produced. + + A domain vocabulary rather than a set of strings: the caller branches on + all four, and the distinction between SKIPPED and CONFLICT is itself a + business rule -- a dropped scheduled occurrence is accounted for, while a + rejected on-demand trigger is reported and leaves no trace. + """ + + LAUNCHED = "launched" + SKIPPED = "skipped" + CONFLICT = "conflict" + FAILED = "failed" + + class TriggerKind(StrEnum): """What caused a dispatch. diff --git a/backend/packages/harness/deerflow/domain/schedule/ports.py b/backend/packages/harness/deerflow/domain/schedule/ports.py index f491f8f35..ae18d2f68 100644 --- a/backend/packages/harness/deerflow/domain/schedule/ports.py +++ b/backend/packages/harness/deerflow/domain/schedule/ports.py @@ -104,7 +104,7 @@ class ScheduledTaskRepository(Protocol): """Remove the task. False when it was absent or owned by someone else.""" ... - async def claim_due(self, *, now: datetime, lease_owner: str, lease_seconds: int, limit: int) -> list[ScheduledTask]: + async def claim_due(self, *, now: datetime, lease_seconds: int, limit: int) -> list[ScheduledTask]: """Atomically take ownership of up to `limit` tasks that are due. A task is due when its next fire time has passed AND either it is @@ -116,9 +116,10 @@ class ScheduledTaskRepository(Protocol): concurrent claimer cannot take the same ones. Returns them in the state they were left in *after* the claim. - `lease_owner` is recorded for diagnostics only and is never read back: - expiry alone decides whether a claim can be taken over. Do not build - ownership enforcement on it without also adding the read side. + Which process is claiming is deliberately absent: it is an identity, + not a rule. The implementation may record one for diagnostics, but + nothing reads it back -- expiry alone decides whether a claim can be + taken over -- so the domain has no reason to carry it. Atomicity is the implementation's responsibility. An in-memory double can satisfy every rule above under single-threaded use while providing diff --git a/backend/packages/harness/deerflow/domain/schedule/service.py b/backend/packages/harness/deerflow/domain/schedule/service.py index 52fb3f98d..5c5b74b9b 100644 --- a/backend/packages/harness/deerflow/domain/schedule/service.py +++ b/backend/packages/harness/deerflow/domain/schedule/service.py @@ -14,14 +14,13 @@ tests that must keep passing unchanged. from __future__ import annotations -from collections.abc import Callable from dataclasses import dataclass, replace from datetime import datetime -from typing import Any, Literal from deerflow.domain.schedule.model import ( ActiveRunConflictError, ContextMode, + DispatchOutcome, LaunchFailedError, RunStatus, ScheduledRun, @@ -48,8 +47,20 @@ from deerflow.domain.schedule.ports import ( _ACTIVE_RUN_CONFLICT_ERROR = "task already has an active run" _SKIP_ACTIVE_RUN_ERROR = "skipped: a previous run of this task is still active" -_UNSET: Any = object() -"""Distinguishes "field omitted" from "field set to None" in update_task.""" + +@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) @@ -57,11 +68,11 @@ class DispatchResult: """What one dispatch attempt produced. `outcome` drives the caller's protocol mapping -- a manual trigger turns - `conflict` into 409 and `failed` into 502 -- so the four values are part of - the contract, not an implementation detail. + CONFLICT into 409 and FAILED into 502 -- so it is domain vocabulary + rather than a bare string. """ - outcome: Literal["launched", "skipped", "conflict", "failed"] + outcome: DispatchOutcome record_id: str | None run_id: str | None thread_id: str @@ -77,16 +88,12 @@ class ScheduleService: launcher: RunLauncher, threads: ThreadLookup, policy: SchedulePolicy, - lease_owner: str, ) -> None: self._tasks = tasks self._runs = runs self._launcher = launcher self._threads = threads self._policy = policy - # Identity of the claiming process. Recorded for diagnostics only -- - # the repository never reads it back. - self._lease_owner = lease_owner # ------------------------------------------------------------------ reads @@ -152,13 +159,16 @@ class ScheduleService: *, user_id: str, now: datetime, - title: str = _UNSET, - prompt: str = _UNSET, - schedule: ScheduleSpec = _UNSET, - context_mode: str | ContextMode = _UNSET, - thread_id: str | None = _UNSET, + title: str | None = None, + prompt: str | None = None, + schedule: ScheduleSpec | None = None, + context: ContextChange | None = None, ) -> ScheduledTask: - """Partially update a task; omitted fields are left alone. + """Partially update a task; `None` 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. Context and schedule are applied through the aggregate's own transitions, so the re-arm rule and the running-task gate cannot be @@ -167,30 +177,25 @@ class ScheduleService: task = await self.get_task(task_id, user_id=user_id) task.ensure_mutable() - if context_mode is not _UNSET or thread_id is not _UNSET: - effective_mode = task.context_mode if context_mode is _UNSET else context_mode - effective_thread = task.thread_id if thread_id is _UNSET else thread_id - task = task.with_context(effective_mode, effective_thread) + if context is not None: + task = task.with_context(context.context_mode, context.thread_id) await self._require_thread(task) - - if schedule is not _UNSET: + if schedule is not None: task = task.with_schedule(schedule, now=now, policy=self._policy) - - if title is not _UNSET: + if title is not None: task = replace(task, title=title) - if prompt is not _UNSET: + if prompt is not None: task = replace(task, prompt=prompt) - saved = await self._tasks.save(task) - if saved is None: - raise TaskNotFoundError("Scheduled task not found") - return saved + return await self._save(task) async def pause_task(self, task_id: str, *, user_id: str) -> ScheduledTask: - return await self._transition(task_id, user_id=user_id, transition=ScheduledTask.paused) + task = await self.get_task(task_id, user_id=user_id) + return await self._save(task.paused()) async def resume_task(self, task_id: str, *, user_id: str) -> ScheduledTask: - return await self._transition(task_id, user_id=user_id, transition=ScheduledTask.resumed) + task = await self.get_task(task_id, user_id=user_id) + return await self._save(task.resumed()) async def delete_task(self, task_id: str, *, user_id: str) -> None: """Deleting is deliberately not gated on the task being idle: the @@ -217,12 +222,7 @@ class ScheduleService: budget = self._policy.max_concurrent_runs - active if budget <= 0: return [] - claimed = await self._tasks.claim_due( - now=now, - lease_owner=self._lease_owner, - lease_seconds=self._policy.lease_seconds, - limit=budget, - ) + claimed = await self._tasks.claim_due(now=now, lease_seconds=self._policy.lease_seconds, limit=budget) return [await self.dispatch_task(task, now=now, trigger=TriggerKind.SCHEDULED) for task in claimed] async def dispatch_task(self, task: ScheduledTask, *, now: datetime, trigger: TriggerKind) -> DispatchResult: @@ -289,9 +289,9 @@ class ScheduleService: # reported as a conflict the caller has to deal with. if trigger is TriggerKind.SCHEDULED and task.skips_on_overlap: return await self._finalize_skip(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, error=str(exc)) - return await self._fail(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, trigger=trigger, error=str(exc), outcome="conflict") + return await self._fail(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, trigger=trigger, error=str(exc), outcome=DispatchOutcome.CONFLICT) except LaunchFailedError as exc: - return await self._fail(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, trigger=trigger, error=str(exc), outcome="failed") + return await self._fail(task, record_id=record.record_id, thread_id=execution_thread_id, now=now, trigger=trigger, error=str(exc), outcome=DispatchOutcome.FAILED) await self._runs.update_status( record.record_id, @@ -314,7 +314,7 @@ class ScheduleService: # Same race as the record write above. protect_terminal=True, ) - return DispatchResult("launched", record.record_id, launched.run_id, launched.thread_id, None) + return DispatchResult(DispatchOutcome.LAUNCHED, record.record_id, launched.run_id, launched.thread_id, None) # ------------------------------------------------------------------ lifecycle @@ -368,9 +368,16 @@ class ScheduleService: if not task.thread_id or not await self._threads.exists_for_user(task.thread_id, task.user_id): raise ThreadNotFoundError("Thread not found") - async def _transition(self, task_id: str, *, user_id: str, transition: Callable[[ScheduledTask], ScheduledTask]) -> ScheduledTask: - task = await self.get_task(task_id, user_id=user_id) - saved = await self._tasks.save(transition(task)) + async def _save(self, task: ScheduledTask) -> ScheduledTask: + """Persist an already-validated aggregate. + + Read-modify-write rather than a conditional UPDATE: pushing the + "not while running" rule into a storage predicate would put it beyond + the reach of a zero-IO test and give it a second home. The pre-existing + code had the same shape, and closing the window properly needs + optimistic locking, which needs a schema change. + """ + saved = await self._tasks.save(task) if saved is None: raise TaskNotFoundError("Scheduled task not found") return saved @@ -381,7 +388,7 @@ class ScheduleService: No history record is written: nothing was scheduled to happen, so there is no occurrence to account for. """ - return DispatchResult("conflict", None, None, thread_id, _ACTIVE_RUN_CONFLICT_ERROR) + return DispatchResult(DispatchOutcome.CONFLICT, None, None, thread_id, _ACTIVE_RUN_CONFLICT_ERROR) async def _record_scheduled_skip(self, task: ScheduledTask, *, thread_id: str, now: datetime, trigger: TriggerKind) -> DispatchResult: """Account for a scheduled occurrence dropped because of an overlap. @@ -423,7 +430,7 @@ class ScheduleService: last_error=error if task.schedule.schedule_type is ScheduleType.ONCE else None, increment_run_count=False, ) - return DispatchResult("skipped", record_id, None, thread_id, error) + return DispatchResult(DispatchOutcome.SKIPPED, record_id, None, thread_id, error) async def _fail( self, @@ -434,7 +441,7 @@ class ScheduleService: now: datetime, trigger: TriggerKind, error: str, - outcome: Literal["conflict", "failed"], + outcome: DispatchOutcome, ) -> DispatchResult: await self._runs.update_status( record_id, diff --git a/backend/tests/schedule_fakes.py b/backend/tests/schedule_fakes.py index 268f7b29e..fb2f91542 100644 --- a/backend/tests/schedule_fakes.py +++ b/backend/tests/schedule_fakes.py @@ -97,7 +97,7 @@ class InMemoryScheduledTaskRepository: del self._rows[task_id] return True - async def claim_due(self, *, now: datetime, lease_owner: str, lease_seconds: int, limit: int) -> list[ScheduledTask]: + async def claim_due(self, *, now: datetime, lease_seconds: int, limit: int) -> list[ScheduledTask]: def claimable(row: _TaskRow) -> bool: task = row.task if task.next_run_at is None or task.next_run_at > now: @@ -115,7 +115,8 @@ class InMemoryScheduledTaskRepository: claimed = [] for row in due: - row.lease_owner = lease_owner + # Stands in for whatever identity a real adapter records. + row.lease_owner = "fake-worker" row.lease_expires_at = now + timedelta(seconds=lease_seconds) row.task = replace(row.task, status=TaskStatus.RUNNING) claimed.append(row.task) diff --git a/backend/tests/test_schedule_domain.py b/backend/tests/test_schedule_domain.py index 96c808d1e..c4e11a1ba 100644 --- a/backend/tests/test_schedule_domain.py +++ b/backend/tests/test_schedule_domain.py @@ -74,7 +74,7 @@ class TestScheduleSpecInvariants: ScheduleSpec.cron_schedule("0 9 * * *", "Mars/Olympus_Mons") def test_cron_without_expression_is_rejected(self): - with pytest.raises(InvalidScheduleError, match="requires schedule_spec.cron"): + with pytest.raises(InvalidScheduleError, match="requires schedule_spec"): ScheduleSpec(ScheduleType.CRON, "UTC") @pytest.mark.parametrize("expr", ["0 9 * *", "0 9 * * * *", "0"]) @@ -96,7 +96,7 @@ class TestScheduleSpecInvariants: assert ScheduleSpec(ScheduleType.CRON, "UTC", cron=" 0 9 * * * ").cron == "0 9 * * *" def test_naive_run_at_is_localized_to_the_schedule_timezone(self): - spec = ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0), "Asia/Shanghai") + spec = ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0), "Asia/Shanghai") # noqa: DTZ001 -- naive input is the subject assert spec.run_at.utcoffset() == timedelta(hours=8) assert spec.run_at.astimezone(UTC) == datetime(2026, 8, 1, 1, 0, tzinfo=UTC) @@ -196,7 +196,7 @@ class TestScheduleSpecEquality: assert ScheduleSpec.cron_schedule(" 0 9 * * * ", "UTC") == ScheduleSpec.cron_schedule("0 9 * * *", "UTC") def test_equivalent_once_inputs_converge(self): - naive = ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0), "Asia/Shanghai") + naive = ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0), "Asia/Shanghai") # noqa: DTZ001 -- naive input is the subject aware = ScheduleSpec.once_at(datetime.fromisoformat("2026-08-01T01:00:00Z"), "Asia/Shanghai") assert naive == aware diff --git a/backend/tests/test_schedule_fakes.py b/backend/tests/test_schedule_fakes.py index 72a14b2c8..21c4d1d17 100644 --- a/backend/tests/test_schedule_fakes.py +++ b/backend/tests/test_schedule_fakes.py @@ -114,12 +114,12 @@ class TestClaimDue: repo = InMemoryScheduledTaskRepository() await repo.add(make_task(next_run_at=NOW - timedelta(minutes=1))) - claimed = await repo.claim_due(now=NOW, lease_owner="worker-1", lease_seconds=120, limit=10) + claimed = await repo.claim_due(now=NOW, lease_seconds=120, limit=10) assert [task.task_id for task in claimed] == ["task-1"] assert claimed[0].status is TaskStatus.RUNNING owner, expires = repo.lease_of("task-1") - assert owner == "worker-1" + assert owner is not None, "an implementation may record who claimed; nothing reads it back" assert expires == NOW + timedelta(seconds=120) @pytest.mark.parametrize( @@ -134,7 +134,7 @@ class TestClaimDue: async def test_does_not_claim(self, label, task_kwargs): repo = InMemoryScheduledTaskRepository() await repo.add(make_task(**task_kwargs)) - assert await repo.claim_due(now=NOW, lease_owner="w", lease_seconds=120, limit=10) == [], label + assert await repo.claim_due(now=NOW, lease_seconds=120, limit=10) == [], label async def test_does_not_steal_a_live_claim(self): repo = InMemoryScheduledTaskRepository() @@ -143,7 +143,7 @@ class TestClaimDue: lease_owner="worker-1", lease_expires_at=NOW + timedelta(seconds=60), ) - assert await repo.claim_due(now=NOW, lease_owner="worker-2", lease_seconds=120, limit=10) == [] + assert await repo.claim_due(now=NOW, lease_seconds=120, limit=10) == [] async def test_reclaims_a_task_stuck_mid_dispatch(self): """The claimer died between claiming and launching: status is running, @@ -155,17 +155,17 @@ class TestClaimDue: lease_expires_at=NOW - timedelta(seconds=1), ) - claimed = await repo.claim_due(now=NOW, lease_owner="worker-2", lease_seconds=120, limit=10) + claimed = await repo.claim_due(now=NOW, lease_seconds=120, limit=10) assert [task.task_id for task in claimed] == ["task-1"] - assert repo.lease_of("task-1")[0] == "worker-2" + assert repo.lease_of("task-1")[1] == NOW + timedelta(seconds=120), "the claim was re-stamped" async def test_claims_the_most_overdue_first_and_honours_the_limit(self): repo = InMemoryScheduledTaskRepository() await repo.add(make_task("late", next_run_at=NOW - timedelta(hours=2))) await repo.add(make_task("later", next_run_at=NOW - timedelta(hours=1))) - claimed = await repo.claim_due(now=NOW, lease_owner="w", lease_seconds=120, limit=1) + claimed = await repo.claim_due(now=NOW, lease_seconds=120, limit=1) assert [task.task_id for task in claimed] == ["late"] diff --git a/backend/tests/test_schedule_service.py b/backend/tests/test_schedule_service.py index e833390ea..d151304c9 100644 --- a/backend/tests/test_schedule_service.py +++ b/backend/tests/test_schedule_service.py @@ -25,6 +25,7 @@ from schedule_fakes import ( from deerflow.domain.schedule.model import ( ContextMode, + DispatchOutcome, InvalidScheduleError, LaunchFailedError, RunStatus, @@ -38,7 +39,7 @@ from deerflow.domain.schedule.model import ( TriggerKind, ) from deerflow.domain.schedule.ports import RunOutcome -from deerflow.domain.schedule.service import ScheduleService +from deerflow.domain.schedule.service import ContextChange, ScheduleService pytestmark = pytest.mark.asyncio @@ -59,7 +60,7 @@ class _BlindRunRepo(InMemoryScheduledRunRepository): rejection on `add` is then the only thing standing in the way. """ - async def has_active(self, task_id: str) -> bool: + async def has_active(self, _task_id: str) -> bool: return False @@ -77,7 +78,6 @@ def make_service( launcher=launcher if launcher is not None else FakeRunLauncher(), threads=threads if threads is not None else FakeThreadLookup(), policy=policy, - lease_owner="test-worker", ) @@ -116,7 +116,7 @@ class TestFullLifecycle: # -- become due, get claimed and dispatched -------------------------- due_at = task.next_run_at + timedelta(seconds=1) results = await service.run_once(now=due_at) - assert [r.outcome for r in results] == ["launched"] + assert [r.outcome for r in results] == [DispatchOutcome.LAUNCHED] assert len(launcher.calls) == 1 launch = launcher.calls[0] assert launch["prompt"] == "summarize" @@ -133,7 +133,7 @@ class TestFullLifecycle: # -- next occurrence overlaps the still-running one ------------------ overlap_at = after_launch.next_run_at + timedelta(seconds=1) overlapped = await service.run_once(now=overlap_at) - assert [r.outcome for r in overlapped] == ["skipped"] + assert [r.outcome for r in overlapped] == [DispatchOutcome.SKIPPED] assert len(launcher.calls) == 1, "the overlapping occurrence must not launch" after_skip = await service.get_task(task.task_id, user_id="user-1") @@ -184,7 +184,7 @@ class TestDispatchOutcomes: result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) - assert result.outcome == "launched" + assert result.outcome is DispatchOutcome.LAUNCHED assert result.run_id == "run-1" assert result.error is None stored = runs.all_runs()[0] @@ -203,7 +203,7 @@ class TestDispatchOutcomes: result = await service.trigger_task(task.task_id, user_id="user-1", now=NOW) - assert result.outcome == "conflict" + assert result.outcome is DispatchOutcome.CONFLICT assert result.record_id is None assert result.error == "task already has an active run" assert len(runs.all_runs()) == before, "no history row for a rejected manual trigger" @@ -216,7 +216,7 @@ class TestDispatchOutcomes: result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) - assert result.outcome == "skipped" + assert result.outcome is DispatchOutcome.SKIPPED assert result.error == "skipped: a previous run of this task is still active" tombstone = next(r for r in runs.all_runs() if r.status is RunStatus.SKIPPED) assert tombstone.is_active is False, "a queued tombstone would collide with the live run" @@ -230,7 +230,7 @@ class TestDispatchOutcomes: result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) - assert result.outcome == "failed" + assert result.outcome is DispatchOutcome.FAILED assert result.run_id is None assert "provider exploded" in result.error assert runs.all_runs()[0].status is RunStatus.FAILED @@ -242,7 +242,7 @@ class TestDispatchOutcomes: result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED) - assert result.outcome == "skipped" + assert result.outcome is DispatchOutcome.SKIPPED async def test_busy_thread_on_a_manual_trigger_stays_a_conflict(self): """Not a skip: the user asked for this one, so it is reported rather @@ -253,7 +253,7 @@ class TestDispatchOutcomes: result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.MANUAL) - assert result.outcome == "conflict" + assert result.outcome is DispatchOutcome.CONFLICT assert result.record_id is not None, "the attempt itself is still recorded" @@ -293,13 +293,13 @@ class TestConflictCollapse: async def test_the_losing_scheduled_dispatch_still_leaves_a_tombstone(self): runs = _BlindRunRepo() result = await self._dispatch_second(runs, TriggerKind.SCHEDULED) - assert result.outcome == "skipped" + assert result.outcome is DispatchOutcome.SKIPPED assert any(r.status is RunStatus.SKIPPED for r in runs.all_runs()) async def test_the_losing_manual_dispatch_leaves_nothing(self): runs = _BlindRunRepo() result = await self._dispatch_second(runs, TriggerKind.MANUAL) - assert result.outcome == "conflict" + assert result.outcome is DispatchOutcome.CONFLICT assert not any(r.status is RunStatus.SKIPPED for r in runs.all_runs()) @@ -374,7 +374,7 @@ class TestRunOnceBudget: results = await service.run_once(now=due_at) assert len(results) == 2 - assert all(r.outcome == "launched" for r in results) + assert all(r.outcome is DispatchOutcome.LAUNCHED for r in results) async def test_a_claimed_task_is_marked_running_before_dispatch(self): """Claiming is what makes the task uneditable while it is being @@ -555,6 +555,42 @@ class TestTaskManagement: assert updated.prompt == task.prompt assert updated.schedule == task.schedule + async def test_context_is_changed_through_the_packaged_value(self): + """context_mode and thread_id move together, so they are supplied + together -- which is what lets every other update field use plain + `None` for "not supplied".""" + 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", + 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", + now=NOW, + context=ContextChange(ContextMode.FRESH_THREAD_PER_RUN), + ) + assert unbound.thread_id is None, "switching to a fresh thread clears the binding" + + async def test_changing_context_to_an_inaccessible_thread_is_rejected(self): + service = make_service(threads=FakeThreadLookup({"thread-1": "someone-else"})) + task = await create_cron_task(service) + + with pytest.raises(ThreadNotFoundError): + await service.update_task( + task.task_id, + user_id="user-1", + now=NOW, + context=ContextChange(ContextMode.REUSE_THREAD, "thread-1"), + ) + async def test_rescheduling_a_terminal_task_re_arms_it(self): tasks = InMemoryScheduledTaskRepository() service = make_service(tasks=tasks)