github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-08-01 19:06:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(schedule): implement the persistence ports in SQL

Browse Source 
Two secondary adapters plus the spec mapping they share, and the first
real payoff of the port boundary: test_schedule_fakes.py becomes a
contract suite that runs all 31 cases against both the in-memory doubles
and the SQL adapters on a real sqlite file. A rule stated in a port
docstring now has to hold for both, and a divergence is a failure rather
than a surprise in production.

The queries come over unchanged. The claim statement's FOR UPDATE SKIP
LOCKED and both protect_terminal conditional writes are this module's
concurrency contract, not style, and the IntegrityError translation in
add() is what lets the service collapse a lost active-slot race into the
same outcome as its own non-atomic fast path.

Two things the adapters own that the domain deliberately does not. The
claiming process's identity is generated here, because who claimed a
task is an identity rather than a rule and nothing reads it back. And
`Unsupported schedule_type` is raised by the mapping, because
ScheduleSpec only accepts the enum -- structural checks belong to the
boundary, value rules to __post_init__, and both surface as the same
domain error so the router maps one family.

_to_domain introduces a failure mode the legacy repository did not have:
a row whose stored schedule no longer parses. Single-row reads let it
propagate; list reads skip and log, so one corrupt row cannot 500 an
entire listing.

The contract suite reaches past the port in exactly one place -- seeding
a task that already carries a claim, a shape claim_due would never
produce -- and says so where it does.
This commit is contained in:
rayhpeng 2026-07-28 14:46:13 +08:00
parent b56e939c4b
commit d4c24e3f6f
6 changed files with 870 additions and 151 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

181
backend/app/infra/persistence/scheduled_task_runs.py Normal file
View File

