From 4c8c45a9b0329524a8f673c014c2e526029e5ce0 Mon Sep 17 00:00:00 2001 From: Paulo Date: Thu, 10 Sep 2026 10:39:47 +0200 Subject: [PATCH] Record domain announcements through announce() (DRU-505) Workflow.announce records the fact in one checkpoint and notifies subscribers in another. Subject and StoredSubject announce through the event log in the current transaction, with the registered app as owner. --- backend/druks/durable/datastructures.py | 5 ++ backend/druks/events/models.py | 31 ++++++++++- backend/druks/models.py | 7 +++ backend/druks/workflows.py | 61 ++++++++++++---------- backend/tests/test_announcements.py | 68 +++++++++++++++++++++++++ backend/tests/test_durable_sdk.py | 55 ++++++++++++++++++++ backend/tests/test_run_state.py | 23 +++++---- docs/writing-an-app.md | 36 +++++++++++-- 8 files changed, 243 insertions(+), 43 deletions(-) create mode 100644 backend/tests/test_announcements.py diff --git a/backend/druks/durable/datastructures.py b/backend/druks/durable/datastructures.py index dc84d9e6..067f56a8 100644 --- a/backend/druks/durable/datastructures.py +++ b/backend/druks/durable/datastructures.py @@ -2,6 +2,7 @@ from dataclasses import dataclass from typing import TYPE_CHECKING, Any, ClassVar, Self +from druks.events.models import Event from druks.models import snake_name if TYPE_CHECKING: @@ -38,6 +39,10 @@ def label(self) -> str: # the handle, not a surrogate key. return self.id + async def announce(self, topic: str, **facts: Any) -> None: + """Record and deliver a domain fact in the current transaction.""" + await Event.announce(self, topic, facts) + @classmethod async def get_for_subject_id(cls, subject_id: str) -> Self | None: """The subject this id names. Ids reach the read side as free text off a URL, diff --git a/backend/druks/events/models.py b/backend/druks/events/models.py index bf6baff6..7333e358 100644 --- a/backend/druks/events/models.py +++ b/backend/druks/events/models.py @@ -1,12 +1,15 @@ from datetime import datetime -from typing import Any +from typing import TYPE_CHECKING, Any from sqlalchemy import Index from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column from druks.database import db_session -from druks.models import Base +from druks.models import Base, StoredSubject + +if TYPE_CHECKING: + from druks.durable.datastructures import Subject class Event(Base): @@ -55,3 +58,27 @@ async def emit( ) ) await db_session().flush() + + @classmethod + async def announce( + cls, subject: "Subject | StoredSubject", topic: str, facts: dict[str, Any] + ) -> None: + """Record a subject's domain fact and notify subscribers, in the current + transaction. A failing subscriber rolls the domain change back with it.""" + # The apps package, the signals bus, and the durable engine are built on this log. + from druks.apps.loader import resolve_workflow_app + from druks.durable.exceptions import WorkflowError + from druks.signals import publish + + try: + app = resolve_workflow_app(type(subject).__module__) + except LookupError: + raise WorkflowError( + f"{type(subject).__module__} declares subject {type(subject).__name__} outside " + "every registered app package. Call register_workflow_package() for the " + "package before importing it." + ) from None + await cls.emit( + type=topic, subject=subject.identity, label=subject.label, payload=facts, app=app + ) + await publish(topic, subject=subject.identity, **facts) diff --git a/backend/druks/models.py b/backend/druks/models.py index b7e38640..a0f1da75 100644 --- a/backend/druks/models.py +++ b/backend/druks/models.py @@ -74,6 +74,13 @@ def get_label(self) -> str: def label(self) -> str: return self.get_label() + async def announce(self, topic: str, **facts: Any) -> None: + """Record and deliver a domain fact in the current transaction.""" + # The event log is built on this module's Base. + from druks.events.models import Event + + await Event.announce(self, topic, facts) + @classmethod async def get_for_subject_id(cls, subject_id: str) -> Self | None: """The row this subject id names. A subject id is free text and reaches the diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py index a57f7828..35ad61a8 100644 --- a/backend/druks/workflows.py +++ b/backend/druks/workflows.py @@ -156,19 +156,18 @@ def __init__(self, subject_class: "type[Subject] | type[StoredSubject] | None") self.subject_class = subject_class def __get__(self, run: "Workflow | None", owner: type) -> Any: - if run is None: - return self.subject_class - - # Live, not a snapshot taken at dispatch: a long-parked run resumes against - # whatever the declared class says then, and finds nothing if it went away. - # Awaitable either way, so ``await self.subject`` is the one shape. - async def resolve() -> Any: - if "subject" in run.__dict__: - return run.__dict__["subject"] - if run._subject: - return await self.subject_class.get_for_subject_id(str(run._subject["id"])) - - return resolve() + if run: + # Live, not a snapshot taken at dispatch: a long-parked run resumes against + # whatever the declared class says then, and finds nothing if it went away. + # Awaitable either way, so ``await self.subject`` is the one shape. + async def resolve() -> Any: + if "subject" in run.__dict__: + return run.__dict__["subject"] + if run._subject: + return await self.subject_class.get_for_subject_id(str(run._subject["id"])) + + return resolve() + return self.subject_class def __set__(self, run: "Workflow", value: Any) -> None: # A test hands the run its subject directly; ``await self.subject`` @@ -777,19 +776,28 @@ def __init__(self) -> None: self._host_secrets_id = "" async def announce(self, topic: str, **facts: Any) -> None: - # The workflow announcing a domain event in its app's vocabulary - # ("pr.opened", pr_number=12, branch="agent/eng-8"). The platform injects - # the routing subscribers filter on, and the publish runs as its own - # retrying checkpoint so a recovery replay doesn't re-fire it. Body-only, - # enforced: the checkpoint is a step, so it can't nest inside one. + """Record a domain fact, then notify subscribers in a separate checkpoint.""" if _in_step.get(): raise WorkflowError("announce() runs in the workflow body, not inside a @step") - async def _fan_out() -> None: + async def record() -> None: + async with step_session(): + run = await Run.get(self.workflow_id) + await Event.emit( + type=topic, + subject=self._subject, + label=run.subject_label, + payload={**facts, "run": self.workflow_id, "kind": self.kind}, + app=self.app, + ) + + await DBOS.run_step_async(StepOptions(name=topic, **_IO_RETRIES), record) + + async def notify() -> None: async with step_session(): await publish(topic, subject=self._subject, kind=self.kind, **facts) - await DBOS.run_step_async(StepOptions(name=topic, **_IO_RETRIES), _fan_out) + await DBOS.run_step_async(StepOptions(name=f"{topic}:propagate", **_IO_RETRIES), notify) async def review( self, *, questions: list[BaseModel] | None = None, context: str = "" @@ -945,14 +953,13 @@ def _validate_subject(cls, subject: "Subject | StoredSubject | None") -> None: if cls.subject: if isinstance(subject, cls.subject): return - given = "nothing" if subject is None else type(subject).__name__ + given = type(subject).__name__ if subject else "nothing" raise WorkflowError(f"{cls.__name__} is about {cls.subject.__name__}, not {given}") - if subject is None: - return - raise WorkflowError( - f"{cls.__name__} declares no subject — declare " - f"``subject = {type(subject).__name__}`` on it, or pass subject=None" - ) + if subject: + raise WorkflowError( + f"{cls.__name__} declares no subject — declare " + f"``subject = {type(subject).__name__}`` on it, or pass subject=None" + ) @classmethod async def cancel(cls, subject: Subject | StoredSubject, *, failure: str | None = None) -> None: diff --git a/backend/tests/test_announcements.py b/backend/tests/test_announcements.py new file mode 100644 index 00000000..e66c9468 --- /dev/null +++ b/backend/tests/test_announcements.py @@ -0,0 +1,68 @@ +import pytest +from druks.apps import loader +from druks.db import db_session +from druks.events.models import Event +from druks.signals import subscribe +from druks.workflows import Subject, WorkflowError +from druks_field_notes.models import Note +from sqlalchemy import select + + +class Report(Subject): + pass + + +async def test_identity_subject_uses_its_registered_app(druks_db, monkeypatch): + monkeypatch.setitem(loader._workflow_packages, __name__, "reports") + report = Report(id="owner/repository#7") + received = [] + + @subscribe("report.published", subject=Report) + async def receive(*, subject: Report, url: str) -> None: + received.append((subject, url)) + + await report.announce("report.published", url="https://example.com/report/7") + + event = (await druks_db.scalars(select(Event).filter_by(type="report.published"))).one() + assert event.app == "reports" + assert event.subject_type == "report" + assert event.subject_id == report.id + assert event.subject_label == report.label + assert event.payload == {"url": "https://example.com/report/7"} + assert received == [(report, "https://example.com/report/7")] + + +async def test_unregistered_subject_names_the_missing_registration(druks_db): + with pytest.raises(WorkflowError, match="register_workflow_package"): + await Report(id="owner/repository#7").announce("report.published") + + +async def test_stored_subject_announces_with_its_app(druks_db): + note = await Note.create(body="An observation") + + await note.announce("note.revised", revision=2) + + event = (await druks_db.scalars(select(Event).filter_by(type="note.revised"))).one() + assert event.app == "field_notes" + assert event.subject_id == str(note.id) + assert event.subject_label == note.label + assert event.payload == {"revision": 2} + + +async def test_subject_delivery_error_rolls_back_with_the_domain_transaction(druks_db): + # The savepoint below is the fixture session's, so the announce must run on it. + db_session.registry.set(druks_db) + note = await Note.create(body="An observation") + + @subscribe("note.delivery_failed", subject=Note) + async def fail(**facts: object) -> None: + raise RuntimeError("Subscriber unavailable") + + with pytest.raises(RuntimeError, match="Subscriber unavailable"): + async with druks_db.begin_nested(): + note.gist = "A summary" + await note.announce("note.delivery_failed") + + await druks_db.refresh(note) + assert not note.gist + assert not list(await druks_db.scalars(select(Event).filter_by(type="note.delivery_failed"))) diff --git a/backend/tests/test_durable_sdk.py b/backend/tests/test_durable_sdk.py index b9798cee..178352e8 100644 --- a/backend/tests/test_durable_sdk.py +++ b/backend/tests/test_durable_sdk.py @@ -5,13 +5,16 @@ import psycopg import pytest +from dbos import DBOS from druks.agents import Agent, AgentOutput from druks.apps.registry import agents, workflows from druks.database import configure_session, get_session from druks.durable import FatalError, Run, RunState from druks.durable.dbos_state import workflow_status from druks.durable.engine import configure_engine, init_dbos, launch, shutdown +from druks.events.models import Event from druks.models import StoredSubject +from druks.signals import subscribe from druks.testing import init_db from druks.user_settings.models import InstallationSettings from druks.workflows import Gate, Subject, Workflow, step, task @@ -228,6 +231,18 @@ async def run_multistep(self) -> None: await Approve.wait() SINK.append(f"acct-after:{self.account_id}") + class AnnounceFlow(Workflow): + subject = Widget + + async def run_multistep(self) -> None: + await self.announce("test.revision", revision=1) + await self.announce("test.revision", revision=2) + marker = f"announced:{self.workflow_id}" + SINK.append(marker) + if SINK.count(marker) == 1: + # A worker interruption leaves the run available for recovery. + raise asyncio.CancelledError("Simulated worker interruption") + return ( SampleFlow, AgentFlow, @@ -239,6 +254,7 @@ async def run_multistep(self) -> None: SubjectlessConfirmFlow, ReviewFlow, AttributedFlow, + AnnounceFlow, ScheduledDispatch, RetryingStepFlow, EnqueueInStepFlow, @@ -304,6 +320,7 @@ async def rt(): subjectless_confirm_flow, review_flow, attributed_flow, + announce_flow, scheduled_dispatch, retrying_step_flow, enqueue_in_step_flow, @@ -328,6 +345,7 @@ async def rt(): SubjectlessConfirmFlow=subjectless_confirm_flow, ReviewFlow=review_flow, AttributedFlow=attributed_flow, + AnnounceFlow=announce_flow, ScheduledDispatch=scheduled_dispatch, RetryingStepFlow=retrying_step_flow, EnqueueInStepFlow=enqueue_in_step_flow, @@ -353,6 +371,7 @@ async def rt(): workflows._items.pop("subjectless_confirm_flow", None) workflows._items.pop("review_flow", None) workflows._items.pop("attributed_flow", None) + workflows._items.pop("announce_flow", None) workflows._items.pop("scheduled_dispatch", None) workflows._items.pop("retrying_step_flow", None) workflows._items.pop("enqueue_in_step_flow", None) @@ -1207,3 +1226,39 @@ async def test_subjectless_run_emits_no_events(rt): await session.close() assert events == [] + + +async def test_announcements_survive_subscriber_retry_and_workflow_replay(rt): + deliveries = [] + + @subscribe("test.revision", workflow=rt.AnnounceFlow) + async def receive(*, subject: Widget, revision: int) -> None: + async with get_session(rt.engine) as session: + events = list(await session.scalars(select(Event).filter_by(type="test.revision"))) + deliveries.append((revision, len(events))) + if len(deliveries) == 1: + raise RuntimeError("Subscriber unavailable") + + workflow_id = await rt.AnnounceFlow.start(subject=Widget(id=7)) + marker = f"announced:{workflow_id}" + try: + await _wait_for(rt.engine, workflow_id, lambda run: SINK.count(marker) == 1) + assert deliveries == [(1, 1), (1, 1), (2, 2)] + + await DBOS.resume_workflow_async(workflow_id) + await _wait_for(rt.engine, workflow_id, lambda run: run.state == RunState.FINISHED) + assert SINK.count(marker) == 2 + + async with get_session(rt.engine) as session: + events = list( + await session.scalars( + select(Event).filter_by(type="test.revision").order_by(Event.id) + ) + ) + assert [event.payload for event in events] == [ + {"revision": 1, "run": workflow_id, "kind": rt.AnnounceFlow.kind}, + {"revision": 2, "run": workflow_id, "kind": rt.AnnounceFlow.kind}, + ] + assert deliveries == [(1, 1), (1, 1), (2, 2)] + finally: + await DBOS.cancel_workflow_async(workflow_id) diff --git a/backend/tests/test_run_state.py b/backend/tests/test_run_state.py index d39507a1..4adc79cf 100644 --- a/backend/tests/test_run_state.py +++ b/backend/tests/test_run_state.py @@ -222,7 +222,7 @@ async def body() -> None: ) ambient_session().expunge_all() - assert (await Run.get(run.id)).failure is None + assert not (await Run.get(run.id)).failure rows = ( await ambient_session().execute(select(Event).filter_by(subject_id=str(item.id))) ).scalars() @@ -258,8 +258,8 @@ async def body() -> None: assert row.failure == "closed at review" # A bare FatalError carries no distinguishing code — only its message. assert row.failure_code == "" - assert row.input_gate is None - assert row.input_request is None + assert not row.input_gate + assert not row.input_request failed = ( await ambient_session().execute( select(Event).filter_by(type="workflow.failed", subject_id=str(item.id)) @@ -364,11 +364,10 @@ async def body() -> None: @pytest.mark.asyncio async def test_announce_carries_the_runs_routing(druks_db): - # The body states its facts; the platform injects what subscribers filter on, - # and the publish rides its own named checkpoint — the boundary that keeps a - # recovery replay from re-firing it. - workflow = Workflow() - workflow._subject = {"type": "note", "id": 7} + note, run = await _item_and_run(druks_db, "running") + workflow = Summarize() + workflow._workflow_id = run.id + workflow._subject = note.identity received = [] checkpoints = [] @@ -383,8 +382,12 @@ async def run_inline(options, fn): with mock.patch("druks.workflows.DBOS.run_step_async", side_effect=run_inline): await workflow.announce("test.announced", pr_number=12) - assert received == [({"type": "note", "id": 7}, {"pr_number": 12})] - assert checkpoints == ["test.announced"] + assert received == [(note.identity, {"pr_number": 12})] + assert checkpoints == ["test.announced", "test.announced:propagate"] + event = (await ambient_session().scalars(select(Event).filter_by(type="test.announced"))).one() + assert event.app == "field_notes" + assert event.subject_label == note.label + assert event.payload == {"pr_number": 12, "run": run.id, "kind": workflow.kind} @pytest.mark.asyncio diff --git a/docs/writing-an-app.md b/docs/writing-an-app.md index c148c8b9..bc6d764a 100644 --- a/docs/writing-an-app.md +++ b/docs/writing-an-app.md @@ -242,15 +242,43 @@ Two rules: ### Announcing domain events -If another component must react to a body action, announce the action: +Announce a domain fact from the workflow body: ```python await self.announce("pr.opened", pr_number=delivery.pr_number, branch=delivery.branch) ``` -The platform routes it to subscribers that filter on your workflow and subject. -The publication is a durable checkpoint. Recovery does not publish it again. -Announce from the body, not inside a `@step`. +Druks records the event in one checkpoint. It notifies subscribers in a second +checkpoint. A subscriber retry cannot insert the completed event again. Recovery +reuses completed checkpoints. An interrupted operation can run again, so +subscribers must remain idempotent. Call this method outside a `@step`. + +A domain method announces through its subject: + +```python +from druks.db import StoredSubject +from sqlalchemy.orm import Mapped + + +class Report(StoredSubject): + __tablename__ = "night_watch_reports" + + published_url: Mapped[str | None] + + async def publish(self, url: str) -> None: + if self.published_url != url: + self.published_url = url + await self.announce("report.published", url=url) +``` + +`Subject` and `StoredSubject` both supply `announce()`. Druks gets the owner from +the registered app package. The call records the subject identity, its current +label, and the supplied facts. It then notifies subscribers in the same +transaction as the domain change. A rollback removes the change and its event. +The app must prevent duplicate domain changes on webhook redelivery. + +Authors supply no app ID, run ID, timestamp, or session. Frontend code owns the +wording. ### Schedules and settings