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 <Tong-bit-art@users.noreply.github.com>

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`.

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

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