deer-flow/backend/tests/test_schedule_service.py

"""Use-case tests for ScheduleService, run entirely on in-memory doubles.

This file is the acceptance criterion of the hexagonal migration: the complete
scheduled-task lifecycle runs here with no HTTP, no database, and no run
runtime -- if any of that were still reachable from the domain, these tests
could not exist.

The dispatch cases mirror the pre-migration behaviour deliberately. Where a
comment says two paths must be indistinguishable, that is a regression the
tests are here to catch, not a description of the implementation.
"""

from __future__ import annotations

from dataclasses import replace
from datetime import UTC, datetime, timedelta

import pytest
from schedule_fakes import (
    FakeRunLauncher,
    FakeThreadLookup,
    InMemoryScheduledRunRepository,
    InMemoryScheduledTaskRepository,
)

from deerflow.domain.schedule.commands import (
    UNSET,
    ContextChange,
    CreateScheduledTask,
    DeleteTask,
    PauseTask,
    ResumeTask,
    TriggerTask,
    UnsetType,
    UpdateScheduledTask,
)
from deerflow.domain.schedule.exceptions import (
    InvalidScheduleError,
    LaunchFailedError,
    TaskNotFoundError,
    TaskNotMutableError,
    ThreadBusyError,
    ThreadNotFoundError,
)
from deerflow.domain.schedule.model import (
    ContextMode,
    DispatchOutcome,
    RunStatus,
    SchedulePolicy,
    ScheduleSpec,
    TaskStatus,
    TriggerKind,
)
from deerflow.domain.schedule.ports import RunOutcome
from deerflow.domain.schedule.service import ScheduleService

pytestmark = pytest.mark.asyncio

NOW = datetime(2026, 7, 27, 12, 0, tzinfo=UTC)
CRON = ScheduleSpec.cron_schedule("0 9 * * *", "UTC")
POLICY = SchedulePolicy(min_once_delay_seconds=60, max_concurrent_runs=3, lease_seconds=120)


def once_spec(*, after_seconds: int = 3600) -> ScheduleSpec:
    return ScheduleSpec.once_at(NOW + timedelta(seconds=after_seconds), "UTC")


class _BlindRunRepo(InMemoryScheduledRunRepository):
    """`has_active` always misses.

    Reproduces the TOCTOU window the fast path cannot close: a concurrent
    dispatch inserted its record after this one looked. The active-slot
    rejection on `add` is then the only thing standing in the way.
    """

    async def has_active(self, _task_id: str) -> bool:
        return False


class _LaunchWritesMidReadTaskRepo(InMemoryScheduledTaskRepository):
    """`record_launch` commits between the completion hook's read and its write.

    That ordering is not exotic -- `dispatch_task` performs its two bookkeeping
    writes after the launch returns, and a run that fails fast reaches the hook
    in between. Modelling it here rather than with real concurrency keeps the
    window deterministic.
    """

    def __init__(self, *, launch_next_run_at: datetime) -> None:
        super().__init__()
        self._launch_next_run_at = launch_next_run_at
        self._armed = False

    def arm(self) -> None:
        """Fire the interleaving on the next read, once."""
        self._armed = True

    async def get(self, task_id: str, *, user_id: str):
        task = await super().get(task_id, user_id=user_id)
        if self._armed:
            self._armed = False
            await self.record_launch(
                task_id,
                status=TaskStatus.ENABLED,
                next_run_at=self._launch_next_run_at,
                last_run_at=NOW,
                last_run_id="run-1",
                last_thread_id="thread-1",
                last_error=None,
                increment_run_count=True,
                protect_terminal=True,
            )
        return task


def make_service(
    *,
    tasks: InMemoryScheduledTaskRepository | None = None,
    runs: InMemoryScheduledRunRepository | None = None,
    launcher: FakeRunLauncher | None = None,
    threads: FakeThreadLookup | None = None,
    policy: SchedulePolicy = POLICY,
) -> ScheduleService:
    return ScheduleService(
        tasks=tasks if tasks is not None else InMemoryScheduledTaskRepository(),
        runs=runs if runs is not None else InMemoryScheduledRunRepository(),
        launcher=launcher if launcher is not None else FakeRunLauncher(),
        threads=threads if threads is not None else FakeThreadLookup(),
        policy=policy,
    )