@ -0,0 +1,181 @@
"""SQL adapter for the schedule context's execution-record repository.
Secondary adapter implementing `ScheduledRunRepository`. Queries are migrated
unchanged from the legacy repository.
The load-bearing piece is the `IntegrityError` translation in `add`: the
partial unique index `uq_scheduled_task_run_active` is the atomic arbiter of
"at most one active run per task", and turning its rejection into
`ActiveRunConflictError` is what lets the service collapse a lost race into
exactly the same outcome as its own non-atomic fast path.
"""
from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.domain.schedule.model import (
ACTIVE_RUN_STATUSES,
TERMINAL_RUN_STATUSES,
ActiveRunConflictError,
RunStatus,
ScheduledRun,
TriggerKind,
)
from deerflow.domain.schedule.ports import ScheduledRunRepository
# Transitional: the ORM row stays in the harness until engine, models, and
# migrations move into app/infra together.
from deerflow.persistence.scheduled_task_runs.model import ScheduledTaskRunRow
_ACTIVE_STATUS_VALUES = tuple(str(status) for status in ACTIVE_RUN_STATUSES)
def _tz_aware(value: datetime | None) -> datetime | None:
"""SQLite drops tzinfo on read; stored values are always UTC."""
return value if value is None or value.tzinfo is not None else value.replace(tzinfo=UTC)
class SqlScheduledRunRepository(ScheduledRunRepository):
"""SQL implementation of the `ScheduledRunRepository` port."""
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
self._sf = session_factory
@staticmethod
def _to_domain(row: ScheduledTaskRunRow) -> ScheduledRun:
return ScheduledRun(
record_id=row.id,
task_id=row.task_id,
thread_id=row.thread_id,
scheduled_for=_tz_aware(row.scheduled_for),
trigger=TriggerKind(row.trigger),
status=RunStatus(row.status),
run_id=row.run_id,
error=row.error,
started_at=_tz_aware(row.started_at),
finished_at=_tz_aware(row.finished_at),
created_at=_tz_aware(row.created_at),
)
async def add(self, run: ScheduledRun) -> ScheduledRun:
row = ScheduledTaskRunRow(
id=run.record_id,
task_id=run.task_id,
thread_id=run.thread_id,
run_id=run.run_id,
scheduled_for=run.scheduled_for,
trigger=str(run.trigger),
status=str(run.status),
error=run.error,
started_at=run.started_at,
finished_at=run.finished_at,
created_at=run.created_at,
)
async with self._sf() as session:
session.add(row)
try:
await session.commit()
except IntegrityError:
await session.rollback()
# Only an active-status insert can trip the partial unique
# index; a terminal row (e.g. a skip tombstone) is outside its
# predicate and cannot conflict, so an IntegrityError there is
# a genuine fault and is re-raised untranslated.
if run.is_active:
raise ActiveRunConflictError(f"scheduled task {run.task_id!r} already has an active run") from None
raise
await session.refresh(row)
return self._to_domain(row)
async def list_by_task(self, task_id: str, *, limit: int = 50, offset: int = 0) -> list[ScheduledRun]:
stmt = (
select(ScheduledTaskRunRow)
.where(ScheduledTaskRunRow.task_id == task_id)
.order_by(
ScheduledTaskRunRow.created_at.desc(),
ScheduledTaskRunRow.id.desc(),
)
.limit(limit)
.offset(offset)
)
async with self._sf() as session:
result = await session.execute(stmt)
return [self._to_domain(row) for row in result.scalars()]
async def count_active(self) -> int:
"""Global count of active rows, used to bound cross-task concurrency."""
stmt = select(func.count()).select_from(ScheduledTaskRunRow).where(ScheduledTaskRunRow.status.in_(_ACTIVE_STATUS_VALUES))
async with self._sf() as session:
result = await session.execute(stmt)
return int(result.scalar() or 0)
async def has_active(self, task_id: str) -> bool:
stmt = (
select(ScheduledTaskRunRow.id)
.where(
ScheduledTaskRunRow.task_id == task_id,
ScheduledTaskRunRow.status.in_(_ACTIVE_STATUS_VALUES),
)
.limit(1)
)
async with self._sf() as session:
result = await session.execute(stmt)
return result.scalars().first() is not None
async def update_status(
self,
record_id: str,
*,
status: RunStatus,
run_id: str | None = None,
error: str | None = None,
started_at: datetime | None = None,
finished_at: datetime | None = None,
protect_terminal: bool = False,
) -> None:
async with self._sf() as session:
row = await session.get(ScheduledTaskRunRow, record_id)
if row is None:
return
if protect_terminal and RunStatus(row.status) in TERMINAL_RUN_STATUSES:
# The launch-path "running" write lost the race against the
# completion hook; keep the terminal status/error and only
# backfill bookkeeping the completion write could not know.
if row.run_id is None and run_id is not None:
row.run_id = run_id
if row.started_at is None and started_at is not None:
row.started_at = started_at
await session.commit()
return
row.status = str(status)
row.run_id = run_id
row.error = error
if started_at is not None:
row.started_at = started_at
if finished_at is not None:
row.finished_at = finished_at
await session.commit()
async def mark_stale_active(self, *, error: str) -> int:
"""Fail-fast bookkeeping for runs orphaned by a process crash.
Agent runs execute in-process, so any active row found at scheduler
startup belongs to a run whose process is gone. Only valid under the
single-scheduler-instance assumption.
"""
stmt = select(ScheduledTaskRunRow).where(ScheduledTaskRunRow.status.in_(_ACTIVE_STATUS_VALUES))
now = datetime.now(UTC)
async with self._sf() as session:
result = await session.execute(stmt)
rows = list(result.scalars())
for row in rows:
row.status = str(RunStatus.INTERRUPTED)
row.error = error
row.finished_at = now
await session.commit()
return len(rows)

281
backend/app/infra/persistence/scheduled_tasks.py Normal file
View File

