human hand off fix

backend/app/main.py

@@ -30,6 +30,15 @@ settings = get_settings()
 inbound_server = InboundSMTPServer()
 
 
+def _index_option_matches(index: dict, options: dict) -> bool:
+    return (
+        bool(index.get("unique")) == bool(options.get("unique"))
+        and bool(index.get("sparse")) == bool(options.get("sparse"))
+        and dict(index.get("partialFilterExpression") or {})
+        == dict(options.get("partialFilterExpression") or {})
+    )
+
+
 async def ensure_mongo_index(collection, keys, **options) -> None:
     """Create an index, replacing any same-name or same-key index with different options."""
     key_list = [(field, direction) for field, direction in keys]
@@ -39,9 +48,6 @@ async def ensure_mongo_index(collection, keys, **options) -> None:
         name = "_".join(f"{field}_{direction}" for field, direction in key_list)
         options["name"] = name
 
-    desired_unique = bool(options.get("unique"))
-    desired_sparse = bool(options.get("sparse"))
-
     async with await collection.list_indexes() as cursor:
         async for index in cursor:
             if index["name"] == "_id_":
@@ -51,17 +57,25 @@ async def ensure_mongo_index(collection, keys, **options) -> None:
             same_name = index["name"] == name
             if not same_key and not same_name:
                 continue
-            same_options = (
+            if same_key and same_name and _index_option_matches(index, options):
-                bool(index.get("unique")) == desired_unique
-                and bool(index.get("sparse")) == desired_sparse
-            )
-            if same_key and same_name and same_options:
                 return
             await collection.drop_index(index["name"])
 
     await collection.create_index(keys, **options)
 
 
+async def cleanup_incomplete_handoff_links(database) -> None:
+    """Remove leftover links that cannot participate in the unique source key."""
+    await database.handoff_message_links.delete_many(
+        {
+            "$or": [
+                {"source_message_id": None},
+                {"source_message_id": {"$exists": False}},
+            ]
+        }
+    )
+
+
 async def backfill_missing_thread_ids(database) -> None:
     cursor = database.mailbox_entries.find({"thread_id": None})
     async for entry in cursor:
@@ -302,6 +316,7 @@ async def lifespan(_: FastAPI):
         [("account_id", 1), ("handoff_thread_id", 1)],
         unique=True,
     )
+    await cleanup_incomplete_handoff_links(mongo_db)
     await ensure_mongo_index(
         mongo_db.handoff_message_links,
         [("handoff_id", 1), ("created_at", -1)],
@@ -310,7 +325,9 @@ async def lifespan(_: FastAPI):
         mongo_db.handoff_message_links,
         [("handoff_id", 1), ("source_message_id", 1), ("relationship", 1)],
         unique=True,
-        sparse=True,
+        partialFilterExpression={
+            "source_message_id": {"$exists": True, "$type": "objectId"}
+        },
     )
     await ensure_mongo_index(
         mongo_db.handoff_delivery_jobs,