From 8bda7c109424c936c5b00f7a8bb3598ff0d1fbb0 Mon Sep 17 00:00:00 2001 From: ordinary being Date: Sun, 25 Oct 2026 22:56:58 +0800 Subject: [PATCH] fix(scheduler): keep manual-trial launch origins stable across retries (#7492) * fix(scheduler): keep manual-trial launch origins stable across retries A manual trial's synthetic queued row omitted scheduled_for, and _launch_provenance used the current now for manual triggers. A retry at a later time therefore produced a different origin, which is part of start_run's per-occurrence idempotent input: the retry either conflicted with the first attempt or submitted a different request. Include the admitted scheduled_for in the synthetic row and use the occurrence time for every trigger, falling back to now only when a legacy caller supplies none. Tests: a manual queued row retried at two later clocks produces the same origin and keeps the admitted scheduled_for; existing manual/scheduled provenance and title tests stay green. Full scheduler suite: 1108 passed / 130 skipped; guidance 0 errors / 0 warnings; ruff clean. * test(scheduler): drive manual retry through requeue and drain The unit provenance test pins the stable-origin function directly. Add a driven case that admits a manual occurrence, hits a busy-thread ConflictError so the row requeues, then drains it at a later clock and asserts both launches carry the admitted scheduled_for and the same full origin. * test(scheduler): pin the immediate queue row in the retry test Capture the queued mappings passed to _attempt_queued_run so the driven test asserts both halves explicitly: the synthetic immediate-launch row and the durable row drained later both retain the admitted scheduled_for. --------- Co-authored-by: Tong-bit-art --- backend/AGENTS.md | 2 +- backend/app/scheduler/service.py | 10 ++- backend/tests/test_scheduled_task_service.py | 80 ++++++++++++++++++++ 3 files changed, 89 insertions(+), 3 deletions(-) diff --git a/backend/AGENTS.md b/backend/AGENTS.md index fb946e7836..50d8a42de2 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -16,7 +16,7 @@ Service roles and ports: [root service topology](../AGENTS.md#service-topology). - The background scheduler is single-instance by default. `scheduler.multi_instance=true` opts into lease-aware recovery across Gateway instances and requires shared Postgres, `run_ownership.heartbeat_enabled=true`, and `run_events.backend=db`; otherwise startup rejects the configuration. Live scheduled runs are preserved when a peer starts; expired launch claims return to the durable queue, expired run leases are atomically taken over, stale launch writes are fenced by lease ownership, and the Postgres advisory-locked budget makes `max_concurrent_runs` a shared global cap for `launching`/`running` rows. - The multi-process startup gate also fires on `deployment.multi_instance: true`; contract in `docs/CONFIGURATION.md`. - Long-running MCP work uses a separate durable task runtime (`McpTaskService` + `mcp_tasks`, lease-based recovery) rather than keeping remote task IDs or status polling inside the Agent loop; only submit remains Agent-visible, the database is the source of truth, and `ThreadState` receives only a bounded current-thread projection. Full contract — including notification retries, dead-lettering, and the cancel endpoint's worker-stopped 503 (leases, cancellation fencing, delivery idempotency, management-tool exposure): [packages/harness/deerflow/mcp/AGENTS.md](packages/harness/deerflow/mcp/AGENTS.md). -- Scheduled-task dispatch permits one active occurrence per task via `uq_scheduled_task_run_active` (`task_id WHERE status IN ('queued','launching','running')`). Durable `queued` rows survive restarts; only lease-fenced `launching` may call Gateway launch; `running` references the durable run. Stable admission idempotency keys reuse that run after recovery. Reused-thread `ConflictError` returns `launching` to `queued`; other launch errors become `failed`. Atomic queue claims enforce `max_concurrent_runs` and the per-owner cap, excluding waiting rows; the budget count and its UPDATE are separate statements, so writers must serialize before the count (Postgres advisory lock; SQLite `BEGIN IMMEDIATE`, whose deferred transaction otherwise reserves the writer only at the UPDATE) or claims on distinct rows overshoot the cap. Repeated triggers coalesce; same-thread FIFO blocks behind older active rows. Queue admission, PATCH/resume, pause and delete lock the parent before the occurrence, freezing active task definitions. Pause/delete atomically cancel `queued` work but reject `launching`/`running`; PATCH/resume reject all active states. Only queued conflicts offer pause cancellation. Manual triggers may queue/run while paused. +- Scheduled-task dispatch permits one active occurrence per task via `uq_scheduled_task_run_active` (`task_id WHERE status IN ('queued','launching','running')`). Durable `queued` rows survive restarts; only lease-fenced `launching` may call Gateway launch; `running` references the durable run. Stable admission idempotency keys reuse that run after recovery. Reused-thread `ConflictError` returns `launching` to `queued`; other launch errors become `failed`. Atomic queue claims enforce `max_concurrent_runs` and the per-owner cap, excluding waiting rows; the budget count and its UPDATE are separate statements, so writers must serialize before the count (Postgres advisory lock; SQLite `BEGIN IMMEDIATE`, whose deferred transaction otherwise reserves the writer only at the UPDATE) or claims on distinct rows overshoot the cap. Repeated triggers coalesce; same-thread FIFO blocks behind older active rows. Queue admission, PATCH/resume, pause and delete lock the parent before the occurrence, freezing active task definitions. Pause/delete atomically cancel `queued` work but reject `launching`/`running`; PATCH/resume reject all active states. Only queued conflicts offer pause cancellation. Manual triggers may queue/run while paused. Manual-trial run origins reuse the admitted `scheduled_for`, so an idempotent retry submits the same request instead of colliding with the first attempt. - Scheduled-task lifecycle, recovery and queue timeout: [persistence guide](packages/harness/deerflow/persistence/AGENTS.md#scheduled-task-lifecycle). - `POST /api/scheduled-tasks/preview-cron` requires authenticated `threads:read`. Bounded cron previews call the shared scheduler calculator in `asyncio.to_thread`, preserving its DST semantics. Capture the optional aware reference once; return UTC and offset-bearing local occurrences without acquiring task/thread/run stores or dispatching work. This advisory API does not reserve execution. - Gateway MCP/skill updates write `extensions_config.json` at runtime. Production Compose mounts it read-write (`config.yaml` stays `:ro`); Helm seeds a writable home-volume copy. Hold both `extensions_config_write_lock` and the sidecar advisory `extensions_config_file_lock` during read-modify-write to prevent lost updates across workers. `atomic_write_extensions_config` uses temp-file replacement; only mount-point `EBUSY` permits a non-atomic in-place overwrite (crashes can truncate). Warn about this fallback once per target; propagate other errors. `get_extensions_config()` revalidates reads by file signature; see [config caching](packages/harness/deerflow/config/AGENTS.md). Tests: `tests/test_compose_extensions_config_writable.py`, `tests/test_extensions_config_atomic_write.py`, `tests/test_helm_extensions_config_writable.py`. diff --git a/backend/app/scheduler/service.py b/backend/app/scheduler/service.py index a7e5082f09..4ae241e93a 100644 --- a/backend/app/scheduler/service.py +++ b/backend/app/scheduler/service.py @@ -645,6 +645,10 @@ class ScheduledTaskService: "task_id": task["id"], "thread_id": execution_thread_id, "trigger": trigger, + # Freeze the admitted occurrence time in the synthetic row: the + # provenance below must be identical if this launch is retried at a + # later now, or start_run's idempotency key sees a different request. + "scheduled_for": now, } return await self._attempt_queued_run(task, queued, now=now) @@ -863,8 +867,10 @@ class ScheduledTaskService: one the run row shows once the launch is accounted. """ trigger = queued["trigger"] - scheduled_for = _as_utc(queued.get("scheduled_for")) if trigger == "scheduled" else None - scheduled_for = scheduled_for or now + # Use the occurrence's admitted time for every trigger. A manual trial + # retried by the queue drain must reproduce the first request's origin; + # falling back to the retry clock changes start_run's idempotent input. + scheduled_for = _as_utc(queued.get("scheduled_for")) or now run_number = None number_lookup = getattr(self._task_run_repo, "run_number", None) if trigger == "scheduled" and callable(number_lookup): diff --git a/backend/tests/test_scheduled_task_service.py b/backend/tests/test_scheduled_task_service.py index 91b38f97d9..1ee34bf8de 100644 --- a/backend/tests/test_scheduled_task_service.py +++ b/backend/tests/test_scheduled_task_service.py @@ -905,6 +905,7 @@ class _StatefulRunRepo: "task_id": kwargs["task_id"], "thread_id": kwargs["thread_id"], "trigger": kwargs["trigger"], + "scheduled_for": kwargs["scheduled_for"], "status": kwargs["status"], "run_id": None, } @@ -1857,6 +1858,85 @@ async def test_launch_passes_user_language_origin_and_run_thread_title(trigger): assert launch["title"] == "检查发布清单 · 10-07 09:00" +@pytest.mark.asyncio +async def test_manual_launch_origin_is_stable_across_retries(): + task_repo = DummyTaskRepo([_provenance_task()]) + run_repo = DummyRunRepo() + service = ScheduledTaskService( + task_repo=task_repo, + task_run_repo=run_repo, + launch_run=lambda **_kwargs: None, + poll_interval_seconds=5, + lease_seconds=120, + max_concurrent_runs=3, + ) + admitted = datetime(2026, 10, 7, 1, 0, tzinfo=UTC) + queued = { + "id": "task-run-prov", + "task_id": "task-prov", + "thread_id": "thread-prov", + "trigger": "manual", + "scheduled_for": admitted, + } + + first, _ = await service._launch_provenance(task_repo.rows[0], queued, now=admitted + timedelta(hours=1)) + second, _ = await service._launch_provenance(task_repo.rows[0], queued, now=admitted + timedelta(hours=2)) + + assert first["scheduled_for"] == admitted.isoformat() + assert first == second + + +@pytest.mark.asyncio +async def test_manual_trial_retry_reuses_the_admitted_origin(): + launches: list[dict] = [] + + async def fake_launch(**kwargs): + launches.append(kwargs) + if len(launches) == 1: + raise ConflictError("Thread busy") + return {"run_id": "run-retried", "thread_id": kwargs["thread_id"]} + + class DrainableRunRepo(_StatefulRunRepo): + async def list_queued_runs(self, *, limit: int, **_kwargs) -> list[dict]: + return [dict(row) for row in self.rows.values() if row["status"] == "queued"][:limit] + + task_repo = DummyTaskRepo([_provenance_task()]) + run_repo = DrainableRunRepo() + service = ScheduledTaskService( + task_repo=task_repo, + task_run_repo=run_repo, + launch_run=fake_launch, + poll_interval_seconds=5, + lease_seconds=120, + max_concurrent_runs=3, + own_stop_available=True, + ) + attempted: list[dict] = [] + original_attempt = service._attempt_queued_run + + async def capture_attempt(task, queued, *, now): + attempted.append(dict(queued)) + return await original_attempt(task, queued, now=now) + + service._attempt_queued_run = capture_attempt + admitted = datetime(2026, 10, 7, 1, 0, tzinfo=UTC) + + first = await service.dispatch_task(task_repo.rows[0], now=admitted, trigger="manual") + assert first["outcome"] == "queued" + + # The immediate launch was busy, so the occurrence was requeued with the + # admitted scheduled_for. Draining it at a later clock must reproduce the + # first request's origin rather than embedding the retry time. + await service._drain_queue(now=admitted + timedelta(hours=2)) + + assert len(launches) == 2, launches + assert attempted[0]["scheduled_for"] == admitted + assert attempted[1]["scheduled_for"] == admitted + assert launches[0]["origin"]["scheduled_for"] == admitted.isoformat() + assert launches[1]["origin"]["scheduled_for"] == admitted.isoformat() + assert launches[0]["origin"] == launches[1]["origin"] + + @pytest.mark.asyncio @pytest.mark.parametrize( ("updates", "trigger", "expected"),