@ -0,0 +1,281 @@
"""SQL adapter for the schedule context's task repository.
Secondary adapter implementing `ScheduledTaskRepository`. SQL/ORM vocabulary
stops at this file: methods exchange domain objects and normalize SQLite's
tz-naive reads.
**The queries are migrated unchanged from the legacy repository.** The claim
statement's `FOR UPDATE SKIP LOCKED` and the `protect_terminal` conditional
write are the module's concurrency contract, not style choices -- rewriting
them is how the invariants silently break.
"""
from __future__ import annotations
import logging
import socket
import uuid
from datetime import UTC, datetime, timedelta
from sqlalchemy import and_, or_, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from app.infra.schedule.spec_mapping import spec_to_domain, spec_to_wire
from deerflow.domain.schedule.model import (
TERMINAL_TASK_STATUSES,
ContextMode,
InvalidScheduleError,
ScheduledTask,
TaskStatus,
)
from deerflow.domain.schedule.ports import ScheduledTaskRepository
# Transitional: the ORM row stays in the harness until engine, models, and
# migrations move into app/infra together.
from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow
logger = logging.getLogger(__name__)
def _tz_aware(value: datetime | None) -> datetime | None:
"""SQLite drops tzinfo on read; stored values are always UTC."""
return value if value is None or value.tzinfo is not None else value.replace(tzinfo=UTC)
class SqlScheduledTaskRepository(ScheduledTaskRepository):
"""SQL implementation of the `ScheduledTaskRepository` port.
Explicit inheritance is a readability aid only: a missing method would
still instantiate fine (Protocol bodies are inherited), so the contract
test suite must cover every port method.
"""
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
self._sf = session_factory
# Diagnostics only. The domain deliberately does not carry this -- who
# claimed a task is an identity, not a rule, and nothing reads it back.
self._lease_owner = f"{socket.gethostname()}:{uuid.uuid4().hex}"
# ------------------------------------------------------------ conversion
@staticmethod
def _to_domain(row: ScheduledTaskRow) -> ScheduledTask:
"""ORM row -> aggregate.
Raises InvalidScheduleError for a row whose stored schedule can no
longer be parsed. That is a **new** failure mode: the legacy repository
returned bare dicts and a corrupt spec only surfaced at dispatch time.
Single-row reads let it propagate; list reads skip and log, so one bad
row cannot 500 an entire listing.
"""
return ScheduledTask(
task_id=row.id,
user_id=row.user_id,
title=row.title,
prompt=row.prompt,
schedule=spec_to_domain(row.schedule_type, row.schedule_spec, row.timezone),
context_mode=ContextMode(row.context_mode),
thread_id=row.thread_id,
assistant_id=row.assistant_id,
status=TaskStatus(row.status),
overlap_policy=row.overlap_policy,
next_run_at=_tz_aware(row.next_run_at),
last_run_at=_tz_aware(row.last_run_at),
last_run_id=row.last_run_id,
last_thread_id=row.last_thread_id,
last_error=row.last_error,
run_count=row.run_count,
created_at=_tz_aware(row.created_at),
updated_at=_tz_aware(row.updated_at),
)
@classmethod
def _to_domain_or_skip(cls, row: ScheduledTaskRow) -> ScheduledTask | None:
try:
return cls._to_domain(row)
except InvalidScheduleError:
logger.warning("Skipping scheduled task %s: stored schedule is unparseable", row.id, exc_info=True)
return None
@staticmethod
def _apply(row: ScheduledTaskRow, task: ScheduledTask) -> None:
"""Aggregate -> ORM row. Explicit field list: a new column stays
invisible here until it is deliberately mapped."""
row.user_id = task.user_id
row.title = task.title
row.prompt = task.prompt
row.schedule_type = str(task.schedule.schedule_type)
row.schedule_spec = spec_to_wire(task.schedule)
row.timezone = task.schedule.timezone
row.context_mode = str(task.context_mode)
row.thread_id = task.thread_id
row.assistant_id = task.assistant_id
row.status = str(task.status)
row.overlap_policy = task.overlap_policy
row.next_run_at = task.next_run_at
row.last_run_at = task.last_run_at
row.last_run_id = task.last_run_id
row.last_thread_id = task.last_thread_id
row.last_error = task.last_error
row.run_count = task.run_count
# ------------------------------------------------------------------ port
async def add(self, task: ScheduledTask) -> ScheduledTask:
now = datetime.now(UTC)
row = ScheduledTaskRow(id=task.task_id, created_at=now, updated_at=now)
self._apply(row, task)
async with self._sf() as session:
session.add(row)
await session.commit()
await session.refresh(row)
return self._to_domain(row)
async def get(self, task_id: str, *, user_id: str) -> ScheduledTask | None:
async with self._sf() as session:
row = await session.get(ScheduledTaskRow, task_id)
if row is None or row.user_id != user_id:
return None
return self._to_domain(row)
async def list_by_user(self, user_id: str) -> list[ScheduledTask]:
stmt = select(ScheduledTaskRow).where(ScheduledTaskRow.user_id == user_id).order_by(ScheduledTaskRow.created_at.desc(), ScheduledTaskRow.id.desc())
async with self._sf() as session:
result = await session.execute(stmt)
return [task for task in (self._to_domain_or_skip(row) for row in result.scalars()) if task is not None]
async def list_by_user_and_thread(self, user_id: str, thread_id: str) -> list[ScheduledTask]:
stmt = (
select(ScheduledTaskRow)
.where(
ScheduledTaskRow.user_id == user_id,
ScheduledTaskRow.thread_id == thread_id,
)
.order_by(ScheduledTaskRow.created_at.desc(), ScheduledTaskRow.id.desc())
)
async with self._sf() as session:
result = await session.execute(stmt)
return [task for task in (self._to_domain_or_skip(row) for row in result.scalars()) if task is not None]
async def save(self, task: ScheduledTask) -> ScheduledTask | None:
async with self._sf() as session:
row = await session.get(ScheduledTaskRow, task.task_id)
if row is None or row.user_id != task.user_id:
return None
self._apply(row, task)
row.updated_at = datetime.now(UTC)
await session.commit()
await session.refresh(row)
return self._to_domain(row)
async def delete(self, task_id: str, *, user_id: str) -> bool:
async with self._sf() as session:
row = await session.get(ScheduledTaskRow, task_id)
if row is None or row.user_id != user_id:
return False
await session.delete(row)
await session.commit()
return True
async def claim_due(self, *, now: datetime, lease_seconds: int, limit: int) -> list[ScheduledTask]:
lease_expires_at = now + timedelta(seconds=lease_seconds)
stmt = (
select(ScheduledTaskRow)
.where(
ScheduledTaskRow.next_run_at.is_not(None),
ScheduledTaskRow.next_run_at <= now,
or_(
and_(
ScheduledTaskRow.status == "enabled",
or_(
ScheduledTaskRow.lease_expires_at.is_(None),
ScheduledTaskRow.lease_expires_at < now,
),
),
# A task stuck in "running" with an expired lease means the
# claiming process died between claim and dispatch; it must
# stay reclaimable or the task is dead forever.
and_(
ScheduledTaskRow.status == "running",
ScheduledTaskRow.lease_expires_at.is_not(None),
ScheduledTaskRow.lease_expires_at < now,
),
),
)
.order_by(ScheduledTaskRow.next_run_at.asc(), ScheduledTaskRow.id.asc())
.limit(limit)
.with_for_update(skip_locked=True)
)
async with self._sf() as session:
result = await session.execute(stmt)
rows = list(result.scalars())
for row in rows:
row.lease_owner = self._lease_owner
row.lease_expires_at = lease_expires_at
row.status = "running"
row.updated_at = datetime.now(UTC)
await session.commit()
return [task for task in (self._to_domain_or_skip(row) for row in rows) if task is not None]
async def record_launch(
self,
task_id: str,
*,
status: TaskStatus,
next_run_at: datetime | None,
last_run_at: datetime | None,
last_run_id: str | None,
last_thread_id: str | None,
last_error: str | None,
increment_run_count: bool,
protect_terminal: bool = False,
) -> None:
async with self._sf() as session:
row = await session.get(ScheduledTaskRow, task_id)
if row is None:
return
if protect_terminal and TaskStatus(row.status) in TERMINAL_TASK_STATUSES:
# A fast-failing run can reach the completion hook (which
# finalizes a `once` task) before this launch-path write
# commits; keep the hook's status/error and only record the
# launch bookkeeping.
pass
else:
row.status = str(status)
row.last_error = last_error
row.next_run_at = next_run_at
row.last_run_at = last_run_at
row.last_run_id = last_run_id
row.last_thread_id = last_thread_id
if increment_run_count:
row.run_count += 1
row.lease_owner = None
row.lease_expires_at = None
row.updated_at = datetime.now(UTC)
await session.commit()
async def cancel_stuck_once_tasks(self, *, error: str) -> int:
"""Reconcile `once` tasks orphaned in `running` by a process crash.
A launched `once` task stays `running` until the in-process completion
hook moves it to a terminal status; its lease was cleared at launch, so
the claim query's expired-lease branch never sees it. After a crash the
hook is gone and the task would be stuck forever. Tasks still holding a
lease are left alone -- they were claimed but not launched, and
expired-lease reclaim recovers them safely.
"""
stmt = select(ScheduledTaskRow).where(
ScheduledTaskRow.schedule_type == "once",
ScheduledTaskRow.status == "running",
ScheduledTaskRow.lease_expires_at.is_(None),
)
async with self._sf() as session:
result = await session.execute(stmt)
rows = list(result.scalars())
now = datetime.now(UTC)
for row in rows:
row.status = "cancelled"
row.last_error = error
row.updated_at = now
await session.commit()
return len(rows)

