fix(schedule): stop a corrupt stored row surfacing as a client error

SqlScheduledTaskRepository._to_domain raised InvalidScheduleError for a
row whose stored schedule no longer parses -- the same error the
aggregate raises for a client-submitted schedule, which the router maps
to 422. A stored fault therefore told the client its perfectly fine
request was wrong, and made the row unrepairable over HTTP: PATCH reads
the task before writing, so the fix path 422'd too. Enum rebuild
failures were worse -- a raw ValueError crossed the boundary
untranslated.

The rebuild is now translated to a dedicated CorruptStoredScheduleError,
raised only by the persistence adapter and deliberately absent from the
router's status table, so it falls through to the unclassified-500
branch: a server-side fault reported as one. List reads keep skipping
and logging the bad row. Pinned by tests/test_schedule_corrupt_rows.py
against a real sqlite database.

backend/app/adapters/schedule/scheduled_task_repository.py

@@ -26,7 +26,7 @@ from datetime import UTC, datetime, timedelta
 from sqlalchemy import and_, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
-from deerflow.domain.schedule.exceptions import InvalidScheduleError
+from deerflow.domain.schedule.exceptions import CorruptStoredScheduleError, ScheduleError
 from deerflow.domain.schedule.model import TERMINAL_TASK_STATUSES, ContextMode, ScheduledTask, ScheduleSpec, ScheduleType, TaskStatus
 from deerflow.domain.schedule.ports import ScheduledTaskRepository
 
@@ -92,39 +92,46 @@ class SqlScheduledTaskRepository(ScheduledTaskRepository):
     def _to_domain(row: ScheduledTaskRow) -> ScheduledTask:
         """ORM row -> aggregate.
 
-        Raises InvalidScheduleError for a row whose stored schedule can no
-        longer be parsed. That is a **new** failure mode: the legacy repository
-        returned bare dicts and a corrupt spec only surfaced at dispatch time.
-        Single-row reads let it propagate; list reads skip and log, so one bad
-        row cannot 500 an entire listing.
+        Raises CorruptStoredScheduleError for a row that can no longer be
+        rebuilt -- unparseable schedule, unknown enum value. The rebuild goes
+        through the aggregate's own validation, so a hand-damaged row blows up
+        here rather than flowing on; the translation to the dedicated error is
+        what keeps it off ``InvalidScheduleError``'s client-facing 422 mapping
+        (a stored fault is the server's problem, and PATCH -- which reads
+        before writing -- must not become unusable for the one row it could
+        repair). Single-row reads let it propagate; list reads skip and log,
+        so one bad row cannot take down an entire listing.
         """
-        return ScheduledTask(
-            task_id=row.id,
-            user_id=row.user_id,
-            title=row.title,
-            prompt=row.prompt,
-            schedule=SqlScheduledTaskRepository._spec_from_row(row),
-            context_mode=ContextMode(row.context_mode),
-            thread_id=row.thread_id,
-            assistant_id=row.assistant_id,
-            status=TaskStatus(row.status),
-            overlap_policy=row.overlap_policy,
-            next_run_at=_tz_aware(row.next_run_at),
-            last_run_at=_tz_aware(row.last_run_at),
-            last_run_id=row.last_run_id,
-            last_thread_id=row.last_thread_id,
-            last_error=row.last_error,
-            run_count=row.run_count,
-            created_at=_tz_aware(row.created_at),
-            updated_at=_tz_aware(row.updated_at),
-        )
+        try:
+            return ScheduledTask(
+                task_id=row.id,
+                user_id=row.user_id,
+                title=row.title,
+                prompt=row.prompt,
+                schedule=SqlScheduledTaskRepository._spec_from_row(row),
+                context_mode=ContextMode(row.context_mode),
+                thread_id=row.thread_id,
+                assistant_id=row.assistant_id,
+                status=TaskStatus(row.status),
+                overlap_policy=row.overlap_policy,
+                next_run_at=_tz_aware(row.next_run_at),
+                last_run_at=_tz_aware(row.last_run_at),
+                last_run_id=row.last_run_id,
+                last_thread_id=row.last_thread_id,
+                last_error=row.last_error,
+                run_count=row.run_count,
+                created_at=_tz_aware(row.created_at),
+                updated_at=_tz_aware(row.updated_at),
+            )
+        except (ScheduleError, ValueError) as exc:
+            raise CorruptStoredScheduleError(f"stored scheduled task {row.id} cannot be rebuilt: {exc}") from exc
 
     @classmethod
     def _to_domain_or_skip(cls, row: ScheduledTaskRow) -> ScheduledTask | None:
         try:
             return cls._to_domain(row)
-        except InvalidScheduleError:
-            logger.warning("Skipping scheduled task %s: stored schedule is unparseable", row.id, exc_info=True)
+        except CorruptStoredScheduleError:
+            logger.warning("Skipping scheduled task %s: stored row cannot be rebuilt", row.id, exc_info=True)
             return None
 
     @staticmethod

backend/packages/harness/deerflow/domain/schedule/__init__.py

