diff --git a/codeframe/core/blockers.py b/codeframe/core/blockers.py
index cdce18af..a3a3b3e9 100644
--- a/codeframe/core/blockers.py
+++ b/codeframe/core/blockers.py
@@ -83,6 +83,19 @@ def create(
         Created Blocker
     """
     origin = BlockerOrigin(created_by)
+
+    # Dedupe on (task_id, question, OPEN) — issue #735. An adapter (react/plan/
+    # verification_wrapper) persists the blocker, then runtime.execute_agent creates
+    # a second one from result.blocker_question. Return the existing OPEN blocker so
+    # one escalation yields one blocker and one webhook; first writer's origin wins.
+    # NOTE: read-then-insert with no UNIQUE index — safe here because the two callers
+    # (adapter + runtime) run sequentially in the same process for one run, not
+    # concurrently. A partial UNIQUE index on (workspace_id, task_id, question) WHERE
+    # status='OPEN' would make it atomic against cross-process races.
+    existing = _find_open_duplicate(workspace, task_id, question)
+    if existing is not None:
+        return existing
+
     blocker_id = str(uuid.uuid4())
     now = _utc_now().isoformat()
 
@@ -126,6 +139,33 @@ def create(
     return blocker
 
 
+def _find_open_duplicate(
+    workspace: Workspace, task_id: Optional[str], question: str
+) -> Optional[Blocker]:
+    """Return an existing OPEN blocker with the same (task_id, question), if any."""
+    conn = get_db_connection(workspace)
+    cursor = conn.cursor()
+
+    select = (
+        "SELECT id, workspace_id, task_id, question, answer, status, created_at, "
+        "answered_at, COALESCE(created_by, 'human') as created_by FROM blockers "
+        "WHERE workspace_id = ? AND question = ? AND status = ?"
+    )
+    params: list = [workspace.id, question, BlockerStatus.OPEN.value]
+    # SQL `task_id = NULL` never matches; use `IS NULL` for workspace-level blockers.
+    if task_id is None:
+        select += " AND task_id IS NULL"
+    else:
+        select += " AND task_id = ?"
+        params.append(task_id)
+
+    cursor.execute(select, params)
+    row = cursor.fetchone()
+    conn.close()
+
+    return _row_to_blocker(row) if row else None
+
+
 def _dispatch_blocker_webhook(
     workspace: Workspace, blocker_id: str, task_id: Optional[str]
 ) -> None:
diff --git a/tests/core/test_blockers_webhook.py b/tests/core/test_blockers_webhook.py
index a46fd1e9..23da017e 100644
--- a/tests/core/test_blockers_webhook.py
+++ b/tests/core/test_blockers_webhook.py
@@ -80,3 +80,76 @@ def test_webhook_failure_does_not_break_blocker_create(workspace):
     ):
         blocker = blockers.create(workspace, question="why?")
     assert blocker.id  # still got created
+
+
+def test_duplicate_open_blocker_is_deduped(workspace):
+    """One escalation → exactly one OPEN blocker and one webhook (issue #735).
+
+    Adapter + runtime each call create() with the same (task_id, question); the
+    second must return the existing OPEN blocker without a new row or webhook.
+    """
+    save_notifications_config(
+        workspace,
+        {"webhook_url": "https://example.com/h", "webhook_enabled": True},
+    )
+    with patch(
+        "codeframe.notifications.webhook.WebhookNotificationService"
+    ) as MockSvc:
+        instance = MockSvc.return_value
+        first = blockers.create(
+            workspace, question="Gates failing?", task_id="t-1", created_by="agent"
+        )
+        second = blockers.create(
+            workspace, question="Gates failing?", task_id="t-1"
+        )
+
+    assert second.id == first.id  # same blocker, not a duplicate
+    assert second.created_by == first.created_by  # first writer (agent) wins
+    assert len(blockers.list_open(workspace)) == 1
+    instance.send_event_background.assert_called_once()  # only one webhook
+
+
+def test_dedupe_does_not_collapse_distinct_questions(workspace):
+    """Different questions on the same task remain separate blockers."""
+    a = blockers.create(workspace, question="Q1?", task_id="t-1")
+    b = blockers.create(workspace, question="Q2?", task_id="t-1")
+    assert a.id != b.id
+    assert len(blockers.list_open(workspace)) == 2
+
+
+def test_dedupe_scoped_to_open_status(workspace):
+    """An answered blocker does not suppress a new one with the same question."""
+    first = blockers.create(workspace, question="Same?", task_id="t-1")
+    blockers.answer(workspace, first.id, "resolved")
+    second = blockers.create(workspace, question="Same?", task_id="t-1")
+    assert second.id != first.id
+    assert len(blockers.list_open(workspace)) == 1
+
+
+def test_dedupe_handles_null_task_id(workspace):
+    """Workspace-level (task_id=None) duplicates are deduped too."""
+    first = blockers.create(workspace, question="Global?")
+    second = blockers.create(workspace, question="Global?")
+    assert second.id == first.id
+    assert len(blockers.list_open(workspace)) == 1
+
+
+def test_dedupe_scoped_to_task_id(workspace):
+    """Same question on different tasks stays as separate blockers."""
+    a = blockers.create(workspace, question="Q?", task_id="t-1")
+    b = blockers.create(workspace, question="Q?", task_id="t-2")
+    assert a.id != b.id
+    assert len(blockers.list_open(workspace)) == 2
+
+
+def test_dedupe_does_not_cross_workspaces(tmp_path):
+    """The same question in two workspaces produces two distinct blockers."""
+    questions = "Cross?"
+    ids = []
+    for name in ("ws-a", "ws-b"):
+        ws_path = tmp_path / name
+        ws_path.mkdir(parents=True, exist_ok=True)
+        ws = create_or_load_workspace(ws_path)
+        ids.append(blockers.create(ws, question=questions, task_id="t-1").id)
+        assert len(blockers.list_open(ws)) == 1
+    assert ids[0] != ids[1]