1
backend/app/infra/schedule/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Secondary adapters for the schedule bounded context."""

68
backend/app/infra/schedule/spec_mapping.py Normal file
View File

@ -0,0 +1,68 @@
"""Wire/storage <-> ScheduleSpec mapping.
`schedule_spec` is both an HTTP request field and a JSON column -- one shape,
two boundaries -- so the mapping lives here once and both the router and the
SQL adapter import it, rather than the domain growing a `Mapping[str, Any]` in
its signatures.
The split is deliberate: **structural** checks (is the key present? is it a
str?) belong to this boundary, **value** rules (5-field cron, resolvable
timezone, run_at present) belong to `ScheduleSpec.__post_init__`. That is why
this module can look thin -- most of what could go wrong is caught one layer
in, and reported with the same domain error.
"""
from __future__ import annotations
from collections.abc import Mapping
from datetime import datetime
from typing import Any
from deerflow.domain.schedule.model import InvalidScheduleError, ScheduleSpec, ScheduleType
def spec_to_domain(schedule_type: str, spec: Mapping[str, Any] | None, timezone: str) -> ScheduleSpec:
"""Parse the stored/submitted triple into the value object.
Raises:
InvalidScheduleError: unknown schedule type, or the type's required key
is missing or not a string. Raising a *domain* error from an
adapter is intentional -- domain errors are the vocabulary the
outer ring uses to say "this violates a domain rule", and the
router maps this one family uniformly.
"""
try:
kind = ScheduleType(schedule_type)
except ValueError as exc:
raise InvalidScheduleError(f"Unsupported schedule_type: {schedule_type}") from exc
fields = spec or {}
if kind is ScheduleType.CRON:
raw_cron = fields.get("cron")
if not isinstance(raw_cron, str):
raise InvalidScheduleError("cron schedule requires schedule_spec.cron")
return ScheduleSpec.cron_schedule(raw_cron, timezone)
raw_run_at = fields.get("run_at")
if not isinstance(raw_run_at, str):
raise InvalidScheduleError("once schedule requires run_at")
try:
run_at = datetime.fromisoformat(raw_run_at)
except ValueError as exc:
raise InvalidScheduleError(f"once schedule has an unparseable run_at: {raw_run_at!r}") from exc
return ScheduleSpec.once_at(run_at, timezone)
def spec_to_wire(spec: ScheduleSpec) -> dict[str, str]:
"""Rebuild the persisted/wire JSON shape.
Note this normalizes the stored string rather than echoing the caller's
bytes: the frontend submits an already-UTC-aware ISO value
(`zonedLocalToUtcIso`), so a trailing-Z input round-trips out as "+00:00".
Both forms parse on either side, so the normalization is deliberate --
preferable to carrying the raw dict on the value object just to preserve
the exact input spelling.
"""
if spec.schedule_type is ScheduleType.CRON:
return {"cron": spec.cron or ""}
return {"run_at": spec.run_at.isoformat() if spec.run_at else ""}

412
backend/tests/test_schedule_fakes.py
View File

@ -1,21 +1,26 @@
"""Semantics of the in-memory schedule doubles.
"""Contract suite for the schedule repository ports.
These are the rules ``docs`` calls the port contract: what `claim_due` selects,
when `add` refuses, what `protect_terminal` preserves. They are asserted here
against the fakes so the doubles cannot drift from the ports they stand in for
while the service tests lean on them.
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.
This file is the seed of the contract suite: when the SQL adapters land, the
same cases run against both implementations and any divergence between them
becomes a failure rather than a surprise in production. Concurrency stays out
of scope in both tiers -- see the note in ``schedule_fakes``.
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_scheduled_task_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,
@ -23,7 +28,11 @@ from schedule_fakes import (
InMemoryScheduledTaskRepository,
)
from app.infra.persistence.scheduled_task_runs import SqlScheduledRunRepository
from app.infra.persistence.scheduled_tasks import SqlScheduledTaskRepository
from deerflow.config.database_config import DatabaseConfig
from deerflow.domain.schedule.model import (
ACTIVE_RUN_STATUSES,
ActiveRunConflictError,
ContextMode,
RunStatus,
@ -34,6 +43,7 @@ from deerflow.domain.schedule.model import (
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
@ -42,6 +52,67 @@ 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",
*,
@ -67,60 +138,107 @@ def make_task(
class TestProtocolConformance:
"""`runtime_checkable` only checks that the methods exist, not their
signatures -- enough to catch a rename that updates one side only.
signatures -- enough to catch a rename that updates one side only."""
Declared async purely to match this module's asyncio mark; there is
nothing to await.
"""
async def test_task_repository_satisfies_the_port(self):
assert isinstance(InMemoryScheduledTaskRepository(), ScheduledTaskRepository)
async def test_run_repository_satisfies_the_port(self):
assert isinstance(InMemoryScheduledRunRepository(), ScheduledRunRepository)
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):
repo = InMemoryScheduledTaskRepository()
await repo.add(make_task(user_id="owner"))
assert await repo.get("task-1", user_id="owner") is not None
assert await repo.get("task-1", user_id="intruder") is None
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):
repo = InMemoryScheduledTaskRepository()
await repo.add(make_task(user_id="owner"))
assert await repo.save(make_task(user_id="intruder")) is None
assert (await repo.get("task-1", user_id="owner")).user_id == "owner"
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):
repo = InMemoryScheduledTaskRepository()
await repo.add(make_task(user_id="owner"))
assert await repo.delete("task-1", user_id="intruder") is False
assert await repo.get("task-1", user_id="owner") is not None
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_thread_only_matches_bound_tasks(self):
repo = InMemoryScheduledTaskRepository()
await repo.add(make_task("bound", context_mode=ContextMode.REUSE_THREAD, thread_id="thread-1"))
await repo.add(make_task("unbound"))
listed = await repo.list_by_user_and_thread("user-1", "thread-1")
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 TestClaimDue:
async def test_claims_an_enabled_due_task_and_stamps_the_lease(self):
repo = InMemoryScheduledTaskRepository()
await repo.add(make_task(next_run_at=NOW - timedelta(minutes=1)))
class TestRoundTrip:
"""What goes in comes back out -- including the value object, which the
SQL side has to rebuild from three separate columns."""
claimed = await repo.claim_due(now=NOW, lease_seconds=120, limit=10)
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_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
owner, expires = repo.lease_of("task-1")
assert owner is not None, "an implementation may record who claimed; nothing reads it back"
assert expires == NOW + timedelta(seconds=120)
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"),
@ -131,54 +249,41 @@ class TestClaimDue:
("completed", {"status": TaskStatus.COMPLETED, "next_run_at": NOW - timedelta(minutes=1)}),
],
)
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_seconds=120, limit=10) == [], label
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_does_not_steal_a_live_claim(self):
repo = InMemoryScheduledTaskRepository()
repo.seed(
make_task(next_run_at=NOW - timedelta(minutes=1)),
lease_owner="worker-1",
lease_expires_at=NOW + timedelta(seconds=60),
)
assert await repo.claim_due(now=NOW, lease_seconds=120, limit=10) == []
async def test_reclaims_a_task_stuck_mid_dispatch(self):
async def test_reclaims_a_task_stuck_mid_dispatch(self, tasks):
"""The claimer died between claiming and launching: status is running,
the lease has expired, and the task must not stay unreachable."""
repo = InMemoryScheduledTaskRepository()
repo.seed(
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)),
lease_owner="dead-worker",
lease_expires_at=NOW - timedelta(seconds=1),
claimed_until=NOW - timedelta(seconds=1),
)
claimed = await repo.claim_due(now=NOW, lease_seconds=120, limit=10)
claimed = await tasks.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")[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)))
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 repo.claim_due(now=NOW, lease_seconds=120, limit=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_releases_the_claim(self):
repo = InMemoryScheduledTaskRepository()
repo.seed(make_task(next_run_at=NOW), lease_owner="w", lease_expires_at=NOW + timedelta(seconds=60))
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 repo.record_launch(
await tasks.record_launch(
"task-1",
status=TaskStatus.ENABLED,
next_run_at=NOW + timedelta(days=1),
next_run_at=NOW - timedelta(seconds=1),
last_run_at=NOW,
last_run_id="run-1",
last_thread_id="thread-1",
@ -186,19 +291,19 @@ class TestRecordLaunch:
increment_run_count=True,
)
task = await repo.get("task-1", user_id="user-1")
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
assert repo.lease_of("task-1") == (None, None)
# 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):
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."""
repo = InMemoryScheduledTaskRepository()
await repo.add(make_task(status=TaskStatus.COMPLETED, schedule=ONCE))
await tasks.add(make_task(status=TaskStatus.COMPLETED, schedule=ONCE))
await repo.record_launch(
await tasks.record_launch(
"task-1",
status=TaskStatus.RUNNING,
next_run_at=None,
@ -210,17 +315,16 @@ class TestRecordLaunch:
protect_terminal=True,
)
task = await repo.get("task-1", user_id="user-1")
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):
repo = InMemoryScheduledTaskRepository()
await repo.add(make_task(status=TaskStatus.COMPLETED, schedule=ONCE))
async def test_without_protect_terminal_the_write_wins(self, tasks):
await tasks.add(make_task(status=TaskStatus.COMPLETED, schedule=ONCE))
await repo.record_launch(
await tasks.record_launch(
"task-1",
status=TaskStatus.FAILED,
next_run_at=None,
@ -231,13 +335,12 @@ class TestRecordLaunch:
increment_run_count=False,
)
task = await repo.get("task-1", user_id="user-1")
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):
repo = InMemoryScheduledTaskRepository()
await repo.record_launch(
async def test_unknown_task_is_ignored(self, tasks):
await tasks.record_launch(
"nope",
status=TaskStatus.ENABLED,
next_run_at=None,
@ -250,30 +353,29 @@ class TestRecordLaunch:
class TestCancelStuckOnceTasks:
async def test_cancels_a_launched_once_task_with_no_claim(self):
repo = InMemoryScheduledTaskRepository()
repo.seed(make_task(status=TaskStatus.RUNNING, schedule=ONCE))
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 repo.cancel_stuck_once_tasks(error="restarted") == 1
task = await repo.get("task-1", user_id="user-1")
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_lease_expiry(self):
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."""
repo = InMemoryScheduledTaskRepository()
repo.seed(
make_task(status=TaskStatus.RUNNING, schedule=ONCE),
lease_owner="w",
lease_expires_at=NOW + timedelta(seconds=60),
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 repo.cancel_stuck_once_tasks(error="restarted") == 0
assert await tasks.cancel_stuck_once_tasks(error="restarted") == 0
async def test_leaves_cron_tasks_alone(self):
repo = InMemoryScheduledTaskRepository()
repo.seed(make_task(status=TaskStatus.RUNNING, schedule=CRON))
assert await repo.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:
@ -283,75 +385,83 @@ class TestActiveSlot:
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):
repo = InMemoryScheduledRunRepository()
await repo.add(self._queued())
async def test_second_active_record_is_refused(self, runs):
await runs.add(self._queued())
with pytest.raises(ActiveRunConflictError):
await repo.add(self._queued())
await runs.add(self._queued())
async def test_a_tombstone_never_conflicts(self):
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."""
repo = InMemoryScheduledRunRepository()
await repo.add(self._queued())
await repo.add(self._tombstone())
assert await repo.count_active() == 1
await runs.add(self._queued())
await runs.add(self._tombstone())
assert await runs.count_active() == 1
async def test_another_task_is_unaffected(self):
repo = InMemoryScheduledRunRepository()
await repo.add(self._queued("task-1"))
await repo.add(self._queued("task-2"))
assert await repo.count_active() == 2
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):
repo = InMemoryScheduledRunRepository()
first = await repo.add(self._queued())
await repo.update_status(first.record_id, status=RunStatus.SUCCESS, finished_at=NOW)
await repo.add(self._queued())
assert await repo.count_active() == 1
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):
repo = InMemoryScheduledRunRepository()
await repo.add(self._queued("task-1"))
await repo.add(self._queued("task-2"))
assert await repo.has_active("task-1") is True
assert await repo.has_active("task-3") is False
assert await repo.count_active() == 2
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 test_protect_terminal_backfills_without_overwriting(self):
repo = InMemoryScheduledRunRepository()
run = await repo.add(ScheduledRun.queued(task_id="t", thread_id="th", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED))
await repo.update_status(run.record_id, status=RunStatus.FAILED, error="boom", finished_at=NOW)
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 repo.update_status(run.record_id, status=RunStatus.RUNNING, run_id="run-1", started_at=NOW, protect_terminal=True)
await runs.update_status(run.record_id, status=RunStatus.RUNNING, run_id="run-1", started_at=NOW, protect_terminal=True)
stored = repo.all_runs()[0]
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):
repo = InMemoryScheduledRunRepository()
await repo.update_status("nope", status=RunStatus.SUCCESS)
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):
repo = InMemoryScheduledRunRepository()
active = await repo.add(ScheduledRun.queued(task_id="t", thread_id="th", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED))
done = await repo.add(ScheduledRun.skipped_tombstone(task_id="t", thread_id="th", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED))
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 repo.mark_stale_active(error="gateway restarted") == 1
assert await runs.mark_stale_active(error="gateway restarted") == 1
by_id = {run.record_id: run for run in repo.all_runs()}
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
async def test_active_statuses_are_exactly_queued_and_running(self):
assert ACTIVE_RUN_STATUSES == (RunStatus.QUEUED, RunStatus.RUNNING)
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(

78
backend/tests/test_schedule_spec_mapping.py Normal file
View File

@ -0,0 +1,78 @@
"""Boundary tests for the schedule_spec mapping.
This is where a `Mapping[str, Any]` is allowed to exist. The split it enforces
is the point: **structural** problems (key missing, wrong type, unknown
schedule type) are caught here, **value** problems (5-field cron, resolvable
timezone) are left to `ScheduleSpec.__post_init__` -- and both surface as the
same domain error, so the router maps one family.
"""
from __future__ import annotations
from datetime import UTC, datetime
import pytest
from app.infra.schedule.spec_mapping import spec_to_domain, spec_to_wire
from deerflow.domain.schedule.model import InvalidScheduleError, ScheduleSpec, ScheduleType
class TestStructuralChecks:
def test_unknown_schedule_type_is_rejected(self):
"""The one rule the domain cannot state: ScheduleSpec only accepts the
enum, so a bad string has to be caught at the boundary."""
with pytest.raises(InvalidScheduleError, match="Unsupported schedule_type"):
spec_to_domain("teleport", {"cron": "0 9 * * *"}, "UTC")
@pytest.mark.parametrize("spec", [{}, None, {"cron": 5}, {"run_at": "..."}])
def test_cron_without_a_string_cron_is_rejected(self, spec):
with pytest.raises(InvalidScheduleError, match="requires schedule_spec"):
spec_to_domain("cron", spec, "UTC")
@pytest.mark.parametrize("spec", [{}, None, {"run_at": 5}, {"cron": "0 9 * * *"}])
def test_once_without_a_string_run_at_is_rejected(self, spec):
with pytest.raises(InvalidScheduleError, match="requires run_at"):
spec_to_domain("once", spec, "UTC")
def test_an_unparseable_run_at_is_rejected(self):
with pytest.raises(InvalidScheduleError, match="unparseable run_at"):
spec_to_domain("once", {"run_at": "next tuesday"}, "UTC")
class TestValueChecksStayInTheDomain:
"""These are not re-implemented here -- they arrive from __post_init__,
which is why the boundary can stay thin."""
def test_a_bad_cron_expression_still_raises(self):
with pytest.raises(InvalidScheduleError, match="exactly 5 fields"):
spec_to_domain("cron", {"cron": "0 9 * *"}, "UTC")
def test_a_bad_timezone_still_raises(self):
with pytest.raises(InvalidScheduleError, match="Unknown timezone"):
spec_to_domain("cron", {"cron": "0 9 * * *"}, "Mars/Olympus_Mons")
class TestRoundTrip:
def test_cron_round_trips_normalized(self):
spec = spec_to_domain("cron", {"cron": " 0 9 * * * "}, "Asia/Shanghai")
assert spec.schedule_type is ScheduleType.CRON
assert spec_to_wire(spec) == {"cron": "0 9 * * *"}
assert spec_to_domain("cron", spec_to_wire(spec), "Asia/Shanghai") == spec
def test_once_round_trips_to_the_same_instant(self):
spec = spec_to_domain("once", {"run_at": "2026-08-01T09:00:00"}, "Asia/Shanghai")
assert spec.run_at == datetime(2026, 8, 1, 1, 0, tzinfo=UTC)
assert spec_to_domain("once", spec_to_wire(spec), "Asia/Shanghai") == spec
def test_a_trailing_z_input_parses_and_re_emits_as_an_offset(self):
"""The shape the frontend actually submits (`zonedLocalToUtcIso`).
Both spellings parse on either side, so normalizing is deliberate."""
spec = spec_to_domain("once", {"run_at": "2026-08-01T01:00:00Z"}, "Asia/Shanghai")
emitted = spec_to_wire(spec)["run_at"]
assert emitted.endswith("+00:00")
assert spec_to_domain("once", {"run_at": emitted}, "Asia/Shanghai") == spec
def test_wire_output_is_idempotent(self):
spec = ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0, tzinfo=UTC), "UTC")
once = spec_to_wire(spec)
assert spec_to_wire(spec_to_domain("once", once, "UTC")) == once