Record gate replies after validation (DRU-508)

backend/druks/workflows.py

@@ -100,6 +100,7 @@
 _REVIEW_CONTROLS = get_args(_ReviewAction)
 
 T = TypeVar("T")
+GateReply = TypeVar("GateReply", bound="Gate")
 
 # The running workflow instance, so a Gate's on_wait() can reach its app's
 # side-effects (set draft, request review, …) when the gate parks. No default:
@@ -316,8 +317,7 @@ async def _on_wait() -> None:
                 await cls.on_wait(workflow)
 
         await DBOS.run_step_async(StepOptions(name=f"{cls.name}._on_wait"), _on_wait)
-        payload = await _park(workflow, cls.name, input_request, ttl_seconds)
-        reply = cls.model_validate(payload)
+        reply = await _park(workflow, cls, input_request, ttl_seconds)
         workflow.journal.add(reply)
         return reply
 
@@ -335,10 +335,10 @@ class OperatorReply(Gate):
 
 async def _park(
     workflow: "Workflow",
-    gate: str,
+    gate: type[GateReply],
     input_request: dict[str, Any] | None,
     ttl_seconds: float,
-) -> dict[str, Any]:
+) -> GateReply:
     # Shared park core: a park lasts days, so reap the warm VM, then suspend on the
     # gate's channel until Run.resume answers it.
     await workflow._reap_run()
@@ -347,30 +347,30 @@ async def _park(
         RunState.PARKED,
         subject=workflow._subject,
         facts={
-            "input_gate": gate,
+            "input_gate": gate.name,
             "input_request": input_request,
             "input_requested_at": datetime.now(UTC),
         },
     )
     if workflow._subject:
         # Every subjected park notifies the designated destination — no author opt-in.
         await _notify_designated_destination(workflow.workflow_id, workflow._subject)
-    payload = await DBOS.recv_async(gate, timeout_seconds=ttl_seconds)
+    payload = await DBOS.recv_async(gate.name, timeout_seconds=ttl_seconds)
     if payload is None:
-        raise GateTimeout(gate)
+        raise GateTimeout(gate.name)
+    reply = gate.model_validate(payload)
     await _emit_run_event(
         workflow.workflow_id,
         RunState.RUNNING,
         subject=workflow._subject,
         facts={**_GATE_CLEARED, "answer_parked_at": Run.input_requested_at},
+        result=reply,
     )
-    return payload
+    return reply
 
 
 async def _notify_designated_destination(workflow_id: str, subject: dict[str, Any]) -> None:
-    # Reads the ask off the run row the parked step just wrote (the
-    # signal payload carries no ask — producer-side placement is the point);
-    # the settings pointer is the operator's off-switch.
+    # The operator's settings select the destination for the recorded request.
     async def _create() -> str | None:
         async with step_session():
             run = await Run.get(workflow_id)
@@ -554,6 +554,7 @@ async def _transition() -> dict[str, Any] | None:
             # Read before the flush: flushing the update unloads the row's
             # computed columns, and reading one back would be implicit IO.
             label = run.subject_label
+            gate = run.input_gate if state == RunState.RUNNING and result else None
             if facts:
                 for field, value in facts.items():
                     setattr(run, field, value)
@@ -563,7 +564,7 @@ async def _transition() -> dict[str, Any] | None:
                 return {
                     "kind": run.kind,
                     "subject": subject,
-                    "payload": await _log_run_event(run, state, subject, label, result),
+                    "payload": await _log_run_event(run, state, subject, label, result, gate),
                 }
 
     transition = await DBOS.run_step_async(
@@ -592,14 +593,19 @@ async def _log_run_event(
     subject: dict[str, Any],
     label: str | None,
     result: Any = None,
+    gate: str | None = None,
 ) -> dict[str, Any]:
     # One event per transition — the feed's run-level granularity, read off the
     # just-written row so gate and failure ride the transition that set them.
     # The result rides the finished event so reactions read the outcome off the
     # payload instead of artifacts.
     payload: dict[str, Any] = {"run": run.id, "kind": run.kind}
-    if run.input_gate:
-        payload["gate"] = run.input_gate
+    if gate or run.input_gate:
+        payload["gate"] = gate or run.input_gate
+    if gate or state == RunState.PARKED:
+        payload["input_requested_at"] = run.input_requested_at.isoformat()
+    if state == RunState.PARKED:
+        payload["input_request"] = run.input_request
     if run.failure:
         payload["failure"] = run.failure
     if isinstance(result, BaseModel):
@@ -824,8 +830,7 @@ async def review(
         }
         if context:
             request["context"] = context
-        payload = await _park(self, OperatorReply.name, request, GATE_TTL_SECONDS)
-        reply = OperatorReply.model_validate(payload)
+        reply = await _park(self, OperatorReply, request, GATE_TTL_SECONDS)
         self.journal.add(reply)
         return reply
 

backend/tests/test_durable_sdk.py

@@ -196,13 +196,18 @@ async def run(self) -> Decision:
     class DoubleGateFlow(Workflow):
         # Two rounds on the same gate — the shape a stale buffered reply would
         # ghost-resume.
+        subject = Widget
+
         async def run_multistep(self) -> None:
             first = await Approve.wait()
             SINK.append(f"round1:{first.action}")
             second = await Approve.wait()
             SINK.append(f"round2:{second.action}")
             replies = [reply.action for reply in self.journal.filter(Approve)]
             SINK.append(f"gate-journal:{replies}")
+            SINK.append("gate:completed")
+            if SINK.count("gate:completed") == 1:
+                raise asyncio.CancelledError
 
     class ConfirmFlow(Workflow):
         subject = Widget
@@ -505,7 +510,7 @@ async def test_duplicate_replies_to_one_round_collapse(rt):
     not buffer on the topic and ghost-resume the gate's next round unprompted."""
     from sqlalchemy import text
 
-    wfid = await rt.DoubleGateFlow.start(subject=None)
+    wfid = await rt.DoubleGateFlow.start(subject=Widget(id=515151))
     parked = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.PARKED)
     first_asked_at = parked.input_requested_at
 
