A schedule fires dispatch() when the workflow declares one

backend/druks/workflows.py

@@ -503,8 +503,9 @@ class Workflow:
     # What this workflow's runs are about, written ``subject = WorkItem`` on the
     # subclass. None for a workflow about nothing, which says so by silence.
     subject = _DeclaredSubject(None)
-    # When set to a cron string, the workflow also registers a schedule that
-    # fires its run() on that cadence (no subject — a framework cron).
+    # When set to a cron string, the workflow also registers a schedule. It fires
+    # dispatch() on that cadence if the class declares one (the policy starts the
+    # real subject-backed run), else run() with no subject (a framework cron).
     every: ClassVar[str | None] = None
     # True holds one warm VM across the run's agent calls (released at gate parks);
     # False gives each call a throwaway VM.
@@ -561,6 +562,19 @@ def __init_subclass__(cls, **kwargs: Any) -> None:
                     f"parameter(s) {sorted(claimed)} — attribution is platform "
                     "routing, not workflow input; name the parameter differently"
                 )
+        if cls.every and getattr(cls, "dispatch", None):
+            required = [
+                name
+                for name, p in inspect.signature(cls.dispatch).parameters.items()
+                if p.default is inspect.Parameter.empty
+                and p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD, p.KEYWORD_ONLY)
+            ]
+            if required:
+                raise WorkflowError(
+                    f"{cls.__name__}.dispatch() requires {required}, but a scheduled "
+                    "dispatch fires with no arguments — a schedule's dispatch must be "
+                    "nullary"
+                )
         _wrap_steps(cls)
         _register_entry(cls)
         workflows.register(cls)
@@ -881,6 +895,15 @@ async def _run_instance(
         await instance._reap_run()
 
 
+async def _dispatch_instance(cls: type[Workflow], _context: dict[str, Any] | None = None) -> Any:
+    # The cron tick runs no workflow of its own kind — no Run row — it just calls
+    # dispatch(), which start()s the real subject-backed run. Body level, not a
+    # step: DBOS only allows the child-start inside start() from a workflow body.
+    # The session gives dispatch()'s reads a transaction, committed on exit.
+    async with step_session():
+        return await cls.dispatch()
+
+
 def _register_entry(cls: type[Workflow]) -> None:
     # The closure binds cls outside the durable arguments: the DBOS workflow
     # NAME (the kind) is what says which class this is, so recovery rebinds by
@@ -892,4 +915,5 @@ async def _entry(subject: dict[str, Any] | None, input: dict[str, Any]) -> None:
     cls._entry = staticmethod(_entry)  # type: ignore[assignment]
 
     if cls.every:
-        register_schedule(cls, partial(_run_instance, cls))
+        fire = _dispatch_instance if getattr(cls, "dispatch", None) else _run_instance
+        register_schedule(cls, partial(fire, cls))

backend/tests/test_durable_sdk.py

@@ -127,6 +127,19 @@ class DailySweep(Workflow):
         async def run(self) -> None:  # pragma: no cover - not fired in tests
             SINK.append("swept")
 
+    class ScheduledDispatch(Workflow):
+        # every + a dispatch() classmethod: the tick fires dispatch(), never the
+        # subjectless run(). dispatch() start()s for real — the enqueue must work
+        # from the scheduled workflow's body (a step is forbidden the child-start).
+        subject = Widget
+        every = "0 */4 * * *"
+
+        @classmethod
+        async def dispatch(cls) -> str:
+            return await cls.start(subject=Widget.get_for_subject_id("313131"))
+
+        async def run(self) -> None: ...
+
     class SubjectFlow(Workflow):
         # Records the subject the platform threaded in, and returns a BaseModel
         # so the result rides its workflow.finished event.
@@ -185,6 +198,7 @@ async def run_multistep(self) -> None:
         SubjectlessConfirmFlow,
         ReviewFlow,
         AttributedFlow,
+        ScheduledDispatch,
     )
 
 