async def create_cron_task(service: ScheduleService, *, schedule: ScheduleSpec = CRON):
    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,
    )


# ==================================================================== lifecycle


class TestFullLifecycle:
    async def test_create_claim_dispatch_overlap_complete_pause_delete(self):
        """The acceptance criterion: the whole feature, zero IO.

        Every step below is a real use case going through real domain rules --
        only the four ports are doubles.
        """
        tasks = InMemoryScheduledTaskRepository()
        runs = InMemoryScheduledRunRepository()
        launcher = FakeRunLauncher()
        service = make_service(tasks=tasks, runs=runs, launcher=launcher)

        # -- create ---------------------------------------------------------
        task = await create_cron_task(service)
        assert task.status is TaskStatus.ENABLED
        assert task.next_run_at == datetime(2026, 7, 28, 9, 0, tzinfo=UTC)

        # -- 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] == [DispatchOutcome.LAUNCHED]
        assert len(launcher.calls) == 1
        launch = launcher.calls[0]
        assert launch["prompt"] == "summarize"
        assert launch["owner_user_id"] == "user-1"
        assert launch["metadata"]["scheduled_task_id"] == task.task_id
        assert launch["metadata"]["scheduled_trigger"] == "scheduled"

        after_launch = await service.get_task(task.task_id, user_id="user-1")
        assert after_launch.status is TaskStatus.ENABLED, "a cron task stays claimable"
        assert after_launch.run_count == 1
        assert after_launch.last_run_id == "run-1"
        assert after_launch.next_run_at > due_at

        # -- 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] == [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")
        assert after_skip.run_count == 1, "a skip is not an execution"
        assert after_skip.last_run_id == after_launch.last_run_id, "bookkeeping carried over"
        assert after_skip.status is TaskStatus.ENABLED
        # Only a lost one-shot occurrence is worth surfacing; a cron task simply
        # waits for its next turn, so the skip must not leave a user-visible
        # error behind (service.py:455). The `once` half is covered by
        # TestOnceTaskDispatch.test_a_skipped_once_task_is_failed_not_completed.
        assert after_skip.last_error is None, "a routine cron overlap is not an error to report"

        # -- the first run finally finishes ----------------------------------
        record = next(r for r in runs.all_runs() if r.status is RunStatus.RUNNING)
        await service.handle_run_completion(
            RunOutcome(
                task_id=task.task_id,
                record_id=record.record_id,
                run_id="run-1",
                user_id="user-1",
                status=RunStatus.SUCCESS,
                error=None,
            ),
            now=overlap_at,
        )
        assert await runs.count_active() == 0, "the active slot is free again"

        # -- pause / resume ---------------------------------------------------
        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(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(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")


# ==================================================================== dispatch


class TestDispatchOutcomes:
    async def test_launch_records_the_run_and_the_bookkeeping(self):
        runs = InMemoryScheduledRunRepository()
        service = make_service(runs=runs)
        task = await create_cron_task(service)

        result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)

        assert result.outcome is DispatchOutcome.LAUNCHED
        assert result.run_id == "run-1"
        assert result.error is None
        stored = runs.all_runs()[0]
        assert stored.status is RunStatus.RUNNING
        assert stored.run_id == "run-1"
        assert stored.started_at == NOW

    async def test_manual_trigger_against_an_active_run_is_a_conflict_with_no_record(self):
        """Nothing was scheduled to happen, so there is no occurrence to
        account for -- the caller gets a conflict and the history stays clean."""
        runs = InMemoryScheduledRunRepository()
        service = make_service(runs=runs)
        task = await create_cron_task(service)
        await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
        before = len(runs.all_runs())

        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
        assert result.error == "task already has an active run"
        assert len(runs.all_runs()) == before, "no history row for a rejected manual trigger"

    async def test_scheduled_overlap_records_a_terminal_tombstone(self):
        runs = InMemoryScheduledRunRepository()
        service = make_service(runs=runs)
        task = await create_cron_task(service)
        await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)

        result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)

        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"
        assert tombstone.started_at == tombstone.finished_at == NOW

    async def test_launch_failure_is_recorded_as_failed(self):
        runs = InMemoryScheduledRunRepository()
        launcher = FakeRunLauncher(fail_with=LaunchFailedError("provider exploded"))
        service = make_service(runs=runs, launcher=launcher)
        task = await create_cron_task(service)

        result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)

        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

    async def test_a_failed_launch_replaces_the_previous_run_bookkeeping(self):
        """A failed launch is still an execution *attempt*, so it overwrites the
        launch bookkeeping rather than carrying it over the way a skip does
        (contrast `_finalize_skip`, which passes the current values back).

        `record_launch` assigns every field unconditionally, so this is what
        `last_run_id=None` in `_fail` actually means for a task that had already
        run successfully: the id of the previous, unrelated run must not be left
        pointing at a launch that never happened.
        """
        launcher = FakeRunLauncher()
        runs = InMemoryScheduledRunRepository()
        service = make_service(runs=runs, launcher=launcher)
        task = await create_cron_task(service)
        await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
        succeeded = await service.get_task(task.task_id, user_id="user-1")
        assert succeeded.last_run_id == "run-1", "precondition: a successful launch was recorded"

        # The slot has to be free, or the next dispatch is an overlap instead.
        await service.handle_run_completion(
            RunOutcome(
                task_id=task.task_id,
                record_id=runs.all_runs()[0].record_id,
                run_id="run-1",
                user_id="user-1",
                status=RunStatus.SUCCESS,
                error=None,
            ),
            now=NOW,
        )
        launcher.fail_with = LaunchFailedError("provider exploded")
        reloaded = await service.get_task(task.task_id, user_id="user-1")

        await service.dispatch_task(reloaded, now=NOW, trigger=TriggerKind.SCHEDULED)

        after = await service.get_task(task.task_id, user_id="user-1")
        assert after.last_run_id is None, "no run started, so no run id to point at"
        assert after.last_error == "provider exploded"
        assert after.last_run_at == NOW, "the attempt itself is timestamped"
        assert after.run_count == 1, "a failed launch is not a completed execution"

    async def test_busy_thread_on_the_scheduled_path_degrades_to_a_skip(self):
        launcher = FakeRunLauncher(fail_with=ThreadBusyError("thread busy"))
        service = make_service(launcher=launcher)
        task = await create_cron_task(service)

        result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)

        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
        than quietly accounted for. The router maps it to 409, not 502."""
        launcher = FakeRunLauncher(fail_with=ThreadBusyError("thread busy"))
        service = make_service(launcher=launcher)
        task = await create_cron_task(service)

        result = await service.dispatch_task(task, now=NOW, trigger=TriggerKind.MANUAL)

        assert result.outcome is DispatchOutcome.CONFLICT
        assert result.record_id is not None, "the attempt itself is still recorded"


class TestConflictCollapse:
    """The fast path and the active-slot rejection must be indistinguishable.

    Two concurrent dispatches can both pass `has_active`; whichever loses is
    rejected by the repository instead. A caller must not be able to tell which
    of the two mechanisms stopped it, or retry behaviour diverges.
    """

    async def _dispatch_second(self, run_repo, trigger: TriggerKind):
        # reuse_thread so the execution thread is fixed: a fresh-thread task
        # 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_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)
        return await service.dispatch_task(task, now=NOW, trigger=trigger)

    @pytest.mark.parametrize("trigger", [TriggerKind.SCHEDULED, TriggerKind.MANUAL])
    async def test_both_paths_produce_the_same_result(self, trigger):
        via_fast_path = await self._dispatch_second(InMemoryScheduledRunRepository(), trigger)
        via_slot_rejection = await self._dispatch_second(_BlindRunRepo(), trigger)

        # record_id is freshly generated; every other field must match exactly.
        assert replace(via_fast_path, record_id=None) == replace(via_slot_rejection, record_id=None)

    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 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 is DispatchOutcome.CONFLICT
        assert not any(r.status is RunStatus.SKIPPED for r in runs.all_runs())


class TestOnceTaskDispatch:
    async def test_a_once_task_waits_in_running_for_its_completion(self):
        """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_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)

        after = await service.get_task(task.task_id, user_id="user-1")
        assert after.status is TaskStatus.RUNNING

    async def test_a_skipped_once_task_is_failed_not_completed(self):
        """The single occurrence was lost; `completed` would claim an execution
        that never happened."""
        runs = InMemoryScheduledRunRepository()
        service = make_service(runs=runs)
        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)
        reloaded = await service.get_task(task.task_id, user_id="user-1")

        await service.dispatch_task(reloaded, now=NOW, trigger=TriggerKind.SCHEDULED)

        after = await service.get_task(task.task_id, user_id="user-1")
        assert after.status is TaskStatus.FAILED
        assert after.last_error == "skipped: a previous run of this task is still active"


