diff --git a/app/lib/backend/schema/gen/wrapped_task_integrations_wire.g.dart b/app/lib/backend/schema/gen/wrapped_task_integrations_wire.g.dart
index 52f22e88100..81249334421 100644
--- a/app/lib/backend/schema/gen/wrapped_task_integrations_wire.g.dart
+++ b/app/lib/backend/schema/gen/wrapped_task_integrations_wire.g.dart
@@ -107,28 +107,40 @@ class GeneratedOAuthUrlResponse {
 }
 
 class GeneratedCreateTaskResponse {
+  final bool? ambiguous;
   final String? error;
+  final String? errorCode;
   final String? externalTaskId;
+  final bool? retryable;
   final bool success;
 
   const GeneratedCreateTaskResponse({
+    this.ambiguous,
     this.error,
+    this.errorCode,
     this.externalTaskId,
+    this.retryable,
     required this.success,
   });
 
   factory GeneratedCreateTaskResponse.fromJson(Map<String, dynamic> json) {
     return GeneratedCreateTaskResponse(
+      ambiguous: _readFieldValue<bool>(_readField(json, const ["ambiguous"]), "ambiguous", _readBool, requiredField: false, nullable: true),
       error: _readFieldValue<String>(_readField(json, const ["error"]), "error", _readString, requiredField: false, nullable: true),
+      errorCode: _readFieldValue<String>(_readField(json, const ["error_code"]), "error_code", _readString, requiredField: false, nullable: true),
       externalTaskId: _readFieldValue<String>(_readField(json, const ["external_task_id"]), "external_task_id", _readString, requiredField: false, nullable: true),
+      retryable: _readFieldValue<bool>(_readField(json, const ["retryable"]), "retryable", _readBool, requiredField: false, nullable: true),
       success: _required(_readFieldValue<bool>(_readField(json, const ["success"]), "success", _readBool, requiredField: true, nullable: false), "success"),
     );
   }
 
   Map<String, dynamic> toJson() {
     return {
+      'ambiguous': ambiguous,
       'error': error,
+      'error_code': errorCode,
       'external_task_id': externalTaskId,
+      'retryable': retryable,
       'success': success,
     };
   }
diff --git a/backend/database/memories.py b/backend/database/memories.py
index cdabe2ed8ae..4e93ef5e082 100644
--- a/backend/database/memories.py
+++ b/backend/database/memories.py
@@ -1,6 +1,7 @@
 import copy
 import hashlib
 import json
+from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import Any, Callable, Dict, List, Optional, TypedDict, cast
 
@@ -34,6 +35,7 @@ class FirestoreNotFound(Exception):
 
 memories_collection = 'memories'
 users_collection = 'users'
+_DELETE_BATCH_SIZE = 499
 
 
 class MemoryDoc(TypedDict, total=False):
@@ -78,6 +80,15 @@ class MemoryDoc(TypedDict, total=False):
     to_sha256: Optional[str]
 
 
+@dataclass(frozen=True)
+class LegacyMemoryDeleteResult:
+    memory_ids: List[str]
+
+    @property
+    def committed_count(self) -> int:
+        return len(self.memory_ids)
+
+
 # Signature expected by ``prepare_for_read`` for the post-read decrypt hook. The
 # concrete helper accepts/returns Optional[Dict] for direct call sites that may
 # pass ``None``; at decorator sites we cast to this narrower contract.
@@ -562,24 +573,8 @@ def _merge_evidence(  # type: ignore[reportUnusedFunction]  # reserved: thin ali
     return merge_evidence_sets(existing, incoming)
 
 
-def delete_memories(uid: str, *, firestore_client: Any = None) -> None:
-    database = _get_db(firestore_client)
-    user_ref = database.collection(users_collection).document(uid)
-    memories_ref = user_ref.collection(memories_collection)
-    # Chunk deletes to stay under the Firestore 500-writes-per-batch limit. A user with more than
-    # 500 memories would otherwise make the single batch.commit() raise and delete nothing. Mirrors
-    # the chunking in unlock_all_memories.
-    batch = database.batch()
-    count = 0
-    for doc in memories_ref.stream():
-        batch.delete(doc.reference)
-        count += 1
-        if count >= 499:  # Firestore batch limit is 500
-            batch.commit()
-            batch = database.batch()
-            count = 0
-    if count > 0:
-        batch.commit()
+def delete_memories(uid: str, *, firestore_client: Any = None) -> LegacyMemoryDeleteResult:
+    return delete_all_memories(uid, firestore_client=firestore_client)
 
 
 @prepare_for_read(decrypt_func=cast(_DecryptFunc, _prepare_memory_for_read))
@@ -882,15 +877,23 @@ def write_projection(transaction: Any) -> None:
     )
 
 
-def delete_memory(uid: str, memory_id: str, *, firestore_client: Any = None) -> None:
+def delete_memory(uid: str, memory_id: str, *, firestore_client: Any = None) -> LegacyMemoryDeleteResult:
     database = _get_db(firestore_client)
     user_ref = database.collection(users_collection).document(uid)
     memories_ref = user_ref.collection(memories_collection)
     memory_ref = memories_ref.document(memory_id)
-    memory_ref.delete()
+    return _delete_memory_references(
+        [(memory_id, memory_ref)],
+        database=database,
+    )
 
 
-def delete_memories_batch(uid: str, memory_ids: List[str], *, firestore_client: Any = None) -> None:
+def delete_memories_batch(
+    uid: str,
+    memory_ids: List[str],
+    *,
+    firestore_client: Any = None,
+) -> LegacyMemoryDeleteResult:
     """Delete multiple memories in a single batched Firestore write.
 
     The router caps a batch-delete request at MEMORIES_BATCH_MAX (100), well under
@@ -898,41 +901,55 @@ def delete_memories_batch(uid: str, memory_ids: List[str], *, firestore_client:
     delete_all_memories so it stays correct if it is ever reused for larger sets.
     """
     if not memory_ids:
-        return
+        return LegacyMemoryDeleteResult(memory_ids=[])
     database = _get_db(firestore_client)
     user_ref = database.collection(users_collection).document(uid)
     memories_ref = user_ref.collection(memories_collection)
-    batch = database.batch()
-    count = 0
-    for memory_id in memory_ids:
-        batch.delete(memories_ref.document(memory_id))
-        count += 1
-        if count >= 499:  # Firestore batch limit is 500
-            batch.commit()
-            batch = database.batch()
-            count = 0
-    if count > 0:
-        batch.commit()
+    references = [(memory_id, memories_ref.document(memory_id)) for memory_id in dict.fromkeys(memory_ids)]
+    return _delete_memory_references(
+        references,
+        database=database,
+    )
 
 
-def delete_all_memories(uid: str, *, firestore_client: Any = None) -> None:
+def delete_all_memories(
+    uid: str,
+    *,
+    memory_ids: Optional[List[str]] = None,
+    firestore_client: Any = None,
+) -> LegacyMemoryDeleteResult:
+    """Delete one authoritative snapshot and return the exact committed IDs."""
     database = _get_db(firestore_client)
     user_ref = database.collection(users_collection).document(uid)
     memories_ref = user_ref.collection(memories_collection)
-    # Chunk deletes to stay under the Firestore 500-writes-per-batch limit. Account deletion and
-    # "delete all memories" hit this for any user with more than 500 memories: the single
-    # batch.commit() would raise and remove nothing. Mirrors the chunking in unlock_all_memories.
-    batch = database.batch()
-    count = 0
-    for doc in memories_ref.stream():
-        batch.delete(doc.reference)
-        count += 1
-        if count >= 499:  # Firestore batch limit is 500
-            batch.commit()
-            batch = database.batch()
-            count = 0
-    if count > 0:
+    references = (
+        [(memory_id, memories_ref.document(memory_id)) for memory_id in dict.fromkeys(memory_ids)]
+        if memory_ids is not None
+        else [(doc.id, doc.reference) for doc in memories_ref.stream()]
+    )
+    return _delete_memory_references(
+        references,
+        database=database,
+    )
+
+
+def _delete_memory_references(
+    references: List[tuple[str, Any]],
+    *,
+    database: Any,
+) -> LegacyMemoryDeleteResult:
+    if not references:
+        return LegacyMemoryDeleteResult(memory_ids=[])
+
+    committed_ids: List[str] = []
+    for offset in range(0, len(references), _DELETE_BATCH_SIZE):
+        chunk = references[offset : offset + _DELETE_BATCH_SIZE]
+        batch = database.batch()
+        for _memory_id, reference in chunk:
+            batch.delete(reference)
         batch.commit()
+        committed_ids.extend(memory_id for memory_id, _reference in chunk)
+    return LegacyMemoryDeleteResult(memory_ids=committed_ids)
 
 
 def ripple_source_deletion(uid: str, source_id: str, *, firestore_client: Any = None) -> Dict[str, Any]:
diff --git a/backend/database/projection_repair.py b/backend/database/projection_repair.py
index 9cbf6505e44..b8eeeabc4fc 100644
--- a/backend/database/projection_repair.py
+++ b/backend/database/projection_repair.py
@@ -60,6 +60,7 @@ def enqueue_projection_repairs(
     collection_ref: Any = database.collection(users_collection).document(uid).collection(projection_repairs_collection)
     repair_ids: List[str] = []
     reasons_by_fact = _reasons_by_fact(mutations)
+    pending_writes = 0
     for fact_id in fact_ids:
         reasons = reasons_by_fact.get(fact_id, ['unknown'])
         repair_id = f"{commit.get('commit_id')}:{fact_id}"
@@ -82,7 +83,13 @@ def enqueue_projection_repairs(
                 'updated_at': now,
             },
         )
-    batch.commit()
+        pending_writes += 1
+        if pending_writes >= 499:
+            batch.commit()
+            batch = database.batch()
+            pending_writes = 0
+    if pending_writes:
+        batch.commit()
     return repair_ids
 
 
diff --git a/backend/database/redis_db.py b/backend/database/redis_db.py
index 6ac6717ed1d..98c91c6d962 100644
--- a/backend/database/redis_db.py
+++ b/backend/database/redis_db.py
@@ -1,8 +1,11 @@
 import ast
 import base64
+import hashlib
 import json
 import os
-from typing import Any, Callable, Dict, List, Optional, TypeVar, Union, cast
+import secrets
+import threading
+from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar, Union, cast
 from datetime import datetime, timedelta, timezone
 
 import redis
@@ -28,6 +31,24 @@
     password=os.getenv('REDIS_DB_PASSWORD'),
     health_check_interval=30,
 )
+_pusher_delivery_r: Any = None
+_pusher_delivery_r_lock = threading.Lock()
+
+_PUSHER_DELIVERY_COMPLETE_LUA = """
+if redis.call('GET', KEYS[1]) ~= ARGV[1] then
+  return 0
+end
+redis.call('SET', KEYS[1], 'done', 'EX', ARGV[2])
+return 1
+"""
+
+_PUSHER_DELIVERY_ABANDON_LUA = """
+if redis.call('GET', KEYS[1]) ~= ARGV[1] then
+  return 0
+end
+redis.call('DEL', KEYS[1])
+return 1
+"""
 
 
 T = TypeVar("T")
@@ -841,7 +862,7 @@ def remove_conversation_summary_app_id(app_id: str) -> bool:
 # Lua script: atomic increment + TTL in a single round-trip.
 # Returns [current_count, ttl_remaining].  Sets TTL on first hit
 # and self-heals any key that lost its TTL (prevents permanent buckets).
