Record domain announcements through announce() (DRU-505)

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,

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)

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

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:

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")))

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)

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