src/scribe/services/dedup.py

@@ -22,6 +22,7 @@ Two signals, both scoped to the same owner + project + kind:
 """
 from __future__ import annotations
 
+import logging
 from dataclasses import dataclass
 
 from sqlalchemy import func, select
@@ -30,6 +31,8 @@ from scribe.models import async_session
 from scribe.models.note import Note
 from scribe.services import embeddings as embeddings_svc
 
+logger = logging.getLogger(__name__)
+
 # Run the semantic check only when the incoming body has at least this many
 # characters — below it, embeddings are dominated by the title and false-positive.
 _MIN_BODY_FOR_SEMANTIC = 200
@@ -82,25 +85,31 @@ async def find_duplicate_note(
     norm = " ".join((title or "").split()).lower()
 
     # --- Signal 1: normalized-title exact match (same scope) ---
+    # Fail-open: a dedup-check failure (DB down, etc.) must never block a
+    # legitimate create — degrade to "no duplicate found" and let it through.
     if norm:
-        async with async_session() as session:
-            stmt = select(Note).where(
-                Note.user_id == user_id,
-                Note.deleted_at.is_(None),
-                Note.note_type == note_type,
-                func.lower(func.trim(Note.title)) == norm,
-            )
-            if project_id is not None:
-                stmt = stmt.where(Note.project_id == project_id)
-            else:
-                stmt = stmt.where(Note.project_id.is_(None))
-            if is_task is True:
-                stmt = stmt.where(Note.status.isnot(None))
-            elif is_task is False:
-                stmt = stmt.where(Note.status.is_(None))
-            existing = (await session.execute(stmt.limit(1))).scalars().first()
-            if existing is not None:
-                return DuplicateMatch(existing.id, existing.title, 1.0, "title")
+        try:
+            async with async_session() as session:
+                stmt = select(Note).where(
+                    Note.user_id == user_id,
+                    Note.deleted_at.is_(None),
+                    Note.note_type == note_type,
+                    func.lower(func.trim(Note.title)) == norm,
+                )
+                if project_id is not None:
+                    stmt = stmt.where(Note.project_id == project_id)
+                else:
+                    stmt = stmt.where(Note.project_id.is_(None))
+                if is_task is True:
+                    stmt = stmt.where(Note.status.isnot(None))
+                elif is_task is False:
+                    stmt = stmt.where(Note.status.is_(None))
+                existing = (await session.execute(stmt.limit(1))).scalars().first()
+                if existing is not None:
+                    return DuplicateMatch(existing.id, existing.title, 1.0, "title")
+        except Exception:
+            logger.debug("dedup title check skipped — query failed", exc_info=True)
+            return None
 
     # --- Signal 2: semantic similarity (only with a substantial body) ---
     if body and len(body.strip()) >= _MIN_BODY_FOR_SEMANTIC: