diff --git a/README.md b/README.md index 0a75d3c7f2..d7f0e1ce0d 100644 --- a/README.md +++ b/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 diff --git a/backend/app/scheduler/service.py b/backend/app/scheduler/service.py index bebe7c4d78..a7e5082f09 100644 --- a/backend/app/scheduler/service.py +++ b/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 diff --git a/backend/packages/harness/deerflow/persistence/AGENTS.md b/backend/packages/harness/deerflow/persistence/AGENTS.md index 8a296c8657..e52f5ea18f 100644 --- a/backend/packages/harness/deerflow/persistence/AGENTS.md +++ b/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 diff --git a/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py b/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py index 7c910ec6cf..9a0b39fbbc 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py +++ b/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, diff --git a/backend/tests/test_scheduled_task_service.py b/backend/tests/test_scheduled_task_service.py index 34ea9b3403..91b38f97d9 100644 --- a/backend/tests/test_scheduled_task_service.py +++ b/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 diff --git a/backend/tests/test_scheduler_completion_atomicity.py b/backend/tests/test_scheduler_completion_atomicity.py index 1632d0e06c..3a612d6674 100644 --- a/backend/tests/test_scheduler_completion_atomicity.py +++ b/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):