diff --git a/backend/druks/build/scoping/workflows.py b/backend/druks/build/scoping/workflows.py
index cdce5e9..9660629 100644
--- a/backend/druks/build/scoping/workflows.py
+++ b/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:
diff --git a/backend/druks/build/workflows.py b/backend/druks/build/workflows.py
index e803530..33aaef6 100644
--- a/backend/druks/build/workflows.py
+++ b/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,
diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py
index 2cf3ff3..5684067 100644
--- a/backend/druks/workflows.py
+++ b/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:
diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py
index c71d0f9..e6f5b7b 100644
--- a/backend/tests/conftest.py
+++ b/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
diff --git a/backend/tests/druks-field_notes/druks_field_notes/workflows.py b/backend/tests/druks-field_notes/druks_field_notes/workflows.py
index 73cf148..2c86678 100644
--- a/backend/tests/druks-field_notes/druks_field_notes/workflows.py
+++ b/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
diff --git a/backend/tests/test_author_surface.py b/backend/tests/test_author_surface.py
index f25003f..bc4ac89 100644
--- a/backend/tests/test_author_surface.py
+++ b/backend/tests/test_author_surface.py
@@ -19,6 +19,7 @@
         "SubjectSummary",
         "Workflow",
         "WorkflowError",
+        "WorkflowStartResult",
         "get_run_phase",
         "set_run_phase",
         "step",
diff --git a/backend/tests/test_build_dispatch.py b/backend/tests/test_build_dispatch.py
index cc2485a..7e60722 100644
--- a/backend/tests/test_build_dispatch.py
+++ b/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)
diff --git a/backend/tests/test_build_durable.py b/backend/tests/test_build_durable.py
index a66bdc4..12e7c41 100644
--- a/backend/tests/test_build_durable.py
+++ b/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,
diff --git a/backend/tests/test_durable_sdk.py b/backend/tests/test_durable_sdk.py
index 8a7f9bb..d110c2d 100644
--- a/backend/tests/test_durable_sdk.py
+++ b/backend/tests/test_durable_sdk.py
@@ -270,9 +270,11 @@ async def test_attribution_rides_the_run_and_survives_resume(rt):
 
     SINK.clear()
     account_id = _account_id(rt.engine, "op@example.com")
-    wfid = await rt.AttributedFlow.start(
-        subject={"type": "widget", "id": 878787}, account_id=account_id
-    )
+    wfid = (
+        await rt.AttributedFlow.start(
+            subject={"type": "widget", "id": 878787}, account_id=account_id
+        )
+    ).run_id
     parked = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.PENDING_INPUT)
     with rt.engine.connect() as conn:
         attributes = conn.execute(
@@ -295,7 +297,7 @@ async def test_browser_origin_start_inherits_the_ambient_account(rt):
     account_id = _account_id(rt.engine, "ambient@example.com")
     token = current_account_id.set(account_id)
     try:
-        wfid = await rt.RecordFeedback.start(subject=None, repo="owner/ambient")
+        wfid = (await rt.RecordFeedback.start(subject=None, repo="owner/ambient")).run_id
     finally:
         current_account_id.reset(token)
     await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FINISHED)
@@ -308,18 +310,19 @@ async def test_duplicate_start_shares_the_run_across_accounts(rt):
     first = _account_id(rt.engine, "op@example.com")
     second = _account_id(rt.engine, "peer@example.com")
     subject = {"type": "widget", "id": 909090}
-    wfid = await rt.SampleFlow.start(subject=subject, account_id=first, repo="owner/app")
+    wfid = (await rt.SampleFlow.start(subject=subject, account_id=first, repo="owner/app")).run_id
     parked = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.PENDING_INPUT)
 
-    dup = await rt.SampleFlow.start(subject=subject, account_id=second, repo="owner/app")
-    assert dup == wfid
+    duplicate = await rt.SampleFlow.start(subject=subject, account_id=second, repo="owner/app")
+    assert duplicate.run_id == wfid
+    assert duplicate.is_duplicate
 
     await parked.resume(action="merge")
     await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FINISHED)
 
 
 async def test_step_gate_resume_finish(rt):