@@ -18,6 +18,7 @@ from deerflow.domain.schedule.commands import (
 )
 from deerflow.domain.schedule.exceptions import (
     ActiveRunConflictError,
+    CorruptStoredScheduleError,
     InvalidContextModeError,
     InvalidScheduleError,
     LaunchFailedError,
@@ -52,6 +53,7 @@ __all__ = [
     "ActiveRunConflictError",
     "ContextChange",
     "ContextMode",
+    "CorruptStoredScheduleError",
     "CreateScheduledTask",
     "DeleteTask",
     "DispatchOutcome",

backend/packages/harness/deerflow/domain/schedule/exceptions.py

@@ -33,6 +33,17 @@ class ThreadNotFoundError(ScheduleError):
     """reuse_thread points at a thread the user cannot access."""
 
 
+class CorruptStoredScheduleError(ScheduleError):
+    """A stored task row can no longer be rebuilt into a valid aggregate.
+
+    Raised by the persistence adapter, never by the aggregate: it means the
+    *storage* is damaged, not that a client submitted something invalid --
+    which is why it is deliberately absent from the router's status table and
+    falls through to the unclassified-500 branch instead of riding
+    ``InvalidScheduleError``'s 422.
+    """
+
+
 class ActiveRunConflictError(ScheduleError):
     """The task already holds its single active run slot.
 

backend/tests/test_schedule_corrupt_rows.py

@@ -0,0 +1,126 @@
+"""A corrupt stored schedule is an operator problem, not a client error.
+
+``SqlScheduledTaskRepository._to_domain`` rebuilds the aggregate on every
+read, so a row whose stored schedule no longer parses surfaces at read time.
+It used to surface as ``InvalidScheduleError`` -- the same error the aggregate
+raises for a *client-submitted* schedule, which the router maps to 422. That
+double duty told the client "your request is wrong" about a request that was
+perfectly fine, and made the row unrepairable over HTTP: PATCH reads the task
+before writing, so the fix path 422'd too.
+
+``CorruptStoredScheduleError`` splits the vocabulary: it is raised only by the
+persistence adapter, and it is deliberately absent from the router's status
+table so it falls through to the unclassified-500 branch -- a server-side
+fault reported as one.
+
+SQL-only: the in-memory double stores whole aggregates and cannot hold a
+corrupt row, which is exactly why this file exists beside the contract suite
+rather than inside it.
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from datetime import UTC, datetime
+
+import pytest
+import pytest_asyncio
+
+from app.adapters.schedule.scheduled_task_repository import SqlScheduledTaskRepository
+from app.gateway.routers.schedule.router import _STATUS_BY_ERROR
+from deerflow.config.database_config import DatabaseConfig
+from deerflow.domain.schedule.exceptions import CorruptStoredScheduleError, InvalidScheduleError
+from deerflow.domain.schedule.model import ScheduledTask, SchedulePolicy, ScheduleSpec
+from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
+from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow
+
+pytestmark = pytest.mark.asyncio
+
+NOW = datetime(2026, 7, 27, 12, 0, tzinfo=UTC)
+
+
+@pytest_asyncio.fixture
+async def repo(tmp_path) -> AsyncIterator[SqlScheduledTaskRepository]:
+    await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
+    try:
+        sf = get_session_factory()
+        assert sf is not None
+        yield SqlScheduledTaskRepository(sf)
+    finally:
+        await close_engine()
+
+
+async def _add_task(repo: SqlScheduledTaskRepository, *, title: str = "healthy") -> ScheduledTask:
+    return await repo.add(
+        ScheduledTask.create(
+            user_id="user-1",
+            title=title,
+            prompt="p",
+            schedule=ScheduleSpec.cron_schedule("0 9 * * *", "UTC"),
+            context_mode="fresh_thread_per_run",
+            thread_id=None,
+            now=NOW,
+            policy=SchedulePolicy(),
+        )
+    )
+
+
+async def _corrupt(task_id: str, **columns) -> None:
+    """Damage a stored row directly -- the exact shape a bug or a manual edit
+    would leave behind, unreachable through the port."""
+    sf = get_session_factory()
+    assert sf is not None
+    async with sf() as session:
+        row = await session.get(ScheduledTaskRow, task_id)
+        assert row is not None
+        for name, value in columns.items():
+            setattr(row, name, value)
+        await session.commit()
+
+
+class TestSingleRowReads:
+    async def test_a_corrupt_schedule_raises_the_dedicated_error(self, repo):
+        task = await _add_task(repo)
+        await _corrupt(task.task_id, schedule_spec={})
+
+        with pytest.raises(CorruptStoredScheduleError):
+            await repo.get(task.task_id, user_id="user-1")
+
+    async def test_the_dedicated_error_is_not_the_client_facing_one(self, repo):
+        """The router maps InvalidScheduleError to 422; a corrupt row must not
+        ride that mapping."""
+        task = await _add_task(repo)
+        await _corrupt(task.task_id, schedule_spec={})
+
+        with pytest.raises(CorruptStoredScheduleError) as exc_info:
+            await repo.get(task.task_id, user_id="user-1")
+        assert not isinstance(exc_info.value, InvalidScheduleError)
+
+    async def test_a_corrupt_context_mode_is_reported_the_same_way(self, repo):
+        """Enum rebuild failures are the same fault as spec parse failures: a
+        raw ValueError out of the adapter would be a technical exception
+        crossing the boundary untranslated."""
+        task = await _add_task(repo)
+        await _corrupt(task.task_id, context_mode="not-a-mode")
+
+        with pytest.raises(CorruptStoredScheduleError):
+            await repo.get(task.task_id, user_id="user-1")
+
+
+class TestListReads:
+    async def test_a_corrupt_row_does_not_take_down_the_listing(self, repo):
+        healthy = await _add_task(repo, title="healthy")
+        broken = await _add_task(repo, title="broken")
+        await _corrupt(broken.task_id, schedule_spec={})
+
+        listed = await repo.list_by_user("user-1")
+
+        assert [t.task_id for t in listed] == [healthy.task_id]
+
+
+class TestRouterMapping:
+    def test_the_corrupt_row_error_is_unclassified_on_purpose(self):
+        """Absent from the status table means the router's fallthrough turns
+        it into a 500 -- a new protocol decision would have to add it here
+        deliberately."""
+        assert CorruptStoredScheduleError not in _STATUS_BY_ERROR