@@ -538,8 +543,34 @@ async def test_duplicate_replies_to_one_round_collapse(rt):
     assert "round1:first" in SINK
 
     # A fresh reply to the new round is a new key, so it still gets through.
+    second_asked_at = parked.input_requested_at
     await parked.resume(action="second")
+    for _ in range(100):
+        if "gate:completed" in SINK:
+            break
+        await asyncio.sleep(0.1)
+    assert SINK.count("gate:completed") == 1
+    await asyncio.sleep(0.2)
+    await DBOS.resume_workflow_async(wfid)
     await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FINISHED)
+    assert SINK.count("gate:completed") == 2
+    async with get_session(rt.engine) as session:
+        events = list(
+            await session.scalars(
+                select(Event).where(Event.payload["run"].astext == wfid).order_by(Event.id)
+            )
+        )
+    requests = [event for event in events if event.type == "workflow.parked"]
+    receipts = [
+        event for event in events if event.type == "workflow.running" and "result" in event.payload
+    ]
+    rounds = [first_asked_at.isoformat(), second_asked_at.isoformat()]
+    assert [event.payload["input_requested_at"] for event in requests] == rounds
+    assert [event.payload["input_requested_at"] for event in receipts] == rounds
+    assert [event.payload["result"] for event in receipts] == [
+        {"action": "first"},
+        {"action": "second"},
+    ]
     assert "round2:second" in SINK
     assert "round2:duplicate" not in SINK
     # Both replies landed on the journal, in reply order.

backend/tests/test_gate_receipt.py

@@ -1,87 +1,103 @@
+from unittest.mock import AsyncMock
+
 import pytest
 from dbos import DBOS
 from dbos._error import DBOSWorkflowCancelledError
 from druks.durable.exceptions import GateTimeout
 from druks.durable.models import Run
+from druks.events.models import Event
 from druks.testing import seed_run
-from druks.workflows import _park
+from druks.workflows import OperatorReply, current_workflow
+from druks_field_notes.models import Note
 from druks_field_notes.workflows import Summarize
+from pydantic import ValidationError
+from sqlalchemy import select
 
-_ASK = {"presentation": "in_app", "controls": ["approve"], "questions": []}
-
-
-class _ParkedWorkflow:
-    # The slice of Workflow that _park touches. A subjectless run keeps the emit to
-    # its facts write (no feed event, no notification) — the receipt path under
-    # test is exactly that write.
-    def __init__(self, workflow_id: str) -> None:
-        self.workflow_id = workflow_id
-        self._subject = None
-
-    async def _reap_run(self) -> None:
-        return
+_ASK = {"presentation": "in_app", "controls": ["approve", "request_changes"], "questions": []}
 
 
 @pytest.fixture
