fix(scheduler): keep a skipped dispatch from failing its own write-back

`_finalize_skip` carries a task's launch bookkeeping over unchanged by
reading the current values off the task dict and passing them back into
`update_after_launch`. But repository dicts hold ISO *strings* for
timestamps -- `_row_to_dict` runs every datetime column through
`coerce_iso` -- while `scheduled_tasks.last_run_at` is a DateTime
column, so that round trip fed a string into a datetime bind parameter
and raised StatementError.

It only reproduces once the task has launched at least once: before
that `last_run_at` is NULL, which the column accepts. Every existing
dispatch test seeds a fresh task, so none of them reached it.

The blast radius was the whole poll cycle rather than the one task: the
exception escapes `dispatch_task` into `_run_loop`, so every task still
queued behind it in that round goes undispatched, and the failing task
holds its lease in `running` until it expires. The skip tombstone was
already written by then, so the run history was left inconsistent with
the task row.

Coerced at the call site instead of loosening the repository's parameter
type: the skip path is the only one that round-trips a stored timestamp,
every other write passes `now` straight through.

backend/app/scheduler/service.py

@@ -21,6 +21,28 @@ _ACTIVE_RUN_CONFLICT_ERROR = "task already has an active run"
 _SKIP_ACTIVE_RUN_ERROR = "skipped: a previous run of this task is still active"
 
 