-_RATE_LIMIT_LUA = r.register_script("""
+_RATE_LIMIT_LUA_SOURCE = """
 local key = KEYS[1]
 local window = tonumber(ARGV[1])
 local current = redis.call('INCR', key)
@@ -854,7 +875,8 @@ def remove_conversation_summary_app_id(app_id: str) -> bool:
     ttl = window
 end
 return {current, ttl}
-""")
+"""
+_RATE_LIMIT_LUA = r.register_script(_RATE_LIMIT_LUA_SOURCE)
 
 # Proactive LLM calls need a reversible reservation: provider/schema failures
 # must not consume a user's successful-completion allowance. Unlike the legacy
@@ -934,7 +956,7 @@ def release_rate_limit(key: str, policy: str) -> None:
 # Burst uses a sorted set keyed by timestamp-ms for sliding-window accuracy,
 # trimmed on every call (O(log n)). Daily char counter auto-expires at midnight
 # UTC (caller passes seconds_until_midnight_utc as the TTL).
-_TTS_RATE_LIMIT_LUA = r.register_script("""
+_TTS_RATE_LIMIT_LUA_SOURCE = """
 local burst_key = KEYS[1]
 local daily_key = KEYS[2]
 local now_ms = tonumber(ARGV[1])
@@ -962,7 +984,8 @@ def release_rate_limit(key: str, policy: str) -> None:
     redis.call('EXPIRE', daily_key, daily_ttl)
 end
 return {0, 0}
-""")
+"""
+_TTS_RATE_LIMIT_LUA = r.register_script(_TTS_RATE_LIMIT_LUA_SOURCE)
 
 
 def _seconds_until_midnight_utc() -> int:
@@ -1009,6 +1032,101 @@ def try_acquire_listen_lock(uid: str, ttl: int = 7) -> bool:
     return result is not None
 
 
+def _get_pusher_delivery_redis() -> Any:
+    """Return a lazy, tightly bounded Redis client for realtime ACK fencing."""
+    global _pusher_delivery_r
+    if _pusher_delivery_r is not None:
+        return _pusher_delivery_r
+    with _pusher_delivery_r_lock:
+        if _pusher_delivery_r is None:
+            _pusher_delivery_r = redis.Redis(
+                host=cast(str, _redis_host),
+                port=int(_redis_port_env) if _redis_port_env is not None else 6379,
+                username='default',
+                password=os.getenv('REDIS_DB_PASSWORD'),
+                health_check_interval=30,
+                socket_connect_timeout=0.5,
+                socket_timeout=0.5,
+                retry_on_timeout=False,
+            )
+    return _pusher_delivery_r
+
+
+def _pusher_delivery_key(uid: str, delivery_id: str) -> str:
+    digest = hashlib.sha256(delivery_id.encode('utf-8')).hexdigest()
+    return f'users:{uid}:pusher_delivery:{digest}'
+
+
+def begin_pusher_delivery(
+    uid: str,
+    delivery_id: str,
+    lease_ttl: int = 600,
+    *,
+    redis_client: Any = None,
+) -> Tuple[str, Optional[str]]:
+    """Acquire a short processing lease for a stable realtime delivery."""
+    client = redis_client or _get_pusher_delivery_redis()
+    key = _pusher_delivery_key(uid, delivery_id)
+    lease_token = secrets.token_urlsafe(18)
+    processing_value = f'processing:{lease_token}'
+    try:
+        if client.set(key, processing_value, ex=lease_ttl, nx=True) is not None:
+            return 'claimed', lease_token
+        state = client.get(key)
+        if state is not None and _decode_redis_value(state) == 'done':
+            return 'done', None
+        return 'busy', None
+    except Exception as exc:
+        logger.warning('pusher delivery lease unavailable uid=%s error=%s', uid, type(exc).__name__)
+        return 'unavailable', None
+
+
+def complete_pusher_delivery(
+    uid: str,
+    delivery_id: str,
+    lease_token: str,
+    retention_ttl: int = 604800,
+    *,
+    redis_client: Any = None,
+) -> bool:
+    """Replace a processing lease with a bounded done marker."""
+    client = redis_client or _get_pusher_delivery_redis()
+    try:
+        result = client.eval(
+            _PUSHER_DELIVERY_COMPLETE_LUA,
+            1,
+            _pusher_delivery_key(uid, delivery_id),
+            f'processing:{lease_token}',
+            retention_ttl,
+        )
+        return int(result) == 1
+    except Exception as exc:
+        logger.warning('pusher delivery completion unavailable uid=%s error=%s', uid, type(exc).__name__)
+        return False
+
+
+def abandon_pusher_delivery(
+    uid: str,
+    delivery_id: str,
+    lease_token: str,
+    *,
+    redis_client: Any = None,
+) -> bool:
+    """Release this worker's lease after an effect fails or is cancelled."""
+    client = redis_client or _get_pusher_delivery_redis()
+    try:
+        result = client.eval(
+            _PUSHER_DELIVERY_ABANDON_LUA,
+            1,
+            _pusher_delivery_key(uid, delivery_id),
+            f'processing:{lease_token}',
+        )
+        return int(result) == 1
+    except Exception as exc:
+        logger.warning('pusher delivery abandon unavailable uid=%s error=%s', uid, type(exc).__name__)
+        return False
+
+
 def try_acquire_client_device_write_lock(uid: str, client_device_id: str, ttl: int = 600) -> bool:
     """Throttle client_devices registry upserts to once per (uid, device) every `ttl` seconds."""
     try:
diff --git a/backend/database/webhook_health.py b/backend/database/webhook_health.py
index e865ad4392f..038317d5cdc 100644
--- a/backend/database/webhook_health.py
+++ b/backend/database/webhook_health.py
@@ -428,3 +428,24 @@ def record_dev_webhook_success(uid: str, wtype: object):
         r.expire(key, _HEALTH_TTL)
     except Exception as e:
         logger.warning(f'record_dev_webhook_success redis error uid={uid} type={wtype}: {e}')
+
+
+def reset_dev_webhook_health(uid: str, wtype: object):
+    """Clear delivery failures without claiming that an HTTP request succeeded."""
+    try:
+        wtype_str = getattr(wtype, 'value') if hasattr(wtype, 'value') else str(wtype)
+        key = f'dev_webhook_health:{uid}:{wtype_str}'
+        r.hset(
+            key,
+            mapping={
+                'failure_count': '0',
+                'last_failure_at': '',
+                'last_success_at': '',
+                'last_status': '',
+                'last_error': '',
+                'disabled': '0',
+            },
+        )
+        r.expire(key, _HEALTH_TTL)
+    except Exception as e:
+        logger.warning(f'reset_dev_webhook_health redis error uid={uid} type={wtype}: {e}')
diff --git a/backend/routers/task_integrations.py b/backend/routers/task_integrations.py
index 10d93e3a402..09a40a2f4c8 100644
--- a/backend/routers/task_integrations.py
+++ b/backend/routers/task_integrations.py
@@ -343,6 +343,9 @@ class CreateTaskResponse(BaseModel):
     success: bool
     external_task_id: Optional[str] = None
     error: Optional[str] = None
+    error_code: Optional[str] = None
+    retryable: Optional[bool] = None
+    ambiguous: Optional[bool] = None
 
 
 @router.post("/v1/task-integrations/{app_key}/tasks", response_model=CreateTaskResponse, tags=['task-integrations'])
@@ -391,6 +394,9 @@ async def create_task_via_integration(
         success=result.get("success", False),
         external_task_id=result.get("external_task_id"),
         error=result.get("error"),
+        error_code=result.get("error_code"),
+        retryable=result.get("retryable"),
+        ambiguous=result.get("ambiguous"),
     )
 
 
diff --git a/backend/routers/users.py b/backend/routers/users.py
index 6edf15a950f..6cb1836949a 100644
--- a/backend/routers/users.py
+++ b/backend/routers/users.py
@@ -26,7 +26,6 @@
 from services.users.data_export import iter_user_data_export
 from services.users.account_deletion import background_wipe_user_data, start_account_deletion
 from database.app_review_config import should_hide_subscription_ui
-from database.webhook_health import record_dev_webhook_success
 from database.conversations import get_in_progress_conversation, get_conversation
 from database.redis_db import (
     cache_user_geolocation,
@@ -123,7 +122,7 @@
     delete_user_person_speech_samples,
     delete_user_person_speech_sample,
 )
-from utils.webhooks import webhook_first_time_setup
+from utils.webhooks import reset_user_webhook_delivery_health, webhook_first_time_setup
 from utils.byok import has_byok_keys, invalidate_byok_state_cache, peppered_fingerprint
 import logging
 
@@ -487,8 +486,7 @@ def set_user_webhook_endpoint(
     if not webhook_url_from_setting(wtype, url):
         disable_user_webhook_db(uid, wtype)
     else:
-        enable_user_webhook_db(uid, wtype)
-        record_dev_webhook_success(uid, wtype)
+        enable_user_webhook_endpoint(wtype, uid)
     return {'status': 'ok'}
 
 
@@ -506,7 +504,7 @@ def disable_user_webhook_endpoint(wtype: WebhookType, uid: str = Depends(auth.ge
 @router.post('/v1/users/developer/webhook/{wtype}/enable', tags=['v1'], response_model=UserStatusResponse)
 def enable_user_webhook_endpoint(wtype: WebhookType, uid: str = Depends(auth.get_current_user_uid)):
     enable_user_webhook_db(uid, wtype)
-    record_dev_webhook_success(uid, wtype.value)
+    reset_user_webhook_delivery_health(uid, wtype, get_user_webhook_db(uid, wtype))
     return {'status': 'ok'}
 
 
diff --git a/backend/testing/e2e/test_task_integrations.py b/backend/testing/e2e/test_task_integrations.py
index 9476f9a8283..c14523c25a7 100644
--- a/backend/testing/e2e/test_task_integrations.py
+++ b/backend/testing/e2e/test_task_integrations.py
@@ -72,7 +72,14 @@ def handler(request):
         _close_async_client(fake_client)
 
     assert created.status_code == 200, created.text
-    assert created.json() == {"success": True, "external_task_id": "todo-123", "error": None}
+    assert created.json() == {
+        "success": True,
+        "external_task_id": "todo-123",
+        "error": None,
+        "error_code": None,
+        "retryable": None,
+        "ambiguous": None,
+    }
     assert len(requests) == 1
     request = requests[0]
     assert str(request.url) == "https://api.todoist.com/rest/v2/tasks"
@@ -145,6 +152,9 @@ def handler(request):
         "success": False,
         "external_task_id": None,
         "error": "Todoist API error: 500",
+        "error_code": "api_error",
+        "retryable": False,
+        "ambiguous": True,
     }
     assert len(requests) == 1
     assert _get_todoist_integration(client, auth_headers)["connected"] is True
@@ -196,5 +206,8 @@ def handler(request):
     assert response.json() == {
         "success": False,
         "external_task_id": None,
-        "error": "deterministic Todoist timeout",
+        "error": "ConnectTimeout",
+        "error_code": "transport_error",
+        "retryable": True,
+        "ambiguous": False,
     }
diff --git a/backend/testing/e2e/test_webhooks.py b/backend/testing/e2e/test_webhooks.py
index 1ca9f14b37b..6ed2de7bb81 100644
--- a/backend/testing/e2e/test_webhooks.py
+++ b/backend/testing/e2e/test_webhooks.py
@@ -80,6 +80,7 @@ async def handler(request):
 def test_realtime_webhook_does_not_call_provider_when_disabled(client, auth_headers, monkeypatch, fake_redis):
     _configure_realtime_webhook(client, auth_headers)
     _disable_realtime_webhook(client, auth_headers)
+    health_before = _health(fake_redis)
     requests = []
 
     async def handler(request):
@@ -89,8 +90,10 @@ async def handler(request):
     _run_realtime_delivery(monkeypatch, handler)
 
     assert requests == []
+    # The user's off toggle must not touch delivery health: skipped deliveries record nothing,
+    # and a user-disable is not the auto-disable flag.
+    assert _health(fake_redis) == health_before
     assert _health(fake_redis)["disabled"] == "0"
-    assert _health(fake_redis)["last_status"] == "200"
 
 
 @pytest.mark.parametrize(
diff --git a/backend/testing/workflow_contracts.json b/backend/testing/workflow_contracts.json
index 69897aeb94c..7af1c0ffcc1 100644
--- a/backend/testing/workflow_contracts.json
+++ b/backend/testing/workflow_contracts.json
@@ -130,7 +130,9 @@
         "tests/unit/test_ws_c_backfill.py",
         "tests/unit/test_backfill_legacy_memories_cli.py"
       ],
-      "checks": ["no_large_tuple_results"],
+      "checks": [
+        "no_large_tuple_results"
+      ],
       "invariants": [
         "completed checkpoints recognize required-processing and quarantined pending-admission destinations without derived side effects",
         "partial checkpoints resume idempotently",
@@ -144,12 +146,16 @@
     {
       "id": "canonical_memory_fanout",
       "risk": "high",
-      "sources": ["backend/utils/memory/canonical_memory_adapter.py"],
+      "sources": [
+        "backend/utils/memory/canonical_memory_adapter.py"
+      ],
       "tests": [
         "tests/unit/test_ws_j_delete_privacy.py",
         "tests/unit/test_canonical_kg_promotion.py"
       ],
-      "checks": ["no_large_tuple_results"],
+      "checks": [
+        "no_large_tuple_results"
+      ],
       "invariants": [
         "domain writes use injected stores",
         "derived keyword/vector/KG fanout is either durable or explicitly recoverable"
@@ -174,7 +180,9 @@
         "tests/unit/test_ws_m_atom_keyword_index.py",
         "tests/unit/test_ws_n_graph_traversal.py"
       ],
-      "checks": ["no_large_tuple_results"],
+      "checks": [
+        "no_large_tuple_results"
+      ],
       "invariants": [
         "explicit submissions remain pending short-term until a content-bound processing receipt exists",
         "pending text is visible only to the first-party memory list and never to agent, developer, MCP, search, vector, or KG reads",
@@ -250,7 +258,9 @@
         "backend/scripts/sync_ledger_fence_cutover.py",
         ".github/workflows/sync_ledger_fence_cutover.yml"
       ],
-      "tests": ["tests/unit/test_sync_ledger_fence_cutover.py"],
+      "tests": [
+        "tests/unit/test_sync_ledger_fence_cutover.py"
+      ],
       "checks": [],
       "invariants": [
         "standby admission traffic reaches every sync surface before Cloud Tasks queues are paused",
@@ -270,7 +280,9 @@
         "tests/unit/test_vector_repair_outbox_worker.py",
         "tests/unit/test_vector_repair_outbox_infra.py"
       ],
-      "checks": ["no_large_tuple_results"],
+      "checks": [
+        "no_large_tuple_results"
+      ],
       "invariants": [
         "leased work is recoverable after worker death",
         "completed and dead-lettered work is never reclaimed"
@@ -309,9 +321,15 @@
     {
       "id": "projection_repair",
       "risk": "high",
-      "sources": ["backend/database/projection_repair.py"],
-      "tests": ["tests/unit/test_memory_ledger.py"],
-      "checks": ["no_large_tuple_results"],
+      "sources": [
+        "backend/database/projection_repair.py"
+      ],
+      "tests": [
+        "tests/unit/test_memory_ledger.py"
+      ],
+      "checks": [
+        "no_large_tuple_results"
+      ],
       "invariants": [
         "repair enqueue is idempotent",
         "repair processing uses injected stores and has retry/dead-letter accounting"
@@ -339,7 +357,9 @@
         "tests/unit/test_canonical_short_term_maintenance_cron.py",
         "tests/unit/test_memory_outbox_worker.py"
       ],
-      "checks": ["no_large_tuple_results"],
+      "checks": [
+        "no_large_tuple_results"
+      ],
       "invariants": [
         "every eligible short-term item receives exactly one terminal L2 route",
         "long-term admission and its graph assertion commit atomically against one exact short-term revision",
@@ -350,20 +370,39 @@
     {
       "id": "webhook_delivery_health",
       "risk": "high",
-      "sources": ["backend/utils/webhooks.py", "backend/database/webhook_health.py"],
-      "tests": ["tests/unit/test_async_webhooks.py", "tests/unit/test_webhook_auto_disable.py"],
+      "sources": [
+        "backend/utils/webhooks.py",
+        "backend/utils/http_client.py",
+        "backend/database/webhook_health.py",
+        "backend/routers/users.py"
+      ],
+      "tests": [
+        "tests/unit/test_async_webhooks.py",
+        "tests/unit/test_async_http_infrastructure.py",
+        "tests/unit/test_webhook_auto_disable.py",
+        "tests/unit/test_users_webhook_url_validation.py"
+      ],
       "checks": [],
       "invariants": [
         "delivery failures are counted durably",
-        "auto-disable state is endpoint-aware and race-safe"
+        "auto-disable state is endpoint-aware and race-safe",
+        "manual enable resets persistent and process-local failure gates without recording a synthetic delivery success"
       ]
     },
     {
       "id": "memory_ingestion_export_runner",
       "risk": "high",
-      "sources": ["backend/utils/memory_ingestion/export_runner.py", "backend/utils/memory_ingestion/pipeline.py"],
-      "tests": ["tests/unit/test_memory_ingestion_pipeline.py", "tests/unit/test_production_like_memory_model.py"],
-      "checks": ["no_large_tuple_results"],
+      "sources": [
+        "backend/utils/memory_ingestion/export_runner.py",
+        "backend/utils/memory_ingestion/pipeline.py"
+      ],
+      "tests": [
+        "tests/unit/test_memory_ingestion_pipeline.py",
+        "tests/unit/test_production_like_memory_model.py"
+      ],
+      "checks": [
+        "no_large_tuple_results"
+      ],
       "invariants": [
         "resume config mismatches fail closed",
         "failed shards remain visible in run summaries"
@@ -399,10 +438,14 @@
         "backend/utils/memory/**",
         "backend/utils/memory_ingestion/**"
       ],
-      "tests": ["testing/e2e/test_canonical_memory_pipeline.py"],
-      "checks": ["no_large_tuple_results"],
+      "tests": [
+        "testing/e2e/test_canonical_memory_pipeline.py"
+      ],
+      "checks": [
+        "no_large_tuple_results"
+      ],
       "invariants": [
-        "capture→consolidate→promote→read with archive excluded from default reads",
+        "capture\u2192consolidate\u2192promote\u2192read with archive excluded from default reads",
         "canonical fail-closed without legacy bleed"
       ]
     },
@@ -438,7 +481,9 @@
         "tests/unit/test_recording_sessions.py",
         "tests/unit/test_listen_finalization_cloud_tasks.py"
       ],
-      "checks": ["conversation-lifecycle-write-guard"],
+      "checks": [
+        "conversation-lifecycle-write-guard"
+      ],
       "invariants": [
         "only one finalizer claim succeeds for an in-progress conversation",
         "finalization outbox reconnects reuse one durable job and lease fences stale workers before any fanout",
@@ -455,8 +500,12 @@
         "backend/routers/transcribe.py",
         "backend/routers/listen/**",
         "backend/routers/pusher.py",
+        "backend/database/redis_db.py",
         "backend/utils/listen_pusher_session.py",
+        "backend/utils/speaker_identification.py",
         "backend/utils/pusher.py",
+        "backend/utils/app_integrations.py",
+        "backend/utils/webhooks.py",
         "backend/utils/stt/live_failure.py",
         "backend/utils/stt/streaming.py",
         "backend/config/stt_provider_policy.py",
@@ -473,6 +522,12 @@
       "tests": [
         "tests/unit/test_listen_pipeline.py",
         "tests/unit/test_pusher_heartbeat.py",
+        "tests/unit/test_pusher_readiness_drain.py",
+        "tests/unit/test_redis_db_cache_serialization.py",
+        "tests/unit/test_speaker_identification_delivery.py",
+        "tests/unit/test_listen_finalization_cloud_tasks.py",
+        "tests/unit/test_async_app_integrations.py",
+        "tests/unit/test_async_webhooks.py",
         "tests/unit/test_pusher_conversation_retry.py",
         "tests/unit/test_live_stt_failure.py",
         "tests/unit/test_listen_runtime_regressions.py",
@@ -490,6 +545,13 @@
         "teardown flushes tail audio before pusher close",
         "pending conversation requests retry until ack or give-up limit",
         "pusher heartbeat prevents idle connection drops",
+        "draining pusher pods reject new sockets cleanly while preserving established-session shutdown",
+        "transcript and speaker deliveries retain stable identity until a local-invocation completion acknowledgement on negotiated sockets",
+        "route-stamped audio survives explicit send failure and cancellation while ambiguous local completion remains documented as best effort",
+        "pusher leases stable deliveries only when a worker owns the effect and never evicts an accepted unacknowledged frame",
+        "done markers suppress stable delivery replays across sockets and replicas while Redis failure remains observable and fail-open",
+        "graceful close performs a bounded acknowledgement drain after the listen runtime becomes inactive",
+        "downstream realtime webhooks receive stable idempotency keys while opcode 202 acknowledges local worker invocation rather than downstream HTTP delivery",
         "a client Parakeet preference cannot route a live session to an incompatible model",
         "listen-to-pusher boundary changes run the credential-free local Firestore, Redis, and Parakeet-stub stack gauntlet in PR CI"
       ]
@@ -617,7 +679,9 @@
         "backend/charts/vad/**",
         "backend/scripts/validate_rendered_deployment_contract.py"
       ],
-      "tests": ["tests/unit/test_rendered_deployment_contract.py"],
+      "tests": [
+        "tests/unit/test_rendered_deployment_contract.py"
+      ],
       "checks": [],
       "invariants": [
         "every first-party GKE workload renders an explicit immutable image identity",
diff --git a/backend/tests/unit/test_async_app_integrations.py b/backend/tests/unit/test_async_app_integrations.py
index b7df57d283d..1332c167520 100644
--- a/backend/tests/unit/test_async_app_integrations.py
+++ b/backend/tests/unit/test_async_app_integrations.py
@@ -11,6 +11,8 @@
 
 import pytest
 
+from testing.import_isolation import load_module_fresh
+
 os.environ.setdefault(
     "ENCRYPTION_SECRET",
     "omi_ZwB2ZNqB2HHpMK6wStk7sTpavJiPTFg7gXUHnc4tFABPU6pZ2c2DKgehtfgi4RZv",
@@ -269,9 +271,10 @@ async def _run_blocking(_executor, func, *args, **kwargs):
 
 _executors_mod.run_blocking = _run_blocking
 
-import importlib
-
-app_integrations = importlib.import_module("utils.app_integrations")
+app_integrations = load_module_fresh(
+    "utils.app_integrations",
+    os.path.join(_BACKEND_DIR, "utils", "app_integrations.py"),
+)
 _restore_stub_modules()
 
 
@@ -564,6 +567,72 @@ async def test_12_apps_sent_in_two_chunks(self):
 class TestAsyncTriggerRealtimeIntegrations:
     """Test async realtime integration fan-out."""
 
+    @pytest.mark.asyncio
+    async def test_retryable_status_reuses_one_receiver_visible_delivery_key(self):
+        unavailable = MagicMock(status_code=503)
+        accepted = MagicMock(status_code=204)
+        client = AsyncMock()
+        client.post = AsyncMock(side_effect=[unavailable, accepted])
+
+        with patch.object(app_integrations, "get_webhook_client", return_value=client):
+            result = await app_integrations._post_realtime_app_webhook(
+                "app-1",
+                "https://app.test/hook",
+                idempotency_key="delivery-1",
+                retry_delays=(0,),
+                json={"segments": [{"text": "hi"}]},
+            )
+
+        assert result is accepted
+        assert client.post.await_count == 2
+        assert [call.kwargs["headers"]["X-Omi-Idempotency-Key"] for call in client.post.await_args_list] == [
+            "delivery-1",
+            "delivery-1",
+        ]
+
+    @pytest.mark.asyncio
+    async def test_transport_retry_generates_one_stable_key_for_legacy_caller(self):
+        accepted = MagicMock(status_code=200)
+        client = AsyncMock()
+        client.post = AsyncMock(
+            side_effect=[
+                app_integrations.httpx.ConnectError("connection unavailable"),
+                accepted,
+            ]
+        )
+
+        with patch.object(app_integrations, "get_webhook_client", return_value=client):
+            result = await app_integrations._post_realtime_app_webhook(
+                "app-1",
+                "https://app.test/hook",
+                retry_delays=(0,),
+                json={"segments": [{"text": "hi"}]},
+            )
+
+        assert result is accepted
+        keys = [call.kwargs["headers"]["X-Omi-Idempotency-Key"] for call in client.post.await_args_list]
+        assert len(keys) == 2
+        assert keys[0] == keys[1]
+        assert keys[0]
+
+    @pytest.mark.asyncio
+    async def test_permanent_client_error_is_not_retried(self):
+        rejected = MagicMock(status_code=400)
+        client = AsyncMock()
+        client.post = AsyncMock(return_value=rejected)
+
+        with patch.object(app_integrations, "get_webhook_client", return_value=client):
+            result = await app_integrations._post_realtime_app_webhook(
+                "app-1",
+                "https://app.test/hook",
+                idempotency_key="delivery-1",
+                retry_delays=(0, 0),
+                json={"segments": [{"text": "hi"}]},
+            )
+
+        assert result is rejected
+        client.post.assert_awaited_once()
+
     @pytest.mark.asyncio
     async def test_no_apps_returns_empty(self):
         """No apps and no mentor → empty result."""
@@ -594,6 +663,26 @@ async def test_multiple_apps_called_concurrently(self):
 
         assert mock_client.post.call_count == 2
 
+    @pytest.mark.asyncio
+    async def test_stable_delivery_id_reaches_realtime_app_webhook(self):
+        app = _make_app("a1", "https://app1.test/hook", triggers_realtime=True)
+        response = MagicMock(status_code=200, text="")
+        response.json.return_value = {}
+        client = AsyncMock()
+        client.post = AsyncMock(return_value=response)
+
+        with patch.object(app_integrations, "get_available_apps", return_value=[app]), patch.object(
+            app_integrations, "process_mentor_notification", return_value=None
+        ), patch.object(app_integrations, "get_webhook_client", return_value=client):
+            await app_integrations.trigger_realtime_integrations(
+                "uid-1",
+                [{"text": "hi"}],
+                "conv-1",
+                idempotency_key="delivery-1",
+            )
+
+        assert client.post.await_args.kwargs["headers"] == {"X-Omi-Idempotency-Key": "delivery-1"}
+
     @pytest.mark.asyncio
     async def test_app_response_message_triggers_notification(self):
         """App returning a message > 5 chars triggers notification."""
diff --git a/backend/tests/unit/test_async_http_infrastructure.py b/backend/tests/unit/test_async_http_infrastructure.py
index 946e092a1f2..c283dbf2d62 100644
--- a/backend/tests/unit/test_async_http_infrastructure.py
+++ b/backend/tests/unit/test_async_http_infrastructure.py
@@ -56,6 +56,7 @@ def _drop_stale_module(name, required_attrs):
 _drop_stale_module("utils.http_client", ["WebhookCircuitBreaker", "get_webhook_circuit_breaker"])
 _drop_stale_module("utils.executors", ["critical_executor", "storage_executor", "shutdown_executors"])
 
+import utils.http_client as http_client_module
 from utils.http_client import (
     WebhookCircuitBreaker,
     get_webhook_circuit_breaker,
@@ -71,6 +72,7 @@ def _drop_stale_module(name, required_attrs):
     _SEMAPHORE_CACHE_MAX,
     _CIRCUIT_BREAKER_FAILURE_THRESHOLD,
     _CIRCUIT_BREAKER_RECOVERY_TIMEOUT,
+    reset_webhook_circuit_breaker,
 )
 from utils.executors import critical_executor, storage_executor
 
@@ -219,6 +221,18 @@ def test_invalid_url_fallback(self):
         assert cb is not None
         assert cb.state == 'closed'
 
+    def test_same_path_url_replacement_is_allowed_immediately(self):
+        old_cb = get_webhook_circuit_breaker("https://example.com/hook?version=old")
+        for _ in range(_CIRCUIT_BREAKER_FAILURE_THRESHOLD):
+            old_cb.record_failure()
+        assert old_cb.allow_request() is False
+
+        reset_webhook_circuit_breaker("https://example.com/hook?version=new")
+
+        replacement_cb = get_webhook_circuit_breaker("https://example.com/hook?version=new")
+        assert replacement_cb is not old_cb
+        assert replacement_cb.allow_request() is True
+
 
 # ============================================================================
 # Latest-wins dropping
@@ -638,15 +652,12 @@ def test_stale_breaker_evicted(self):
         assert 'https://stale.test/hook' not in _webhook_circuit_breakers
         _webhook_circuit_breakers.clear()
 
-    def test_allow_request_updates_access_time(self):
+    def test_allow_request_updates_access_time(self, monkeypatch):
         """allow_request() must update _last_access_time."""
-        import time
-        from utils.http_client import _webhook_circuit_breakers, get_webhook_circuit_breaker
-
         _webhook_circuit_breakers.clear()
         cb = get_webhook_circuit_breaker('https://test.test/hook')
-        old_access = cb._last_access_time
-        time.sleep(0.01)
+        cb._last_access_time = 100.0
+        monkeypatch.setattr(http_client_module.time, 'monotonic', lambda: 101.0)
         cb.allow_request()
-        assert cb._last_access_time > old_access
+        assert cb._last_access_time == 101.0
         _webhook_circuit_breakers.clear()
diff --git a/backend/tests/unit/test_async_webhooks.py b/backend/tests/unit/test_async_webhooks.py
index 39b03e58046..c0ba387b6b8 100644
--- a/backend/tests/unit/test_async_webhooks.py
+++ b/backend/tests/unit/test_async_webhooks.py
@@ -9,6 +9,7 @@
 import re
 from unittest.mock import MagicMock, AsyncMock, patch
 
+import httpx
 import pytest
 
 import utils.webhooks as webhooks_module
@@ -32,6 +33,82 @@ def _stub_webhook_db_helpers(monkeypatch):
     monkeypatch.setattr(webhooks_module, "record_dev_webhook_failure", MagicMock(return_value=False))
 
 
+class TestPostDevWebhookRetryPolicy:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize('status_code', [400, 401, 403, 404, 409, 410, 422])
+    async def test_permanent_4xx_is_not_retried(self, status_code):
+        response = MagicMock(status_code=status_code)
+        mock_client = AsyncMock()
+        mock_client.post = AsyncMock(return_value=response)
+        mock_sleep = AsyncMock()
+
+        with patch.object(webhooks_module, 'get_webhook_client', return_value=mock_client), patch.object(
+            webhooks_module.asyncio, 'sleep', new=mock_sleep
+        ):
+            actual = await webhooks_module._post_dev_webhook(
+                'test_webhook',
+                'https://example.com/webhook',
+                retry_delays=(1.0,),
+                json={'event': 'conversation.completed'},
+            )
+
+        assert actual is response
+        mock_client.post.assert_awaited_once()
+        mock_sleep.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize('status_code', [408, 425, 429, 500, 503])
+    async def test_retryable_http_status_retries_with_stable_idempotency_key(self, status_code):
+        retryable_response = MagicMock(status_code=status_code)
+        success_response = MagicMock(status_code=204)
+        mock_client = AsyncMock()
+        mock_client.post = AsyncMock(side_effect=[retryable_response, success_response])
+        mock_sleep = AsyncMock()
+
+        with patch.object(webhooks_module, 'get_webhook_client', return_value=mock_client), patch.object(
+            webhooks_module.asyncio, 'sleep', new=mock_sleep
+        ):
+            actual = await webhooks_module._post_dev_webhook(
+                'test_webhook',
+                'https://example.com/webhook',
+                retry_delays=(1.0,),
+                json={'event': 'conversation.completed'},
+            )
+
+        assert actual is success_response
+        assert mock_client.post.await_count == 2
+        mock_sleep.assert_awaited_once_with(1.0)
+        idempotency_keys = [call.kwargs['headers']['Idempotency-Key'] for call in mock_client.post.await_args_list]
+        assert idempotency_keys[0]
+        assert idempotency_keys[0] == idempotency_keys[1]
+
+    @pytest.mark.asyncio
+    async def test_network_failure_retries_with_stable_idempotency_key(self):
+        request = httpx.Request('POST', 'https://example.com/webhook')
+        network_error = httpx.ConnectError('connection failed', request=request)
+        success_response = MagicMock(status_code=200)
+        mock_client = AsyncMock()
+        mock_client.post = AsyncMock(side_effect=[network_error, success_response])
+        mock_sleep = AsyncMock()
+
+        with patch.object(webhooks_module, 'get_webhook_client', return_value=mock_client), patch.object(
+            webhooks_module.asyncio, 'sleep', new=mock_sleep
+        ):
+            actual = await webhooks_module._post_dev_webhook(
+                'test_webhook',
+                'https://example.com/webhook',
+                retry_delays=(1.0,),
+                json={'event': 'conversation.completed'},
+            )
+
+        assert actual is success_response
+        assert mock_client.post.await_count == 2
+        mock_sleep.assert_awaited_once_with(1.0)
+        idempotency_keys = [call.kwargs['headers']['Idempotency-Key'] for call in mock_client.post.await_args_list]
+        assert idempotency_keys[0]
+        assert idempotency_keys[0] == idempotency_keys[1]
+
+
 class TestRealtimeTranscriptWebhook:
     """Test realtime_transcript_webhook uses httpx async."""
 
@@ -52,6 +129,21 @@ async def test_success_sends_via_httpx(self):
         call_args = mock_client.post.call_args
         assert "segments" in call_args.kwargs.get("json", {})
 
+    @pytest.mark.asyncio
+    async def test_stable_delivery_id_reaches_realtime_webhook(self):
+        mock_response = MagicMock(status_code=204)
+        mock_client = AsyncMock()
+        mock_client.post = AsyncMock(return_value=mock_response)
+
+        with patch.object(webhooks_module, "get_webhook_client", return_value=mock_client):
+            await realtime_transcript_webhook(
+                "uid-1",
+                [{"text": "hello"}],
+                idempotency_key="delivery-1",
+            )
+
+        assert mock_client.post.await_args.kwargs["headers"]["Idempotency-Key"] == "delivery-1"
+
     @pytest.mark.asyncio
     async def test_notification_on_200_with_message(self):
         """Verify webhook notification sent when response has message > 5 chars."""
@@ -104,7 +196,7 @@ async def test_timeout_error_handled(self):
         mock_client.post = AsyncMock(side_effect=httpx.TimeoutException("connect timeout"))
 
         with patch("utils.webhooks.get_webhook_client", return_value=mock_client), patch(
-            "utils.webhooks._get_dev_webhook_retry_delays", return_value=()
+            "utils.webhooks._REALTIME_DEV_WEBHOOK_RETRY_DELAYS", ()
         ):
             # Should not raise
             await realtime_transcript_webhook("uid-1", [{"text": "hello"}])
@@ -397,7 +489,7 @@ async def test_transcript_webhook_records_failure_on_exception(self):
 
         with patch("utils.webhooks.get_webhook_circuit_breaker", return_value=mock_cb), patch(
             "utils.webhooks.get_webhook_client", return_value=mock_client
-        ), patch("utils.webhooks._get_dev_webhook_retry_delays", return_value=()):
+        ), patch("utils.webhooks._REALTIME_DEV_WEBHOOK_RETRY_DELAYS", ()):
             await realtime_transcript_webhook("uid-1", [{"text": "hello"}])
             mock_cb.record_failure.assert_called_once()
 
diff --git a/backend/tests/unit/test_language_catalog.py b/backend/tests/unit/test_language_catalog.py
index cd08513a23b..1cf13ff7685 100644
--- a/backend/tests/unit/test_language_catalog.py
+++ b/backend/tests/unit/test_language_catalog.py
@@ -24,7 +24,9 @@ def test_offered_codes_round_trip_unchanged():
     # A code that normalizes to something else would leave the picker showing one
     # language while the account stores another.
     drifted = [
-        (code, normalize_user_language(code)) for code, _ in PRIMARY_LANGUAGE_OPTIONS if normalize_user_language(code) != code
+        (code, normalize_user_language(code))
+        for code, _ in PRIMARY_LANGUAGE_OPTIONS
+        if normalize_user_language(code) != code
     ]
     assert drifted == [], f"code changes on save: {drifted}"
 
diff --git a/backend/tests/unit/test_redis_db_cache_serialization.py b/backend/tests/unit/test_redis_db_cache_serialization.py
index 10374d62722..07a35c6dbe6 100644
--- a/backend/tests/unit/test_redis_db_cache_serialization.py
+++ b/backend/tests/unit/test_redis_db_cache_serialization.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from typing import Any, Dict, List, Optional
+from unittest.mock import MagicMock
 
 import pytest
 
@@ -13,8 +14,11 @@ class _FakeRedis:
     def __init__(self) -> None:
         self._store: Dict[str, Any] = {}
 
-    def set(self, key: str, value: Any, ex: Optional[int] = None) -> None:
+    def set(self, key: str, value: Any, ex: Optional[int] = None, nx: bool = False) -> Optional[bool]:
+        if nx and key in self._store:
+            return None
         self._store[key] = value
+        return True
 
     def get(self, key: str) -> Optional[Any]:
         return self._store.get(key)
@@ -25,6 +29,15 @@ def expire(self, key: str, ttl: int) -> None:
     def mget(self, keys: List[str]) -> List[Optional[Any]]:
         return [self._store.get(key) for key in keys]
 
+    def eval(self, script: str, _numkeys: int, key: str, expected: str, *args: Any) -> int:
+        if self._store.get(key) != expected:
+            return 0
+        if "redis.call('DEL'" in script:
+            del self._store[key]
+        else:
+            self._store[key] = 'done'
+        return 1
+
 
 @pytest.fixture
 def fake_redis(monkeypatch: pytest.MonkeyPatch) -> _FakeRedis:
@@ -104,3 +117,73 @@ def test_apps_reviews_batch_round_trip(fake_redis: _FakeRedis) -> None:
         "app-b": {"uid-2": {"rating": 5}},
         "app-missing": {},
     }
+
+
+def test_pusher_delivery_lease_reaches_done_with_a_bounded_key(fake_redis: _FakeRedis) -> None:
+    delivery_id = f"delivery-{'x' * 512}"
+
+    state, lease_token = redis_db.begin_pusher_delivery("uid-1", delivery_id, redis_client=fake_redis)
+    assert state == 'claimed'
+    assert lease_token
+    assert redis_db.begin_pusher_delivery("uid-1", delivery_id, redis_client=fake_redis) == ('busy', None)
+    assert (
+        redis_db.complete_pusher_delivery(
+            "uid-1",
+            delivery_id,
+            lease_token,
+            redis_client=fake_redis,
+        )
+        is True
+    )
+    assert redis_db.begin_pusher_delivery("uid-1", delivery_id, redis_client=fake_redis) == ('done', None)
+    assert redis_db.begin_pusher_delivery("uid-2", delivery_id, redis_client=fake_redis)[0] == 'claimed'
+    assert all(delivery_id not in key for key in fake_redis._store)
+    assert all(len(key) < 128 for key in fake_redis._store)
+
+
+def test_pusher_failed_effect_releases_only_its_own_lease(fake_redis: _FakeRedis) -> None:
+    state, lease_token = redis_db.begin_pusher_delivery("uid-1", "delivery-1", redis_client=fake_redis)
+    assert state == 'claimed'
+    assert lease_token
+    assert (
+        redis_db.abandon_pusher_delivery(
+            "uid-1",
+            "delivery-1",
+            "wrong-token",
+            redis_client=fake_redis,
+        )
+        is False
+    )
+    assert (
+        redis_db.abandon_pusher_delivery(
+            "uid-1",
+            "delivery-1",
+            lease_token,
+            redis_client=fake_redis,
+        )
+        is True
+    )
+    assert redis_db.begin_pusher_delivery("uid-1", "delivery-1", redis_client=fake_redis)[0] == 'claimed'
+
+
+def test_pusher_delivery_lease_fails_open_when_redis_is_unavailable() -> None:
+    class _UnavailableRedis:
+        def set(self, *args: Any, **kwargs: Any) -> None:
+            raise ConnectionError("redis unavailable")
+
+    client = _UnavailableRedis()
+
+    assert redis_db.begin_pusher_delivery("uid-1", "delivery-1", redis_client=client) == ('unavailable', None)
+    assert redis_db.complete_pusher_delivery("uid-1", "delivery-1", "lease-1", redis_client=client) is False
+    assert redis_db.abandon_pusher_delivery("uid-1", "delivery-1", "lease-1", redis_client=client) is False
+
+
+def test_pusher_delivery_client_has_bounded_network_timeouts(monkeypatch: pytest.MonkeyPatch) -> None:
+    constructor = MagicMock(return_value=object())
+    monkeypatch.setattr(redis_db.redis, 'Redis', constructor)
+    monkeypatch.setattr(redis_db, '_pusher_delivery_r', None)
+
+    assert redis_db._get_pusher_delivery_redis() is constructor.return_value
+    assert constructor.call_args.kwargs['socket_connect_timeout'] == 0.5
+    assert constructor.call_args.kwargs['socket_timeout'] == 0.5
+    assert constructor.call_args.kwargs['retry_on_timeout'] is False
diff --git a/backend/tests/unit/test_speaker_identification_delivery.py b/backend/tests/unit/test_speaker_identification_delivery.py
new file mode 100644
index 00000000000..242ea89fdb9
--- /dev/null
+++ b/backend/tests/unit/test_speaker_identification_delivery.py
@@ -0,0 +1,238 @@
+import pytest
+from unittest.mock import AsyncMock, MagicMock
+
+import utils.speaker_identification as speaker_identification
+from utils.other import storage as storage_utils
+
+
+async def _inline_run_blocking(_executor, func, *args, **kwargs):
+    return func(*args, **kwargs)
+
+
+@pytest.fixture
+def anyio_backend():
+    return 'asyncio'
+
+
+@pytest.mark.anyio
+async def test_missing_audio_metadata_is_retryable_instead_of_false_success(monkeypatch):
+    monkeypatch.setattr(speaker_identification, 'run_blocking', _inline_run_blocking)
+    monkeypatch.setattr(speaker_identification.users_db, 'get_person', lambda _uid, _person_id: {})
+    monkeypatch.setattr(
+        speaker_identification.users_db,
+        'get_person_speech_samples_count',
+        lambda _uid, _person_id: 0,
+    )
+    monkeypatch.setattr(
+        speaker_identification.conversations_db,
+        'get_conversation',
+        lambda _uid, _conversation_id: {
+            'started_at': 1.0,
+            'transcript_segments': [{'id': 'segment-1', 'start': 0.0, 'end': 10.0}],
+            'audio_files': [],
+        },
+    )
+
+    result = await speaker_identification.extract_speaker_samples(
+        uid='uid-1',
+        person_id='person-1',
+        conversation_id='conversation-1',
+        segment_ids=['segment-1'],
+    )
+
+    assert result.status == 'retryable'
+    assert result.reason == 'audio_files_not_ready'
+
+
+@pytest.mark.anyio
+async def test_missing_requested_segment_is_retryable_instead_of_false_success(monkeypatch):
+    monkeypatch.setattr(speaker_identification, 'run_blocking', _inline_run_blocking)
+    monkeypatch.setattr(speaker_identification.users_db, 'get_person', lambda _uid, _person_id: {})
+    monkeypatch.setattr(
+        speaker_identification.users_db,
+        'get_person_speech_samples_count',
+        lambda _uid, _person_id: 0,
+    )
+    monkeypatch.setattr(
+        speaker_identification.conversations_db,
+        'get_conversation',
+        lambda _uid, _conversation_id: {
+            'started_at': 1.0,
+            'transcript_segments': [],
+            'audio_files': [{'chunk_timestamps': [1.0]}],
+        },
+    )
+
+    result = await speaker_identification.extract_speaker_samples(
+        uid='uid-1',
+        person_id='person-1',
+        conversation_id='conversation-1',
+        segment_ids=['segment-not-persisted-yet'],
+    )
+
+    assert result.status == 'retryable'
+    assert result.reason == 'transcript_segments_not_ready'
+
+
+@pytest.mark.anyio
+async def test_unhandled_extraction_error_is_retryable_instead_of_false_success(monkeypatch):
+    monkeypatch.setattr(speaker_identification, 'run_blocking', _inline_run_blocking)
+
+    def fail(_uid, _person_id):
+        raise RuntimeError('database unavailable')
+
+    monkeypatch.setattr(speaker_identification.users_db, 'get_person', fail)
+
+    result = await speaker_identification.extract_speaker_samples(
+        uid='uid-1',
+        person_id='person-1',
+        conversation_id='conversation-1',
+        segment_ids=['segment-1'],
+    )
+
+    assert result.status == 'retryable'
+    assert result.reason == 'extraction_failed'
+
+
+@pytest.mark.anyio
+async def test_missing_person_is_retryable_before_upload_side_effects(monkeypatch):
+    monkeypatch.setattr(speaker_identification, 'run_blocking', _inline_run_blocking)
+    monkeypatch.setattr(speaker_identification.users_db, 'get_person', lambda _uid, _person_id: None)
+    get_conversation = MagicMock()
+    upload_sample = MagicMock()
+    monkeypatch.setattr(speaker_identification.conversations_db, 'get_conversation', get_conversation)
+    monkeypatch.setattr(speaker_identification, 'upload_person_speech_sample_from_bytes', upload_sample)
+
+    result = await speaker_identification.extract_speaker_samples(
+        uid='uid-1',
+        person_id='person-1',
+        conversation_id='conversation-1',
+        segment_ids=['segment-1'],
+    )
+
+    assert result.status == 'retryable'
+    assert result.reason == 'person_not_ready'
+    get_conversation.assert_not_called()
+    upload_sample.assert_not_called()
+
+
+@pytest.mark.anyio
+async def test_sample_append_enforces_the_single_sample_limit_transactionally(monkeypatch):
+    monkeypatch.setattr(speaker_identification, 'run_blocking', _inline_run_blocking)
+    monkeypatch.setattr(speaker_identification.users_db, 'get_person', lambda _uid, _person_id: {})
+    monkeypatch.setattr(
+        speaker_identification.users_db,
+        'get_person_speech_samples_count',
+        lambda _uid, _person_id: 0,
+    )
+    monkeypatch.setattr(
+        speaker_identification.conversations_db,
+        'get_conversation',
+        lambda _uid, _conversation_id: {
+            'started_at': 1_000.0,
+            'transcript_segments': [
+                {
+                    'id': 'segment-1',
+                    'start': 0.0,
+                    'end': 10.0,
+                    'text': 'hello there',
+                    'speaker_id': 1,
+                }
+            ],
+            'audio_files': [{'chunk_timestamps': [1_000.0]}],
+        },
+    )
+    monkeypatch.setattr(
+        speaker_identification,
+        'download_audio_chunks_and_merge',
+        lambda *_args, **_kwargs: b'audio',
+    )
+    monkeypatch.setattr(
+        speaker_identification,
+        '_trim_pcm_audio',
+        lambda *_args, **_kwargs: b'\x00\x00' * (16_000 * 10),
+    )
+    monkeypatch.setattr(
+        speaker_identification,
+        'verify_and_transcribe_sample',
+        AsyncMock(return_value=('hello there', True, '')),
+    )
+    upload_sample = MagicMock(return_value='users/uid-1/people/person-1/sample.pcm')
+    monkeypatch.setattr(speaker_identification, 'upload_person_speech_sample_from_bytes', upload_sample)
+    add_sample = MagicMock(return_value=True)
+    monkeypatch.setattr(speaker_identification.users_db, 'add_person_speech_sample', add_sample)
+    monkeypatch.setattr(
+        speaker_identification,
+        'extract_embedding_from_bytes',
+        MagicMock(side_effect=RuntimeError('embedding unavailable')),
+    )
+
+    result = await speaker_identification.extract_speaker_samples(
+        uid='uid-1',
+        person_id='person-1',
+        conversation_id='conversation-1',
+        segment_ids=['segment-1'],
+        sample_rate=16_000,
+        delivery_id='delivery-1',
+    )
+
+    assert result.status == 'stored'
+    upload_sample.assert_called_once_with(
+        b'\x00\x00' * (16_000 * 10),
+        'uid-1',
+        'person-1',
+        16_000,
+        'speaker-sample\0uid-1\0person-1\0delivery-1',
+    )
+    add_sample.assert_called_once_with(
+        'uid-1',
+        'person-1',
+        'users/uid-1/people/person-1/sample.pcm',
+        transcript='hello there',
+        max_samples=1,
+    )
+
+
+def test_speech_sample_upload_reuses_one_hashed_object_for_a_stable_delivery(monkeypatch):
+    uploaded_paths = []
+
+    class FakeBlob:
+        def __init__(self, path):
+            self.path = path
+
+        def upload_from_string(self, _payload, *, content_type):
+            assert content_type == 'audio/wav'
+            uploaded_paths.append(self.path)
+
+    class FakeBucket:
+        def blob(self, path):
+            return FakeBlob(path)
+
+    monkeypatch.setattr(storage_utils, '_get_speech_profiles_bucket', lambda **_kwargs: FakeBucket())
+
+    stable_key = 'speaker-sample\0uid-1\0person-1\0delivery-1'
+    first = storage_utils.upload_person_speech_sample_from_bytes(
+        b'\x00\x00',
+        'uid-1',
+        'person-1',
+        deduplication_key=stable_key,
+    )
+    replay = storage_utils.upload_person_speech_sample_from_bytes(
+        b'\x00\x00',
+        'uid-1',
+        'person-1',
+        deduplication_key=stable_key,
+    )
+    another = storage_utils.upload_person_speech_sample_from_bytes(
+        b'\x00\x00',
+        'uid-1',
+        'person-1',
+        deduplication_key='speaker-sample\0uid-1\0person-1\0delivery-2',
+    )
+    legacy_first = storage_utils.upload_person_speech_sample_from_bytes(b'\x00\x00', 'uid-1', 'person-1')
+    legacy_second = storage_utils.upload_person_speech_sample_from_bytes(b'\x00\x00', 'uid-1', 'person-1')
+
+    assert first == replay, 'one logical speaker delivery must overwrite one stable GCS object'
+    assert another != first
+    assert legacy_first != legacy_second
+    assert uploaded_paths == [first, replay, another, legacy_first, legacy_second]
diff --git a/backend/tests/unit/test_task_integration_due_date_validation.py b/backend/tests/unit/test_task_integration_due_date_validation.py
index a37ed216964..7482bce6786 100644
--- a/backend/tests/unit/test_task_integration_due_date_validation.py
+++ b/backend/tests/unit/test_task_integration_due_date_validation.py
@@ -72,3 +72,33 @@ def test_missing_due_date_is_allowed(app_client, monkeypatch):
 
     assert resp.status_code == 200
     assert created.await_args.kwargs["due_date"] is None
+
+
+def test_provider_failure_metadata_is_preserved_in_public_response(app_client, monkeypatch):
+    client, ti = app_client
+    monkeypatch.setattr(ti.users_db, "get_task_integration", _connected)
+    monkeypatch.setattr(
+        ti,
+        "create_task_internal",
+        AsyncMock(
+            return_value={
+                "success": False,
+                "error": "ReadTimeout",
+                "error_code": "transport_error",
+                "retryable": False,
+                "ambiguous": True,
+            }
+        ),
+    )
+
+    resp = client.post("/v1/task-integrations/todoist/tasks", json={"title": "ambiguous task"})
+
+    assert resp.status_code == 200
+    assert resp.json() == {
+        "success": False,
+        "external_task_id": None,
+        "error": "ReadTimeout",
+        "error_code": "transport_error",
+        "retryable": False,
+        "ambiguous": True,
+    }
diff --git a/backend/tests/unit/test_task_integrations_ops.py b/backend/tests/unit/test_task_integrations_ops.py
index 2039c396164..6300bb80a31 100644
--- a/backend/tests/unit/test_task_integrations_ops.py
+++ b/backend/tests/unit/test_task_integrations_ops.py
@@ -71,6 +71,7 @@ async def test_create_task_todoist_api_error_marks_disconnected():
 
     assert result["success"] is False
     assert result["error_code"] == "api_error"
+    assert result["retryable"] is False
     mock_run_blocking.assert_awaited_once()
     saved = mock_run_blocking.call_args[0][4]
     assert saved["connected"] is False
@@ -88,9 +89,145 @@ async def test_create_task_missing_access_token():
         "success": False,
         "error": "No access token for todoist",
         "error_code": "no_access_token",
+        "retryable": False,
+        "ambiguous": False,
     }
 
 
+@pytest.mark.asyncio
+async def test_create_task_todoist_server_failure_is_ambiguous_not_retryable():
+    client = AsyncMock(spec=httpx.AsyncClient)
+    client.post.return_value = _mock_response(503, text="Unavailable")
+
+    result = await ops.create_task_internal(
+        uid="uid-4",
+        app_key="todoist",
+        integration={"connected": True, "access_token": "token"},
+        title="Retry task",
+        client=client,
+    )
+
+    assert result["success"] is False
+    assert result["status_code"] == 503
+    assert result["retryable"] is False
+    assert result["ambiguous"] is True
+
+
+@pytest.mark.asyncio
+async def test_create_task_todoist_rate_limit_is_safe_to_retry():
+    client = AsyncMock(spec=httpx.AsyncClient)
+    client.post.return_value = _mock_response(429, text="Rate limited")
+
+    result = await ops.create_task_internal(
+        uid="uid-4",
+        app_key="todoist",
+        integration={"connected": True, "access_token": "token"},
+        title="Retry task",
+        client=client,
+    )
+
+    assert result["retryable"] is True
+    assert result["ambiguous"] is False
+
+
+@pytest.mark.asyncio
+async def test_create_task_todoist_missing_identity_is_ambiguous_not_string_none():
+    client = AsyncMock(spec=httpx.AsyncClient)
+    client.post.return_value = _mock_response(201, {})
+
+    result = await ops.create_task_internal(
+        uid="uid-4",
+        app_key="todoist",
+        integration={"connected": True, "access_token": "token"},
+        title="Ambiguous task",
+        client=client,
+    )
+
+    assert result == {
+        "success": False,
+        "error": "Provider response omitted task identity",
+        "error_code": "invalid_provider_response",
+        "retryable": False,
+        "ambiguous": True,
+    }, "provider success without task identity must be reported as ambiguous"
+
+
+@pytest.mark.asyncio
+async def test_create_task_unexpected_success_status_is_ambiguous():
+    client = AsyncMock(spec=httpx.AsyncClient)
+    client.post.return_value = _mock_response(202, {"id": "possibly-created"})
+
+    result = await ops.create_task_internal(
+        uid="uid-4",
+        app_key="todoist",
+        integration={"connected": True, "access_token": "token"},
+        title="Ambiguous task",
+        client=client,
+    )
+
+    assert result == {
+        "success": False,
+        "error": "Todoist response did not contain a completed task",
+        "error_code": "invalid_provider_response",
+        "status_code": 202,
+        "retryable": False,
+        "ambiguous": True,
+    }
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "error",
+    [
+        httpx.ReadTimeout("response timed out"),
+        httpx.ReadError("response failed"),
+        httpx.WriteTimeout("request write timed out"),
+        httpx.WriteError("request write failed"),
+    ],
+)
+async def test_create_task_transport_failure_is_ambiguous_not_blindly_retryable(error):
+    client = AsyncMock(spec=httpx.AsyncClient)
+    client.post.side_effect = error
+
+    result = await ops.create_task_internal(
+        uid="uid-4",
+        app_key="todoist",
+        integration={"connected": True, "access_token": "token"},
+        title="Ambiguous task",
+        client=client,
+    )
+
+    assert result["error_code"] == "transport_error"
+    assert result["retryable"] is False
+    assert result["ambiguous"] is True
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "error",
+    [
+        httpx.PoolTimeout("connection pool exhausted"),
+        httpx.ConnectTimeout("connection timed out"),
+        httpx.ConnectError("connection failed"),
+    ],
+)
+async def test_create_task_pre_send_transport_failure_is_safe_to_retry(error):
+    client = AsyncMock(spec=httpx.AsyncClient)
+    client.post.side_effect = error
+
+    result = await ops.create_task_internal(
+        uid="uid-4",
+        app_key="todoist",
+        integration={"connected": True, "access_token": "token"},
+        title="Retry task",
+        client=client,
+    )
+
+    assert result["error_code"] == "transport_error"
+    assert result["retryable"] is True
+    assert result["ambiguous"] is False
+
+
 @pytest.mark.asyncio
 async def test_asana_retry_reuses_injected_client_for_refresh_and_retry():
     client = AsyncMock(spec=httpx.AsyncClient)
diff --git a/backend/tests/unit/test_users_webhook_url_validation.py b/backend/tests/unit/test_users_webhook_url_validation.py
index 2d021b4ed10..7df63e83ff2 100644
--- a/backend/tests/unit/test_users_webhook_url_validation.py
+++ b/backend/tests/unit/test_users_webhook_url_validation.py
@@ -126,7 +126,8 @@ def test_valid_url_sets():
         patch.object(users_mod, 'set_user_webhook_db') as setdb,
         patch.object(users_mod, 'disable_user_webhook_db') as disable,
         patch.object(users_mod, 'enable_user_webhook_db') as enable,
-        patch.object(users_mod, 'record_dev_webhook_success') as reset_health,
+        patch.object(users_mod, 'get_user_webhook_db', return_value='http://x'),
+        patch.object(users_mod, 'reset_user_webhook_delivery_health') as reset_health,
     ):
         result = users_mod.set_user_webhook_endpoint(
             wtype='audio_bytes', data=SetUserWebhookUrlRequest(url='http://x'), uid='u1'
@@ -134,7 +135,7 @@ def test_valid_url_sets():
     assert result['status'] == 'ok'
     setdb.assert_called_once()
     enable.assert_called_once_with('u1', 'audio_bytes')
-    reset_health.assert_called_once_with('u1', 'audio_bytes')
+    reset_health.assert_called_once_with('u1', 'audio_bytes', 'http://x')
     disable.assert_not_called()
 
 
@@ -143,7 +144,7 @@ def test_empty_url_disables_without_resetting_health():
         patch.object(users_mod, 'set_user_webhook_db') as setdb,
         patch.object(users_mod, 'disable_user_webhook_db') as disable,
         patch.object(users_mod, 'enable_user_webhook_db') as enable,
-        patch.object(users_mod, 'record_dev_webhook_success') as reset_health,
+        patch.object(users_mod, 'reset_user_webhook_delivery_health') as reset_health,
     ):
         result = users_mod.set_user_webhook_endpoint(
             wtype='audio_bytes', data=SetUserWebhookUrlRequest(url=''), uid='u1'
@@ -165,7 +166,7 @@ def test_cleared_audio_bytes_url_keeping_delay_disables():
         patch.object(users_mod, 'set_user_webhook_db') as setdb,
         patch.object(users_mod, 'disable_user_webhook_db') as disable,
         patch.object(users_mod, 'enable_user_webhook_db') as enable,
-        patch.object(users_mod, 'record_dev_webhook_success') as reset_health,
+        patch.object(users_mod, 'reset_user_webhook_delivery_health') as reset_health,
     ):
         result = users_mod.set_user_webhook_endpoint(
             wtype='audio_bytes', data=SetUserWebhookUrlRequest(url=',5'), uid='u1'
@@ -182,7 +183,8 @@ def test_audio_bytes_url_with_delay_still_enables():
         patch.object(users_mod, 'set_user_webhook_db'),
         patch.object(users_mod, 'disable_user_webhook_db') as disable,
         patch.object(users_mod, 'enable_user_webhook_db') as enable,
-        patch.object(users_mod, 'record_dev_webhook_success'),
+        patch.object(users_mod, 'get_user_webhook_db', return_value='http://x,5'),
+        patch.object(users_mod, 'reset_user_webhook_delivery_health'),
     ):
         users_mod.set_user_webhook_endpoint(
             wtype='audio_bytes', data=SetUserWebhookUrlRequest(url='http://x,5'), uid='u1'
@@ -196,7 +198,7 @@ def test_blank_url_disables_for_non_audio_webhooks():
         patch.object(users_mod, 'set_user_webhook_db'),
         patch.object(users_mod, 'disable_user_webhook_db') as disable,
         patch.object(users_mod, 'enable_user_webhook_db') as enable,
-        patch.object(users_mod, 'record_dev_webhook_success'),
+        patch.object(users_mod, 'reset_user_webhook_delivery_health'),
     ):
         users_mod.set_user_webhook_endpoint(wtype='memory_created', data=SetUserWebhookUrlRequest(url='   '), uid='u1')
     disable.assert_called_once_with('u1', 'memory_created')
@@ -209,7 +211,8 @@ def test_comma_in_non_audio_url_is_not_a_delay_separator():
         patch.object(users_mod, 'set_user_webhook_db'),
         patch.object(users_mod, 'disable_user_webhook_db') as disable,
         patch.object(users_mod, 'enable_user_webhook_db') as enable,
-        patch.object(users_mod, 'record_dev_webhook_success'),
+        patch.object(users_mod, 'get_user_webhook_db', return_value='https://h/i?ids=1,2'),
+        patch.object(users_mod, 'reset_user_webhook_delivery_health'),
     ):
         users_mod.set_user_webhook_endpoint(
             wtype='realtime_transcript', data=SetUserWebhookUrlRequest(url='https://h/i?ids=1,2'), uid='u1'
diff --git a/backend/tests/unit/test_webhook_auto_disable.py b/backend/tests/unit/test_webhook_auto_disable.py
index 4a04b0fca8c..7ce642650a8 100644
--- a/backend/tests/unit/test_webhook_auto_disable.py
+++ b/backend/tests/unit/test_webhook_auto_disable.py
@@ -15,6 +15,7 @@
 import httpx
 import pytest
 
+import database.webhook_health as webhook_health_db
 from testing.import_isolation import load_module_fresh, stub_modules
 from utils.apps import validate_app_endpoints_for_reenable
 
@@ -318,7 +319,7 @@ async def test_dev_webhook_disabled_on_threshold(self):
             patch("utils.webhooks.record_dev_webhook_failure", return_value=True) as mock_fail,
             patch("utils.webhooks.disable_user_webhook_db") as mock_disable,
             patch("utils.webhooks.send_notification") as mock_notify,
-            patch("utils.webhooks._DEV_WEBHOOK_RETRY_DELAYS", ()),
+            patch("utils.webhooks._REALTIME_DEV_WEBHOOK_RETRY_DELAYS", ()),
         ):
             await realtime_transcript_webhook("uid-1", [{"text": "hello"}])
             mock_fail.assert_called_once()
@@ -367,7 +368,7 @@ async def test_dev_webhook_exception_records_failure(self):
             patch("utils.webhooks.get_webhook_client", return_value=mock_client),
             patch("utils.webhooks.get_webhook_circuit_breaker", return_value=mock_cb),
             patch("utils.webhooks.record_dev_webhook_failure", return_value=False) as mock_fail,
-            patch("utils.webhooks._DEV_WEBHOOK_RETRY_DELAYS", ()),
+            patch("utils.webhooks._REALTIME_DEV_WEBHOOK_RETRY_DELAYS", ()),
         ):
             await realtime_transcript_webhook("uid-1", [{"text": "hello"}])
             mock_fail.assert_called_once()
@@ -406,7 +407,7 @@ async def fake_sleep(delay):
             patch("utils.webhooks.get_webhook_semaphore", return_value=mock_sem),
             patch("utils.webhooks.record_dev_webhook_success") as mock_success,
             patch("utils.webhooks.record_dev_webhook_failure") as mock_fail,
-            patch("utils.webhooks._DEV_WEBHOOK_RETRY_DELAYS", (0.01,)),
+            patch("utils.webhooks._REALTIME_DEV_WEBHOOK_RETRY_DELAYS", (0.01,)),
             patch("utils.webhooks.asyncio.sleep", side_effect=fake_sleep),
         ):
             await realtime_transcript_webhook("uid-1", [{"text": "hello"}])
@@ -1379,18 +1380,17 @@ def test_non_disabled_app_loaded(self):
 class TestDevWebhookManualReEnable:
     """Test that manual dev webhook re-enable clears health state."""
 
-    def test_success_on_enable_clears_state(self):
-        """record_dev_webhook_success called on manual enable should reset all fields."""
-        from database.webhook_health import record_dev_webhook_success
-
+    def test_reset_on_enable_clears_state_without_faking_success(self):
         mock_r = MagicMock()
-        with patch("database.webhook_health.r", mock_r):
-            record_dev_webhook_success("uid-1", "realtime_transcript")
+        with patch.object(webhook_health_db, 'r', mock_r):
+            webhook_health_db.reset_dev_webhook_health("uid-1", "realtime_transcript")
 
         mapping = mock_r.hset.call_args.kwargs.get('mapping') or mock_r.hset.call_args[1].get('mapping')
         assert mapping['failure_count'] == '0'
         assert mapping['disabled'] == '0'
         assert mapping['last_error'] == ''
+        assert mapping['last_success_at'] == ''
+        assert mapping['last_status'] == ''
         mock_r.expire.assert_called_once()
 
 
diff --git a/backend/utils/app_integrations.py b/backend/utils/app_integrations.py
index 8f6bdbb5b1e..cac05b9ae54 100644
--- a/backend/utils/app_integrations.py
+++ b/backend/utils/app_integrations.py
@@ -3,6 +3,7 @@
 from typing import List
 import os
 import time
+import uuid
 
 import httpx
 
@@ -87,6 +88,58 @@ def _delivery_failure_is_retryable(status_code: int) -> bool:
     return status_code >= 500 or status_code in _RETRYABLE_DELIVERY_STATUSES
 
 
+_REALTIME_APP_WEBHOOK_RETRY_DELAYS = (0.5, 2.0)
+
+
+async def _post_realtime_app_webhook(
+    app_id: str,
+    webhook_url: str,
+    *,
+    idempotency_key: str | None = None,
+    retry_delays: tuple[float, ...] = _REALTIME_APP_WEBHOOK_RETRY_DELAYS,
+    **request_kwargs,
+):
+    """Retry a realtime app delivery without changing its receiver-visible identity."""
+    headers = dict(request_kwargs.pop('headers', {}) or {})
+    headers.setdefault('X-Omi-Idempotency-Key', idempotency_key or str(uuid.uuid4()))
+    request_kwargs['headers'] = headers
+    client = get_webhook_client()
+    attempts = len(retry_delays) + 1
+    last_response = None
+    last_exception: httpx.TransportError | None = None
+
+    for attempt_index in range(attempts):
+        try:
+            async with get_webhook_semaphore():
+                response = await client.post(webhook_url, **request_kwargs)
+            last_response = response
+            last_exception = None
+            if 200 <= response.status_code < 300:
+                return response
+            if not _delivery_failure_is_retryable(response.status_code):
+                return response
+        except httpx.TransportError as error:
+            last_response = None
+            last_exception = error
+
+        if attempt_index < len(retry_delays):
+            delay = retry_delays[attempt_index]
+            logger.warning(
+                'Realtime app webhook retry app=%s attempt=%s/%s delay=%ss',
+                app_id,
+                attempt_index + 1,
+                attempts,
+                f'{delay:g}',
+            )
+            await asyncio.sleep(delay)
+
+    if last_response is not None:
+        return last_response
+    if last_exception is not None:
+        raise last_exception
+    raise RuntimeError('Realtime app webhook failed without a response')
+
+
 def _notify_app_owner(app_id: str, title: str, body: str):
     """Send a push notification to the app owner about webhook health."""
     try:
@@ -342,10 +395,18 @@ async def trigger_realtime_integrations(
     segments: list[dict],
     conversation_id: str | None,
     source: str | None = None,
+    *,
+    idempotency_key: str | None = None,
 ):
     logger.info(f"trigger_realtime_integrations {uid}")
     """REALTIME STREAMING"""
-    return await _async_trigger_realtime_integrations(uid, segments, conversation_id, source=source)
+    return await _async_trigger_realtime_integrations(
+        uid,
+        segments,
+        conversation_id,
+        source=source,
+        idempotency_key=idempotency_key,
+    )
 
 
 async def trigger_realtime_audio_bytes(uid: str, sample_rate: int, data: bytearray):
@@ -779,6 +840,8 @@ async def _async_trigger_realtime_integrations(
     segments: List[dict],
     conversation_id: str | None,
     source: str | None = None,
+    *,
+    idempotency_key: str | None = None,
 ) -> dict:
     # Paywall: skip mentor + third-party proactive notifications when this
     # transcription session belongs to a paywalled desktop user.
@@ -842,15 +905,15 @@ async def _single(app: App):
             return
 
         try:
-            async with get_webhook_semaphore():
-                client = get_webhook_client()
-                response = await client.post(
-                    pinned_url,
-                    json={"session_id": uid, "segments": segments},
-                    headers=pin_kwargs['headers'],
-                    extensions=pin_kwargs['extensions'],
-                    follow_redirects=False,
-                )
+            response = await _post_realtime_app_webhook(
+                app.id,
+                pinned_url,
+                json={"session_id": uid, "segments": segments},
+                headers=pin_kwargs['headers'],
+                extensions=pin_kwargs['extensions'],
+                follow_redirects=False,
+                idempotency_key=idempotency_key,
+            )
             if response.status_code < 200 or response.status_code >= 300:
                 cb.record_failure()
                 error_str = f'HTTP {response.status_code}'
diff --git a/backend/utils/http_client.py b/backend/utils/http_client.py
index 1818523c6c3..551060e133f 100644
--- a/backend/utils/http_client.py
+++ b/backend/utils/http_client.py
@@ -195,6 +195,13 @@ def record_failure(self):
 _CIRCUIT_BREAKER_IDLE_TTL = 3600  # seconds — evict entries idle for 1 hour
 
 
+def _webhook_circuit_breaker_key(url: str) -> str:
+    try:
+        return url.split('?')[0].split('#')[0]
+    except (IndexError, AttributeError):
+        return url
+
+
 def get_webhook_circuit_breaker(url: str) -> WebhookCircuitBreaker:
     """Get or create a circuit breaker for a webhook target URL.
 
@@ -202,11 +209,7 @@ def get_webhook_circuit_breaker(url: str) -> WebhookCircuitBreaker:
     different webhook endpoints on the same host are isolated from each other.
     Evicts stale entries when the registry grows beyond _CIRCUIT_BREAKER_MAX_ENTRIES.
     """
-    try:
-        # Strip query params but keep scheme + host + path
-        key = url.split('?')[0].split('#')[0]
-    except (IndexError, AttributeError):
-        key = url
+    key = _webhook_circuit_breaker_key(url)
     if key not in _webhook_circuit_breakers:
         if len(_webhook_circuit_breakers) > _CIRCUIT_BREAKER_MAX_ENTRIES:
             _evict_stale_circuit_breakers()
@@ -214,6 +217,11 @@ def get_webhook_circuit_breaker(url: str) -> WebhookCircuitBreaker:
     return _webhook_circuit_breakers[key]
 
 
+def reset_webhook_circuit_breaker(url: str) -> None:
+    """Forget prior failures when a user explicitly replaces or re-enables a target."""
+    _webhook_circuit_breakers.pop(_webhook_circuit_breaker_key(url), None)
+
+
 def _evict_stale_circuit_breakers():
     """Remove circuit breaker entries not accessed for longer than _CIRCUIT_BREAKER_IDLE_TTL.
 
diff --git a/backend/utils/other/storage.py b/backend/utils/other/storage.py
index 83edbbfdfb3..d70f61f8450 100644
--- a/backend/utils/other/storage.py
+++ b/backend/utils/other/storage.py
@@ -214,6 +214,7 @@ def upload_person_speech_sample_from_bytes(
     uid: str,
     person_id: str,
     sample_rate: int = 16000,
+    deduplication_key: Optional[str] = None,
 ) -> str:
     """Upload PCM audio bytes as WAV speech sample. Returns GCS path."""
     import uuid as uuid_module
@@ -224,14 +225,13 @@ def upload_person_speech_sample_from_bytes(
         wav_file.setsampwidth(2)  # 16-bit audio
         wav_file.setframerate(sample_rate)
         wav_file.writeframes(audio_bytes)
-
     bucket = _get_speech_profiles_bucket(required=True)
     assert bucket is not None  # required=True raises if missing
-    filename = f"{uuid_module.uuid4()}.wav"
+    filename_id = hashlib.sha256(deduplication_key.encode()).hexdigest() if deduplication_key else uuid_module.uuid4()
+    filename = f"{filename_id}.wav"
     path = f'{uid}/people_profiles/{person_id}/{filename}'
     blob = bucket.blob(path)
     blob.upload_from_string(wav_buffer.getvalue(), content_type='audio/wav')
-
     return path
 
 
diff --git a/backend/utils/speaker_identification.py b/backend/utils/speaker_identification.py
index ac464a2fa24..9659a26ffbc 100644
--- a/backend/utils/speaker_identification.py
+++ b/backend/utils/speaker_identification.py
@@ -1,7 +1,8 @@
 import io
 import re
 import wave
-from typing import Any, Dict, List, Optional, cast
+from dataclasses import dataclass
+from typing import Any, Dict, List, Literal, Optional, cast
 
 import av
 import numpy as np
@@ -21,6 +22,16 @@
 logger = logging.getLogger(__name__)
 
 
+@dataclass(frozen=True)
+class SpeakerSampleExtractionResult:
+    status: Literal['stored', 'already_present', 'terminal_no_sample', 'retryable']
+    reason: str
+
+    @property
+    def retryable(self) -> bool:
+        return self.status == 'retryable'
+
+
 def _pcm_to_wav_bytes(pcm_data: bytes, sample_rate: int) -> bytes:
     """
     Convert PCM16 mono audio to WAV format bytes.
@@ -327,7 +338,8 @@ async def extract_speaker_samples(
     conversation_id: str,
     segment_ids: List[str],
     sample_rate: int = 16000,
-):
+    delivery_id: Optional[str] = None,
+) -> SpeakerSampleExtractionResult:
     """
     Extract speech samples from segments and store as speaker profiles.
     Fetches conversation from DB to get started_at and segment details.
@@ -337,6 +349,8 @@ async def extract_speaker_samples(
         # Run lazy migration for samples before checking count
         # (migration may drop invalid samples, freeing up space)
         person = await run_blocking(db_executor, users_db.get_person, uid, person_id)
+        if person is None:
+            return SpeakerSampleExtractionResult('retryable', 'person_not_ready')
         if person:
             person = await maybe_migrate_person_samples(uid, person)
 
@@ -344,18 +358,18 @@ async def extract_speaker_samples(
         sample_count = await run_blocking(db_executor, users_db.get_person_speech_samples_count, uid, person_id)
         if sample_count >= 1:
             logger.warning(f"Person {person_id} already has {sample_count} samples, skipping {uid} {conversation_id}")
-            return
+            return SpeakerSampleExtractionResult('already_present', 'sample_already_present')
 
         # Fetch conversation to get started_at and segment details
         conversation = await run_blocking(db_executor, conversations_db.get_conversation, uid, conversation_id)
         if not conversation:
             logger.warning(f"Conversation {conversation_id} not found {uid}")
-            return
+            return SpeakerSampleExtractionResult('retryable', 'conversation_not_ready')
 
         started_at = conversation.get('started_at')
         if not started_at:
             logger.info(f"Conversation {conversation_id} has no started_at {uid}")
-            return
+            return SpeakerSampleExtractionResult('retryable', 'conversation_not_ready')
 
         started_at_ts = started_at.timestamp() if hasattr(started_at, 'timestamp') else float(started_at)
 
@@ -367,7 +381,7 @@ async def extract_speaker_samples(
         audio_files = conversation.get('audio_files', [])
         if not audio_files:
             logger.warning(f"No audio files found for {conversation_id}, skipping speaker sample extraction {uid}")
-            return
+            return SpeakerSampleExtractionResult('retryable', 'audio_files_not_ready')
 
         # Collect all chunk timestamps from audio files
         all_timestamps: List[Any] = []
@@ -377,13 +391,14 @@ async def extract_speaker_samples(
 
         if not all_timestamps:
             logger.warning(f"No chunk timestamps found for {conversation_id}, skipping speaker sample extraction {uid}")
-            return
+            return SpeakerSampleExtractionResult('retryable', 'audio_timestamps_not_ready')
 
         # Build chunks list in expected format
         chunks: List[Dict[str, Any]] = [{'timestamp': ts} for ts in sorted(set(all_timestamps))]
 
         samples_added = 0
         max_samples_to_add = 1 - sample_count
+        retryable_reason: Optional[str] = None
 
         # Build ordered list with index lookup for expansion
         ordered_segments = [s for s in conv_segments if s.get('id')]
@@ -396,6 +411,7 @@ async def extract_speaker_samples(
             seg = segment_map.get(seg_id)
             if not seg:
                 logger.warning(f"Segment {seg_id} not found in conversation {uid} {conversation_id}")
+                retryable_reason = 'transcript_segments_not_ready'
                 continue
 
             segment_start = seg.get('start')
@@ -464,6 +480,7 @@ async def extract_speaker_samples(
                 logger.info(
                     f"No relevant chunks for segment {segment_start:.1f}-{segment_end:.1f}s {uid} {conversation_id}"
                 )
+                retryable_reason = 'audio_chunks_not_ready'
                 continue
 
             # Download, merge, and extract (sync_executor avoids parent-child deadlock on storage_executor, #7387)
@@ -503,15 +520,30 @@ async def extract_speaker_samples(
             transcript, is_valid, reason = await verify_and_transcribe_sample(wav_bytes, sample_rate, expected_text)
             if not is_valid:
                 logger.error(f"Sample failed quality check: {reason} {uid} {conversation_id}")
+                if reason.startswith('transcription_failed'):
+                    retryable_reason = 'transcription_failed'
                 continue  # Try next segment
 
             # Upload and store
+            sample_deduplication_key = f'speaker-sample\0{uid}\0{person_id}\0{delivery_id}' if delivery_id else None
             path = await run_blocking(
-                storage_executor, upload_person_speech_sample_from_bytes, sample_audio, uid, person_id, sample_rate
+                storage_executor,
+                upload_person_speech_sample_from_bytes,
+                sample_audio,
+                uid,
+                person_id,
+                sample_rate,
+                sample_deduplication_key,
             )
 
             success = await run_blocking(
-                db_executor, users_db.add_person_speech_sample, uid, person_id, path, transcript=transcript
+                db_executor,
+                users_db.add_person_speech_sample,
+                uid,
+                person_id,
+                path,
+                transcript=transcript,
+                max_samples=1,
             )
             if success:
                 samples_added += 1
@@ -533,9 +565,20 @@ async def extract_speaker_samples(
                     )
                 except Exception as emb_err:
                     logger.error(f"Failed to extract/store speaker embedding: {emb_err} {uid} {conversation_id}")
+                return SpeakerSampleExtractionResult('stored', 'sample_stored')
             else:
                 logger.error(f"Failed to add speech sample for person {person_id} {uid} {conversation_id}")
-                break  # Likely hit limit
+                current_count = await run_blocking(
+                    db_executor, users_db.get_person_speech_samples_count, uid, person_id
+                )
+                if current_count >= 1:
+                    return SpeakerSampleExtractionResult('already_present', 'sample_added_concurrently')
+                return SpeakerSampleExtractionResult('retryable', 'sample_persistence_failed')
 
     except Exception as e:
         logger.error(f"Error extracting speaker samples: {e} {uid} {conversation_id}")
+        return SpeakerSampleExtractionResult('retryable', 'extraction_failed')
+
+    if retryable_reason is not None:
+        return SpeakerSampleExtractionResult('retryable', retryable_reason)
+    return SpeakerSampleExtractionResult('terminal_no_sample', 'no_eligible_segment')
diff --git a/backend/utils/task_integrations_ops.py b/backend/utils/task_integrations_ops.py
index c4f731d1641..96b411ada31 100644
--- a/backend/utils/task_integrations_ops.py
+++ b/backend/utils/task_integrations_ops.py
@@ -28,6 +28,51 @@
 http_client: Optional[httpx.AsyncClient] = None
 
 
+def _provider_create_success(external_task_id: Any) -> dict:
+    task_id = str(external_task_id).strip() if external_task_id is not None else ''
+    if not task_id:
+        return {
+            'success': False,
+            'error': 'Provider response omitted task identity',
+            'error_code': 'invalid_provider_response',
+            'retryable': False,
+            'ambiguous': True,
+        }
+    return {'success': True, 'external_task_id': task_id}
+
+
+def _provider_create_http_failure(provider: str, status_code: int) -> dict:
+    if 200 <= status_code < 300:
+        return {
+            'success': False,
+            'error': f'{provider} response did not contain a completed task',
+            'error_code': 'invalid_provider_response',
+            'status_code': status_code,
+            'retryable': False,
+            'ambiguous': True,
+        }
+    ambiguous = status_code in {408, 425} or status_code >= 500
+    return {
+        'success': False,
+        'error': f'{provider} API error: {status_code}',
+        'error_code': 'api_error',
+        'status_code': status_code,
+        'retryable': status_code == 429,
+        'ambiguous': ambiguous,
+    }
+
+
+def _provider_create_transport_failure(error: httpx.TransportError) -> dict:
+    safe_before_send = isinstance(error, (httpx.PoolTimeout, httpx.ConnectTimeout, httpx.ConnectError))
+    return {
+        'success': False,
+        'error': type(error).__name__,
+        'error_code': 'transport_error',
+        'retryable': safe_before_send,
+        'ambiguous': not safe_before_send,
+    }
+
+
 def get_http_client() -> httpx.AsyncClient:
     """Get or create the HTTP client instance."""
     global http_client
@@ -187,6 +232,48 @@ async def ensure_valid_oauth_token(
     return integration
 
 
+def _task_create_configuration_failure(app_key: str, integration: dict) -> Optional[dict]:
+    if app_key not in OAUTH_CONFIGS:
+        return {
+            'success': False,
+            'error': f'Unsupported integration: {app_key}',
+            'error_code': 'unsupported',
+            'retryable': False,
+            'ambiguous': False,
+        }
+    if integration.get('connected') is False:
+        name = OAUTH_CONFIGS[app_key]['name']
+        return {
+            'success': False,
+            'error': f'{name} token refresh failed',
+            'error_code': 'token_refresh_failed',
+            'retryable': False,
+            'ambiguous': False,
+        }
+    if not integration.get('access_token'):
+        return {
+            'success': False,
+            'error': f'No access token for {app_key}',
+            'error_code': 'no_access_token',
+            'retryable': False,
+            'ambiguous': False,
+        }
+    required_field = {
+        'asana': ('workspace_gid', 'No workspace configured', 'no_workspace'),
+        'google_tasks': ('default_list_id', 'No task list configured', 'no_list'),
+        'clickup': ('list_id', 'No list configured', 'no_list'),
+    }.get(app_key)
+    if required_field and not integration.get(required_field[0]):
+        return {
+            'success': False,
+            'error': required_field[1],
+            'error_code': required_field[2],
+            'retryable': False,
+            'ambiguous': False,
+        }
+    return None
+
+
 async def perform_request_with_token_retry(
     uid: str,
     app_key: str,
@@ -224,7 +311,7 @@ async def create_task_internal(
     Returns:
         dict: {"success": bool, "external_task_id": str, "error": str, "error_code": str}
     """