-    wfid = await rt.SampleFlow.start(subject=None, repo="owner/app")
+    wfid = (await rt.SampleFlow.start(subject=None, repo="owner/app")).run_id
 
     parked = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.PENDING_INPUT)
     assert parked.input_gate == "approve"
@@ -336,7 +339,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=None)).run_id
     parked = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.PENDING_INPUT)
     first_asked_at = parked.input_requested_at
 
@@ -378,7 +381,7 @@ async def test_duplicate_replies_to_one_round_collapse(rt):
 async def test_fail_branch(rt):
     from sqlalchemy import text
 
-    wfid = await rt.SampleFlow.start(subject=None, repo="owner/app")
+    wfid = (await rt.SampleFlow.start(subject=None, repo="owner/app")).run_id
     parked = await _wait_for(rt.engine, wfid, lambda r: r.input_gate == "approve")
 
     await parked.resume(action="close")
@@ -397,7 +400,7 @@ async def test_fail_branch(rt):
 async def test_subjectless_gate_fails_loudly(rt):
     """A gate with no on_wait override fails a subjectless run now, instead of
     parking it unseen for the whole gate TTL."""
-    wfid = await rt.ConfirmFlow.start(subject=None)
+    wfid = (await rt.ConfirmFlow.start(subject=None)).run_id
     failed = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FAILED)
     assert failed.failure
     assert "'confirm'" in failed.failure
@@ -408,7 +411,7 @@ async def test_subjectless_gate_fails_loudly(rt):
 async def test_subject_gate_parks_unchanged(rt):
     """The same no-on_wait gate still parks and resumes for a subject run — the
     subject's watchers are the ones who'd see it, feed-side."""
-    wfid = await rt.ConfirmFlow.start(subject={"type": "widget", "id": 636363})
+    wfid = (await rt.ConfirmFlow.start(subject={"type": "widget", "id": 636363})).run_id
     parked = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.PENDING_INPUT)
     assert parked.input_gate == "confirm"
     # start() stamped the subject as workflow attributes — the keying every
@@ -427,7 +430,7 @@ async def test_subject_gate_parks_unchanged(rt):
 async def test_subjectless_review_fails_loudly(rt):
     """review() is a human gate too: a subjectless run fails instead of parking
     an in-app ask nobody would ever see."""
-    wfid = await rt.ReviewFlow.start(subject=None)
+    wfid = (await rt.ReviewFlow.start(subject=None)).run_id
     failed = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FAILED)
     assert failed.failure
     assert "'review'" in failed.failure
@@ -477,7 +480,7 @@ async def _fake_render(*_a, **_k):
     monkeypatch.setattr("druks.sandbox.client.Client.ephemeral", _fake_ephemeral)
     monkeypatch.setattr("druks.agents.render_prompt", _fake_render)
 
-    wfid = await rt.AgentFlow.start(subject=None, repo="owner/app")
+    wfid = (await rt.AgentFlow.start(subject=None, repo="owner/app")).run_id
     failed = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FAILED)
     assert failed.failure == "stopped by agent"
     assert seen and seen[0]["artifact_dir"].name == f"run-{wfid}"
@@ -673,7 +676,7 @@ async def test_input_is_validated_at_start(rt):
     with pytest.raises(WorkflowError):
         await rt.SubjectFlow.start(subject=None, repo="x")  # takes no input
 
-    wfid = await rt.RecordFeedback.start(subject=None, repo="owner/flat")
+    wfid = (await rt.RecordFeedback.start(subject=None, repo="owner/flat")).run_id
     await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FINISHED)
     assert "owner/flat" in SINK
 
@@ -735,7 +738,7 @@ async def run_multistep(self) -> None: ...
 async def test_subject_reaches_body_and_result_rides_finished_event(rt):
     from druks.events.models import Event
 
-    wfid = await rt.SubjectFlow.start(subject={"type": "widget", "id": 7})
+    wfid = (await rt.SubjectFlow.start(subject={"type": "widget", "id": 7})).run_id
     await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FINISHED)
 
     assert "subj-id:7" in SINK  # the platform threaded subject into self.subject