+def _as_datetime(value: Any) -> datetime | None:
+    """Coerce a timestamp read back off a task dict into a real datetime.
+
+    ``ScheduledTaskRepository._row_to_dict`` runs every timestamp column
+    through ``coerce_iso``, so what a caller reads is an ISO *string* while the
+    column it goes back into is a ``DateTime``. Only the skip path performs
+    that round trip -- every other write passes ``now`` straight through -- so
+    the coercion lives here rather than being pushed into the repository, which
+    would loosen the type of an otherwise strict boundary.
+    """
+    if value is None or isinstance(value, datetime):
+        return value
+    try:
+        parsed = datetime.fromisoformat(str(value))
+    except (TypeError, ValueError):
+        # Display-only bookkeeping: an unparseable stored timestamp must not
+        # take down the poll cycle along with every task queued behind it.
+        logger.warning("Dropping unparseable timestamp %r while recording a skipped occurrence", value)
+        return None
+    return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC)
+
+
 class ScheduledTaskService:
     def __init__(
         self,
@@ -291,7 +313,11 @@ class ScheduledTaskService:
             task["id"],
             status=self._task_status_for_skip(task),
             next_run_at=next_at,
-            last_run_at=task.get("last_run_at"),
+            # A skip is not an execution, so the launch bookkeeping is carried
+            # over unchanged. `update_after_launch` assigns unconditionally, so
+            # "unchanged" has to be spelled out by passing the current values
+            # back -- and `last_run_at` comes back as an ISO string.
+            last_run_at=_as_datetime(task.get("last_run_at")),
             last_run_id=task.get("last_run_id"),
             last_thread_id=task.get("last_thread_id"),
             last_error=error if task["schedule_type"] == "once" else None,

backend/tests/test_scheduled_task_skip_writeback.py

@@ -0,0 +1,148 @@
+"""Regression test for the write-back on an overlap-skipped dispatch.
+
+``ScheduledTaskService._finalize_skip`` preserves a task's launch bookkeeping
+by reading the current values off the task dict and passing them straight back
+into ``update_after_launch``. But repository dicts carry ISO *strings* for
+timestamps (``ScheduledTaskRepository._row_to_dict`` runs every datetime column
+through ``coerce_iso``), while ``scheduled_tasks.last_run_at`` is a
+``DateTime`` column -- so the round trip fed a string into a datetime bind
+parameter and the write raised.
+
+It only reproduces once the task has actually launched at least once: before
+that ``last_run_at`` is NULL, and NULL is accepted by the column. That is why
+the existing dispatch tests, which all seed a fresh task, never hit it.
+
+Uses the REAL repositories against a file-backed ``sqlite+aiosqlite`` DB,
+because an in-memory fake stores whatever object it is handed and cannot
+reject the type.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from app.scheduler.service import ScheduledTaskService
+from deerflow.config.database_config import DatabaseConfig
+from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
+from deerflow.persistence.scheduled_task_runs import ScheduledTaskRunRepository
+from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
+
+pytestmark = pytest.mark.asyncio
+
+
+def _make_service(task_repo, run_repo, launched: list) -> ScheduledTaskService:
+    async def fake_launch(**kwargs):
+        launched.append(kwargs)
+        return {"run_id": f"run-{len(launched)}", "thread_id": kwargs["thread_id"]}
+
+    return ScheduledTaskService(
+        task_repo=task_repo,
+        task_run_repo=run_repo,
+        launch_run=fake_launch,
+        poll_interval_seconds=5,
+        lease_seconds=120,
+        max_concurrent_runs=10,
+    )
+
+
+async def _seed_task(task_repo: ScheduledTaskRepository, task_id: str, *, schedule_type: str = "cron") -> dict:
+    spec = {"cron": "*/5 * * * *"} if schedule_type == "cron" else {"run_at": "2099-01-01T00:00:00+00:00"}
+    await task_repo.create(
+        task_id=task_id,
+        user_id="user-1",
+        thread_id=None,
+        context_mode="fresh_thread_per_run",
+        assistant_id="lead_agent",
+        title=task_id,
+        prompt="do the thing",
+        schedule_type=schedule_type,
+        schedule_spec=spec,
+        timezone="UTC",
+        next_run_at=None,
+    )
+    task = await task_repo.get(task_id, user_id="user-1")
+    assert task is not None
+    return task
+
+
+async def test_scheduled_skip_preserves_bookkeeping_of_a_previously_launched_task(tmp_path):
+    """A cron task that already ran once, then overlaps, must record the skip
+    without disturbing (or failing to write) its launch bookkeeping."""
+    await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
+    try:
+        sf = get_session_factory()
+        assert sf is not None
+        task_repo = ScheduledTaskRepository(sf)
+        run_repo = ScheduledTaskRunRepository(sf)
+        launched: list = []
+        service = _make_service(task_repo, run_repo, launched)
+        task = await _seed_task(task_repo, "task-skip-cron")
+
+        first_at = datetime(2026, 7, 27, 12, 0, tzinfo=UTC)
+        first = await service.dispatch_task(dict(task), now=first_at, trigger="scheduled")
+        assert first["outcome"] == "launched"
+
+        after_launch = await task_repo.get("task-skip-cron", user_id="user-1")
+        assert after_launch is not None
+        # The precondition the old dispatch tests never established.
+        assert after_launch["last_run_at"] is not None
+
+        # The fake launcher never terminates the run, so its row is still
+        # active and this second dispatch overlaps. `after_launch` is passed
+        # verbatim, exactly as the poller passes what claim_due_tasks returned.
+        second = await service.dispatch_task(after_launch, now=first_at + timedelta(minutes=5), trigger="scheduled")
+
+        assert second["outcome"] == "skipped"
+        assert len(launched) == 1, "the overlapping dispatch must not launch"
+
+        after_skip = await task_repo.get("task-skip-cron", user_id="user-1")
+        assert after_skip is not None
+        # A skip is not an execution: the launch bookkeeping is carried over
+        # untouched and the counter does not move.
+        assert after_skip["last_run_at"] == after_launch["last_run_at"]
+        assert after_skip["last_run_id"] == after_launch["last_run_id"]
+        assert after_skip["last_thread_id"] == after_launch["last_thread_id"]
+        assert after_skip["run_count"] == after_launch["run_count"] == 1
+        # A cron task stays claimable; only `once` burns its occurrence.
+        assert after_skip["status"] == "enabled"
+        assert after_skip["last_error"] is None
+        # The skip itself is recorded as a terminal tombstone.
+        rows = await run_repo.list_by_task("task-skip-cron", limit=10)
+        assert sorted(row["status"] for row in rows) == ["running", "skipped"]
+    finally:
+        await close_engine()
+
+
+async def test_scheduled_skip_of_a_previously_launched_once_task_records_the_lost_occurrence(tmp_path):
+    """Same write-back path, `once` branch: the single occurrence is lost, so
+    the task ends `failed` and carries the skip reason."""
+    await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
+    try:
+        sf = get_session_factory()
+        assert sf is not None
+        task_repo = ScheduledTaskRepository(sf)
+        run_repo = ScheduledTaskRunRepository(sf)
+        launched: list = []
+        service = _make_service(task_repo, run_repo, launched)
+        task = await _seed_task(task_repo, "task-skip-once", schedule_type="once")
+
+        first_at = datetime(2026, 7, 27, 12, 0, tzinfo=UTC)
+        first = await service.dispatch_task(dict(task), now=first_at, trigger="scheduled")
+        assert first["outcome"] == "launched"
+
+        after_launch = await task_repo.get("task-skip-once", user_id="user-1")
+        assert after_launch is not None
+        assert after_launch["last_run_at"] is not None
+
+        second = await service.dispatch_task(after_launch, now=first_at + timedelta(minutes=5), trigger="scheduled")
+
+        assert second["outcome"] == "skipped"
+        after_skip = await task_repo.get("task-skip-once", user_id="user-1")
+        assert after_skip is not None
+        assert after_skip["last_run_at"] == after_launch["last_run_at"]
+        assert after_skip["status"] == "failed"
+        assert after_skip["last_error"] == "skipped: a previous run of this task is still active"
+    finally:
+        await close_engine()