-    if app_key in ['google_tasks', 'asana']:
+    if app_key in {'google_tasks', 'asana'}:
         integration = await ensure_valid_oauth_token(
             uid,
             app_key,
@@ -232,18 +319,13 @@ async def create_task_internal(
             refresh_if_missing_expires_at=(app_key == 'google_tasks'),
             client=client,
         )
-        # Use `is False` so a missing key (None) falls through to access_token
-        # validation below instead of blocking valid tokens on legacy records.
-        if integration.get('connected') is False:
-            name = OAUTH_CONFIGS.get(app_key, {'name': app_key}).get('name', app_key)
-            return {"success": False, "error": f"{name} token refresh failed", "error_code": "token_refresh_failed"}
-
-    access_token = integration.get('access_token')
-    if not access_token:
-        return {"success": False, "error": f"No access token for {app_key}", "error_code": "no_access_token"}
+    preflight_error = _task_create_configuration_failure(app_key, integration)
+    if preflight_error is not None:
+        return preflight_error
 
     try:
         client = client or get_http_client()
+        access_token = str(integration['access_token'])
 
         if app_key == 'todoist':
             body = {'content': title, 'priority': 2}
@@ -260,7 +342,7 @@ async def create_task_internal(
 
             if response.status_code in [200, 201]:
                 task_data = response.json()
-                return {"success": True, "external_task_id": str(task_data.get('id'))}
+                return _provider_create_success(task_data.get('id'))
             else:
                 if response.status_code == 401:
                     await run_blocking(
@@ -270,20 +352,13 @@ async def create_task_internal(
                         'todoist',
                         {'connected': False},
                     )
-                return {
-                    "success": False,
-                    "error": f"Todoist API error: {response.status_code}",
-                    "error_code": "api_error",
-                }
+                return _provider_create_http_failure('Todoist', response.status_code)
 
         elif app_key == 'asana':
-            workspace_gid = integration.get('workspace_gid')
+            workspace_gid = str(integration['workspace_gid'])
             project_gid = integration.get('project_gid')
             user_gid = integration.get('user_gid')
 
-            if not workspace_gid:
-                return {"success": False, "error": "No workspace configured", "error_code": "no_workspace"}
-
             task_data = {'name': title, 'workspace': workspace_gid}
             if description:
                 task_data['notes'] = description
@@ -305,22 +380,21 @@ async def _asana_post(c, token):
                 uid, app_key, integration, _asana_post, client=client
             )
             if retry_err:
-                return {"success": False, "error": "Asana token refresh failed", "error_code": "token_refresh_failed"}
+                return {
+                    "success": False,
+                    "error": "Asana token refresh failed",
+                    "error_code": "token_refresh_failed",
+                    "retryable": False,
+                }
 
             if response.status_code in [200, 201]:
                 result = response.json()
-                return {"success": True, "external_task_id": result.get('data', {}).get('gid')}
+                return _provider_create_success(result.get('data', {}).get('gid'))
             else:
-                return {
-                    "success": False,
-                    "error": f"Asana API error: {response.status_code}",
-                    "error_code": "api_error",
-                }
+                return _provider_create_http_failure('Asana', response.status_code)
 
         elif app_key == 'google_tasks':
-            list_id = integration.get('default_list_id')
-            if not list_id:
-                return {"success": False, "error": "No task list configured", "error_code": "no_list"}
+            list_id = str(integration['default_list_id'])
 
             task_data = {'title': title}
             if description:
@@ -343,22 +417,17 @@ async def _google_tasks_post(c, token):
                     "success": False,
                     "error": "Google Tasks token refresh failed",
                     "error_code": "token_refresh_failed",
+                    "retryable": False,
                 }
 
             if response.status_code in [200, 201]:
                 result = response.json()
-                return {"success": True, "external_task_id": result.get('id')}
+                return _provider_create_success(result.get('id'))
             else:
-                return {
-                    "success": False,
-                    "error": f"Google Tasks API error: {response.status_code}",
-                    "error_code": "api_error",
-                }
+                return _provider_create_http_failure('Google Tasks', response.status_code)
 
         elif app_key == 'clickup':
-            list_id = integration.get('list_id')
-            if not list_id:
-                return {"success": False, "error": "No list configured", "error_code": "no_list"}
+            list_id = str(integration['list_id'])
 
             task_data: dict[str, Any] = {'name': title}
             if description:
@@ -374,17 +443,27 @@ async def _google_tasks_post(c, token):
 
             if response.status_code in [200, 201]:
                 result = response.json()
-                return {"success": True, "external_task_id": result.get('id')}
+                return _provider_create_success(result.get('id'))
             else:
-                return {
-                    "success": False,
-                    "error": f"ClickUp API error: {response.status_code}",
-                    "error_code": "api_error",
-                }
-
+                return _provider_create_http_failure('ClickUp', response.status_code)
         else:
-            return {"success": False, "error": f"Unsupported integration: {app_key}", "error_code": "unsupported"}
+            return {
+                'success': False,
+                'error': f'Unsupported integration: {app_key}',
+                'error_code': 'unsupported',
+                'retryable': False,
+                'ambiguous': False,
+            }
 
+    except httpx.TransportError as e:
+        logger.error(f"Error creating task in {app_key}: {e}")
+        return _provider_create_transport_failure(e)
     except Exception as e:
         logger.error(f"Error creating task in {app_key}: {e}")
-        return {"success": False, "error": str(e)}
+        return {
+            "success": False,
+            "error": type(e).__name__,
+            "error_code": "internal_error",
+            "retryable": False,
+            "ambiguous": True,
+        }
diff --git a/backend/utils/webhooks.py b/backend/utils/webhooks.py
index 10cbfb5f1a2..f91bb74bb13 100644
--- a/backend/utils/webhooks.py
+++ b/backend/utils/webhooks.py
@@ -6,6 +6,8 @@
 from typing import List, Optional
 from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
 
+import httpx
+
 from database.redis_db import (
     get_user_webhook_db,
     user_webhook_status_db,
@@ -13,20 +15,32 @@
     enable_user_webhook_db,
     set_user_webhook_db,
 )
-from database.webhook_health import record_dev_webhook_failure, record_dev_webhook_success, _DEV_FAILURE_THRESHOLD
+from database.webhook_health import (
+    record_dev_webhook_failure,
+    record_dev_webhook_success,
+    reset_dev_webhook_health,
+    _DEV_FAILURE_THRESHOLD,
+)
 from models.conversation import Conversation
 from models.users import WebhookType, webhook_url_from_setting
 import database.notifications as notification_db
 from utils.conversations.render import populate_speaker_names, populate_folder_names
 from utils.conversations.render import conversation_to_dict
 from utils.executors import db_executor, run_blocking
-from utils.http_client import get_webhook_client, get_webhook_circuit_breaker, get_webhook_semaphore
+from utils.http_client import (
+    get_webhook_client,
+    get_webhook_circuit_breaker,
+    get_webhook_semaphore,
+    reset_webhook_circuit_breaker,
+)
 from utils.notifications import send_notification
 import logging
 
 logger = logging.getLogger(__name__)
 
 _DEV_WEBHOOK_RETRY_DELAYS = (1.0, 5.0, 30.0)
+_DEV_WEBHOOK_RETRYABLE_STATUS_CODES = frozenset({408, 425, 429})
+_REALTIME_DEV_WEBHOOK_RETRY_DELAYS = (0.5, 2.0)
 
 
 def _get_dev_webhook_retry_delays() -> tuple[float, ...]:
@@ -50,6 +64,20 @@ def _append_query_params(url: str, params: dict) -> str:
     return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query_items), parts.fragment))
 
 