@@ -769,9 +772,9 @@ async def test_run_events_carry_subject(rt):
     # run-state transition emits a run-level event keyed to it.
     from druks.events.models import Event
 
-    wfid = await rt.RecordFeedback.start(
-        subject={"type": "widget", "id": 4242}, repo="owner/evented"
-    )
+    wfid = (
+        await rt.RecordFeedback.start(subject={"type": "widget", "id": 4242}, repo="owner/evented")
+    ).run_id
     await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FINISHED)
 
     session = get_session(rt.engine)
@@ -791,10 +794,14 @@ async def test_duplicate_start_returns_the_live_run(rt):
     # hands back the live run's id. The slot frees at the terminal outcome, so
     # the subject can run again.
     subject = {"type": "widget", "id": 515151}
-    wfid = await rt.SampleFlow.start(subject=subject, repo="owner/app")
+    start_result = await rt.SampleFlow.start(subject=subject, repo="owner/app")
+    assert not start_result.is_duplicate
+    wfid = start_result.run_id
     parked = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.PENDING_INPUT)
 
-    assert await rt.SampleFlow.start(subject=subject, repo="owner/app") == wfid
+    duplicate = await rt.SampleFlow.start(subject=subject, repo="owner/app")
+    assert duplicate.run_id == wfid
+    assert duplicate.is_duplicate
 
     await parked.resume(action="merge")
     await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FINISHED)
@@ -803,7 +810,7 @@ async def test_duplicate_start_returns_the_live_run(rt):
     deadline = asyncio.get_event_loop().time() + 10.0
     fresh = wfid
     while fresh == wfid and asyncio.get_event_loop().time() < deadline:
-        fresh = await rt.SampleFlow.start(subject=subject, repo="owner/app")
+        fresh = (await rt.SampleFlow.start(subject=subject, repo="owner/app")).run_id
         if fresh == wfid:
             await asyncio.sleep(0.1)
     assert fresh != wfid
@@ -822,7 +829,7 @@ async def _boom(*args, **kwargs):
         with pytest.raises(RuntimeError, match="queue down"):
             await rt.SubjectFlow.start(subject=subject)
 
-    wfid = await rt.SubjectFlow.start(subject=subject)
+    wfid = (await rt.SubjectFlow.start(subject=subject)).run_id
     await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FINISHED)
 
 
@@ -831,7 +838,7 @@ async def test_subjectless_run_emits_no_events(rt):
     # must not write run-level events into the feed.
     from druks.events.models import Event
 
-    wfid = await rt.RecordFeedback.start(subject=None, repo="owner/quiet")
+    wfid = (await rt.RecordFeedback.start(subject=None, repo="owner/quiet")).run_id
     await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FINISHED)
 
     session = get_session(rt.engine)
diff --git a/backend/tests/test_notifications_durable.py b/backend/tests/test_notifications_durable.py
index 3d3d5ac..5e41a53 100644
--- a/backend/tests/test_notifications_durable.py
+++ b/backend/tests/test_notifications_durable.py
@@ -399,7 +399,9 @@ async def test_in_app_park_notifies_with_actions(rt, deliver_spy):
     )
     _set_gate_park_pointer(rt, destination.id)
 
-    workflow_id = await rt.InAppFlow.start(subject={"type": "notification_probe", "id": 9001})
+    workflow_id = (
+        await rt.InAppFlow.start(subject={"type": "notification_probe", "id": 9001})
+    ).run_id
     parked = await _wait_run(rt, workflow_id, lambda run: run.state == RunState.PENDING_INPUT)
 
     notification = await _wait_notification(rt, workflow_id, "delivered")
@@ -429,7 +431,9 @@ async def test_external_park_notifies_without_actions(rt, deliver_spy):
     )
     _set_gate_park_pointer(rt, destination.id)
 
-    workflow_id = await rt.ExternalFlow.start(subject={"type": "notification_probe", "id": 9002})
+    workflow_id = (
+        await rt.ExternalFlow.start(subject={"type": "notification_probe", "id": 9002})
+    ).run_id
     await _wait_run(rt, workflow_id, lambda run: run.state == RunState.PENDING_INPUT)
 
     notification = await _wait_notification(rt, workflow_id, "delivered")