# ==================================================================== run_once


class TestRunOnceBudget:
    async def test_claims_nothing_when_the_global_budget_is_exhausted(self):
        """The cap is on active runs across all tasks, not on one poll's batch:
        long runs accumulate across cycles."""
        runs = InMemoryScheduledRunRepository()
        service = make_service(runs=runs, policy=replace(POLICY, max_concurrent_runs=1))
        first = await create_cron_task(service)
        await service.dispatch_task(first, now=NOW, trigger=TriggerKind.SCHEDULED)

        second = await create_cron_task(service)
        results = await service.run_once(now=second.next_run_at + timedelta(seconds=1))

        assert results == []

    async def test_claims_only_into_the_remaining_budget(self):
        tasks = InMemoryScheduledTaskRepository()
        service = make_service(tasks=tasks, policy=replace(POLICY, max_concurrent_runs=2))
        for _ in range(3):
            await create_cron_task(service)

        due_at = datetime(2026, 7, 28, 9, 0, 1, tzinfo=UTC)
        results = await service.run_once(now=due_at)

        assert len(results) == 2
        assert all(r.outcome is DispatchOutcome.LAUNCHED for r in results)

    async def test_the_claim_is_held_across_the_launch_and_released_after(self):
        """Claiming is what makes the task uneditable while it is being
        dispatched, so the claim has to be live *at the moment of the launch* --
        not merely taken at some point and released by the end.

        The launch is the only place that ordering is observable, so the
        launcher double reads the repository from inside `launch`. Asserting
        only on the end state would pass even if the claim were taken after the
        run had already started.
        """
        tasks = InMemoryScheduledTaskRepository()
        observed: dict = {}

        class _ObservingLauncher(FakeRunLauncher):
            async def launch(self, **kwargs):
                task_id = kwargs["metadata"]["scheduled_task_id"]
                observed["lease"] = tasks.lease_of(task_id)
                observed["status"] = (await tasks.get(task_id, user_id="user-1")).status
                return await super().launch(**kwargs)

        service = make_service(tasks=tasks, launcher=_ObservingLauncher())
        task = await create_cron_task(service)

        await service.run_once(now=task.next_run_at + timedelta(seconds=1))

        assert observed["status"] is TaskStatus.RUNNING, "the claim marks the task running before the launch"
        assert observed["lease"][1] is not None, "and the lease is still held while the run starts"
        assert tasks.lease_of(task.task_id) == (None, None), "released only once the dispatch is done"


