fix(scheduler): reconcile running occurrences from terminal durable runs (#7269)

* fix(scheduler): reconcile running occurrences from terminal durable runs

A completion hook can fail after the durable run has already committed its terminal status (transient DB, observer, or notification error). Single-instance run_once only recovered expired launching claims, so the occurrence stayed running: a recurring schedule stopped firing and a once task never reached its terminal parent state until a restart.

Add a non-destructive ScheduledTaskRepository.reconcile_terminal_runs pass. It only selects running occurrences whose linked runs row is already terminal and whose scheduled metadata matches, then finalizes through the existing finalize_occurrence path with the run's status, error, and goal verdict. Live runs are never taken over or interrupted; multi-instance reconciliation already handles terminal runs through lease ownership. Single-instance run_once calls it after expired-launch recovery. No schema change.

Tests: a terminal success finalizes, a live run is left alone, error/timeout/interrupted map correctly, and a success with an unsatisfied goal verdict becomes unmet. Full scheduler suite: 1041 passed / 130 skipped; guidance 0 errors / 0 warnings; ruff clean.

* fix(scheduler): isolate terminal-run reconciliation failures per occurrence

A poisoned occurrence (the same observer or database failure that made the sweep necessary) ran in the same transaction as every other candidate, so one raised exception rolled back the whole pass and escaped run_once before admission and queue draining. Give each candidate its own session/transaction: a failure rolls back only that occurrence, logs its id, and the remaining candidates still commit. The service also catches a remaining pass-level failure so the rest of the poller runs.

Extend coverage: scheduled-trigger finalization (the motivating case), a second reconcile returning 0 with no state change, once-task completed projection, and a poisoned-observer candidate ordered before a healthy one whose success still commits. Full scheduler suite 1044 passed / 130 skipped.

---------

Co-authored-by: Tong-bit-art <Tong-bit-art@users.noreply.github.com>

README.md

@@ -3478,6 +3478,7 @@ Current MVP capabilities:
 - Run background scheduled executions as non-interactive DeerFlow runs (`ask_clarification` is not exposed there)
 - Persist a due execution as `queued` when its reused thread or the global execution budget is busy, then launch it when capacity is available; queued occurrences survive Gateway restarts and fail after `scheduler.queue_timeout_seconds`
 - Retrying a busy scheduled task keeps its execution slot while an earlier recovery pass is still in progress.
+- A scheduled run whose completion hook fails is finalized from its durable terminal run on the next scheduler poll instead of staying `running` until a restart.
 - Share execution slots fairly between task owners: one owner may have at most `scheduler.max_concurrent_runs_per_user` scheduled runs starting or running at a time (default 2, never more than `max_concurrent_runs`; `0` turns the per-owner cap off), and the waiting queue is drained owner by owner, so one owner's backlog never holds back another owner's run. A run that waits longer than `scheduler.queue_timeout_seconds` is skipped, and its history reads "Skipped: it waited too long for a free slot"
 - Freeze a task's definition while an occurrence is `queued`, `launching`, or `running`, so a durable occurrence cannot silently pick up a different prompt, thread, or schedule; transitioning a task to paused or deleting it cancels an existing waiting occurrence, while `launching`/`running` work must finish before those mutations are retried and an explicit manual trigger may still wait and run without resuming a paused schedule
 - Pause, resume, trigger, inspect history, and delete tasks

backend/app/scheduler/service.py

@@ -454,6 +454,15 @@ class ScheduledTaskService:
                 error=_LEASE_RECOVERY_ERROR,
                 now=now,
             )
+            try:
+                finalized = await self._task_repo.reconcile_terminal_runs(now=now)
+                if finalized:
+                    logger.info("Finalized %d scheduled occurrence(s) from terminal durable runs", finalized)
+            except Exception:
+                # The pass already isolates per-occurrence failures; a remaining
+                # failure (e.g. the candidate query) must not skip admission or
+                # queue draining for this poll.
+                logger.exception("Failed to reconcile terminal scheduled runs")
         await self._expire_waiting_runs(now=now)
         await self._drain_queue(now=now)
         # Admission and execution capacity are separate. Due occurrences are