+def _is_retryable_dev_webhook_status(status_code: int) -> bool:
+    return status_code in _DEV_WEBHOOK_RETRYABLE_STATUS_CODES or 500 <= status_code < 600
+
+
+def reset_user_webhook_delivery_health(uid: str, wtype: WebhookType, webhook_url: Optional[str]) -> None:
+    """Reset persisted and process-local failure gates after an explicit enable."""
+    reset_dev_webhook_health(uid, wtype)
+    target_url = webhook_url or ''
+    if wtype == WebhookType.audio_bytes:
+        target_url = target_url.split(',', 1)[0]
+    if target_url:
+        reset_webhook_circuit_breaker(target_url)
+
+
 async def _post_dev_webhook(
     webhook_name: str,
     webhook_url: str,
@@ -85,7 +113,13 @@ async def _post_dev_webhook(
                 )
                 return response
             failure_reason = f'HTTP {response.status_code}'
-        except Exception as e:
+            if not _is_retryable_dev_webhook_status(response.status_code):
+                logger.error(
+                    f'{webhook_name}: delivery failed status={response.status_code} '
+                    f'attempt={attempt_number}/{attempts} retryable=false'
+                )
+                return response
+        except httpx.TransportError as e:
             last_response = None
             last_exception = e
             failure_reason = type(e).__name__
@@ -110,13 +144,13 @@ async def _post_dev_webhook(
     raise last_exception
 
 
-async def _handle_dev_webhook_disable(uid: str, wtype: str, should_disable: bool):
+async def _handle_dev_webhook_disable(uid: str, wtype: WebhookType | str, should_disable: bool):
     if should_disable:
         logger.warning(
             f'Dev webhook auto-disabled: uid={uid} type={wtype} after {_DEV_FAILURE_THRESHOLD} consecutive failures'
         )
         await run_blocking(db_executor, disable_user_webhook_db, uid, wtype)
-        wtype_str = wtype.value if hasattr(wtype, 'value') else str(wtype)
+        wtype_str = wtype.value if isinstance(wtype, WebhookType) else str(wtype)
         await run_blocking(
             db_executor,
             send_notification,
@@ -134,7 +168,7 @@ def _build_conversation_webhook_payload_sync(uid: str, memory: Conversation) ->
     return payload
 
 
-async def conversation_created_webhook(uid, memory: Conversation):
+async def conversation_created_webhook(uid: str, memory: Conversation):
     if memory.is_locked:
         return
 
@@ -237,7 +271,12 @@ async def day_summary_webhook(uid, summary: str, summary_json: Optional[dict] =
         return
 
 
-async def realtime_transcript_webhook(uid, segments: List[dict]):
+async def realtime_transcript_webhook(
+    uid,
+    segments: List[dict],
+    *,
+    idempotency_key: Optional[str] = None,
+):
     logger.info(f"realtime_transcript_webhook {uid}")
     toggled = await run_blocking(db_executor, user_webhook_status_db, uid, WebhookType.realtime_transcript)
 
@@ -256,6 +295,8 @@ async def realtime_transcript_webhook(uid, segments: List[dict]):
                 webhook_url,
                 json={'segments': segments, 'session_id': uid},
                 headers={'Content-Type': 'application/json'},
+                idempotency_key=idempotency_key,
+                retry_delays=_REALTIME_DEV_WEBHOOK_RETRY_DELAYS,
             )
             if response.status_code >= 200 and response.status_code < 300:
                 cb.record_success()
@@ -331,6 +372,7 @@ async def send_audio_bytes_developer_webhook(uid: str, sample_rate: int, data: b
                 webhook_url,
                 content=bytes(data),
                 headers={'Content-Type': 'application/octet-stream'},
+                retry_delays=_REALTIME_DEV_WEBHOOK_RETRY_DELAYS,
             )
             if response.status_code >= 200 and response.status_code < 300:
                 cb.record_success()
diff --git a/desktop/windows/src/renderer/src/lib/omiApi.generated.ts b/desktop/windows/src/renderer/src/lib/omiApi.generated.ts
index dfb94afe710..8dde2f02497 100644
--- a/desktop/windows/src/renderer/src/lib/omiApi.generated.ts
+++ b/desktop/windows/src/renderer/src/lib/omiApi.generated.ts
@@ -1339,8 +1339,11 @@ export interface CreateTaskRequest {
 }
 
 export interface CreateTaskResponse {
+  ambiguous?: boolean | null;
   error?: string | null;
+  error_code?: string | null;
   external_task_id?: string | null;
+  retryable?: boolean | null;
   success: boolean;
 }
 
diff --git a/docs/api-reference/app-client-openapi.json b/docs/api-reference/app-client-openapi.json
index af787b85efa..84637a89ad4 100644
--- a/docs/api-reference/app-client-openapi.json
+++ b/docs/api-reference/app-client-openapi.json
@@ -8413,6 +8413,17 @@
       "CreateTaskResponse": {
         "description": "Response for task creation",
         "properties": {
+          "ambiguous": {
+            "anyOf": [
+              {
+                "type": "boolean"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Ambiguous"
+          },
           "error": {
             "anyOf": [
               {
@@ -8424,6 +8435,17 @@
             ],
             "title": "Error"
           },
+          "error_code": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Error Code"
+          },
           "external_task_id": {
             "anyOf": [
               {
@@ -8435,6 +8457,17 @@
             ],
             "title": "External Task Id"
           },
+          "retryable": {
+            "anyOf": [
+              {
+                "type": "boolean"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Retryable"
+          },
           "success": {
             "title": "Success",
             "type": "boolean"
diff --git a/web/admin/lib/services/omi-api/omiApi.generated.ts b/web/admin/lib/services/omi-api/omiApi.generated.ts
index dfb94afe710..8dde2f02497 100644
--- a/web/admin/lib/services/omi-api/omiApi.generated.ts
+++ b/web/admin/lib/services/omi-api/omiApi.generated.ts
@@ -1339,8 +1339,11 @@ export interface CreateTaskRequest {
 }
 
 export interface CreateTaskResponse {
+  ambiguous?: boolean | null;
   error?: string | null;
+  error_code?: string | null;
   external_task_id?: string | null;
+  retryable?: boolean | null;
   success: boolean;
 }
 
diff --git a/web/app/src/lib/omiApi.generated.ts b/web/app/src/lib/omiApi.generated.ts
index dfb94afe710..8dde2f02497 100644
--- a/web/app/src/lib/omiApi.generated.ts
+++ b/web/app/src/lib/omiApi.generated.ts
@@ -1339,8 +1339,11 @@ export interface CreateTaskRequest {
 }
 
 export interface CreateTaskResponse {
+  ambiguous?: boolean | null;
   error?: string | null;
+  error_code?: string | null;
   external_task_id?: string | null;
+  retryable?: boolean | null;
   success: boolean;
 }
 
diff --git a/web/personas-open-source/src/lib/omiApi.generated.ts b/web/personas-open-source/src/lib/omiApi.generated.ts
index dfb94afe710..8dde2f02497 100644
--- a/web/personas-open-source/src/lib/omiApi.generated.ts
+++ b/web/personas-open-source/src/lib/omiApi.generated.ts
@@ -1339,8 +1339,11 @@ export interface CreateTaskRequest {
 }
 
 export interface CreateTaskResponse {
+  ambiguous?: boolean | null;
   error?: string | null;
+  error_code?: string | null;
   external_task_id?: string | null;
+  retryable?: boolean | null;
   success: boolean;
 }