diff --git a/backend/druks/apps/registry.py b/backend/druks/apps/registry.py index a8afa439..9c92b213 100644 --- a/backend/druks/apps/registry.py +++ b/backend/druks/apps/registry.py @@ -8,7 +8,7 @@ # imports exactly these (``routes`` defines routers the loader mounts; the rest # fire registration as an import side effect). The set is the single source of # truth for what "a capability module" is named. -_ROLES = frozenset({"webhooks", "subscribers", "workflows", "routes", "services"}) +_ROLES = frozenset({"webhooks", "subscribers", "workflows", "tasks", "routes", "services"}) class Registry: diff --git a/backend/druks/core/workflows.py b/backend/druks/core/tasks.py similarity index 75% rename from backend/druks/core/workflows.py rename to backend/druks/core/tasks.py index 9fb94e9d..c334b424 100644 --- a/backend/druks/core/workflows.py +++ b/backend/druks/core/tasks.py @@ -5,35 +5,29 @@ from druks.harnesses.registry import get_harnesses from druks.sandbox import gate from druks.user_settings.models import HarnessSettings, UserSettings -from druks.workflows import Workflow +from druks.workflows import task logger = logging.getLogger(__name__) -class RefreshTokens(Workflow): - every = "*/15 * * * *" +@task(every="*/15 * * * *") +async def refresh_tokens() -> None: + # Every 15 min. With an ~8h Claude TTL refreshed at <2h remaining (and + # codex ~10d at <24h), this keeps both tokens alive with a wide margin + # while doing almost nothing on most ticks. + await _refresh() - async def run(self) -> dict[str, object]: - # Every 15 min. With an ~8h Claude TTL refreshed at <2h remaining (and - # codex ~10d at <24h), this keeps both tokens alive with a wide margin - # while doing almost nothing on most ticks. - return await _refresh() - -class RefreshModels(Workflow): - every = "0 6 * * *" - - async def run(self) -> dict[str, object]: - fallback_id = UserSettings.get().fallback_account_id - results = [] - for harness in get_harnesses(): - connections = HarnessConnection.list_for_harness(harness.name) - if not connections: - continue - preferred = [c for c in connections if c.account_id == fallback_id] - settings = HarnessSettings.require(harness.name) - results.append(await settings.refresh_models((preferred or connections)[0])) - return {"results": results} +@task(every="0 6 * * *") +async def refresh_models() -> None: + fallback_id = UserSettings.get().fallback_account_id + for harness in get_harnesses(): + connections = HarnessConnection.list_for_harness(harness.name) + if not connections: + continue + preferred = [c for c in connections if c.account_id == fallback_id] + settings = HarnessSettings.require(harness.name) + await settings.refresh_models((preferred or connections)[0]) async def _refresh() -> dict[str, object]: diff --git a/backend/druks/doctor.py b/backend/druks/doctor.py index 7afeeb07..cde86c6e 100644 --- a/backend/druks/doctor.py +++ b/backend/druks/doctor.py @@ -11,6 +11,7 @@ from urllib.parse import urlparse import httpx +from dbos._dbos import _get_or_create_dbos_registry from drukbox_sdk import SandboxAPI from sqlalchemy import select, text from sqlalchemy.orm import Session @@ -28,7 +29,7 @@ from .settings import Settings, load_settings from .user_settings.models import UserSettings from .webhooks.base import Webhook -from .workflows import Workflow +from .workflows import Workflow, _Task @dataclass(frozen=True) @@ -330,6 +331,8 @@ def _defined_capability(module: ModuleType) -> tuple[str, str] | None: for value in vars(module).values(): if isinstance(value, type) and issubclass(value, Workflow) and value.__module__ == name: return "workflows", value.kind + if isinstance(value, _Task) and value.module == name: + return "tasks", value.name if ( isinstance(value, type) and issubclass(value, Webhook) @@ -369,6 +372,8 @@ def check_capability_modules(settings: Settings) -> CheckResult: # an off-canon module below (which self-registers too) can't mask a stray. autodiscover(package) discovered = {role: set(registry._items) for role, registry in by_role.items()} + # Tasks register straight into DBOS's own map — it is their registry. + discovered["tasks"] = set(_get_or_create_dbos_registry().workflow_info_map) pkg = importlib.import_module(package) for info in pkgutil.walk_packages(pkg.__path__, prefix=f"{package}."): if info.ispkg or info.name.rsplit(".", 1)[-1] in _ROLES: diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py index ead5755e..d7e984d7 100644 --- a/backend/druks/workflows.py +++ b/backend/druks/workflows.py @@ -1,14 +1,24 @@ import inspect -from collections.abc import Callable +from collections.abc import Awaitable, Callable from contextlib import nullcontext, suppress from contextvars import ContextVar from datetime import UTC, datetime from functools import partial -from typing import TYPE_CHECKING, Any, ClassVar, Literal, Self, TypeVar, get_args, get_type_hints +from typing import ( + TYPE_CHECKING, + Any, + ClassVar, + Literal, + Self, + TypeVar, + get_args, + get_type_hints, + overload, +) from croniter import croniter -from dbos import DBOS, SetEnqueueOptions, SetWorkflowAttributes, SetWorkflowID, StepOptions -from dbos._dbos import _get_dbos_instance +from dbos import DBOS, Queue, SetEnqueueOptions, SetWorkflowAttributes, SetWorkflowID, StepOptions +from dbos._dbos import _get_dbos_instance, _get_or_create_dbos_registry from dbos._error import ( DBOSAwaitedWorkflowCancelledError, DBOSQueueDeduplicatedError, @@ -48,8 +58,8 @@ from druks.user_settings.models import SettingsOverride, UserSettings from druks.workspaces import Workspace -# druks.workflows is the author door for workflow authoring: the bases (Workflow, -# Gate, step) defined below plus the author-facing read contracts. The engine +# druks.workflows is the author door for workflow authoring: Workflow, Gate, +# step, task, and the author-facing read contracts. The engine # itself (druks.durable) stays internal. __all__ = [ "AgentCall", @@ -69,6 +79,7 @@ "WorkflowEvent", "set_run_phase", "step", + "task", ] if TYPE_CHECKING: @@ -93,6 +104,8 @@ # step, so it skips wrapping itself; outside, it wraps itself in its own step. _in_step: ContextVar[bool] = ContextVar("_in_step", default=False) +task_queue = Queue("druks_tasks") + # Reserved so _entry's arity and old checkpoints stay untouched; a body # parameter may not claim it. _ACCOUNT_INPUT_KEY = "__account_id__" @@ -117,6 +130,7 @@ def _resolve_body_method(cls: type["Workflow"]) -> str: ) method._durable_step = True method._step_name = None + method._step_retries = 0 return "run" if has_multistep: method = cls.__dict__["run_multistep"] @@ -166,29 +180,28 @@ def _declare_subject(cls: type["Workflow"]) -> None: cls.subject = _DeclaredSubject(declared) -def _input_model_from_signature(cls: type["Workflow"]) -> type[BaseModel] | None: - # A workflow's input IS its body's signature: plain annotated parameters, - # Python's native way to declare inputs. The SDK synthesizes a pydantic - # model from them (the wire contract) — start() validates kwargs against it - # and dumps to JSON (it crosses a JSONB row and a DBOS checkpoint; never a - # live or pickled object), and the entry re-validates. A parameter without a - # default is required at start(); a cron-scheduled workflow must default - # every parameter. - method_name = cls._body_method - method = getattr(cls, method_name) - parameters = [p for name, p in inspect.signature(method).parameters.items() if name != "self"] +def _input_model_from_signature( + function: Callable, *, owner: str, model_name: str +) -> type[BaseModel] | None: + # A body's input IS its signature: plain annotated parameters, Python's + # native way to declare inputs. The synthesized model is the wire contract — + # the caller validates kwargs against it and dumps to JSON (input crosses a + # JSONB row and a DBOS checkpoint; never a live or pickled object), and the + # entry re-validates. A parameter without a default is required; a + # cron-scheduled body must default every parameter. + parameters = [p for name, p in inspect.signature(function).parameters.items() if name != "self"] if not parameters: return - hints = get_type_hints(method) + hints = get_type_hints(function) fields: dict[str, Any] = {} for p in parameters: if p.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD): - raise WorkflowError(f"{cls.__name__}.{method_name}() cannot take *args/**kwargs") + raise WorkflowError(f"{owner} cannot take *args/**kwargs") if p.name not in hints: - raise WorkflowError(f"{cls.__name__}.{method_name}() parameter {p.name!r} needs a type") + raise WorkflowError(f"{owner} parameter {p.name!r} needs a type") default = ... if p.default is inspect.Parameter.empty else p.default fields[p.name] = (hints[p.name], default) - return create_model(f"{cls.__name__}Input", **fields) + return create_model(model_name, **fields) class Journal: @@ -358,20 +371,146 @@ async def _create() -> str | None: await notifications_queue.enqueue_async(send_notification, notification_id) -def step(method: Callable | None = None, *, name: str | None = None) -> Callable: +def step(method: Callable | None = None, *, name: str | None = None, retries: int = 0) -> Callable: """Mark a ``run_multistep()`` helper method as a durable, replay-safe step — its body runs once and its return is memoized for recovery. `name=` pins the - durable step name independent of the method name. A leading `_` carries no - semantics — it's just Python privacy.""" + durable step name independent of the method name. `retries=` sets retries + after the first attempt. A leading `_` is only Python privacy.""" def stamp(m: Callable) -> Callable: m._durable_step = True # type: ignore[attr-defined] m._step_name = name # type: ignore[attr-defined] + m._step_retries = retries # type: ignore[attr-defined] return m return stamp(method) if method else stamp +class _Task: + def __init__( + self, + function: Callable[..., Awaitable[Any]], + *, + every: str | None, + retries: int, + ) -> None: + self._function = function + self.module = function.__module__ + try: + app = resolve_workflow_app(function.__module__) + except LookupError: + raise WorkflowError( + f"{function.__module__} declares task {function.__name__} outside every " + "registered app package — task modules load through " + "druks.apps.loader; a module the loader doesn't own must " + "register_workflow_package() before importing" + ) from None + self.name = f"{app}.{function.__name__}" if app else function.__name__ + self._retries = retries + self._input_model = _input_model_from_signature( + function, owner=f"task {self.name}", model_name=f"{function.__name__}_input" + ) + self._scheduled_entry: Callable[..., Any] | None = None + + # DBOS only warns on a duplicate durable name and lets the last + # registration win — enqueues would silently run the other body. + if self.name in _get_or_create_dbos_registry().workflow_info_map: + raise WorkflowError( + f"task {self.name} shares its durable name with a registered workflow " + "or task — two capabilities can't share a durable identity; rename one" + ) + if every and self._input_model: + required = [ + name + for name, field in self._input_model.model_fields.items() + if field.is_required() + ] + if required: + raise WorkflowError( + f"task {self.name} requires {required}, but a scheduled task fires " + "with no arguments — a scheduled task must be nullary" + ) + + @DBOS.workflow(name=self.name) + async def _entry(input: dict[str, Any]) -> None: + await self._run(input) + + self._entry = _entry + + if every: + + @DBOS.scheduled(every) + @DBOS.workflow(name=f"{self.name}.scheduled") + async def _scheduled_entry(scheduled_at: datetime, started_at: datetime | None) -> None: + await self._run({}) + + self._scheduled_entry = _scheduled_entry + + async def enqueue(self, **input: Any) -> None: + if _in_step.get(): + raise WorkflowError( + "enqueue() cannot run inside a @step — a retried step would enqueue the task again" + ) + wire: dict[str, Any] = {} + if self._input_model: + wire = self._input_model.model_validate(input).model_dump(mode="json") + elif input: + raise WorkflowError(f"task {self.name} takes no input") + await task_queue.enqueue_async(self._entry, wire) + + async def _run(self, input: dict[str, Any]) -> None: + kwargs: dict[str, Any] = {} + if self._input_model: + validated = self._input_model.model_validate(input) + kwargs = {name: getattr(validated, name) for name in type(validated).model_fields} + + async def _do() -> None: + async with step_session(): + await self._function(**kwargs) + + await DBOS.run_step_async( + StepOptions( + name=self.name, + retries_allowed=self._retries > 0, + max_attempts=self._retries + 1, + ), + _do, + ) + + +@overload +def task( + function: Callable[..., Awaitable[Any]], + *, + every: str | None = None, + retries: int = 0, +) -> _Task: ... + + +@overload +def task( + function: None = None, + *, + every: str | None = None, + retries: int = 0, +) -> Callable[[Callable[..., Awaitable[Any]]], _Task]: ... + + +def task( + function: Callable[..., Awaitable[Any]] | None = None, + *, + every: str | None = None, + retries: int = 0, +) -> _Task | Callable[[Callable[..., Awaitable[Any]]], _Task]: + """Make an async function a durable background task. ``every=`` adds a UTC + cron, and ``retries=`` sets retries after the first attempt.""" + + def decorate(declared: Callable[..., Awaitable[Any]]) -> _Task: + return _Task(declared, every=every, retries=retries) + + return decorate(function) if function else decorate + + # Lifecycle steps do real IO (DB, Redis, event subscribers), so a # transient failure must not become a failed run — a lost reaction has no # redelivery, unlike a webhook. @@ -571,7 +710,11 @@ def __init_subclass__(cls, **kwargs: Any) -> None: validate_settings_declaration(cls.Settings) cls._body_method = _resolve_body_method(cls) # Before _wrap_steps: run()'s wrapper signature is (*args, **kwargs). - cls._run_input_model = _input_model_from_signature(cls) + cls._run_input_model = _input_model_from_signature( + getattr(cls, cls._body_method), + owner=f"{cls.__name__}.{cls._body_method}()", + model_name=f"{cls.__name__}Input", + ) if cls._run_input_model: claimed = {"account_id"} & set(cls._run_input_model.model_fields) if claimed: @@ -846,10 +989,11 @@ def _wrap_steps(cls: type[Workflow]) -> None: for method_name, method in list(vars(cls).items()): if getattr(method, "_durable_step", False): name = getattr(method, "_step_name", None) or method_name - setattr(cls, method_name, _make_step(cls.kind, name, method)) + retries = getattr(method, "_step_retries", 0) + setattr(cls, method_name, _make_step(cls.kind, name, method, retries)) -def _make_step(kind: str, name: str, method: Callable) -> Callable: +def _make_step(kind: str, name: str, method: Callable, retries: int) -> Callable: # Run the method inside its own session via a zero-arg closure, so `self` is # never serialized into the DBOS checkpoint. async def _step(self: Workflow, *args: Any, **kwargs: Any) -> Any: @@ -861,7 +1005,11 @@ async def _do() -> Any: finally: _in_step.reset(token) - return await DBOS.run_step_async(StepOptions(name=f"{kind}.{name}"), _do) + options = StepOptions(name=f"{kind}.{name}") + if retries > 0: + options["retries_allowed"] = True + options["max_attempts"] = retries + 1 + return await DBOS.run_step_async(options, _do) _step.__name__ = name _step.__wrapped__ = method # lets a test call run() without DBOS diff --git a/backend/tests/test_api_settings.py b/backend/tests/test_api_settings.py index 90da27bd..e5ffb330 100644 --- a/backend/tests/test_api_settings.py +++ b/backend/tests/test_api_settings.py @@ -411,7 +411,6 @@ def test_incoherent_app_save_is_rejected_and_rolled_back_before_schedules( "/api/settings/apps", json={ "agentModels": {"generate_plan": "claude-opus-4-7"}, - "workflowSettings": {"core.refresh_tokens": {"schedule": "0 9 * * *"}}, "appSettings": {"review": {"app_id": "42"}}, }, ) @@ -424,7 +423,6 @@ def test_incoherent_app_save_is_rejected_and_rolled_back_before_schedules( assert _review_settings_fields(client)["app_id"]["secretSet"] is False agents = {agent["name"]: agent for agent in _ship_app(client)["agents"]} assert agents["generate_plan"]["model"] == "gpt-5.5" - assert _refresh_tokens_fields(client)["schedule"]["value"] == "*/15 * * * *" def test_clearing_the_identity_deletes_its_overrides_and_stays_coherent(tmp_path: Path): @@ -550,65 +548,6 @@ def test_build_review_code_is_a_workflow_setting(tmp_path: Path): assert fields["review_code"]["overridden"] is True -def _app(client: TestClient, name: str) -> dict: - body = client.get("/api/settings/apps").json() - return next(m for m in body["apps"] if m["name"] == name) - - -def _refresh_tokens_fields(client: TestClient) -> dict: - workflows = {w["kind"]: w for w in _app(client, "core")["workflows"]} - return {f["name"]: f for f in workflows["core.refresh_tokens"]["fields"]} - - -def test_scheduled_workflow_surfaces_schedule_fields(tmp_path: Path): - """A workflow's every= surfaces as two ordinary settings fields on the - app that owns it.""" - with _build_client(tmp_path) as client: - fields = _refresh_tokens_fields(client) - assert fields["schedule"]["value"] == "*/15 * * * *" - assert fields["schedule"]["default"] == "*/15 * * * *" - assert fields["schedule"]["overridden"] is False - assert fields["schedule_enabled"]["value"] is True - assert fields["schedule_enabled"]["type"] == "bool" - - -def test_schedule_override_persists_and_reconciles(tmp_path: Path, monkeypatch): - """Overriding the cadence or pausing persists like any workflow setting and - repoints the DBOS crons now, not at the next launch.""" - reconciled = [] - monkeypatch.setattr( - "druks.user_settings.routes.apply_schedules", lambda: reconciled.append(True) - ) - with _build_client(tmp_path) as client: - patch = client.patch( - "/api/settings/apps", - json={ - "workflowSettings": { - "core.refresh_tokens": { - "schedule": "0 9 * * *", - "schedule_enabled": False, - } - } - }, - ) - assert patch.status_code == 200 - assert reconciled - fields = _refresh_tokens_fields(client) - assert fields["schedule"]["value"] == "0 9 * * *" - assert fields["schedule"]["overridden"] is True - assert fields["schedule_enabled"]["value"] is False - - -def test_schedule_rejects_invalid_cron(tmp_path: Path): - # A malformed cron would be silently never-fired by DBOS — reject at the write. - with _build_client(tmp_path) as client: - patch = client.patch( - "/api/settings/apps", - json={"workflowSettings": {"core.refresh_tokens": {"schedule": "not a cron"}}}, - ) - assert patch.status_code == 422 - - def test_apps_clearing_an_override_reverts_to_the_family_default(tmp_path: Path): with _build_client(tmp_path) as client: client.patch( diff --git a/backend/tests/test_author_surface.py b/backend/tests/test_author_surface.py index a71cdded..69abc835 100644 --- a/backend/tests/test_author_surface.py +++ b/backend/tests/test_author_surface.py @@ -34,6 +34,7 @@ "WorkflowEvent", "set_run_phase", "step", + "task", }, "druks.db": {"Base", "StoredSubject", "db_session"}, "druks.schemas": {"BaseResponse"}, diff --git a/backend/tests/test_core_workflows.py b/backend/tests/test_core_workflows.py index da36eb51..a7e0baf7 100644 --- a/backend/tests/test_core_workflows.py +++ b/backend/tests/test_core_workflows.py @@ -1,6 +1,6 @@ import logging -from druks.core.workflows import _log_result +from druks.core.tasks import _log_result from druks.harnesses.datastructures import RotationResult diff --git a/backend/tests/test_doctor_capability_modules.py b/backend/tests/test_doctor_capability_modules.py index d5792cbc..e1fd701f 100644 --- a/backend/tests/test_doctor_capability_modules.py +++ b/backend/tests/test_doctor_capability_modules.py @@ -25,7 +25,9 @@ def get_action(self) -> str: """ -def _temp_capability_package(tmp_path: Path, monkeypatch, *, module_name: str) -> str: +def _temp_capability_package( + tmp_path: Path, monkeypatch, *, module_name: str, source: str = _CAPABILITY_SOURCE +) -> str: """A real, importable one-module package whose capability lives in ``{module_name}.py`` — canonical (``webhooks``) or off-canon (``webhook``). Points the check's package walk at it and returns the package name.""" @@ -33,7 +35,7 @@ def _temp_capability_package(tmp_path: Path, monkeypatch, *, module_name: str) - pkg_dir = tmp_path / package pkg_dir.mkdir() (pkg_dir / "__init__.py").write_text("") - (pkg_dir / f"{module_name}.py").write_text(_CAPABILITY_SOURCE) + (pkg_dir / f"{module_name}.py").write_text(source) monkeypatch.syspath_prepend(str(tmp_path)) for name in list(sys.modules): @@ -59,6 +61,25 @@ def test_capability_under_off_canon_filename_is_flagged(tmp_path: Path, monkeypa assert "rename to webhooks.py" in result.detail +def test_task_under_off_canon_filename_is_flagged(tmp_path: Path, monkeypatch) -> None: + from druks.apps.loader import register_workflow_package + + _temp_capability_package( + tmp_path, + monkeypatch, + module_name="task", + source="from druks.workflows import task\n\n\n@task\nasync def tock() -> None: ...\n", + ) + register_workflow_package("doctorprobe", None) + settings = make_settings(tmp_path) + + result = doctor.check_capability_modules(settings) + + assert not result.ok + assert "doctorprobe.task" in result.detail + assert "rename to tasks.py" in result.detail + + def test_capability_under_canonical_filename_passes(tmp_path: Path, monkeypatch) -> None: _temp_capability_package(tmp_path, monkeypatch, module_name="webhooks") settings = make_settings(tmp_path) diff --git a/backend/tests/test_durable_sdk.py b/backend/tests/test_durable_sdk.py index a4c5403a..920ef09a 100644 --- a/backend/tests/test_durable_sdk.py +++ b/backend/tests/test_durable_sdk.py @@ -13,7 +13,7 @@ from druks.durable.engine import configure_engine, init_dbos, launch, shutdown from druks.models import StoredSubject from druks.testing import init_db -from druks.workflows import Gate, Subject, Workflow, step +from druks.workflows import Gate, Subject, Workflow, step, task from pydantic import BaseModel from sqlalchemy import create_engine, select @@ -49,6 +49,8 @@ class RepoCfg(BaseModel): SINK: list[str] = [] +TASK_RETRY_ATTEMPTS = 0 +STEP_RETRY_ATTEMPTS = 0 class Widget(StoredSubject): @@ -65,6 +67,22 @@ class Gadget(Subject): # run_multistep() below for fixtures using @step/a gate; run() for the rest. def _build_units(): + @task + async def record_task(repo: str) -> None: + SINK.append(f"task:{repo}") + + @task(retries=1) + async def retry_task() -> None: + global TASK_RETRY_ATTEMPTS + TASK_RETRY_ATTEMPTS += 1 + if TASK_RETRY_ATTEMPTS == 1: + raise RuntimeError("retry task") + SINK.append("task:retried") + + @task(every="0 6 * * *") + async def scheduled_task() -> None: + SINK.append("task:scheduled") + class Approve(Gate): # The on_wait override is what lets the subjectless flows below park # here at all — without it every wait() would fail as SubjectlessGate. @@ -120,6 +138,27 @@ async def note_repo(self, repo: str) -> None: async def run_multistep(self, repo: str) -> None: await self.note_repo(repo) + class RetryingStepFlow(Workflow): + @step(retries=1) + async def unreliable(self) -> None: + global STEP_RETRY_ATTEMPTS + STEP_RETRY_ATTEMPTS += 1 + if STEP_RETRY_ATTEMPTS == 1: + raise RuntimeError("retry step") + SINK.append("step:retried") + + async def run_multistep(self) -> None: + await self.unreliable() + + class EnqueueInStepFlow(Workflow): + # A retried step would enqueue again, so enqueue() must refuse in-step. + @step + async def misuse(self) -> None: + await record_task.enqueue(repo="from-a-step") + + async def run_multistep(self) -> None: + await self.misuse() + # every= so launch()'s apply_schedules has a schedule to create (smoke). class DailySweep(Workflow): every = "0 6 * * *" @@ -199,6 +238,11 @@ async def run_multistep(self) -> None: ReviewFlow, AttributedFlow, ScheduledDispatch, + RetryingStepFlow, + EnqueueInStepFlow, + record_task, + retry_task, + scheduled_task, ) @@ -258,6 +302,11 @@ def rt(): review_flow, attributed_flow, scheduled_dispatch, + retrying_step_flow, + enqueue_in_step_flow, + record_task, + retry_task, + scheduled_task, ) = _build_units() os.environ["DRUKS_DATABASE_URL"] = URL init_dbos() @@ -276,6 +325,11 @@ def rt(): ReviewFlow=review_flow, AttributedFlow=attributed_flow, ScheduledDispatch=scheduled_dispatch, + RetryingStepFlow=retrying_step_flow, + EnqueueInStepFlow=enqueue_in_step_flow, + record_task=record_task, + retry_task=retry_task, + scheduled_task=scheduled_task, ) finally: shutdown() @@ -295,6 +349,8 @@ def rt(): workflows._items.pop("review_flow", None) workflows._items.pop("attributed_flow", None) workflows._items.pop("scheduled_dispatch", None) + workflows._items.pop("retrying_step_flow", None) + workflows._items.pop("enqueue_in_step_flow", None) if db_url_snap is None: os.environ.pop("DRUKS_DATABASE_URL", None) else: @@ -668,6 +724,80 @@ async def test_task_enqueue(rt): assert "owner/queued" in SINK +async def test_durable_task_enqueue(rt): + SINK.clear() + await rt.record_task.enqueue(repo="owner/queued") + deadline = asyncio.get_event_loop().time() + 15 + while "task:owner/queued" not in SINK and asyncio.get_event_loop().time() < deadline: + await asyncio.sleep(0.1) + assert "task:owner/queued" in SINK + + +async def test_durable_task_retries(rt): + global TASK_RETRY_ATTEMPTS + TASK_RETRY_ATTEMPTS = 0 + SINK.clear() + await rt.retry_task.enqueue() + deadline = asyncio.get_event_loop().time() + 15 + while "task:retried" not in SINK and asyncio.get_event_loop().time() < deadline: + await asyncio.sleep(0.1) + assert "task:retried" in SINK + assert TASK_RETRY_ATTEMPTS > 1 + + +async def test_step_retries(rt): + global STEP_RETRY_ATTEMPTS + STEP_RETRY_ATTEMPTS = 0 + SINK.clear() + workflow_id = await rt.RetryingStepFlow.start(subject=None) + await _wait_for(rt.engine, workflow_id, lambda run: run.state == RunState.FINISHED) + assert "step:retried" in SINK + assert STEP_RETRY_ATTEMPTS > 1 + + +async def test_enqueue_inside_a_step_fails_the_run(rt): + wfid = await rt.EnqueueInStepFlow.start(subject=None) + failed = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FAILED) + assert "inside a @step" in failed.failure + + +async def test_scheduled_task_runs_nullary_body(rt): + from datetime import UTC, datetime + + SINK.clear() + await rt.scheduled_task._scheduled_entry(datetime.now(UTC), None) + assert "task:scheduled" in SINK + + +async def test_scheduled_task_must_be_nullary(rt): + from druks.durable.exceptions import WorkflowError + + with pytest.raises(WorkflowError, match="nullary"): + + @task(every="0 6 * * *") + async def needs_argument(target: str) -> None: ... + + +async def test_task_name_uses_declaring_app(rt): + from druks.apps.loader import register_workflow_package + + register_workflow_package("plain_task_package", None) + register_workflow_package("app_task_package", "alpha") + + async def bare_task() -> None: ... + + bare_task.__module__ = "plain_task_package.tasks" + bare = task(bare_task) + + async def summarize() -> None: ... + + summarize.__module__ = "app_task_package.tasks" + namespaced = task(summarize) + + assert bare.name == "bare_task" + assert namespaced.name == "alpha.summarize" + + async def test_every_registers_schedule(rt): # A Workflow with every= registers (schedule_name, cron, fn) so launch()'s # apply_schedules creates the DBOS cron. The fn must satisfy DBOS's diff --git a/backend/tests/test_gate.py b/backend/tests/test_gate.py index f8a04cec..92976fb2 100644 --- a/backend/tests/test_gate.py +++ b/backend/tests/test_gate.py @@ -3,7 +3,7 @@ import druks.redis import pytest -from druks.core import workflows as harness_workflows +from druks.core import tasks as harness_workflows from druks.harnesses.datastructures import RotationResult from druks.sandbox import gate diff --git a/backend/tests/test_workflow_identity.py b/backend/tests/test_workflow_identity.py index 8f88a5ff..93a27e0b 100644 --- a/backend/tests/test_workflow_identity.py +++ b/backend/tests/test_workflow_identity.py @@ -3,13 +3,13 @@ from druks.apps.exceptions import MalformedApp from druks.apps.loader import register_workflow_package, resolve_workflow_app from druks.apps.registry import workflows -from druks.core.workflows import RefreshTokens +from druks.core.tasks import refresh_tokens from druks.durable.enums import RunState from druks.durable.exceptions import WorkflowError from druks.durable.models import Run from druks.durable.schemas import get_display_label from druks.events.models import Event -from druks.workflows import Gate, Workflow, _log_run_event, step +from druks.workflows import Gate, Workflow, _log_run_event, step, task from druks_field_notes.models import Note @@ -60,6 +60,39 @@ def test_resolution_matches_package_boundaries(): resolve_workflow_app("alpha_pkg_sibling.workflows") +def _task_in(module: str, body, name: str | None = None): + body.__module__ = module + if name: + body.__name__ = name + return task(body) + + +def test_duplicate_task_name_is_rejected(): + # DBOS only warns on a duplicate durable name and lets the last registration + # win — enqueues of one task would silently run the other's body. + register_workflow_package("collide_pkg", None) + + async def clash() -> None: ... + + _task_in("collide_pkg.tasks", clash) + + async def other() -> None: ... + + with pytest.raises(WorkflowError, match="durable identity"): + _task_in("collide_pkg.nested.tasks", other, name="clash") + + +def test_task_input_is_typed(): + # The signature is the wire contract — enqueue validates against it and + # dumps to JSON, so an unannotated parameter has no wire shape. + register_workflow_package("untyped_pkg", None) + + async def send(recipient) -> None: ... # noqa: ANN001 + + with pytest.raises(WorkflowError, match="needs a type"): + _task_in("untyped_pkg.tasks", send) + + def test_unregistered_module_fails_at_class_definition(): # The error carries the invariant: load through the loader, or register first. with pytest.raises(WorkflowError, match="druks.apps.loader"): @@ -98,12 +131,10 @@ def test_none_owned_package_keeps_bare_kinds(): def test_in_tree_identities_are_stable(): - # These kinds are durable identities (DBOS workflow names, settings keys, - # dedup prefixes, step-name prefixes) — byte-for-byte pins. + # These are durable DBOS names — byte-for-byte pins. assert workflows.get("ship.build") is not None assert workflows.get("ship.profile") is not None - assert RefreshTokens.kind == "core.refresh_tokens" - assert (workflows.get("ship.build").app, RefreshTokens.app) == ("ship", "core") + assert refresh_tokens.name == "core.refresh_tokens" def test_steps_capture_the_namespaced_kind(): diff --git a/docs/writing-an-app.md b/docs/writing-an-app.md index e55d6619..b7c7d69e 100644 --- a/docs/writing-an-app.md +++ b/docs/writing-an-app.md @@ -57,6 +57,7 @@ package modules: | --- | --- | | `app.py` | `App` subclass, agents, app settings | | `workflows.py` | durable `Workflow` and `Gate` subclasses | +| `tasks.py` | `@task` background functions; add when needed | | `models.py` | SQLAlchemy models with `_` table names, `StoredSubject` among them | | `contracts.py` | `AgentOutput` contracts | | `schemas.py` | HTTP responses and subject summaries | @@ -67,7 +68,7 @@ package modules: | `migrations/versions/` | this distribution's Alembic history | | `dist/` | optional built frontend module, mounted inside the shell (served under `/app/`) | -Druks recursively discovers leaf modules named `workflows`, `routes`, +Druks recursively discovers leaf modules named `workflows`, `tasks`, `routes`, `subscribers`, `webhooks`, and `services`. A capability hidden in `workflow.py` is not discovered. Ordinary names such as `policy.py` and `workspace.py` have no import side effect unless a discovered module imports them. @@ -240,6 +241,35 @@ 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. +### Background tasks + +A `Workflow` is the right home for work you want on a subject's timeline — a run +with agent calls, gates, and operator-tunable settings. Plumbing that wants none +of that — periodic maintenance, a fire-and-forget side effect — is a `task`: + +```python +from druks.workflows import task + + +@task(every="*/15 * * * *") +async def refresh_tokens() -> None: + ... + + +@task(retries=4) +async def sync_labels(pull_request_id: int) -> None: + ... +``` + +Call `await sync_labels.enqueue(pull_request_id=7)` to run one durably in the +background — from a route, a subscriber, or a workflow body (never inside a +`@step`). Like a workflow, the signature is the wire contract: parameters are +annotated, and `enqueue()` validates them and stores JSON. A task keeps no run +row and never reaches the timeline; it has no subject, gate, or operator +settings, and it cannot make agent calls. `every=` runs it on a fixed UTC cadence +the code owns — a workflow's `every=` is the one an operator can retune. +`retries=` sets retries after the first attempt, both here and on `@step`. + A workflow may declare its own operator settings: ```python