backend/packages/harness/deerflow/persistence/AGENTS.md

@@ -31,6 +31,8 @@ The per-owner run cap `scheduler.max_concurrent_runs_per_user` is effective `min
 
 Recovery locks task/run pairs in task-id/run-id order and restores `run_id`, `started_at` and live errors before releasing launch claims. Launch/failure/timeout updates use one parent-first transaction to prevent interleaved claims. Queue timeout fails the occurrence and advances scheduled work to prevent immediate requeue.
 
+Single-instance polling also performs a non-destructive terminal-run reconciliation: a `running` occurrence whose durable `runs` row is already terminal is finalized from that row's status, error and goal verdict. Live runs are never taken over or interrupted; multi-instance reconciliation already covers terminal runs through lease ownership. This repairs a completion hook that failed after the run committed its terminal status.
+
 Expired-launch scans select candidates, not ownership. After taking the parent
 and occurrence locks, recheck launch status and lease expiry before recovery.
 A busy launch may have requeued and a manual trigger reclaimed the same row

backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py

@@ -15,12 +15,14 @@ from deerflow.persistence.scheduled_task_runs.finalization import FinalizationOb
 from deerflow.persistence.scheduled_task_runs.model import ScheduledTaskRunRow
 from deerflow.persistence.scheduled_task_runs.projection import account_launch, can_project
 from deerflow.persistence.scheduled_tasks.model import ACTIVE_RUN_STATUSES, LIVE_TASK_STATUSES, ONCE_TASK_STATUS_BY_RUN_STATUS, TERMINAL_RUN_STATUSES, TERMINAL_TASK_STATUSES, ScheduledTaskRow
-from deerflow.scheduler.host_notes import RUN_ERROR_END_REACHED
+from deerflow.scheduler.host_notes import RUN_ERROR_END_REACHED, RUN_ERROR_INTERRUPTED
 from deerflow.utils.goal_objective import normalize_goal_objective
 from deerflow.utils.time import coerce_iso
 
 logger = logging.getLogger(__name__)
 
+_TERMINAL_DURABLE_RUN_STATUSES = frozenset({"success", "error", "timeout", "interrupted"})
+
 
 class ScheduledTaskQuotaExceeded(ValueError):
     """A tool-created task would exceed its owner's live schedule quota."""
@@ -651,6 +653,78 @@ class ScheduledTaskRepository:
             await session.commit()
             return accepted
 
+    async def reconcile_terminal_runs(self, *, now: datetime, limit: int = 100) -> int:
+        """Finalize running occurrences whose durable run already finished.
+
+        A completion hook can fail after the durable run committed its terminal
+        status, leaving the occurrence in ``running`` until a restart. This
+        non-destructive pass only consumes already-terminal ``runs`` rows; it
+        never takes over, interrupts, or otherwise touches a live run.
+
+        Each candidate commits in its own transaction. A poisoned occurrence
+        (for example, the same observer/database failure that made the sweep
+        necessary) rolls back only its own savepoint, is logged, and cannot
+        prevent later candidates from finalizing or the rest of the poller from
+        running.
+        """
+        async with self._sf() as session:
+            candidates = list(
+                (
+                    await session.execute(
+                        select(ScheduledTaskRunRow.id, ScheduledTaskRunRow.task_id)
+                        .join(RunRow, RunRow.run_id == ScheduledTaskRunRow.run_id)
+                        .where(
+                            ScheduledTaskRunRow.status == "running",
+                            ScheduledTaskRunRow.run_id.is_not(None),
+                            RunRow.status.in_(tuple(_TERMINAL_DURABLE_RUN_STATUSES)),
+                            RunRow.operation_kind == "run",
+                        )
+                        .order_by(ScheduledTaskRunRow.created_at.asc(), ScheduledTaskRunRow.id.asc())
+                        .limit(limit)
+                    )
+                ).all()
+            )
+        finalized = 0
+        for occurrence_id, task_id in candidates:
+            try:
+                async with self._sf() as session:
+                    task = await self._lock_task(session, task_id)
+                    occurrence = await session.get(ScheduledTaskRunRow, occurrence_id, with_for_update=True)
+                    if occurrence is None or occurrence.task_id != task_id or occurrence.status != "running" or occurrence.run_id is None:
+                        continue
+                    run = await session.get(RunRow, occurrence.run_id)
+                    if run is None or run.status not in _TERMINAL_DURABLE_RUN_STATUSES or run.operation_kind != "run":
+                        continue
+                    metadata = run.metadata_json if isinstance(run.metadata_json, dict) else {}
+                    if metadata.get("scheduled_task_id") != task_id or metadata.get("scheduled_task_run_id") != occurrence_id:
+                        continue
+                    if run.status == "success":
+                        status, error = "success", None
+                    elif run.status == "interrupted":
+                        status, error = "interrupted", run.error or RUN_ERROR_INTERRUPTED
+                    else:
+                        status, error = "failed", run.error
+                    accepted = await finalize_occurrence(
+                        session,
+                        task,
+                        occurrence,
+                        status=status,
+                        error=error,
+                        finished_at=now,
+                        run_id=run.run_id,
+                        goal_verdict=run.goal_verdict,
+                        observer=self._finalization_observer,
+                    )
+                    await session.commit()
+                    if accepted:
+                        finalized += 1
+            except Exception:
+                logger.exception(
+                    "Failed to reconcile scheduled occurrence %s from its terminal run; leaving it for the next poll",
+                    occurrence_id,
+                )
+        return finalized
+
     async def claim_dispatch_lease(
         self,
         task_id: str,

backend/tests/test_scheduled_task_service.py

@@ -48,6 +48,9 @@ class DummyTaskRepo:
     async def update_after_launch(self, *args, **kwargs):
         self.updated = (args, kwargs)
 
+    async def reconcile_terminal_runs(self, **_kwargs):
+        return 0
+
     async def complete_run(self, task_id, **kwargs):
         self.completions.append((task_id, kwargs))
         return True

backend/tests/test_scheduler_completion_atomicity.py

@@ -6,6 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn
 
 from app.scheduler.service import ScheduledTaskService
 from deerflow.persistence.base import Base
+from deerflow.persistence.run.model import RunRow
 from deerflow.persistence.scheduled_task_runs import ScheduledTaskRunRepository
 from deerflow.persistence.scheduled_task_runs.model import ScheduledTaskRunRow
 from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
@@ -13,6 +14,7 @@ from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow
 from deerflow.runtime import RunStatus
 from deerflow.runtime.runs.manager import RunRecord
 from deerflow.runtime.runs.schemas import DisconnectMode
+from deerflow.scheduler.host_notes import RUN_ERROR_INTERRUPTED
 
 
 @asynccontextmanager
@@ -20,19 +22,22 @@ async def _database(tmp_path):
     engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'completion.db'}")
     try:
         async with engine.begin() as connection:
-            await connection.run_sync(lambda sync: Base.metadata.create_all(sync, tables=[ScheduledTaskRow.__table__, ScheduledTaskRunRow.__table__]))
+            await connection.run_sync(lambda sync: Base.metadata.create_all(sync, tables=[ScheduledTaskRow.__table__, ScheduledTaskRunRow.__table__, RunRow.__table__]))
         yield engine, async_sessionmaker(engine, expire_on_commit=False)
     finally:
         await engine.dispose()
 
 
-async def _launching_occurrence(sf, *, schedule_type):
+async def _launching_occurrence(sf, *, schedule_type, suffix="completion", trigger="manual"):
     now = datetime(2026, 9, 9, 8, tzinfo=UTC)
     tasks = ScheduledTaskRepository(sf)
     occurrences = ScheduledTaskRunRepository(sf)
+    task_id = f"task-{suffix}"
+    occurrence_id = f"occurrence-{suffix}"
+    thread_id = f"thread-{suffix}"
     spec = {"run_at": now.isoformat()} if schedule_type == "once" else {"cron": "0 9 * * *"}
     await tasks.create(
-        task_id="task-completion",
+        task_id=task_id,
         user_id="user-1",
         thread_id=None,
         context_mode="fresh_thread_per_run",
@@ -45,15 +50,15 @@ async def _launching_occurrence(sf, *, schedule_type):
         next_run_at=now,
     )
     await occurrences.create(
-        run_record_id="occurrence-completion",
-        task_id="task-completion",
-        thread_id="thread-completion",
+        run_record_id=occurrence_id,
+        task_id=task_id,
+        thread_id=thread_id,
         scheduled_for=now,
-        trigger="manual",
+        trigger=trigger,
         status="queued",
     )
     claimed = await occurrences.claim_queued_run(
-        "occurrence-completion",
+        occurrence_id,
         lease_owner="launcher",
         now=now,
         lease_seconds=120,
@@ -61,7 +66,7 @@ async def _launching_occurrence(sf, *, schedule_type):
     )
     assert claimed is not None
     await tasks.update(
-        "task-completion",
+        task_id,
         user_id="user-1",
         updates={"status": "running" if schedule_type == "once" else "enabled", "last_error": "previous error"},
     )
@@ -92,10 +97,10 @@ def _completion(status, error):
     )
 
 
-async def _snapshot(sf):
+async def _snapshot(sf, task_id="task-completion", occurrence_id="occurrence-completion"):
     async with sf() as session:
-        task = await session.get(ScheduledTaskRow, "task-completion")
-        occurrence = await session.get(ScheduledTaskRunRow, "occurrence-completion")
+        task = await session.get(ScheduledTaskRow, task_id)
+        occurrence = await session.get(ScheduledTaskRunRow, occurrence_id)
         assert task is not None and occurrence is not None
         return task.to_dict(), occurrence.to_dict()
 
@@ -157,6 +162,168 @@ async def test_completion_before_launch_bookkeeping_is_durable_and_counted_once(
         assert task_after_retry["last_error"] == expected_error
 
 
+async def _running_occurrence_with_terminal_run(
+    sf,
+    *,
+    run_status: str,
+    error: str | None = None,
+    goal_objective: str | None = None,
+    goal_verdict: dict | None = None,
+    suffix: str = "completion",
+    trigger: str = "manual",
+    schedule_type: str = "cron",
+):
+    tasks, occurrences, now = await _launching_occurrence(sf, schedule_type=schedule_type, suffix=suffix, trigger=trigger)
+    occurrence_id = f"occurrence-{suffix}"
+    task_id = f"task-{suffix}"
+    thread_id = f"thread-{suffix}"
+    run_id = f"run-{suffix}"
+    await occurrences.update_status(
+        occurrence_id,
+        status="running",
+        run_id=run_id,
+        started_at=now,
+    )
+    await tasks.update(
+        task_id,
+        user_id="user-1",
+        updates={"next_run_at": now if schedule_type == "once" else now + timedelta(days=1), "goal_objective": goal_objective},
+    )
+    async with sf() as session:
+        if goal_objective is not None:
+            occurrence = await session.get(ScheduledTaskRunRow, occurrence_id)
+            assert occurrence is not None
+            occurrence.goal_objective = goal_objective
+        session.add(
+            RunRow(
+                run_id=run_id,
+                thread_id=thread_id,
+                user_id="user-1",
+                status=run_status,
+                error=error,
+                metadata_json={"scheduled_task_id": task_id, "scheduled_task_run_id": occurrence_id},
+                goal_verdict=goal_verdict,
+            )
+        )
+        await session.commit()
+    return tasks, occurrences, now
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("trigger", ["manual", "scheduled"])
+async def test_run_once_finalizes_running_occurrence_from_terminal_run(tmp_path, trigger):
+    async with _database(tmp_path) as (_engine, sf):
+        tasks, occurrences, now = await _running_occurrence_with_terminal_run(sf, run_status="success", trigger=trigger)
+        service = _service(tasks, occurrences)
+
+        await service.run_once(now=now)
+
+        task, occurrence = await _snapshot(sf)
+        assert occurrence["status"] == "success"
+        assert task["status"] == "enabled"
+        assert task["run_count"] == 1
+        assert await tasks.reconcile_terminal_runs(now=now) == 0
+        assert await _snapshot(sf) == (task, occurrence)
+
+
+@pytest.mark.asyncio
+async def test_reconcile_terminal_runs_isolates_a_poisoned_candidate(tmp_path, caplog):
+    import logging
+
+    async with _database(tmp_path) as (_engine, sf):
+        tasks, occurrences, now = await _running_occurrence_with_terminal_run(sf, run_status="success", suffix="aaa-poisoned", trigger="scheduled")
+        await _running_occurrence_with_terminal_run(sf, run_status="success", suffix="zzz-healthy", trigger="scheduled")
+        service = _service(tasks, occurrences)
+        original = tasks._finalization_observer
+
+        async def poisoned_observer(session, task, occurrence, *, events):
+            if task is not None and task.id == "task-aaa-poisoned":
+                raise RuntimeError("poisoned finalization observer")
+            if original is not None:
+                await original(session, task, occurrence, events=events)
+
+        tasks.set_finalization_observer(poisoned_observer)
+
+        with caplog.at_level(logging.ERROR):
+            await service.run_once(now=now)
+
+        healthy_task, healthy_occurrence = await _snapshot(sf, "task-zzz-healthy", "occurrence-zzz-healthy")
+        poisoned_task, poisoned_occurrence = await _snapshot(sf, "task-aaa-poisoned", "occurrence-aaa-poisoned")
+        assert healthy_occurrence["status"] == "success"
+        assert healthy_task["status"] == "enabled"
+        assert poisoned_occurrence["status"] == "running"
+        assert poisoned_task["status"] == "enabled"
+        assert "Failed to reconcile scheduled occurrence occurrence-aaa-poisoned" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_run_once_projects_once_task_from_terminal_run(tmp_path):
+    async with _database(tmp_path) as (_engine, sf):
+        tasks, occurrences, now = await _running_occurrence_with_terminal_run(sf, run_status="success", schedule_type="once", trigger="scheduled")
+        service = _service(tasks, occurrences)
+
+        await service.run_once(now=now)
+
+        task, occurrence = await _snapshot(sf)
+        assert occurrence["status"] == "success"
+        assert task["status"] == "completed"
+
+
+@pytest.mark.asyncio
+async def test_run_once_uses_terminal_run_goal_verdict(tmp_path):
+    async with _database(tmp_path) as (_engine, sf):
+        tasks, occurrences, now = await _running_occurrence_with_terminal_run(
+            sf,
+            run_status="success",
+            goal_objective="ship it",
+            goal_verdict={"satisfied": False, "stand_down_reason": "not_met"},
+        )
+        service = _service(tasks, occurrences)
+
+        await service.run_once(now=now)
+
+        task, occurrence = await _snapshot(sf)
+        assert occurrence["status"] == "unmet"
+        assert occurrence["error"] == "not_met"
+        assert task["last_error"] == "not_met"
+
+
+@pytest.mark.asyncio
+async def test_run_once_leaves_running_occurrence_with_live_run_alone(tmp_path):
+    async with _database(tmp_path) as (_engine, sf):
+        tasks, occurrences, now = await _running_occurrence_with_terminal_run(sf, run_status="running")
+        service = _service(tasks, occurrences)
+
+        await service.run_once(now=now)
+
+        task, occurrence = await _snapshot(sf)
+        assert occurrence["status"] == "running"
+        assert task["status"] == "enabled"
+        assert task["run_count"] == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("run_status", "error", "occurrence_status", "expected_error"),
+    [
+        ("error", "boom", "failed", "boom"),
+        ("timeout", "time limit", "failed", "time limit"),
+        ("interrupted", None, "interrupted", RUN_ERROR_INTERRUPTED),
+    ],
+)
+async def test_run_once_finalizes_terminal_run_outcomes(tmp_path, run_status, error, occurrence_status, expected_error):
+    async with _database(tmp_path) as (_engine, sf):
+        tasks, occurrences, now = await _running_occurrence_with_terminal_run(sf, run_status=run_status, error=error)
+        service = _service(tasks, occurrences)
+
+        await service.run_once(now=now)
+
+        task, occurrence = await _snapshot(sf)
+        assert occurrence["status"] == occurrence_status
+        assert task["status"] == "enabled"
+        assert task["last_error"] == expected_error
+
+
 @pytest.mark.asyncio
 async def test_completion_commit_failure_rolls_back_outcome_and_accounting(tmp_path):
     async with _database(tmp_path) as (engine, sf):