# ==================================================================== completion


class TestRunCompletion:
    async def _launched_once_task(self):
        runs = InMemoryScheduledRunRepository()
        service = make_service(runs=runs)
        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]
        return service, task, record

    @pytest.mark.parametrize(
        ("outcome_status", "expected"),
        [
            (RunStatus.SUCCESS, TaskStatus.COMPLETED),
            (RunStatus.FAILED, TaskStatus.FAILED),
            (RunStatus.INTERRUPTED, TaskStatus.CANCELLED),
        ],
    )
    async def test_once_task_terminal_mapping(self, outcome_status, expected):
        service, task, record = await self._launched_once_task()

        await service.handle_run_completion(
            RunOutcome(
                task_id=task.task_id,
                record_id=record.record_id,
                run_id="run-1",
                user_id="user-1",
                status=outcome_status,
                error=None if outcome_status is RunStatus.SUCCESS else "boom",
            ),
            now=NOW,
        )

        after = await service.get_task(task.task_id, user_id="user-1")
        assert after.status is expected

    async def test_a_cron_task_keeps_its_status_but_records_the_error(self):
        runs = InMemoryScheduledRunRepository()
        service = make_service(runs=runs)
        task = await create_cron_task(service)
        await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
        record = runs.all_runs()[0]

        await service.handle_run_completion(
            RunOutcome(
                task_id=task.task_id,
                record_id=record.record_id,
                run_id="run-1",
                user_id="user-1",
                status=RunStatus.FAILED,
                error="boom",
            ),
            now=NOW,
        )

        after = await service.get_task(task.task_id, user_id="user-1")
        assert after.status is TaskStatus.ENABLED, "the schedule outlives any single run"
        assert after.last_error == "boom"

    async def test_a_task_deleted_mid_flight_still_finalizes_its_run(self):
        """A task deleted while its run was in flight has nothing to write back,
        and that is not an error -- but the *run* record must still be closed.

        The hook writes the record first and only then reads the task, so
        asserting on the record is what keeps that ordering honest; a test that
        only checked "no exception" would stay green if the record write were
        dropped entirely.
        """
        runs = InMemoryScheduledRunRepository()
        service = make_service(runs=runs)
        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(DeleteTask(task_id=task.task_id, user_id="user-1"))

        await service.handle_run_completion(
            RunOutcome(
                task_id=task.task_id,
                record_id=record.record_id,
                run_id="run-1",
                user_id="user-1",
                status=RunStatus.SUCCESS,
                error=None,
            ),
            now=NOW,
        )

        stored = runs.all_runs()[0]
        assert stored.status is RunStatus.SUCCESS, "the execution record is finalized regardless"
        assert stored.finished_at == NOW
        assert await runs.count_active() == 0, "the active slot is freed"

    async def test_a_cron_task_survives_a_launch_write_landing_mid_completion(self):
        """The interleaving `dispatch_task` already admits is possible.

        A run that fails fast reaches this hook while the dispatch path has not
        yet written its bookkeeping, so the hook reads a task still carrying the
        elapsed `next_run_at`. If the write-back replays that whole snapshot,
        the launch path's fresh fire time is rolled back and the task lands in
        `running` with no live claim -- a shape neither `claim_due` branch nor
        `cancel_stuck_once_tasks` can reach, i.e. permanently unschedulable.
        """
        tasks = _LaunchWritesMidReadTaskRepo(launch_next_run_at=NOW + timedelta(days=1))
        runs = InMemoryScheduledRunRepository()
        service = make_service(tasks=tasks, runs=runs)
        task = await create_cron_task(service)
        await service.dispatch_task(task, now=NOW, trigger=TriggerKind.SCHEDULED)
        record = runs.all_runs()[0]
        tasks.arm()

        await service.handle_run_completion(
            RunOutcome(
                task_id=task.task_id,
                record_id=record.record_id,
                run_id="run-1",
                user_id="user-1",
                status=RunStatus.FAILED,
                error="boom",
            ),
            now=NOW,
        )

        after = await service.get_task(task.task_id, user_id="user-1")
        assert after.last_error == "boom", "the completion still records its verdict"
        assert after.next_run_at == NOW + timedelta(days=1), "the launch path's fire time must survive"
        claimed = await tasks.claim_due(now=NOW + timedelta(days=2), lease_seconds=120, limit=10)
        assert [t.task_id for t in claimed] == [task.task_id], "the task must remain schedulable"


