mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-13 16:28:38 +00:00
6f84a4094d
The outer ring for the domain added in #4597: SQL repositories, the run launcher, the thread lookup, and the run-completion listener implementing the ports it declared, plus the HTTP router and the poller that drive them. All of it is instantiated in one composition root, so no route or lifespan hook builds an adapter of its own. With the ports filled, the pre-hexagonal implementation is deleted rather than left alongside: `app/scheduler/service.py` and its router mixed policy, persistence, and HTTP into one class, which is why its rules were only reachable through a live database. Keeping both would leave two implementations of the same rules writing to the same table. Three of the domain's contracts needed real work on this side rather than a straight port of the pre-#4597 adapters: - The launcher now distinguishes certain failure from doubt. Only a 4xx is certain enough to raise LaunchFailedError, which releases the task's single active slot; a 5xx, an arbitrary exception, or a reply whose identity will not decode all raise LaunchIndeterminateError and keep the slot held. Guessing "failed" after the launch request was sent is what re-opens #4452's duplicate execution. - The task repository implements the optimistic token. `save` is a conditional UPDATE on `version` rather than read-check-write, because the latter lets two savers observe the same version and both commit; every other committed write increments it. This needs a column, so it ships with migration 0011 -- the only schema change in the slice, and the reason the alembic head pins move. - The router builds commands with plain `None` for "not supplied", and maps ConcurrentUpdateError onto a retryable 409. The concurrency invariants are pinned by contract suites that run each port against both the in-memory double and real sqlite -- including a new TestOptimisticConcurrency covering what invalidates an earlier read -- plus the dispatch-race tests against a real database. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
682 lines
29 KiB
Python
682 lines
29 KiB
Python
"""Contract suite for the schedule repository ports.
|
|
|
|
Every case here runs **twice**: once against the in-memory doubles and once
|
|
against the SQL adapters on a real file-backed sqlite database. That is the
|
|
point -- a rule stated in a port docstring has to hold for both, and a
|
|
divergence becomes a failure rather than a surprise in production.
|
|
|
|
What the contract owns is single-threaded semantics: which rows `claim_due`
|
|
selects, that `add` refuses a second active record, what `protect_terminal`
|
|
preserves. What it deliberately does **not** own is atomicity -- the doubles
|
|
provide none, and a green run here says nothing about two dispatchers racing.
|
|
That is covered against a real database in
|
|
``test_schedule_dispatch_race.py``. Do not read a passing contract suite
|
|
as licence to run more than one scheduler.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from schedule_fakes import (
|
|
FakeRunLauncher,
|
|
FakeThreadLookup,
|
|
InMemoryScheduledRunRepository,
|
|
InMemoryScheduledTaskRepository,
|
|
)
|
|
|
|
from app.adapters.schedule.scheduled_run_repository import SqlScheduledRunRepository
|
|
from app.adapters.schedule.scheduled_task_repository import SqlScheduledTaskRepository
|
|
from deerflow.config.database_config import DatabaseConfig
|
|
from deerflow.domain.schedule.exceptions import ActiveRunConflictError, ConcurrentUpdateError
|
|
from deerflow.domain.schedule.model import ContextMode, RunStatus, ScheduledRun, ScheduledTask, ScheduleSpec, TaskStatus, TriggerKind
|
|
from deerflow.domain.schedule.ports import ScheduledRunRepository, ScheduledTaskRepository
|
|
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
NOW = datetime(2026, 7, 27, 12, 0, tzinfo=UTC)
|
|
CRON = ScheduleSpec.cron_schedule("0 9 * * *", "UTC")
|
|
ONCE = ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0, tzinfo=UTC), "UTC")
|
|
|
|
|
|
@pytest_asyncio.fixture(params=["memory", "sql"])
|
|
async def repos(request, tmp_path) -> AsyncIterator[tuple[ScheduledTaskRepository, ScheduledRunRepository]]:
|
|
"""One parametrized fixture, two implementations of the same ports."""
|
|
if request.param == "memory":
|
|
yield InMemoryScheduledTaskRepository(), InMemoryScheduledRunRepository()
|
|
return
|
|
await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
|
|
try:
|
|
sf = get_session_factory()
|
|
assert sf is not None
|
|
yield SqlScheduledTaskRepository(sf), SqlScheduledRunRepository(sf)
|
|
finally:
|
|
await close_engine()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def tasks(repos) -> ScheduledTaskRepository:
|
|
return repos[0]
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def runs(repos) -> ScheduledRunRepository:
|
|
return repos[1]
|
|
|
|
|
|
async def seed(
|
|
repo: ScheduledTaskRepository,
|
|
task: ScheduledTask,
|
|
*,
|
|
claimed_until: datetime | None = None,
|
|
) -> ScheduledTask:
|
|
"""Install a task, optionally already carrying a claim.
|
|
|
|
A claim cannot be installed through the port -- `claim_due` only stamps
|
|
tasks that are actually due, and these cases need shapes that claiming
|
|
would never produce (running with an *expired* claim, running with a live
|
|
one). So each implementation is set up directly: the fake exposes a seed
|
|
helper, and the SQL side writes the row. That is the one place this suite
|
|
reaches past the port, and it is why the assertions that follow go back
|
|
through it.
|
|
"""
|
|
stored = await repo.add(task)
|
|
if claimed_until is None:
|
|
return stored
|
|
|
|
if isinstance(repo, InMemoryScheduledTaskRepository):
|
|
repo.seed(task, lease_owner="prior-worker", lease_expires_at=claimed_until)
|
|
return task
|
|
|
|
from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow
|
|
|
|
factory = get_session_factory()
|
|
assert factory is not None
|
|
async with factory() as session:
|
|
row = await session.get(ScheduledTaskRow, task.task_id)
|
|
row.lease_owner = "prior-worker"
|
|
row.lease_expires_at = claimed_until
|
|
await session.commit()
|
|
return task
|
|
|
|
|
|
def make_task(
|
|
task_id: str = "task-1",
|
|
*,
|
|
user_id: str = "user-1",
|
|
status: TaskStatus = TaskStatus.ENABLED,
|
|
next_run_at: datetime | None = None,
|
|
schedule: ScheduleSpec = CRON,
|
|
thread_id: str | None = None,
|
|
context_mode: ContextMode = ContextMode.FRESH_THREAD_PER_RUN,
|
|
) -> ScheduledTask:
|
|
return ScheduledTask(
|
|
task_id=task_id,
|
|
user_id=user_id,
|
|
title=task_id,
|
|
prompt="do the thing",
|
|
schedule=schedule,
|
|
status=status,
|
|
next_run_at=next_run_at,
|
|
thread_id=thread_id,
|
|
context_mode=context_mode,
|
|
)
|
|
|
|
|
|
class TestProtocolConformance:
|
|
"""`runtime_checkable` only checks that the methods exist, not their
|
|
signatures -- enough to catch a rename that updates one side only."""
|
|
|
|
async def test_repositories_satisfy_their_ports(self, tasks, runs):
|
|
assert isinstance(tasks, ScheduledTaskRepository)
|
|
assert isinstance(runs, ScheduledRunRepository)
|
|
|
|
|
|
class TestTaskOwnershipIsolation:
|
|
"""Another user's task must read as absent, never as forbidden."""
|
|
|
|
async def test_get_hides_another_users_task(self, tasks):
|
|
await tasks.add(make_task(user_id="owner"))
|
|
assert await tasks.get("task-1", user_id="owner") is not None
|
|
assert await tasks.get("task-1", user_id="intruder") is None
|
|
|
|
async def test_save_refuses_another_users_task(self, tasks):
|
|
await tasks.add(make_task(user_id="owner"))
|
|
assert await tasks.save(make_task(user_id="intruder")) is None
|
|
stored = await tasks.get("task-1", user_id="owner")
|
|
assert stored.user_id == "owner"
|
|
|
|
async def test_delete_refuses_another_users_task(self, tasks):
|
|
await tasks.add(make_task(user_id="owner"))
|
|
assert await tasks.delete("task-1", user_id="intruder") is False
|
|
assert await tasks.get("task-1", user_id="owner") is not None
|
|
|
|
async def test_list_by_user_excludes_other_owners(self, tasks):
|
|
await tasks.add(make_task("mine", user_id="owner"))
|
|
await tasks.add(make_task("theirs", user_id="someone-else"))
|
|
assert [t.task_id for t in await tasks.list_by_user("owner")] == ["mine"]
|
|
|
|
async def test_list_by_thread_only_matches_bound_tasks(self, tasks):
|
|
await tasks.add(make_task("bound", context_mode=ContextMode.REUSE_THREAD, thread_id="thread-1"))
|
|
await tasks.add(make_task("unbound"))
|
|
listed = await tasks.list_by_user_and_thread("user-1", "thread-1")
|
|
assert [task.task_id for task in listed] == ["bound"]
|
|
|
|
|
|
class TestRoundTrip:
|
|
"""What goes in comes back out -- including the value object, which the
|
|
SQL side has to rebuild from three separate columns."""
|
|
|
|
async def test_a_cron_task_round_trips(self, tasks):
|
|
original = await tasks.add(make_task(schedule=ScheduleSpec.cron_schedule("*/5 * * * *", "Asia/Shanghai")))
|
|
stored = await tasks.get(original.task_id, user_id="user-1")
|
|
assert stored.schedule == original.schedule
|
|
assert stored.schedule.cron == "*/5 * * * *"
|
|
assert stored.schedule.timezone == "Asia/Shanghai"
|
|
|
|
async def test_a_once_task_round_trips_to_the_same_instant(self, tasks):
|
|
original = await tasks.add(make_task(schedule=ONCE))
|
|
stored = await tasks.get(original.task_id, user_id="user-1")
|
|
assert stored.schedule == original.schedule
|
|
|
|
async def test_timestamps_come_back_timezone_aware(self, tasks):
|
|
"""SQLite drops tzinfo on read; a naive datetime downstream would
|
|
compare wrong against an aware `now`."""
|
|
await tasks.add(make_task(next_run_at=NOW))
|
|
stored = await tasks.get("task-1", user_id="user-1")
|
|
assert stored.next_run_at.tzinfo is not None
|
|
assert stored.created_at.tzinfo is not None
|
|
|
|
async def test_add_persists_the_aggregates_construction_instant(self, tasks):
|
|
"""`created_at` has one source of truth: the aggregate's construction
|
|
instant. An adapter that mints its own timestamp on insert gives the
|
|
same fact two values -- domain tests and listings would disagree with
|
|
the stored row by however long the insert took."""
|
|
task = make_task()
|
|
await tasks.add(task)
|
|
stored = await tasks.get(task.task_id, user_id="user-1")
|
|
assert stored.created_at == task.created_at
|
|
assert stored.updated_at == task.updated_at
|
|
|
|
async def test_save_replaces_the_whole_aggregate(self, tasks):
|
|
from dataclasses import replace
|
|
|
|
await tasks.add(make_task())
|
|
stored = await tasks.get("task-1", user_id="user-1")
|
|
|
|
await tasks.save(replace(stored, title="renamed", status=TaskStatus.PAUSED, run_count=7))
|
|
|
|
reloaded = await tasks.get("task-1", user_id="user-1")
|
|
assert reloaded.title == "renamed"
|
|
assert reloaded.status is TaskStatus.PAUSED
|
|
assert reloaded.run_count == 7
|
|
|
|
|
|
class TestClaimDue:
|
|
async def test_claims_an_enabled_due_task(self, tasks):
|
|
await tasks.add(make_task(next_run_at=NOW - timedelta(minutes=1)))
|
|
|
|
claimed = await tasks.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
|
|
|
|
async def test_a_claimed_task_is_not_claimed_again(self, tasks):
|
|
"""The lease is not readable through the port, so the rule is asserted
|
|
the way it actually matters: a second claimer comes back empty."""
|
|
await tasks.add(make_task(next_run_at=NOW - timedelta(minutes=1)))
|
|
assert len(await tasks.claim_due(now=NOW, lease_seconds=120, limit=10)) == 1
|
|
|
|
assert await tasks.claim_due(now=NOW, lease_seconds=120, limit=10) == []
|
|
|
|
async def test_the_claim_expires(self, tasks):
|
|
await tasks.add(make_task(next_run_at=NOW - timedelta(minutes=1)))
|
|
await tasks.claim_due(now=NOW, lease_seconds=60, limit=10)
|
|
|
|
later = NOW + timedelta(seconds=61)
|
|
reclaimed = await tasks.claim_due(now=later, lease_seconds=60, limit=10)
|
|
|
|
assert [task.task_id for task in reclaimed] == ["task-1"]
|
|
|
|
@pytest.mark.parametrize(
|
|
("label", "task_kwargs"),
|
|
[
|
|
("not yet due", {"next_run_at": NOW + timedelta(minutes=1)}),
|
|
("never scheduled", {"next_run_at": None}),
|
|
("paused", {"status": TaskStatus.PAUSED, "next_run_at": NOW - timedelta(minutes=1)}),
|
|
("completed", {"status": TaskStatus.COMPLETED, "next_run_at": NOW - timedelta(minutes=1)}),
|
|
],
|
|
)
|
|
async def test_does_not_claim(self, tasks, label, task_kwargs):
|
|
await tasks.add(make_task(**task_kwargs))
|
|
assert await tasks.claim_due(now=NOW, lease_seconds=120, limit=10) == [], label
|
|
|
|
async def test_reclaims_a_task_stuck_mid_dispatch(self, tasks):
|
|
"""The claimer died between claiming and launching: status is running,
|
|
the claim has expired, and the task must not stay unreachable."""
|
|
await seed(
|
|
tasks,
|
|
make_task(status=TaskStatus.RUNNING, next_run_at=NOW - timedelta(minutes=1)),
|
|
claimed_until=NOW - timedelta(seconds=1),
|
|
)
|
|
|
|
claimed = await tasks.claim_due(now=NOW, lease_seconds=120, limit=10)
|
|
|
|
assert [task.task_id for task in claimed] == ["task-1"]
|
|
|
|
async def test_claims_the_most_overdue_first_and_honours_the_limit(self, tasks):
|
|
await tasks.add(make_task("late", next_run_at=NOW - timedelta(hours=2)))
|
|
await tasks.add(make_task("later", next_run_at=NOW - timedelta(hours=1)))
|
|
|
|
claimed = await tasks.claim_due(now=NOW, lease_seconds=120, limit=1)
|
|
|
|
assert [task.task_id for task in claimed] == ["late"]
|
|
|
|
|
|
class TestRecordLaunch:
|
|
async def test_writes_bookkeeping_and_frees_the_task_for_the_next_round(self, tasks):
|
|
await tasks.add(make_task(next_run_at=NOW - timedelta(minutes=1)))
|
|
await tasks.claim_due(now=NOW, lease_seconds=120, limit=10)
|
|
|
|
await tasks.record_launch(
|
|
"task-1",
|
|
status=TaskStatus.ENABLED,
|
|
next_run_at=NOW - timedelta(seconds=1),
|
|
last_run_at=NOW,
|
|
last_run_id="run-1",
|
|
last_thread_id="thread-1",
|
|
last_error=None,
|
|
increment_run_count=True,
|
|
)
|
|
|
|
task = await tasks.get("task-1", user_id="user-1")
|
|
assert task.status is TaskStatus.ENABLED
|
|
assert task.last_run_id == "run-1"
|
|
assert task.run_count == 1
|
|
# The claim was released, so the next round can take it again.
|
|
assert len(await tasks.claim_due(now=NOW, lease_seconds=120, limit=10)) == 1
|
|
|
|
async def test_protect_terminal_keeps_a_concurrently_finalized_verdict(self, tasks):
|
|
"""A fast-failing run's completion hook lands before the launch path's
|
|
own write; the completion is authoritative."""
|
|
await tasks.add(make_task(status=TaskStatus.COMPLETED, schedule=ONCE))
|
|
|
|
await tasks.record_launch(
|
|
"task-1",
|
|
status=TaskStatus.RUNNING,
|
|
next_run_at=None,
|
|
last_run_at=NOW,
|
|
last_run_id="run-1",
|
|
last_thread_id="thread-1",
|
|
last_error="stale",
|
|
increment_run_count=True,
|
|
protect_terminal=True,
|
|
)
|
|
|
|
task = await tasks.get("task-1", user_id="user-1")
|
|
assert task.status is TaskStatus.COMPLETED, "the terminal status must survive"
|
|
assert task.last_error is None, "the terminal error must survive"
|
|
assert task.last_run_id == "run-1", "bookkeeping is still recorded"
|
|
assert task.run_count == 1
|
|
|
|
async def test_without_protect_terminal_the_write_wins(self, tasks):
|
|
await tasks.add(make_task(status=TaskStatus.COMPLETED, schedule=ONCE))
|
|
|
|
await tasks.record_launch(
|
|
"task-1",
|
|
status=TaskStatus.FAILED,
|
|
next_run_at=None,
|
|
last_run_at=NOW,
|
|
last_run_id=None,
|
|
last_thread_id=None,
|
|
last_error="boom",
|
|
increment_run_count=False,
|
|
)
|
|
|
|
task = await tasks.get("task-1", user_id="user-1")
|
|
assert task.status is TaskStatus.FAILED
|
|
assert task.last_error == "boom"
|
|
|
|
async def test_unknown_task_is_ignored(self, tasks):
|
|
await tasks.record_launch(
|
|
"nope",
|
|
status=TaskStatus.ENABLED,
|
|
next_run_at=None,
|
|
last_run_at=None,
|
|
last_run_id=None,
|
|
last_thread_id=None,
|
|
last_error=None,
|
|
increment_run_count=False,
|
|
)
|
|
|
|
|
|
class TestRecordCompletion:
|
|
"""The completion hook's write.
|
|
|
|
Deliberately as narrow as `record_launch` is, and for the same reason: the
|
|
two race, so neither may write through the whole aggregate. This one owns
|
|
the terminal verdict and nothing else -- every scheduling field belongs to
|
|
the launch path, which may commit at any point around it.
|
|
"""
|
|
|
|
async def test_records_the_terminal_status_and_error(self, tasks):
|
|
await tasks.add(make_task(status=TaskStatus.RUNNING, schedule=ONCE))
|
|
|
|
await tasks.record_completion("task-1", user_id="user-1", status=TaskStatus.FAILED, error="boom")
|
|
|
|
task = await tasks.get("task-1", user_id="user-1")
|
|
assert task.status is TaskStatus.FAILED
|
|
assert task.last_error == "boom"
|
|
|
|
async def test_a_none_status_records_the_error_and_leaves_the_status(self, tasks):
|
|
"""A cron task's schedule outlives any single run, so only what went
|
|
wrong is recorded."""
|
|
await tasks.add(make_task(status=TaskStatus.ENABLED, schedule=CRON))
|
|
|
|
await tasks.record_completion("task-1", user_id="user-1", status=None, error="boom")
|
|
|
|
task = await tasks.get("task-1", user_id="user-1")
|
|
assert task.status is TaskStatus.ENABLED
|
|
assert task.last_error == "boom"
|
|
|
|
async def test_never_rolls_back_a_concurrent_launch_write(self, tasks):
|
|
"""The regression this method exists to prevent.
|
|
|
|
A fast-failing run reaches the completion hook while the dispatch path
|
|
is still writing its bookkeeping. Whichever lands second must not undo
|
|
the other: the launch owns the schedule, the completion owns the
|
|
verdict.
|
|
"""
|
|
await tasks.add(make_task(next_run_at=NOW - timedelta(minutes=1)))
|
|
await tasks.claim_due(now=NOW, lease_seconds=120, limit=10)
|
|
|
|
next_at = NOW + timedelta(days=1)
|
|
await tasks.record_launch(
|
|
"task-1",
|
|
status=TaskStatus.ENABLED,
|
|
next_run_at=next_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,
|
|
)
|
|
|
|
await tasks.record_completion("task-1", user_id="user-1", status=None, error="boom")
|
|
|
|
task = await tasks.get("task-1", user_id="user-1")
|
|
assert task.next_run_at == next_at, "the launch path's next fire time must survive"
|
|
assert task.run_count == 1, "the launch path's run count must survive"
|
|
assert task.last_run_id == "run-1"
|
|
assert task.last_thread_id == "thread-1"
|
|
assert task.last_error == "boom", "the completion still records its verdict"
|
|
# The whole point: the task is still reachable by the next poll.
|
|
claimed = await tasks.claim_due(now=NOW + timedelta(days=2), lease_seconds=120, limit=10)
|
|
assert [t.task_id for t in claimed] == ["task-1"]
|
|
|
|
async def test_another_users_task_is_untouched(self, tasks):
|
|
await tasks.add(make_task(status=TaskStatus.ENABLED))
|
|
|
|
await tasks.record_completion("task-1", user_id="someone-else", status=TaskStatus.FAILED, error="boom")
|
|
|
|
task = await tasks.get("task-1", user_id="user-1")
|
|
assert task.status is TaskStatus.ENABLED
|
|
assert task.last_error is None
|
|
|
|
async def test_unknown_task_is_ignored(self, tasks):
|
|
await tasks.record_completion("nope", user_id="user-1", status=TaskStatus.FAILED, error="boom")
|
|
|
|
|
|
class TestOptimisticConcurrency:
|
|
"""`save` is a compare-and-set on `version`; every other committed write
|
|
bumps it.
|
|
|
|
This is what stops a read-modify-write from rolling back the fields the
|
|
dispatch and completion paths own. The API reads a task, the user edits the
|
|
title, and meanwhile the poller launches it -- committing the edited
|
|
snapshot wholesale would restore the pre-launch `next_run_at` and
|
|
`run_count`, re-arming an occurrence that already ran.
|
|
|
|
Atomicity is still out of scope here (see the module docstring): these
|
|
cases interleave by hand, single-threaded. What they pin is that both
|
|
implementations agree on *when* a write is refused.
|
|
"""
|
|
|
|
async def test_a_stored_task_starts_at_version_zero(self, tasks):
|
|
stored = await tasks.add(make_task())
|
|
assert stored.version == 0
|
|
|
|
async def test_save_returns_the_incremented_version(self, tasks):
|
|
from dataclasses import replace
|
|
|
|
stored = await tasks.add(make_task())
|
|
saved = await tasks.save(replace(stored, title="renamed"))
|
|
assert saved.version == stored.version + 1
|
|
|
|
async def test_a_second_save_of_a_stale_snapshot_is_refused(self, tasks):
|
|
from dataclasses import replace
|
|
|
|
stored = await tasks.add(make_task())
|
|
await tasks.save(replace(stored, title="first"))
|
|
|
|
with pytest.raises(ConcurrentUpdateError):
|
|
await tasks.save(replace(stored, title="second"))
|
|
|
|
async def test_the_refused_write_leaves_the_stored_state_alone(self, tasks):
|
|
from dataclasses import replace
|
|
|
|
stored = await tasks.add(make_task())
|
|
await tasks.save(replace(stored, title="first"))
|
|
|
|
with pytest.raises(ConcurrentUpdateError):
|
|
await tasks.save(replace(stored, title="second"))
|
|
|
|
reloaded = await tasks.get(stored.task_id, user_id="user-1")
|
|
assert reloaded.title == "first"
|
|
|
|
async def test_record_launch_invalidates_an_earlier_read(self, tasks):
|
|
"""The race the token exists for: a snapshot read before the dispatch
|
|
path committed must not be saveable afterwards."""
|
|
from dataclasses import replace
|
|
|
|
stored = await tasks.add(make_task(next_run_at=NOW))
|
|
await tasks.record_launch(
|
|
stored.task_id,
|
|
status=TaskStatus.ENABLED,
|
|
next_run_at=NOW + timedelta(days=1),
|
|
last_run_at=NOW,
|
|
last_run_id="run-1",
|
|
last_thread_id="thread-1",
|
|
last_error=None,
|
|
increment_run_count=True,
|
|
)
|
|
|
|
with pytest.raises(ConcurrentUpdateError):
|
|
await tasks.save(replace(stored, title="renamed"))
|
|
|
|
reloaded = await tasks.get(stored.task_id, user_id="user-1")
|
|
assert reloaded.run_count == 1, "the launch write survived"
|
|
assert reloaded.last_run_id == "run-1"
|
|
|
|
async def test_record_completion_invalidates_an_earlier_read(self, tasks):
|
|
from dataclasses import replace
|
|
|
|
stored = await tasks.add(make_task())
|
|
await tasks.record_completion(stored.task_id, user_id="user-1", status=None, error="boom")
|
|
|
|
with pytest.raises(ConcurrentUpdateError):
|
|
await tasks.save(replace(stored, title="renamed"))
|
|
|
|
async def test_claim_due_invalidates_an_earlier_read(self, tasks):
|
|
from dataclasses import replace
|
|
|
|
stored = await tasks.add(make_task(next_run_at=NOW - timedelta(minutes=1)))
|
|
claimed = await tasks.claim_due(now=NOW, lease_seconds=60, limit=10)
|
|
assert [task.task_id for task in claimed] == [stored.task_id]
|
|
|
|
with pytest.raises(ConcurrentUpdateError):
|
|
await tasks.save(replace(stored, title="renamed"))
|
|
|
|
async def test_a_freshly_read_snapshot_saves_again(self, tasks):
|
|
"""The token refuses stale writes, not repeated ones -- re-reading is
|
|
the whole recovery path, so it has to work."""
|
|
from dataclasses import replace
|
|
|
|
await tasks.add(make_task())
|
|
for title in ("first", "second", "third"):
|
|
current = await tasks.get("task-1", user_id="user-1")
|
|
await tasks.save(replace(current, title=title))
|
|
|
|
assert (await tasks.get("task-1", user_id="user-1")).title == "third"
|
|
|
|
async def test_an_absent_task_is_none_not_a_conflict(self, tasks):
|
|
""" "No such task" and "someone else wrote first" are different answers;
|
|
the caller maps them to 404 and 409 respectively."""
|
|
assert await tasks.save(make_task(task_id="task-absent")) is None
|
|
|
|
async def test_another_users_task_is_none_not_a_conflict(self, tasks):
|
|
await tasks.add(make_task())
|
|
stored = await tasks.get("task-1", user_id="user-1")
|
|
|
|
from dataclasses import replace
|
|
|
|
assert await tasks.save(replace(stored, user_id="user-2")) is None
|
|
|
|
|
|
class TestCancelStuckOnceTasks:
|
|
async def test_cancels_a_launched_once_task_with_no_claim(self, tasks):
|
|
"""Launched, so the claim was released; the completion hook then died
|
|
with the process. Expired-claim reclaim can never see this one."""
|
|
await tasks.add(make_task(status=TaskStatus.RUNNING, schedule=ONCE))
|
|
|
|
assert await tasks.cancel_stuck_once_tasks(error="restarted") == 1
|
|
task = await tasks.get("task-1", user_id="user-1")
|
|
assert task.status is TaskStatus.CANCELLED
|
|
assert task.last_error == "restarted"
|
|
|
|
async def test_leaves_a_claimed_task_to_claim_expiry(self, tasks):
|
|
"""Claimed but not launched -- expired-claim reclaim recovers it, and
|
|
cancelling here would throw away a dispatch that never happened."""
|
|
await seed(
|
|
tasks,
|
|
make_task(status=TaskStatus.RUNNING, schedule=ONCE, next_run_at=NOW - timedelta(minutes=1)),
|
|
claimed_until=NOW + timedelta(seconds=60),
|
|
)
|
|
assert await tasks.cancel_stuck_once_tasks(error="restarted") == 0
|
|
|
|
async def test_leaves_cron_tasks_alone(self, tasks):
|
|
await tasks.add(make_task(status=TaskStatus.RUNNING, schedule=CRON))
|
|
assert await tasks.cancel_stuck_once_tasks(error="restarted") == 0
|
|
|
|
|
|
class TestActiveSlot:
|
|
def _queued(self, task_id: str = "task-1") -> ScheduledRun:
|
|
return ScheduledRun.queued(task_id=task_id, thread_id="thread-1", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED)
|
|
|
|
def _tombstone(self, task_id: str = "task-1") -> ScheduledRun:
|
|
return ScheduledRun.skipped_tombstone(task_id=task_id, thread_id="thread-1", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED)
|
|
|
|
async def test_second_active_record_is_refused(self, runs):
|
|
await runs.add(self._queued())
|
|
with pytest.raises(ActiveRunConflictError):
|
|
await runs.add(self._queued())
|
|
|
|
async def test_a_tombstone_never_conflicts(self, runs):
|
|
"""Terminal from birth, so it sits outside the active-slot rule -- this
|
|
is why the skip path cannot reuse the queued factory."""
|
|
await runs.add(self._queued())
|
|
await runs.add(self._tombstone())
|
|
assert await runs.count_active() == 1
|
|
|
|
async def test_another_task_is_unaffected(self, runs):
|
|
await runs.add(self._queued("task-1"))
|
|
await runs.add(self._queued("task-2"))
|
|
assert await runs.count_active() == 2
|
|
|
|
async def test_slot_frees_up_once_the_record_terminalizes(self, runs):
|
|
first = await runs.add(self._queued())
|
|
await runs.update_status(first.record_id, status=RunStatus.SUCCESS, finished_at=NOW)
|
|
await runs.add(self._queued())
|
|
assert await runs.count_active() == 1
|
|
|
|
async def test_has_active_is_scoped_to_one_task_while_count_is_global(self, runs):
|
|
await runs.add(self._queued("task-1"))
|
|
await runs.add(self._queued("task-2"))
|
|
assert await runs.has_active("task-1") is True
|
|
assert await runs.has_active("task-3") is False
|
|
assert await runs.count_active() == 2
|
|
|
|
async def test_a_run_round_trips(self, runs):
|
|
stored = await runs.add(self._queued())
|
|
listed = await runs.list_by_task("task-1", limit=10, offset=0)
|
|
assert [r.record_id for r in listed] == [stored.record_id]
|
|
assert listed[0].trigger is TriggerKind.SCHEDULED
|
|
assert listed[0].scheduled_for.tzinfo is not None
|
|
|
|
|
|
class TestRunStatusWrites:
|
|
async def _one_queued(self, runs) -> ScheduledRun:
|
|
return await runs.add(ScheduledRun.queued(task_id="t", thread_id="th", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED))
|
|
|
|
async def test_protect_terminal_backfills_without_overwriting(self, runs):
|
|
run = await self._one_queued(runs)
|
|
await runs.update_status(run.record_id, status=RunStatus.FAILED, error="boom", finished_at=NOW)
|
|
|
|
# The launch path's write arrives late.
|
|
await runs.update_status(run.record_id, status=RunStatus.RUNNING, run_id="run-1", started_at=NOW, protect_terminal=True)
|
|
|
|
stored = (await runs.list_by_task("t", limit=10, offset=0))[0]
|
|
assert stored.status is RunStatus.FAILED
|
|
assert stored.error == "boom"
|
|
assert stored.run_id == "run-1", "the id the completion could not know is backfilled"
|
|
assert stored.started_at == NOW
|
|
|
|
async def test_unknown_record_is_ignored(self, runs):
|
|
await runs.update_status("nope", status=RunStatus.SUCCESS)
|
|
|
|
async def test_mark_stale_active_terminalizes_orphans(self, runs):
|
|
active = await self._one_queued(runs)
|
|
done = await runs.add(ScheduledRun.skipped_tombstone(task_id="t", thread_id="th", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED))
|
|
|
|
assert await runs.mark_stale_active(error="gateway restarted") == 1
|
|
|
|
by_id = {run.record_id: run for run in await runs.list_by_task("t", limit=10, offset=0)}
|
|
assert by_id[active.record_id].status is RunStatus.INTERRUPTED
|
|
assert by_id[active.record_id].error == "gateway restarted"
|
|
assert by_id[done.record_id].status is RunStatus.SKIPPED
|
|
|
|
|
|
class TestLauncherAndThreadLookup:
|
|
"""Fake-only: these two ports have no SQL implementation -- one starts a
|
|
run and the other asks the thread store, so both land in later adapters."""
|
|
|
|
async def test_launcher_records_the_call_and_echoes_the_thread(self):
|
|
launcher = FakeRunLauncher()
|
|
launched = await launcher.launch(
|
|
thread_id="thread-1",
|
|
assistant_id="lead_agent",
|
|
prompt="go",
|
|
owner_user_id="user-1",
|
|
metadata={"scheduled_task_id": "task-1"},
|
|
)
|
|
assert launched.thread_id == "thread-1"
|
|
assert launcher.calls[0]["metadata"] == {"scheduled_task_id": "task-1"}
|
|
|
|
async def test_launcher_can_be_driven_into_either_failure_branch(self):
|
|
boom = RuntimeError("nope")
|
|
launcher = FakeRunLauncher(fail_with=boom)
|
|
with pytest.raises(RuntimeError):
|
|
await launcher.launch(thread_id="t", assistant_id=None, prompt="p", owner_user_id=None, metadata={})
|
|
assert len(launcher.calls) == 1, "the attempt is still recorded"
|
|
|
|
async def test_thread_lookup_requires_both_existence_and_ownership(self):
|
|
lookup = FakeThreadLookup({"thread-1": "user-1"})
|
|
assert await lookup.exists_for_user("thread-1", "user-1") is True
|
|
assert await lookup.exists_for_user("thread-1", "user-2") is False
|
|
assert await lookup.exists_for_user("missing", "user-1") is False
|