diff --git a/backend/druks/api/runs.py b/backend/druks/api/runs.py index c4b4ef22..5524b62d 100644 --- a/backend/druks/api/runs.py +++ b/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") diff --git a/backend/druks/events/models.py b/backend/druks/events/models.py index 7333e358..2da1144a 100644 --- a/backend/druks/events/models.py +++ b/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( diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py index 35ad61a8..12f27498 100644 --- a/backend/druks/workflows.py +++ b/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. diff --git a/backend/tests/test_durable_sdk.py b/backend/tests/test_durable_sdk.py index 178352e8..af43e1b5 100644 --- a/backend/tests/test_durable_sdk.py +++ b/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) diff --git a/backend/tests/test_operator_stops.py b/backend/tests/test_operator_stops.py new file mode 100644 index 00000000..5e7939fa --- /dev/null +++ b/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)))