-def _direct_steps(monkeypatch):
-    # Run each durable step body inline — the test exercises _park's own logic,
-    # not DBOS checkpointing.
-    async def _call_through(options, func, *args, **kwargs):
-        return await func(*args, **kwargs)
-
-    monkeypatch.setattr(DBOS, "run_step_async", _call_through)
-
-
-async def _reload(druks_db, run_id: str) -> Run:
-    druks_db.expunge_all()
-    return await druks_db.get(Run, run_id)
-
-
-async def test_answer_stamps_the_receipt_beside_the_gate_clear(
-    druks_db, _direct_steps, monkeypatch
+def direct_steps(monkeypatch):
+    async def call_through(options, function, *args, **kwargs):
+        return await function(*args, **kwargs)
+
+    monkeypatch.setattr(DBOS, "run_step_async", call_through)
+    monkeypatch.setattr("druks.workflows._notify_designated_destination", AsyncMock())
+
+
+@pytest.fixture(params=["gate", "review"])
+async def request_reply(request, druks_db, direct_steps):
+    note = await Note.create(body="A request to review")
+    run = await seed_run(druks_db, kind=Summarize.kind, subject=note)
+    workflow = Summarize()
+    workflow._workflow_id = run.id
+    workflow._subject = note.identity
+    token = current_workflow.set(workflow)
+    try:
+        call = (
+            OperatorReply.wait(input_request=_ASK) if request.param == "gate" else workflow.review()
+        )
+        yield run.id, call
+    finally:
+        current_workflow.reset(token)
+
+
+async def test_valid_reply_records_the_request_round_before_current_fields_clear(
+    druks_db, request_reply, monkeypatch
 ):
-    run = await seed_run(druks_db, kind=Summarize.kind, run_id="run-receipt-answer")
+    run_id, call = request_reply
+    monkeypatch.setattr(DBOS, "recv_async", AsyncMock(return_value={"action": "approve"}))
 
-    async def _answer(topic, timeout_seconds):
-        return {"action": "approve"}
+    reply = await call
 
-    monkeypatch.setattr(DBOS, "recv_async", _answer)
-    payload = await _park(_ParkedWorkflow(run.id), "review", _ASK, ttl_seconds=1.0)
-
-    assert payload == {"action": "approve"}
-    run = await _reload(druks_db, run.id)
-    # The receipt is the round the answer cleared: the same stamp the park
-    # wrote, which _GATE_CLEARED preserves on the row.
-    assert run.input_requested_at
+    assert reply == OperatorReply(action="approve")
+    druks_db.expunge_all()
+    run = await druks_db.get(Run, run_id)
     assert run.answer_parked_at == run.input_requested_at
     assert not run.input_gate
     assert not run.input_request
+    events = list(await druks_db.scalars(select(Event).order_by(Event.id)))
+    assert [event.type for event in events] == ["workflow.parked", "workflow.running"]
+    request, receipt = events
+    for event in events:
+        assert event.payload["run"] == run_id
+        assert event.payload["gate"] == "review"
+        assert event.payload["input_requested_at"] == run.input_requested_at.isoformat()
+    assert request.payload["input_request"] == _ASK
+    assert receipt.payload["result"] == {"action": "approve", "answers": {}, "note": ""}
+
+
+@pytest.mark.parametrize("payload", [{"action": "merge"}, {}, None])
+async def test_invalid_reply_or_timeout_records_no_receipt(
+    druks_db, request_reply, monkeypatch, payload
+):
+    run_id, call = request_reply
+    monkeypatch.setattr(DBOS, "recv_async", AsyncMock(return_value=payload))
 
+    with pytest.raises(GateTimeout if payload is None else ValidationError):
+        await call
 
-async def test_timeout_never_writes_the_receipt(druks_db, _direct_steps, monkeypatch):
-    run = await seed_run(druks_db, kind=Summarize.kind, run_id="run-receipt-timeout")
-
-    async def _lapse(topic, timeout_seconds):
-        return None
-
-    monkeypatch.setattr(DBOS, "recv_async", _lapse)
-    with pytest.raises(GateTimeout):
-        await _park(_ParkedWorkflow(run.id), "review", _ASK, ttl_seconds=1.0)
-
-    run = await _reload(druks_db, run.id)
+    druks_db.expunge_all()
+    run = await druks_db.get(Run, run_id)
     assert not run.answer_parked_at
+    assert run.input_gate == "review"
+    assert run.input_request == _ASK
     assert run.input_requested_at
+    events = list(await druks_db.scalars(select(Event)))
+    assert [event.type for event in events] == ["workflow.parked"]
 
 
-async def test_cancel_never_writes_the_receipt(druks_db, _direct_steps, monkeypatch):
-    run = await seed_run(druks_db, kind=Summarize.kind, run_id="run-receipt-cancel")
+async def test_cancel_records_no_receipt(druks_db, request_reply, monkeypatch):
+    run_id, call = request_reply
+    monkeypatch.setattr(
+        DBOS, "recv_async", AsyncMock(side_effect=DBOSWorkflowCancelledError(run_id))
+    )
 
-    async def _cancelled(topic, timeout_seconds):
-        raise DBOSWorkflowCancelledError(run.id)
-
-    monkeypatch.setattr(DBOS, "recv_async", _cancelled)
     with pytest.raises(DBOSWorkflowCancelledError):
-        await _park(_ParkedWorkflow(run.id), "review", _ASK, ttl_seconds=1.0)
+        await call
 
-    run = await _reload(druks_db, run.id)
+    druks_db.expunge_all()
+    run = await druks_db.get(Run, run_id)
     assert not run.answer_parked_at
+    events = list(await druks_db.scalars(select(Event)))
+    assert [event.type for event in events] == ["workflow.parked"]

backend/tests/test_run_state.py

@@ -162,7 +162,11 @@ async def _raises(**_: object) -> None:
             run.id,
             RunState.PARKED,
             subject={"type": "work_item", "id": item.id},
-            facts={"input_gate": "review_work", "input_request": {"label": "Review"}},
+            facts={
+                "input_gate": "review_work",
+                "input_request": {"label": "Review"},
+                "input_requested_at": datetime.now(UTC),
+            },
         )
 
     ambient_session().expunge_all()