fix: deduplicate backlog work items

desloppify/engine/_work_queue/snapshot.py

@@ -505,21 +505,25 @@ def _build_backlog(
     p: _Partitions,
     execution_ids: set[str],
 ) -> list[WorkQueueItem]:
-    return [
-        item
-        for item in (
-            [
-                *p.objective_items,
-                *p.initial_review_items,
-                *p.postflight_assessment_items,
-                *p.review_issue_items,
-                *p.scan_items,
-                *p.postflight_workflow_items,
-                *p.triage_items,
-            ]
-        )
-        if item.get("id", "") not in execution_ids
+    """Return visible backlog items once, preserving partition priority."""
+    seen_ids: set[str] = set()
+    backlog: list[WorkQueueItem] = []
+    candidates = [
+        *p.objective_items,
+        *p.initial_review_items,
+        *p.postflight_assessment_items,
+        *p.review_issue_items,
+        *p.scan_items,
+        *p.postflight_workflow_items,
+        *p.triage_items,
     ]
+    for item in candidates:
+        item_id = str(item.get("id", ""))
+        if item_id in execution_ids or item_id in seen_ids:
+            continue
+        seen_ids.add(item_id)
+        backlog.append(item)
+    return backlog
 
 
 # ---------------------------------------------------------------------------

desloppify/tests/review/test_work_queue_plan_order_and_triage.py

@@ -530,6 +530,40 @@ def test_backlog_queue_excludes_execution_objective_items():
     assert "workflow::run-scan" not in ids
 
 
+def test_backlog_queue_deduplicates_mechanical_triage_findings():
+    """A mechanical finding shared with triage appears once in the backlog."""
+    from desloppify.engine._plan.schema import empty_plan
+    from desloppify.engine._state.issue_semantics import is_triage_finding
+
+    finding = _issue(
+        "dict_keys::src/a.py::phantom_read::payload::missing",
+        detector="dict_keys",
+        tier=2,
+        confidence="high",
+    )
+    assert is_triage_finding(finding)
+
+    state = _state([finding])
+    plan = empty_plan()
+    # A stale plan puts the snapshot in scan mode, where the finding belongs
+    # in backlog through both the objective and triage partitions.
+    plan["queue_order"] = ["missing::planned"]
+    plan["plan_start_scores"] = {"strict": 80.0}
+    plan["refresh_state"] = {"lifecycle_phase": "execute"}
+
+    queue = build_backlog_queue(
+        state,
+        options=QueueBuildOptions(
+            count=None,
+            include_subjective=False,
+            plan=plan,
+        ),
+    )
+
+    ids = [item["id"] for item in queue["items"]]
+    assert ids == [finding["id"]]
+
+
 def test_unplanned_objective_items_dont_block_postflight():
     """Unplanned objective items don't keep the system in EXECUTE phase.