From a6388bc967e1747ed65086f24dea956efc1eaa4d Mon Sep 17 00:00:00 2001 From: rayhpeng Date: Wed, 29 Jul 2026 19:49:48 +0800 Subject: [PATCH] fix(schedule): give each bookkeeping timestamp a single owner created_at was minted twice for one construction: the aggregate's field default read the clock, then the adapter's add() read it again and stored its own value -- so the aggregate a caller held disagreed with the stored row by however long the insert took, and the model docstring pointed at a write path that no longer exists. ScheduledTask.create now stamps created_at and updated_at from the explicit now= it already receives (one clock reading, the rule input), and the adapter persists the aggregate's instants verbatim on insert -- matching the feedback reference implementation. updated_at's later life stays deliberately with the adapter's write paths: the CAS methods (record_launch / record_completion) write the row without ever holding an aggregate, so only storage can stamp 'last written'. The model docstring now states that ownership split instead of citing the deleted legacy repository. Pinned by a domain test on the factory and a contract case on add() across both implementations. --- .../schedule/scheduled_task_repository.py | 5 +++-- .../deerflow/domain/schedule/model/task.py | 19 +++++++++++++++---- backend/tests/test_schedule_domain.py | 18 ++++++++++++++++++ backend/tests/test_schedule_fakes.py | 11 +++++++++++ 4 files changed, 47 insertions(+), 6 deletions(-) diff --git a/backend/app/adapters/schedule/scheduled_task_repository.py b/backend/app/adapters/schedule/scheduled_task_repository.py index 2513c33be..6741bc573 100644 --- a/backend/app/adapters/schedule/scheduled_task_repository.py +++ b/backend/app/adapters/schedule/scheduled_task_repository.py @@ -159,8 +159,9 @@ class SqlScheduledTaskRepository(ScheduledTaskRepository): # ------------------------------------------------------------------ port async def add(self, task: ScheduledTask) -> ScheduledTask: - now = datetime.now(UTC) - row = ScheduledTaskRow(id=task.task_id, created_at=now, updated_at=now) + # The aggregate's construction instant is the one truth for both + # bookkeeping stamps on insert; the later write paths own updated_at. + row = ScheduledTaskRow(id=task.task_id, created_at=task.created_at, updated_at=task.updated_at) self._apply(row, task) async with self._sf() as session: session.add(row) diff --git a/backend/packages/harness/deerflow/domain/schedule/model/task.py b/backend/packages/harness/deerflow/domain/schedule/model/task.py index 678be725d..e526b70b9 100644 --- a/backend/packages/harness/deerflow/domain/schedule/model/task.py +++ b/backend/packages/harness/deerflow/domain/schedule/model/task.py @@ -51,10 +51,16 @@ class ScheduledTask: router/service pair scattered across three files; nothing here knows about HTTP, SQL, or the run runtime. - The `with_*` / `paused` / `resumed` transitions return a new aggregate and - deliberately leave `updated_at` alone: the repository stamps it on write - (scheduled_tasks/sql.py:97), and a second clock read here would make the - domain both impure and a competing source of truth for the same column. + The two bookkeeping timestamps have one owner each. `created_at` is the + aggregate's construction instant -- `create` stamps it from its explicit + `now` input and the adapter persists it verbatim, never minting its own. + `updated_at` belongs to the *storage* write path: the CAS methods + (`record_launch` / `record_completion`) update the row without ever + holding an aggregate, so only the adapter can stamp "last written" -- the + `with_*` / `paused` / `resumed` transitions therefore deliberately leave + `updated_at` alone, and the field's value on a read-back is the adapter's + stamp, seeded at construction time. Neither timestamp is a rule input; + the clock the rules see is always an explicit `now=` parameter. """ task_id: str @@ -116,6 +122,11 @@ class ScheduledTask: context_mode=mode, thread_id=effective_thread, next_run_at=schedule.ensure_launchable(now, policy), + # The clock is already an explicit rule input here; reading it a + # second time through the field defaults would give one + # construction two instants. + created_at=now, + updated_at=now, ) @property diff --git a/backend/tests/test_schedule_domain.py b/backend/tests/test_schedule_domain.py index b8ec54f7d..302b6eed3 100644 --- a/backend/tests/test_schedule_domain.py +++ b/backend/tests/test_schedule_domain.py @@ -346,6 +346,24 @@ class TestScheduledTaskInvariants: assert task.assistant_id == "lead_agent" assert task.next_run_at == datetime(2026, 7, 28, 9, 0, tzinfo=UTC) + def test_create_stamps_the_bookkeeping_timestamps_from_now(self): + """The factory already receives the clock as a rule input; reading it + a second time through the field defaults would give the aggregate two + slightly different construction instants -- and the stored row a third + if the adapter minted its own. One explicit `now`, one truth.""" + task = ScheduledTask.create( + user_id="user-1", + title="t", + prompt="p", + schedule=cron_spec("0 9 * * *", "UTC"), + context_mode=ContextMode.FRESH_THREAD_PER_RUN, + thread_id=None, + now=NOW, + policy=NO_DELAY, + ) + assert task.created_at == NOW + assert task.updated_at == NOW + def test_create_rejects_a_once_schedule_inside_the_delay_floor(self): with pytest.raises(InvalidScheduleError, match="at least 60 seconds"): ScheduledTask.create( diff --git a/backend/tests/test_schedule_fakes.py b/backend/tests/test_schedule_fakes.py index 46d6a1387..68c4b1972 100644 --- a/backend/tests/test_schedule_fakes.py +++ b/backend/tests/test_schedule_fakes.py @@ -191,6 +191,17 @@ class TestRoundTrip: 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