ENG-812 - Ship: ticket intake is not idempotent — a re-delivered "Ready for Agent" webhook re-builds a merged item

backend/druks/contrib/ship/subscribers.py

@@ -77,4 +77,9 @@ async def ticket_transition_drives_the_funnel(*, payload: dict) -> None:
     """Dispatch a build when a ticket from the chosen tracker enters its trigger status."""
     settings = Ship.settings()
     if payload["source"] == settings.tracker and payload["status"] == settings.trigger_status:
-        await Build.dispatch(ticket=payload)
+        item = WorkItem.get_for_ticket_key(
+            source=payload["source"],
+            ticket_key=payload["identifier"],
+        )
+        if not item or (not item.resolution and not item.get_status(workflow=Build).is_active):
+            await Build.dispatch(ticket=payload)

backend/druks/durable/schemas.py

@@ -6,7 +6,7 @@
 
 from druks.schemas import BaseResponse
 
-from .enums import AgentCallStatus, RunState
+from .enums import ACTIVE_STATES, AgentCallStatus, RunState
 
 if TYPE_CHECKING:
     from .models import AgentCall, Artifact, Run
@@ -188,6 +188,10 @@ class SubjectStatus(BaseResponse):
     triggered_at: datetime | None = None
     account_username: str | None = None
 
+    @property
+    def is_active(self) -> bool:
+        return bool(self.kind) and self.state in ACTIVE_STATES
+
     @property
     def is_parked(self) -> bool:
         return self.state == RunState.PARKED

backend/tests/ship/test_webhooks_jira.py

@@ -11,6 +11,8 @@
 from druks.webhooks.router import router as webhooks_router
 from fastapi import HTTPException
 
+from ship.factories import make_test_work_item, seed_build_run
+
 
 def _provider(tmp_path, *, payload, headers=None):
     events = JiraEvents(
@@ -150,7 +152,9 @@ def _pin_settings(monkeypatch, **over):
     monkeypatch.setattr(subs.Ship, "settings", classmethod(lambda cls: settings))
 
 
-async def test_trigger_status_dispatches_build_with_the_webhook_payload(tmp_path, monkeypatch):
+async def test_trigger_status_dispatches_build_with_the_webhook_payload(
+    tmp_path, druks_db, monkeypatch
+):
     """The build funnel receives the normalized ticket payload without a refetch."""
     _pin_settings(monkeypatch, jira_trigger_status="Ready")
     build = AsyncMock()
@@ -162,6 +166,60 @@ async def test_trigger_status_dispatches_build_with_the_webhook_payload(tmp_path
     build.assert_awaited_once_with(ticket=payload)
 
 
+async def test_trigger_status_does_not_redispatch_a_resolved_item(druks_db, monkeypatch):
+    item = make_test_work_item(
+        repo="octo/alfred",
+        source="jira",
+        ticket_key="IT-12",
+        title="Add an endpoint",
+    )
+    item.resolution = "merged"
+    druks_db.flush()
+    _pin_settings(monkeypatch, jira_trigger_status="Ready")
+    build = AsyncMock()
+    monkeypatch.setattr(subs.Build, "dispatch", build)
+
+    await subs.ticket_transition_drives_the_funnel(payload=_jira_payload(status="Ready"))
+
+    build.assert_not_awaited()
+
+
+@pytest.mark.parametrize("state", ["scheduled", "running", "parked"])
+async def test_trigger_status_does_not_redispatch_an_active_build(state, druks_db, monkeypatch):
+    item = make_test_work_item(
+        repo="octo/alfred",
+        source="jira",
+        ticket_key="IT-12",
+        title="Add an endpoint",
+    )
+    seed_build_run(druks_db, work_item_id=item.id, state=state)
+    _pin_settings(monkeypatch, jira_trigger_status="Ready")
+    build = AsyncMock()
+    monkeypatch.setattr(subs.Build, "dispatch", build)
+
+    await subs.ticket_transition_drives_the_funnel(payload=_jira_payload(status="Ready"))
+
+    build.assert_not_awaited()
+
+
+async def test_trigger_status_redispatches_after_a_failed_build(druks_db, monkeypatch):
+    item = make_test_work_item(
+        repo="octo/alfred",
+        source="jira",
+        ticket_key="IT-12",
+        title="Add an endpoint",
+    )
+    seed_build_run(druks_db, work_item_id=item.id, state="failed", failure="boom")
+    _pin_settings(monkeypatch, jira_trigger_status="Ready")
+    build = AsyncMock()
+    monkeypatch.setattr(subs.Build, "dispatch", build)
+    payload = _jira_payload(status="Ready")
+
+    await subs.ticket_transition_drives_the_funnel(payload=payload)
+
+    build.assert_awaited_once_with(ticket=payload)
+
+
 async def test_trigger_status_routes_a_new_ticket_by_label(tmp_path, druks_db, monkeypatch):
     """No work item yet: the label names the repo, the registry routes it."""
     from druks.contrib.ship.models import Project, ProjectRepo, WorkItem

backend/tests/test_durable_schemas.py

@@ -1,3 +1,4 @@
+import pytest
 from druks.accounts.models import Account
 from druks.durable.enums import RunState
 from druks.durable.exceptions import GateTimeout
@@ -30,6 +31,32 @@ def _status_of(runs, calls=None):
     return _status(runs[0], calls or [])
 
 
+@pytest.mark.parametrize(
+    ("state", "expected"),
+    [
+        (RunState.SCHEDULED, True),
+        (RunState.RUNNING, True),
+        (RunState.PARKED, True),
+        (RunState.FINISHED, False),
+        (RunState.FAILED, False),
+        (RunState.CANCELLED, False),
+        (RunState.ORPHANED, False),
+    ],
+)
+def test_subject_status_is_active_only_for_actual_active_runs(state, expected):
+    status = _status_of([_run("run", "ship.build", state)])
+
+    assert status.is_active is expected
+
+
+def test_subject_status_without_a_driving_run_is_not_active():
+    status = _status(None, [])
+
+    assert status.state == RunState.SCHEDULED
+    assert not status.kind
+    assert not status.is_active
+
+
 def test_subject_state_takes_the_newest_run():
     runs = [
         _run("new", "ship.build", RunState.RUNNING),