@@ -446,7 +450,9 @@ async def test_external_park_with_declared_url_sets_deep_link(rt, deliver_spy):
     )
     _set_gate_park_pointer(rt, destination.id)
 
-    workflow_id = await rt.ExternalUrlFlow.start(subject={"type": "notification_probe", "id": 9003})
+    workflow_id = (
+        await rt.ExternalUrlFlow.start(subject={"type": "notification_probe", "id": 9003})
+    ).run_id
     await _wait_run(rt, workflow_id, lambda run: run.state == RunState.PENDING_INPUT)
 
     notification = await _wait_notification(rt, workflow_id, "delivered")
@@ -457,8 +463,12 @@ async def test_external_park_with_declared_url_sets_deep_link(rt, deliver_spy):
 async def test_no_designated_destination_notifies_nothing(rt, deliver_spy):
     _set_gate_park_pointer(rt, None)
 
-    in_app_id = await rt.InAppFlow.start(subject={"type": "notification_probe", "id": 9004})
-    external_id = await rt.ExternalFlow.start(subject={"type": "notification_probe", "id": 9014})
+    in_app_id = (
+        await rt.InAppFlow.start(subject={"type": "notification_probe", "id": 9004})
+    ).run_id
+    external_id = (
+        await rt.ExternalFlow.start(subject={"type": "notification_probe", "id": 9014})
+    ).run_id
     await _wait_run(rt, in_app_id, lambda run: run.state == RunState.PENDING_INPUT)
     await _wait_run(rt, external_id, lambda run: run.state == RunState.PENDING_INPUT)
 
@@ -476,7 +486,9 @@ async def test_deleted_designated_destination_notifies_nothing(rt, deliver_spy):
     _set_gate_park_pointer(rt, destination.id)
     _seed(rt, lambda: Destination.get(destination.id).delete())
 
-    workflow_id = await rt.ExternalFlow.start(subject={"type": "notification_probe", "id": 9005})
+    workflow_id = (
+        await rt.ExternalFlow.start(subject={"type": "notification_probe", "id": 9005})
+    ).run_id
     await _wait_run(rt, workflow_id, lambda run: run.state == RunState.PENDING_INPUT)
 
     await asyncio.sleep(1.0)
@@ -496,7 +508,7 @@ async def test_subjectless_park_notifies_nothing(rt, deliver_spy):
     )
     _set_gate_park_pointer(rt, destination.id)
 
-    workflow_id = await rt.ExternalFlow.start(subject=None)
+    workflow_id = (await rt.ExternalFlow.start(subject=None)).run_id
     await _wait_run(rt, workflow_id, lambda run: run.state == RunState.PENDING_INPUT)
 
     await asyncio.sleep(1.0)
@@ -512,7 +524,9 @@ async def test_failed_delivery_leaves_run_parked_and_resumable(rt, deliver_spy,
     )
     _set_gate_park_pointer(rt, destination.id)
 
-    workflow_id = await rt.InAppFlow.start(subject={"type": "notification_probe", "id": 9006})
+    workflow_id = (
+        await rt.InAppFlow.start(subject={"type": "notification_probe", "id": 9006})
+    ).run_id
     await _wait_run(rt, workflow_id, lambda run: run.state == RunState.PENDING_INPUT)
     notification = await _wait_notification(rt, workflow_id, "failed")
     assert _WEBHOOK_URL not in notification.last_error
@@ -530,7 +544,9 @@ async def test_replayed_park_notifies_once(rt, deliver_spy):
     )
     _set_gate_park_pointer(rt, destination.id)
 
-    workflow_id = await rt.ExternalFlow.start(subject={"type": "notification_probe", "id": 9007})
+    workflow_id = (
+        await rt.ExternalFlow.start(subject={"type": "notification_probe", "id": 9007})
+    ).run_id
     await _wait_run(rt, workflow_id, lambda run: run.state == RunState.PENDING_INPUT)
     await _wait_notification(rt, workflow_id, "delivered")
     assert len(deliver_spy.calls) == 1
