Return WorkflowStartResult from Workflow.start()

backend/druks/build/scoping/workflows.py

@@ -27,14 +27,14 @@ class Scope(Workflow):
     async def dispatch(cls, *, ticket: Ticket) -> str | None:
         # The scoped label is the done marker — remove it to force a re-scope.
         if ticket.has_label(Build.settings().scoper_scoped_label):
-            return None
+            return
         item = WorkItem.get_for_remote_key(source=ticket.provider, remote_key=ticket.key)
         if not item:
             target = ProjectRepo.lookup(project_name=ticket.project_name, labels=ticket.labels)
             project = Project.get_for_repo(target.full_name) if target else None
             if not project:
                 logger.info("No project routes %s; not scoping.", ticket.key)
-                return None
+                return
             item = WorkItem.create(
                 project_id=project.id,
                 source=ticket.provider,
@@ -46,12 +46,13 @@ async def dispatch(cls, *, ticket: Ticket) -> str | None:
         assignee = None
         if ticket.assignee_email:
             assignee = Account.get_for_username(ticket.assignee_email.strip())
-        return await cls.start(
+        start_result = await cls.start(
             subject=WorkItem.subject_for(item.id),
             account_id=assignee.id if assignee else None,
             remote_key=ticket.key,
             source=ticket.provider,
         )
+        return start_result.run_id
 
     @classmethod
     def parked_for(cls, work_item_id: int) -> Run | None:

backend/druks/build/workflows.py

@@ -148,7 +148,7 @@ async def dispatch(
             raise ValueError(f"dispatching a build for unknown work item {work_item_id}")
         # The assignee's account runs the calls; the owner fields stay input.
         assignee = Account.get_for_username(assignee_email.strip()) if assignee_email else None
-        run_id = await cls.start(
+        start_result = await cls.start(
             subject=WorkItem.subject_for(item.id),
             account_id=assignee.id if assignee else None,
             repo=item.repo,
@@ -159,11 +159,11 @@ async def dispatch(
             task_owner_email=assignee_email,
             task_owner_name=assignee_name,
         )
-        item.update(build_run_id=run_id)
+        item.update(build_run_id=start_result.run_id)
         # Back onto the active board: a scoped item re-enters flight when its
         # build starts. History is for items at rest in a handoff lane.
         item.set_status(None)
-        return run_id
+        return start_result.run_id
 
     async def run_multistep(
         self,

backend/druks/workflows.py

@@ -56,6 +56,7 @@
     "SubjectSummary",
     "Workflow",
     "WorkflowError",
+    "WorkflowStartResult",
     "get_run_phase",
     "set_run_phase",
     "step",
@@ -143,7 +144,7 @@ def _input_model_from_signature(cls: type["Workflow"]) -> type[BaseModel] | None
     method = getattr(cls, method_name)
     parameters = [p for name, p in inspect.signature(method).parameters.items() if name != "self"]
     if not parameters:
-        return None
+        return
     hints = get_type_hints(method)
     fields: dict[str, Any] = {}
     for p in parameters:
@@ -251,9 +252,8 @@ async def _notify_designated_destination(workflow_id: str, subject: dict[str, An
     async def _create() -> str | None:
         async with step_session():
             destination_id = UserSettings.get().gate_park_destination_id
-            if not destination_id:
-                return None
-            return Run.get(workflow_id).create_park_notification(destination_id, subject)
+            if destination_id:
+                return Run.get(workflow_id).create_park_notification(destination_id, subject)
 
     notification_id = await DBOS.run_step_async(
         StepOptions(name="notifications.gate_park", **_IO_RETRIES), _create
@@ -306,14 +306,13 @@ async def _transition() -> dict[str, Any] | None:
                 for field, value in facts.items():
                     setattr(run, field, value)
                 session.flush()
-            if not subject:
-                # Subjectless framework crons are plumbing: no feed entry.
-                return None
-            return {
-                "kind": run.kind,
-                "subject": subject,
-                "payload": _log_run_event(run, event, subject, result),
-            }
+            # Subjectless framework crons are plumbing: no feed entry.
+            if subject:
+                return {
+                    "kind": run.kind,
+                    "subject": subject,
+                    "payload": _log_run_event(run, event, subject, result),
+                }
 
     transition = await DBOS.run_step_async(
         StepOptions(name=f"run.{event.value}", **_IO_RETRIES), _transition
@@ -402,6 +401,13 @@ async def _execute_run(
     return result
 
 
+class WorkflowStartResult(BaseModel):
+    # The run start() resolved to; is_duplicate marks the subject's already-live
+    # run handed back instead of a fresh enqueue.
+    run_id: str
+    is_duplicate: bool
+
+
 class Workflow:
     kind: ClassVar[str] = ""
     # The extension that declares this workflow — class identity, resolved from
@@ -554,7 +560,7 @@ async def _ensure_host(self) -> str | None:
         # The warm VM, provisioned once per segment; state is carried in git, so
         # only the host-id matters across steps — held-across-steps never fights replay.
         if not self.steps_reuse_sandbox:
-            return None
+            return
         if self._host and self._host.expires_at:
             remaining = (self._host.expires_at - datetime.now(UTC)).total_seconds()
             if remaining < SANDBOX_HOST_ROTATE_BEFORE_SECONDS:
@@ -626,15 +632,15 @@ async def start(
         subject: dict[str, Any] | None,
         account_id: str | None = None,
         **input: Any,
-    ) -> str:
+    ) -> WorkflowStartResult:
         # Mint the id, write the projection row, enqueue the body. Returns the
-        # workflow id; an extension that wants one-active-run-per-subject enforces
-        # that on its own side before calling this. Enqueuing (not start_workflow)
-        # routes execution onto the shared queue, so the process that kicks a run
-        # off — often the web process — doesn't have to be the one that runs it;
-        # any launched executor picks it up. The input kwargs mirror the body's own
-        # signature and validate against the model synthesized from it, so a bad
-        # shape fails at start, not inside the run.
+        # run to follow — freshly enqueued, or the subject's already-live run
+        # with is_duplicate set. Enqueuing (not start_workflow) routes execution
+        # onto the shared queue, so the process that kicks a run off — often the
+        # web process — doesn't have to be the one that runs it; any launched
+        # executor picks it up. The input kwargs mirror the body's own signature
+        # and validate against the model synthesized from it, so a bad shape
+        # fails at start, not inside the run.
         # subject is required (no default) so a run can't silently lose its
         # timeline by omission — pass subject=None explicitly for a background run.
         if account_id is None:
@@ -643,12 +649,11 @@ async def start(
             account_id = current_account_id.get()
         if subject is not None:
             _Subject.model_validate(subject)  # raises on a bad shape (wrong/extra keys, types)
-        if cls._run_input_model is None:
-            if input:
-                raise WorkflowError(f"{cls.__name__}.{cls._body_method}() takes no input")
-            wire: dict[str, Any] = {}
-        else:
+        wire: dict[str, Any] = {}
+        if cls._run_input_model:
             wire = cls._run_input_model.model_validate(input).model_dump(mode="json")
+        elif input:
+            raise WorkflowError(f"{cls.__name__}.{cls._body_method}() takes no input")
         if account_id:
             wire[_ACCOUNT_INPUT_KEY] = account_id
         workflow_id = str(uuid7())
@@ -681,7 +686,7 @@ async def start(
                 run_queue.name, duplicate.deduplication_id
             )
             if holder:
-                return holder
+                return WorkflowStartResult(run_id=holder, is_duplicate=True)
             # The holder reached terminal between the rejection and the lookup —
             # the slot is free now, so this start goes through.
             return await cls.start(subject=subject, account_id=account_id, **input)
@@ -690,7 +695,7 @@ async def start(
         Run.create_row(
             _step_engine(), workflow_id=workflow_id, kind=cls.kind, account_id=account_id
         )
-        return workflow_id
+        return WorkflowStartResult(run_id=workflow_id, is_duplicate=False)
 
 
 def _wrap_steps(cls: type[Workflow]) -> None:

backend/tests/conftest.py

@@ -144,10 +144,10 @@ def _no_durable_dispatch(request):
         yield
         return
 
-    from druks.workflows import Workflow
+    from druks.workflows import Workflow, WorkflowStartResult
 
     async def _noop(*args, **kwargs):
-        return ""
+        return WorkflowStartResult(run_id="", is_duplicate=False)
 
     async def _phase_noop(*args, **kwargs):
         pass

backend/tests/druks-field_notes/druks_field_notes/workflows.py

@@ -10,7 +10,7 @@ class Summarize(Workflow):
 
     async def run(self, note_id: int) -> None:
         note = Note.get(note_id)
-        assert note is not None  # dispatched against a note the route just created
+        assert note  # dispatched against a note the route just created
         # The note body is the agent's prompt context; the summary it returns is the
         # extension's own domain result, saved onto the note.
         result = await FieldNotes.summarize(note_body=note.body)
@@ -20,7 +20,8 @@ async def run(self, note_id: int) -> None:
     async def dispatch(cls, *, note_id: int) -> str:
         # Launch policy for a note: one run per note, keyed by its subject; the
         # signed-in requester attributes it ambiently.
-        return await cls.start(
+        start_result = await cls.start(
             subject={"type": FieldNotes.subject_type, "id": note_id},
             note_id=note_id,
         )
+        return start_result.run_id

backend/tests/test_author_surface.py

@@ -19,6 +19,7 @@
         "SubjectSummary",
         "Workflow",
         "WorkflowError",
+        "WorkflowStartResult",
         "get_run_phase",
         "set_run_phase",
         "step",

backend/tests/test_build_dispatch.py

@@ -1,6 +1,7 @@
 from conftest import make_test_work_item, seed_run
 from druks.build.enums import HandoffStatus
 from druks.build.workflows import BuildWorkflow
+from druks.workflows import WorkflowStartResult
 
 
 async def test_dispatch_pulls_scoped_item_back_onto_the_board(db_session, monkeypatch) -> None:
@@ -12,7 +13,7 @@ async def test_dispatch_pulls_scoped_item_back_onto_the_board(db_session, monkey
     seed_run(db_session, "run-1")
 
     async def fake_start(cls, **kwargs):
-        return "run-1"
+        return WorkflowStartResult(run_id="run-1", is_duplicate=False)
 
     monkeypatch.setattr(BuildWorkflow, "start", classmethod(fake_start))
     run_id = await BuildWorkflow.dispatch(work_item_id=item.id)

backend/tests/test_build_durable.py

@@ -185,10 +185,12 @@ async def test_happy_path_to_merge(rt, monkeypatch):
     _stub(monkeypatch, rt)
 
     item_id = _seed_work_item(rt.engine, repo="acme/widget")
-    wfid = await rt.flow.start(
-        repo="acme/widget",
-        subject={"type": "work_item", "id": item_id},
-    )
+    wfid = (
+        await rt.flow.start(
+            repo="acme/widget",
+            subject={"type": "work_item", "id": item_id},
+        )
+    ).run_id
 
     parked = await _wait(
         rt.engine,
@@ -233,10 +235,12 @@ async def test_recovery_rebuilds_the_diary_without_rerunning_agents(rt, monkeypa
     invoked = _stub(monkeypatch, rt)
 
     item_id = _seed_work_item(rt.engine, repo="acme/widget")
-    wfid = await rt.flow.start(
-        repo="acme/widget",
-        subject={"type": "work_item", "id": item_id},
-    )
+    wfid = (
+        await rt.flow.start(
+            repo="acme/widget",
+            subject={"type": "work_item", "id": item_id},
+        )
+    ).run_id
     await _wait(
         rt.engine,
         wfid,