Scope duplicate detection to same type (notes vs tasks)

create_note now only checks against existing notes (is_task=False);
create_task only checks against existing tasks (is_task=True). All three
checks (exact title, fuzzy title, semantic similarity) pass the type
filter through to list_notes and semantic_search_notes.

Adds is_task param to semantic_search_notes() so callers can restrict
results to notes or tasks independently.

Prevents a note titled "Design enemy AI" from blocking a task with the
same title, and vice versa.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-10 08:35:54 -04:00
parent da28a2f5c4
commit f5bee4573f
2 changed files with 21 additions and 20 deletions
@@ -80,6 +80,7 @@ async def semantic_search_notes(
limit: int = 8,
threshold: float = _SIMILARITY_THRESHOLD,
project_id: int | None = None,
is_task: bool | None = None,
) -> list[tuple[float, Note]]:
"""Return up to *limit* (score, note) pairs most relevant to *query*.
@@ -102,6 +103,10 @@ async def semantic_search_notes(
)
if project_id is not None:
stmt = stmt.where(Note.project_id == project_id)
if is_task is True:
stmt = stmt.where(Note.status.isnot(None))
elif is_task is False:
stmt = stmt.where(Note.status.is_(None))
if exclude_ids:
stmt = stmt.where(NoteEmbedding.note_id.notin_(exclude_ids))
rows = list((await session.execute(stmt)).all())
+16 -20
View File
@@ -813,26 +813,24 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
return {"success": False, "error": f"Milestone '{milestone_name}' not found in project '{proj.title}'. Use list_milestones to see available milestones."}
pre_fields["milestone_id"] = ms.id
existing = await get_note_by_title(user_id, task_title)
if existing is not None:
item_type = "task" if existing.status is not None else "note"
return {"success": False, "error": f"A {item_type} titled '{task_title}' already exists (id: {existing.id}). Use update_note to modify it instead of creating a duplicate."}
existing_tasks, _ = await list_notes(user_id=user_id, q=task_title, is_task=True, limit=1)
exact = next((t for t in existing_tasks if t.title.lower() == task_title.lower()), None)
if exact is not None:
return {"success": False, "error": f"A task titled '{task_title}' already exists (id: {exact.id}). Use update_note to modify it instead of creating a duplicate."}
clean_q = _PUNCT_RE.sub(" ", task_title).strip()
candidates, _ = await list_notes(user_id=user_id, q=clean_q, limit=20)
candidates, _ = await list_notes(user_id=user_id, q=clean_q, is_task=True, limit=20)
near, ratio = _fuzzy_title_match(task_title, candidates)
if near is not None:
item_type = "task" if near.status is not None else "note"
return {"success": False, "requires_confirmation": True, "similar_note": {"id": near.id, "title": near.title}, "error": f"A {item_type} with a very similar title '{near.title}' already exists (similarity: {ratio:.0%}). Ask the user to confirm before creating a separate entry."}
return {"success": False, "requires_confirmation": True, "similar_note": {"id": near.id, "title": near.title}, "error": f"A task with a very similar title '{near.title}' already exists (similarity: {ratio:.0%}). Ask the user to confirm before creating a separate entry."}
if not arguments.get("confirmed") and len(task_body.strip()) >= 200:
from fabledassistant.services.embeddings import semantic_search_notes as _ssn
sem_query = f"{task_title}\n{task_body}".strip()
sem_hits = await _ssn(user_id, sem_query, limit=3, threshold=0.90)
sem_hits = await _ssn(user_id, sem_query, limit=3, threshold=0.90, is_task=True)
if sem_hits:
best_score, best_note = sem_hits[0]
item_type = "task" if best_note.status is not None else "note"
return {"success": False, "requires_confirmation": True, "similar_note": {"id": best_note.id, "title": best_note.title}, "error": f"A {item_type} with very similar content exists: '{best_note.title}' (semantic similarity: {best_score:.0%}). Ask the user to confirm before creating a separate entry."}
return {"success": False, "requires_confirmation": True, "similar_note": {"id": best_note.id, "title": best_note.title}, "error": f"A task with very similar content exists: '{best_note.title}' (semantic similarity: {best_score:.0%}). Ask the user to confirm before creating a separate entry."}
note = await create_note(
user_id=user_id,
@@ -891,26 +889,24 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
return {"success": False, "error": f"Project '{project_name}' not found. Use list_projects to see available projects or create_project to create one."}
project_id = proj.id
existing = await get_note_by_title(user_id, note_title)
if existing is not None:
item_type = "task" if existing.status is not None else "note"
return {"success": False, "error": f"A {item_type} titled '{note_title}' already exists (id: {existing.id}). Use update_note to modify it instead of creating a duplicate."}
existing_notes, _ = await list_notes(user_id=user_id, q=note_title, is_task=False, limit=1)
exact = next((n for n in existing_notes if n.title.lower() == note_title.lower()), None)
if exact is not None:
return {"success": False, "error": f"A note titled '{note_title}' already exists (id: {exact.id}). Use update_note to modify it instead of creating a duplicate."}
clean_q = _PUNCT_RE.sub(" ", note_title).strip()
candidates, _ = await list_notes(user_id=user_id, q=clean_q, limit=20)
candidates, _ = await list_notes(user_id=user_id, q=clean_q, is_task=False, limit=20)
near, ratio = _fuzzy_title_match(note_title, candidates)
if near is not None:
item_type = "task" if near.status is not None else "note"
return {"success": False, "requires_confirmation": True, "similar_note": {"id": near.id, "title": near.title}, "error": f"A {item_type} with a very similar title '{near.title}' already exists (similarity: {ratio:.0%}). Ask the user to confirm before creating a separate entry."}
return {"success": False, "requires_confirmation": True, "similar_note": {"id": near.id, "title": near.title}, "error": f"A note with a very similar title '{near.title}' already exists (similarity: {ratio:.0%}). Ask the user to confirm before creating a separate entry."}
if not arguments.get("confirmed") and len(note_body.strip()) >= 200:
from fabledassistant.services.embeddings import semantic_search_notes as _ssn
sem_query = f"{note_title}\n{note_body}".strip()
sem_hits = await _ssn(user_id, sem_query, limit=3, threshold=0.90)
sem_hits = await _ssn(user_id, sem_query, limit=3, threshold=0.90, is_task=False)
if sem_hits:
best_score, best_note = sem_hits[0]
item_type = "task" if best_note.status is not None else "note"
return {"success": False, "requires_confirmation": True, "similar_note": {"id": best_note.id, "title": best_note.title}, "error": f"A {item_type} with very similar content exists: '{best_note.title}' (semantic similarity: {best_score:.0%}). Ask the user to confirm before creating a separate entry."}
return {"success": False, "requires_confirmation": True, "similar_note": {"id": best_note.id, "title": best_note.title}, "error": f"A note with very similar content exists: '{best_note.title}' (semantic similarity: {best_score:.0%}). Ask the user to confirm before creating a separate entry."}
note = await create_note(
user_id=user_id,