@@ -582,7 +598,9 @@ async def test_each_park_round_gets_its_own_notification(rt, deliver_spy):
     )
     _set_gate_park_pointer(rt, destination.id)
 
-    workflow_id = await rt.DoubleParkFlow.start(subject={"type": "notification_probe", "id": 9008})
+    workflow_id = (
+        await rt.DoubleParkFlow.start(subject={"type": "notification_probe", "id": 9008})
+    ).run_id
     first_round = await _wait_run(rt, workflow_id, lambda run: run.state == RunState.PENDING_INPUT)
     await _wait_notification(rt, workflow_id, "delivered")
     assert len(deliver_spy.calls) == 1
@@ -651,7 +669,9 @@ async def test_respond_round_trip_finishes_the_run(rt, deliver_spy):
         rt, lambda: Destination.create(name="inbox-respond", kind="slack_webhook", url=_WEBHOOK_URL)
     )
     _set_gate_park_pointer(rt, destination.id)
-    workflow_id = await rt.InAppFlow.start(subject={"type": "notification_probe", "id": 9009})
+    workflow_id = (
+        await rt.InAppFlow.start(subject={"type": "notification_probe", "id": 9009})
+    ).run_id
     await _wait_run(rt, workflow_id, lambda run: run.state == RunState.PENDING_INPUT)
     notification = await _wait_notification(rt, workflow_id, "delivered")
     token = notification.correlation_token
@@ -677,7 +697,9 @@ async def test_concurrent_responds_resolve_to_one_answer(rt, deliver_spy):
         rt, lambda: Destination.create(name="inbox-race", kind="slack_webhook", url=_WEBHOOK_URL)
     )
     _set_gate_park_pointer(rt, destination.id)
-    workflow_id = await rt.InAppFlow.start(subject={"type": "notification_probe", "id": 9010})
+    workflow_id = (
+        await rt.InAppFlow.start(subject={"type": "notification_probe", "id": 9010})
+    ).run_id
     await _wait_run(rt, workflow_id, lambda run: run.state == RunState.PENDING_INPUT)
     notification = await _wait_notification(rt, workflow_id, "delivered")
     token = notification.correlation_token
diff --git a/backend/tests/test_project_repo_routes.py b/backend/tests/test_project_repo_routes.py
index 4f97f61..c41c76d 100644
--- a/backend/tests/test_project_repo_routes.py
+++ b/backend/tests/test_project_repo_routes.py
@@ -1,6 +1,7 @@
 from pathlib import Path
 
 import pytest
+from druks.workflows import WorkflowStartResult
 from fastapi.testclient import TestClient
 
 
@@ -23,7 +24,7 @@ def _stub_profile_start(monkeypatch):
 
     async def _start(cls, *, subject, **input):
         calls.append({"subject": subject, **input})
-        return "fake-run-id"
+        return WorkflowStartResult(run_id="fake-run-id", is_duplicate=False)
 
     monkeypatch.setattr(Profile, "start", classmethod(_start))
     return calls
diff --git a/backend/tests/test_proof_extension.py b/backend/tests/test_proof_extension.py
index 1d85767..021aee0 100644
--- a/backend/tests/test_proof_extension.py
+++ b/backend/tests/test_proof_extension.py
@@ -6,6 +6,7 @@
 import pytest
 from druks.extensions import loader
 from druks.extensions.loader import load_extension
+from druks.workflows import WorkflowStartResult
 
 # druks-field_notes is a real, standalone extension package that lives outside the
 # druks tree (``backend/tests/druks-field_notes``) — a model, a migration, a route, a
@@ -311,7 +312,7 @@ async def test_dispatch_starts_a_run_keyed_to_the_note(installed, monkeypatch):
 
     async def _capture(*, subject, **input):
         started.update(subject=subject, input=input)
-        return "run-1"
+        return WorkflowStartResult(run_id="run-1", is_duplicate=False)
 
     monkeypatch.setattr(Summarize, "start", _capture)
 