@@ -217,7 +231,7 @@ def rt():
         session.flush()
         session.add_all(
             Widget(id=subject_id)
-            for subject_id in (7, 4242, 636363, 424242, 515151, 878787, 909090)
+            for subject_id in (7, 4242, 636363, 424242, 515151, 878787, 909090, 313131)
         )
         session.add(
             HarnessConnection(
@@ -243,6 +257,7 @@ def rt():
         subjectless_confirm_flow,
         review_flow,
         attributed_flow,
+        scheduled_dispatch,
     ) = _build_units()
     os.environ["DRUKS_DATABASE_URL"] = URL
     init_dbos()
@@ -260,6 +275,7 @@ def rt():
             SubjectlessConfirmFlow=subjectless_confirm_flow,
             ReviewFlow=review_flow,
             AttributedFlow=attributed_flow,
+            ScheduledDispatch=scheduled_dispatch,
         )
     finally:
         shutdown()
@@ -278,6 +294,7 @@ def rt():
         workflows._items.pop("subjectless_confirm_flow", None)
         workflows._items.pop("review_flow", None)
         workflows._items.pop("attributed_flow", None)
+        workflows._items.pop("scheduled_dispatch", None)
         if db_url_snap is None:
             os.environ.pop("DRUKS_DATABASE_URL", None)
         else:
@@ -626,6 +643,53 @@ async def test_every_registers_schedule(rt):
     assert len(params) == 2 and params[1].name == "context"
 
 
+async def test_scheduled_tick_fires_dispatch_not_run(rt):
+    # A workflow that declares dispatch() is subject-backed — its run() can't fire
+    # subjectless. The tick must reach dispatch(), and dispatch()'s start() must
+    # enqueue the real run from the scheduled workflow's body.
+    from datetime import UTC, datetime
+
+    from druks.durable.engine import _scheduled
+
+    _, fn = next(row for row in _scheduled if row[0].kind == "scheduled_dispatch")
+    await fn(datetime.now(UTC), None)
+
+    def dispatched_run():
+        session = get_session(rt.engine)
+        try:
+            return session.execute(
+                select(Run).where(Run.kind == "scheduled_dispatch")
+            ).scalar_one_or_none()
+        finally:
+            session.close()
+
+    deadline = asyncio.get_event_loop().time() + 15
+    while asyncio.get_event_loop().time() < deadline:
+        run = dispatched_run()
+        if run and run.state == RunState.FINISHED:
+            break
+        await asyncio.sleep(0.1)
+    assert run.state == RunState.FINISHED
+    assert run.subject_label == "W-313131"  # about its subject, not subjectless
+
+
+async def test_scheduled_dispatch_must_be_nullary(rt):
+    # The tick fires dispatch() with no arguments, so a required parameter is a
+    # declaration error, caught when the class is defined.
+    from druks.durable.exceptions import WorkflowError
+
+    with pytest.raises(WorkflowError, match="nullary"):
+
+        class NeedsArg(Workflow):
+            every = "0 6 * * *"
+
+            async def run(self) -> None: ...  # pragma: no cover
+
+            @classmethod
+            async def dispatch(cls, target: str) -> str:  # pragma: no cover
+                return target
+
+
 async def test_apply_schedules_drops_undeclared(rt):
     # A schedule the sys-db still holds but no Workflow declares (a renamed or
     # removed cron) must be reconciled away, or it keeps firing a dead name.

docs/writing-an-extension.md

@@ -217,10 +217,28 @@ class Sweep(Workflow):
     every = "0 6 * * *"
 ```
 
-Every parameter of a scheduled workflow needs a default because the scheduler
-supplies no input. Druks evaluates cron expressions in the operator timezone.
-The dashboard can retune or disable a declared schedule but cannot invent a new
-workflow schedule.
+The tick fires the workflow's body with no subject and no input, so every body
+parameter needs a default. A workflow whose runs are *about* something (it
+declares a `subject`) shouldn't fire that way. Give it a `dispatch()` classmethod
+and the schedule fires that instead — it resolves the subject and starts the
+real run:
+
+```python
+class Engage(Workflow):
+    subject = Account
+    every = "0 */4 * * *"
+
+    @classmethod
+    async def dispatch(cls) -> str:
+        return await cls.start(subject=Account.get())
+
+    async def run(self) -> None:
+        ...
+```
+
+A scheduled `dispatch()` fires with no arguments, so it must be nullary. Druks
+evaluates cron expressions in the operator timezone. The dashboard can retune or
+disable a declared schedule but cannot invent a new workflow schedule.
 
 A workflow may declare its own operator settings: