Record queue admission and operator stops (DRU-507)

backend/druks/api/runs.py

@@ -10,8 +10,10 @@
     agent_error_responses,
 )
 from druks.api.schemas import CancelRunResponse, ResumeRequest, RetryRunResponse
-from druks.durable.enums import RunState
+from druks.apps.registry import workflows
+from druks.durable.enums import RunState, WorkflowEvent
 from druks.durable.models import Run
+from druks.events.models import Event
 from druks.notifications.exceptions import InvalidChoiceError
 from druks.notifications.services import validate_in_app_answer
 
@@ -72,7 +74,16 @@ async def cancel_run(
         return CancelRunResponse(run=run.id, result="already_cancelled")
     if not run.is_active:
         raise RunNotActive(run_id)
+    subject = await run.get_subject()
+    label = run.subject_label
     await run.cancel(failure=reason)
+    await Event.emit(
+        type=WorkflowEvent.CANCELLED,
+        subject=subject,
+        label=label,
+        payload={"run": run.id, "kind": run.kind, "reason": reason},
+        app=workflows.get(run.kind).app,
+    )
     return CancelRunResponse(run=run.id, result="cancelled")
 
 

backend/druks/events/models.py

@@ -3,6 +3,7 @@
 
 from sqlalchemy import Index
 from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Mapped, mapped_column
 
 from druks.database import db_session
@@ -13,9 +14,7 @@
 
 
 class Event(Base):
-    """The append-only log: one row per run-state transition and (later) domain
-    milestone, keyed to the subject it concerns. An app reads it back as a feed,
-    or folds the newest-per-subject into a status."""
+    """Recorded workflow and domain facts, keyed to their subject."""
 
     __tablename__ = "events"
     # Newest-per-subject is the history/dashboard rollup; the feed orders on the
@@ -45,9 +44,12 @@ async def emit(
         label: str | None = None,
         payload: dict[str, Any] | None = None,
         app: str | None = None,
+        session: AsyncSession | None = None,
     ) -> None:
+        """Record in the supplied session or the current domain transaction."""
+        session = session or db_session()
         subject = subject or {}
-        db_session().add(
+        session.add(
             cls(
                 type=type,
                 subject_type=subject.get("type"),
@@ -57,7 +59,7 @@ async def emit(
                 payload=payload or {},
             )
         )
-        await db_session().flush()
+        await session.flush()
 
     @classmethod
     async def announce(

backend/druks/workflows.py

@@ -35,6 +35,7 @@
     validate_setting_override,
     validate_settings_declaration,
 )
+from druks.database import get_session
 from druks.durable.activity import set_run_phase
 from druks.durable.datastructures import Subject
 from druks.durable.engine import _step_engine, register_schedule, run_queue, step_session
@@ -1035,6 +1036,18 @@ async def start(
                 _step_engine(), workflow_id=workflow_id, kind=cls.kind, account_id=account_id
             )
             if subject:
+                if cls.app:
+                    # Admission must be visible before the caller's transaction commits.
+                    async with get_session(_step_engine()) as session:
+                        await Event.emit(
+                            type=WorkflowEvent.SCHEDULED,
+                            subject=subject_record,
+                            label=subject.label,
+                            payload={"run": workflow_id, "kind": cls.kind},
+                            app=cls.app,
+                            session=session,
+                        )
+                        await session.commit()
                 await publish(WorkflowEvent.SCHEDULED, subject=subject.identity, kind=cls.kind)
             return workflow_id
         # The slot was held — the handle is the subject's live run.

backend/tests/test_durable_sdk.py

@@ -7,8 +7,9 @@
 import pytest
 from dbos import DBOS
 from druks.agents import Agent, AgentOutput
+from druks.apps import loader
 from druks.apps.registry import agents, workflows
-from druks.database import configure_session, get_session
+from druks.database import configure_session, db_session, get_session, session_scope
 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
@@ -1262,3 +1263,77 @@ async def receive(*, subject: Widget, revision: int) -> None:
         assert deliveries == [(1, 1), (1, 1), (2, 2)]
     finally:
         await DBOS.cancel_workflow_async(workflow_id)
+
+
+async def test_admission_commits_before_the_request_and_deduplicates(rt, monkeypatch):
+    monkeypatch.setitem(loader._workflow_packages, "test_admission", "admission")
+
+    class AdmissionFlow(Workflow):
+        __module__ = "test_admission"
+        subject = Widget
+
+        async def run_multistep(self) -> None:
+            await DBOS.recv_async("finish")
+
+    subject = Widget(id=7)
+    workflow_id = ""
+    try:
+        with pytest.raises(ValueError, match="Roll back the request"):
+            async with session_scope(rt.engine):
+                request_session = db_session()
+                await request_session.execute(select(Widget).where(Widget.id == 7))
+                workflow_id = await AdmissionFlow.start(subject=subject)
+                assert await AdmissionFlow.start(subject=subject) == workflow_id
+                assert db_session() is request_session
+                assert request_session.in_transaction()
+
+                async with get_session(rt.engine) as reader:
+                    events = list(
+                        await reader.scalars(
+                            select(Event).filter_by(type="workflow.scheduled", app="admission")
+                        )
+                    )
+                assert len(events) == 1
+                assert events[0].payload == {"run": workflow_id, "kind": AdmissionFlow.kind}
+                assert events[0].subject_id == "7"
+                assert events[0].subject_label == "W-7"
+                raise ValueError("Roll back the request")
+
+        async with get_session(rt.engine) as reader:
+            events = list(
+                await reader.scalars(
+                    select(Event).filter_by(type="workflow.scheduled", app="admission")
+                )
+            )
+        assert len(events) == 1
+    finally:
+        if workflow_id:
+            await DBOS.send_async(workflow_id, "done", topic="finish")
+            await _wait_for(rt.engine, workflow_id, lambda run: run.state == RunState.FINISHED)
+        workflows._items.pop(AdmissionFlow.kind)
+
+
+async def test_failed_retry_attempts_keep_separate_terminal_records(rt):
+    class FailingAttempt(Workflow):
+        subject = Widget
+
+        async def run(self) -> None:
+            raise FatalError("Source unavailable")
+
+    try:
+        first_id = await FailingAttempt.start(subject=Widget(id=7))
+        await _wait_for(rt.engine, first_id, lambda run: run.state == RunState.FAILED)
+        async with session_scope(rt.engine):
+            first_run = await Run.get(first_id)
+            retry_id = await first_run.retry()
+        await _wait_for(rt.engine, retry_id, lambda run: run.state == RunState.FAILED)
+
+        assert retry_id != first_id
+        async with get_session(rt.engine) as reader:
+            events = list(await reader.scalars(select(Event).filter_by(type="workflow.failed")))
+        failures = [event for event in events if event.payload["run"] in {first_id, retry_id}]
+        assert len(failures) == 2
+        assert {event.payload["run"] for event in failures} == {first_id, retry_id}
+        assert {event.payload["failure"] for event in failures} == {"Source unavailable"}
+    finally:
+        workflows._items.pop(FailingAttempt.kind)

backend/tests/test_operator_stops.py

@@ -0,0 +1,70 @@
+import pytest
+from druks.contrib.software_factory.workflows import Build
+from druks.db import db_session
+from druks.durable.enums import WorkflowEvent
+from druks.events.models import Event
+from druks.testing import seed_run
+from druks_field_notes.models import Note
+from druks_field_notes.workflows import Summarize
+from software_factory.factories import make_test_work_item
+from sqlalchemy import select
+
+
+async def test_operator_stop_records_its_reason_and_exact_run_once(druks_client, druks_db):
+    note = await Note.create(body="Stop this work")
+    run = await seed_run(druks_db, kind=Summarize.kind, subject=note)
+
+    response = await druks_client.post(
+        f"/api/runs/{run.id}/cancel", json={"reason": "Wrong source"}
+    )
+    assert response.status_code == 200
+    assert response.json() == {"run": run.id, "result": "cancelled"}
+
+    repeated = await druks_client.post(
+        f"/api/runs/{run.id}/cancel", json={"reason": "Wrong source"}
+    )
+    assert repeated.json() == {"run": run.id, "result": "already_cancelled"}
+    events = list(await druks_db.scalars(select(Event).filter_by(type=WorkflowEvent.CANCELLED)))
+    assert len(events) == 1
+    assert events[0].payload == {"run": run.id, "kind": Summarize.kind, "reason": "Wrong source"}
+    assert events[0].app == "field_notes"
+    assert events[0].subject_id == str(note.id)
+    assert events[0].subject_label == note.label
+
+
+async def test_failed_operator_cancellation_records_no_stop(druks_client, druks_db, monkeypatch):
+    note = await Note.create(body="Cancellation failed")
+    run = await seed_run(druks_db, kind=Summarize.kind, subject=note)
+
+    async def unavailable(workflow_id: str) -> None:
+        raise RuntimeError("Cancellation unavailable")
+
+    monkeypatch.setattr("dbos.DBOS.cancel_workflow_async", unavailable)
+    with pytest.raises(RuntimeError, match="Cancellation unavailable"):
+        await druks_client.post(f"/api/runs/{run.id}/cancel", json={"reason": "Wrong source"})
+
+    assert not list(await druks_db.scalars(select(Event).filter_by(type=WorkflowEvent.CANCELLED)))
+
+
+@pytest.mark.parametrize("state", ["failed", "finished"])
+async def test_inactive_run_has_no_operator_stop(druks_client, druks_db, state):
+    note = await Note.create(body="Finished work")
+    run = await seed_run(druks_db, kind=Summarize.kind, subject=note, state=state)
+
+    response = await druks_client.post(
+        f"/api/runs/{run.id}/cancel", json={"reason": "Wrong source"}
+    )
+
+    assert response.status_code == 409
+    assert not list(await druks_db.scalars(select(Event).filter_by(type=WorkflowEvent.CANCELLED)))
+
+
+@pytest.mark.parametrize("reason", ["Pull request merged", "Pull request closed"])
+async def test_factory_cleanup_creates_no_operator_stop(druks_db, reason):
+    db_session.registry.set(druks_db)
+    item = await make_test_work_item(repo="owner/repo", title="Completed work")
+    await seed_run(druks_db, kind=Build.kind, subject=item)
+
+    await Build.cancel(item, failure=reason)
+
+    assert not list(await druks_db.scalars(select(Event).filter_by(type=WorkflowEvent.CANCELLED)))