diff --git a/backend/tests/test_scope_durable.py b/backend/tests/test_scope_durable.py
index 44faadd..a4f1163 100644
--- a/backend/tests/test_scope_durable.py
+++ b/backend/tests/test_scope_durable.py
@@ -122,11 +122,13 @@ async def _run_agent(self, **kwargs):
 
     monkeypatch.setattr("druks.agents.Agent._run", _run_agent)
 
-    wfid = await rt.flow.start(
-        subject=WorkItem.subject_for(item_id),
-        source="linear",
-        remote_key="ACME-1",
-    )
+    wfid = (
+        await rt.flow.start(
+            subject=WorkItem.subject_for(item_id),
+            source="linear",
+            remote_key="ACME-1",
+        )
+    ).run_id
 
     parked = await _wait(
         rt.engine,
diff --git a/backend/tests/test_scope_reply_rescope.py b/backend/tests/test_scope_reply_rescope.py
index 1e9307c..d783813 100644
--- a/backend/tests/test_scope_reply_rescope.py
+++ b/backend/tests/test_scope_reply_rescope.py
@@ -3,6 +3,7 @@
 from druks.build.scoping.workflows import Scope, ScopeReply
 from druks.durable import Run
 from druks.ticketing.datastructures import Ticket
+from druks.workflows import WorkflowStartResult
 from uuid_utils import uuid7
 
 
@@ -14,7 +15,7 @@ def _stub_enqueue(monkeypatch):
 
     async def _fake_start(**kwargs):
         calls.append(kwargs)
-        return str(uuid7())
+        return WorkflowStartResult(run_id=str(uuid7()), is_duplicate=False)
 
     monkeypatch.setattr("druks.build.scoping.workflows.Scope.start", _fake_start)
     return calls
diff --git a/backend/tests/test_webhooks_push.py b/backend/tests/test_webhooks_push.py
index 0903f63..e42e944 100644
--- a/backend/tests/test_webhooks_push.py
+++ b/backend/tests/test_webhooks_push.py
@@ -3,6 +3,7 @@
 import druks.build.subscribers  # noqa: F401 — registers the repo.pushed subscriber
 from conftest import make_settings
 from druks.core.webhooks.github import GitHubEvents
+from druks.workflows import WorkflowStartResult
 
 
 def _push_payload(*, full_name, default_branch, ref, changed_paths=()):
@@ -28,7 +29,7 @@ def _stub_profile_start(monkeypatch):
 
     async def _start(cls, *, subject, **input):
         calls.append({"subject": subject, **input})
-        return "fake-run-id"
+        return WorkflowStartResult(run_id="fake-run-id", is_duplicate=False)
 
     monkeypatch.setattr(Profile, "start", classmethod(_start))
     return calls
diff --git a/docs/writing-an-extension.md b/docs/writing-an-extension.md
index 192a124..4891dba 100644
--- a/docs/writing-an-extension.md
+++ b/docs/writing-an-extension.md
@@ -124,17 +124,20 @@ control flow and I/O inside steps. See
 Start a workflow with an explicit subject:
 
 ```python
-run_id = await Sweep.start(
+start_result = await Sweep.start(
     subject={"type": "repository", "id": repo_id},
     repo=full_name,
 )
 ```
 
-Pass `subject=None` deliberately for background work. A subject has at most one
-active run of a workflow kind; a duplicate start returns the active run id —
-attribution never changes that (two accounts starting the same subject share
-the one run). Wrap `start()` in a domain `dispatch()` method when the extension
-needs lookup, snapshot, or routing policy before launch.
+Pass `subject=None` deliberately for background work. `start()` returns a
+`WorkflowStartResult`: a subject has at most one active run of a workflow kind,
+and a duplicate start hands back the active run's `run_id` with `is_duplicate`
+set — attribution never changes that (two accounts starting the same subject
+share the one run). Code that consumed the old string return migrates with
+`run_id = (await Workflow.start(...)).run_id`. Wrap `start()` in a domain
+`dispatch()` method when the extension needs lookup, snapshot, or routing
+policy before launch.
 
 A browser-origin start attributes itself: the session gate stamps the
 request's signed-in account, and `start()` inherits it — a route that starts a