class TestReconcileOnStartup:
    async def test_sweeps_orphaned_runs_and_stuck_once_tasks(self):
        runs = InMemoryScheduledRunRepository()
        tasks = InMemoryScheduledTaskRepository()
        service = make_service(tasks=tasks, runs=runs)
        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)
        # The process dies here: the run is active and the once task is parked
        # in running with its claim already released.

        stale_runs, stuck_tasks = await service.reconcile_on_startup(error="gateway restarted")

        assert (stale_runs, stuck_tasks) == (1, 1)
        assert runs.all_runs()[0].status is RunStatus.INTERRUPTED
        after = await service.get_task(task.task_id, user_id="user-1")
        assert after.status is TaskStatus.CANCELLED


# ==================================================================== CRUD


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_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,
            )

    async def test_an_invalid_schedule_is_rejected_before_any_write(self):
        tasks = InMemoryScheduledTaskRepository()
        service = make_service(tasks=tasks)

        with pytest.raises(InvalidScheduleError):
            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") == []

    async 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"

    async 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_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_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

    async def test_a_task_deleted_mid_update_reports_as_missing(self):
        """get_task saw it, save no longer does -- a concurrent delete landed
        in between. The caller gets the same not-found it would have got a
        moment earlier, rather than a None leaking out."""

        class _VanishingRepo(InMemoryScheduledTaskRepository):
            async def save(self, _task):
                return None

        tasks = _VanishingRepo()
        service = make_service(tasks=tasks)
        task = await create_cron_task(service)

        with pytest.raises(TaskNotFoundError):
            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(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
        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_scheduled_task(
            UpdateScheduledTask(task_id=task.task_id, user_id="user-1", context=ContextChange(ContextMode.REUSE_THREAD, "thread-1")),
            now=NOW,
        )
        assert bound.context_mode is ContextMode.REUSE_THREAD
        assert bound.thread_id == "thread-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,
        )
        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_scheduled_task(
                UpdateScheduledTask(task_id=task.task_id, user_id="user-1", context=ContextChange(ContextMode.REUSE_THREAD, "thread-1")),
                now=NOW,
            )

    async def test_rescheduling_a_terminal_task_re_arms_it(self):
        tasks = InMemoryScheduledTaskRepository()
        service = make_service(tasks=tasks)
        task = await create_cron_task(service)
        tasks.seed(replace(task, status=TaskStatus.FAILED))

        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,
        )

        assert updated.status is TaskStatus.ENABLED, "otherwise it would never be claimed again"

    async def test_a_running_task_cannot_be_edited(self):
        tasks = InMemoryScheduledTaskRepository()
        service = make_service(tasks=tasks)
        task = await create_cron_task(service)
        tasks.seed(replace(task, status=TaskStatus.RUNNING))

        with pytest.raises(TaskNotMutableError):
            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(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
        delete -- that asymmetry is preserved."""
        tasks = InMemoryScheduledTaskRepository()
        service = make_service(tasks=tasks)
        task = await create_cron_task(service)
        tasks.seed(replace(task, status=TaskStatus.RUNNING))

        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)
        with pytest.raises(TaskNotFoundError):
            if call == "get":
                await service.get_task(task.task_id, user_id="intruder")
            elif call == "update":
                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(PauseTask(task_id=task.task_id, user_id="intruder"))
            elif call == "resume":
                await service.resume_task(ResumeTask(task_id=task.task_id, user_id="intruder"))
            elif call == "delete":
                await service.delete_task(DeleteTask(task_id=task.task_id, user_id="intruder"))
            else:
                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_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)

        assert len(await service.list_tasks("user-1")) == 2
        assert await service.list_tasks("someone-else") == []
        by_thread = await service.list_tasks_by_thread("user-1", "thread-1")
        assert [t.task_id for t in by_thread] == [bound.task_id]