bvandeusen/FabledScribe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 50 Wiki Activity

fix: fuzzy project resolution and guard against accidental project creation

Browse Source 
- _resolve_project now uses reverse-substring + SequenceMatcher (≥0.55)
  so partial names like "Famous Supply" match "Famous Supply Co." reliably
- create_project runs similar-project checks before the confirmed gate so
  confirmed=true can't bypass them; threshold lowered to 0.55 to catch
  abbreviated names
- Error messages on project-not-found no longer suggest create_project,
  directing the model to list_projects first
- Tool description updated to explicitly prohibit calling create_project
  in response to a lookup failure
- tasks.py: fire-and-forget embedding update on create/update
- briefing_pipeline.py: await is_caldav_configured (was sync call)
- notes.py: remove erroneous duplicate count_query filter

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Van Deusen
2026-03-16 07:44:27 -04:00
parent 0d41b8be34
commit a58dbe79f2
4 changed files with 59 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/fabledassistant/routes/tasks.py
+9
View File
@@ -1,8 +1,11 @@
import asyncio
from quart import Blueprint, jsonify, request
from fabledassistant.auth import login_required, get_current_user_id
from fabledassistant.models.note import TaskPriority, TaskStatus
from fabledassistant.routes.utils import not_found, parse_iso_date
from fabledassistant.services.embeddings import upsert_note_embedding
from fabledassistant.services.notes import (
create_note,
delete_note,
@@ -93,6 +96,9 @@ async def create_task_route():
milestone_id=data.get("milestone_id"),
parent_id=data.get("parent_id"),
)
text = f"{task.title}\n{task.body}".strip() if task.body else (task.title or "")
if text:
asyncio.create_task(upsert_note_embedding(task.id, uid, text))
return jsonify(task.to_dict()), 201
@@ -147,6 +153,9 @@ async def update_task_route(task_id: int):
task = await update_note(uid, task_id, **fields)
if task is None:
return not_found("Task")
text = f"{task.title}\n{task.body}".strip() if task.body else (task.title or "")
if text:
asyncio.create_task(upsert_note_embedding(task.id, uid, text))
return jsonify(task.to_dict())
src/fabledassistant/services/briefing_pipeline.py
+1 -1
View File
@@ -80,7 +80,7 @@ async def _gather_internal(user_id: int) -> dict:
# Calendar events today
calendar_events = []
try:
if is_caldav_configured(user_id):
if await is_caldav_configured(user_id):
events = await list_events(user_id, start=today, end=today)
calendar_events = [
f"{e.get('summary', 'Event')} at {e.get('dtstart', 'unknown time')}"
src/fabledassistant/services/notes.py
-1
View File
@@ -157,7 +157,6 @@ async def list_notes(
if no_project:
query = query.where(Note.project_id.is_(None))
count_query = count_query.where(Note.project_id.is_(None))
count_query = count_query.where(Note.parent_id == parent_id)
sort_col = getattr(Note, sort, Note.updated_at)
if order == "asc":
src/fabledassistant/services/tools.py
+49 -64
View File
@@ -30,6 +30,35 @@ def _schedule_embedding(note_id: int, user_id: int, title: str, body: str) -> No
asyncio.create_task(upsert_note_embedding(note_id, user_id, text))
async def _resolve_project(user_id: int, project_name: str):
"""Exact-then-fuzzy project lookup. Returns the Project or None.
Resolution order:
1. Exact title match (case-insensitive via DB)
2. project_name is a substring of an existing title ("Famous Supply" in "Famous Supply Co.")
3. Existing title is a substring of project_name ("Famous Supply Co." in "Famous Supply Co. UK")
4. SequenceMatcher ratio >= 0.55 (catches abbreviations / partial names)
"""
from fabledassistant.services.projects import get_project_by_title, list_projects
proj = await get_project_by_title(user_id, project_name)
if proj is not None:
return proj
needle = project_name.lower().strip()
all_p = await list_projects(user_id)
# Substring checks first (deterministic)
for p in all_p:
haystack = p.title.lower().strip()
if needle in haystack or haystack in needle:
return p
# SequenceMatcher fallback for abbreviations / partial names
best, best_r = None, 0.0
for p in all_p:
r = SequenceMatcher(None, needle, p.title.lower().strip()).ratio()
if r >= 0.55 and r > best_r:
best, best_r = p, r
return best
# Core tools — always available
_CORE_TOOLS = [
{
@@ -358,7 +387,7 @@ _CORE_TOOLS = [
"type": "function",
"function": {
"name": "create_project",
"description": "Create a new project. IMPORTANT: You MUST ask the user to confirm the project name and intent before calling this tool. Only call with confirmed=true after the user has explicitly approved creation.",
"description": "Create a brand-new project. IMPORTANT: NEVER call this because a project name was not found — always use list_projects first to confirm no similar project exists. Only call this when the user has explicitly asked to create a new project AND confirmed it. Call with confirmed=true only after the user has explicitly approved. The system will block creation if any project with a similar name already exists.",
"parameters": {
"type": "object",
"properties": {
@@ -860,14 +889,9 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
parent_task_name = arguments.get("parent_task")
pre_fields: dict = {}
if project_name:
from fabledassistant.services.projects import get_project_by_title as _gpbt, list_projects as _lp
proj = await _gpbt(user_id, project_name)
proj = await _resolve_project(user_id, project_name)
if proj is None:
all_p = await _lp(user_id)
matches = [p for p in all_p if project_name.lower() in p.title.lower()]
proj = matches[0] if matches else None
if proj is None:
return {"success": False, "error": f"Project '{project_name}' not found. Use list_projects to see available projects or create_project to create one."}
return {"success": False, "error": f"Project '{project_name}' not found. Use list_projects to find the correct project name and retry with the exact name."}
pre_fields["project_id"] = proj.id
if milestone_name:
from fabledassistant.services.milestones import get_milestone_by_title as _gmbt
@@ -942,14 +966,9 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
project_name = arguments.get("project")
project_id = None
if project_name:
from fabledassistant.services.projects import get_project_by_title as _gpbt, list_projects as _lp
proj = await _gpbt(user_id, project_name)
proj = await _resolve_project(user_id, project_name)
if proj is None:
all_p = await _lp(user_id)
matches = [p for p in all_p if project_name.lower() in p.title.lower()]
proj = matches[0] if matches else None
if proj is None:
return {"success": False, "error": f"Project '{project_name}' not found. Use list_projects to see available projects or create_project to create one."}
return {"success": False, "error": f"Project '{project_name}' not found. Use list_projects to find the correct project name and retry with the exact name."}
project_id = proj.id
existing_notes, _ = await list_notes(user_id=user_id, q=note_title, is_task=False, limit=1)
@@ -1042,12 +1061,7 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
if "project" in arguments:
project_name = arguments["project"]
if project_name:
from fabledassistant.services.projects import get_project_by_title as _gpbt, list_projects as _lp
proj = await _gpbt(user_id, project_name)
if proj is None:
all_p = await _lp(user_id)
matches = [p for p in all_p if project_name.lower() in p.title.lower()]
proj = matches[0] if matches else None
proj = await _resolve_project(user_id, project_name)
if proj is None:
return {"success": False, "error": f"Project '{project_name}' not found. Use list_projects to see available projects."}
update_fields["project_id"] = proj.id
@@ -1082,13 +1096,7 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
project_name = arguments.get("project")
milestone_name = arguments.get("milestone")
if project_name:
from fabledassistant.services.projects import get_project_by_title, list_projects as _lp
proj = await get_project_by_title(user_id, project_name)
if proj is None:
all_p = await _lp(user_id)
matches = [p for p in all_p if project_name.lower() in p.title.lower()]
if matches:
proj = matches[0]
proj = await _resolve_project(user_id, project_name)
if proj:
project_id = proj.id
if milestone_name:
@@ -1150,12 +1158,7 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
# Resolve project name → id
search_project_id: int | None = None
if project_name:
from fabledassistant.services.projects import get_project_by_title as _gpbt2, list_projects as _lp2
proj = await _gpbt2(user_id, project_name)
if proj is None:
all_p = await _lp2(user_id)
matches = [p for p in all_p if project_name.lower() in p.title.lower()]
proj = matches[0] if matches else None
proj = await _resolve_project(user_id, project_name)
if proj:
search_project_id = proj.id
@@ -1334,12 +1337,7 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
project_id = None
project_name = arguments.get("project")
if project_name:
from fabledassistant.services.projects import get_project_by_title as _gpbt, list_projects as _lp
proj = await _gpbt(user_id, project_name)
if proj is None:
all_p = await _lp(user_id)
matches = [p for p in all_p if project_name.lower() in p.title.lower()]
proj = matches[0] if matches else None
proj = await _resolve_project(user_id, project_name)
if proj is None:
return {"success": False, "error": f"Project '{project_name}' not found."}
project_id = proj.id
@@ -1412,7 +1410,6 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
}
elif tool_name == "create_milestone":
from fabledassistant.services.projects import get_project_by_title as _gpbt, list_projects as _lp
from fabledassistant.services.milestones import create_milestone as _cm, get_project_milestone_summary
project_name = arguments.get("project", "")
ms_title_pre = arguments.get("title", "Untitled Milestone")
@@ -1420,13 +1417,9 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
return {"success": False, "error": "project is required"}
if not arguments.get("confirmed"):
return {"success": False, "requires_confirmation": True, "error": f"Milestone creation requires explicit user confirmation. Ask the user: 'Shall I create a milestone titled \"{ms_title_pre}\" in project \"{project_name}\"?' Then retry with confirmed=true if they agree."}
proj = await _gpbt(user_id, project_name)
proj = await _resolve_project(user_id, project_name)
if proj is None:
all_p = await _lp(user_id)
matches = [p for p in all_p if project_name.lower() in p.title.lower()]
proj = matches[0] if matches else None
if proj is None:
return {"success": False, "error": f"Project '{project_name}' not found. Use list_projects to see available projects or create_project to create one."}
return {"success": False, "error": f"Project '{project_name}' not found. Use list_projects to find the correct project name and retry with the exact name."}
ms_title = ms_title_pre
from fabledassistant.services.milestones import get_milestone_by_title as _gmbt2, list_milestones as _lms
existing_ms = await _gmbt2(user_id, proj.id, ms_title)
@@ -1445,17 +1438,12 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
return {"success": True, "type": "milestone", "data": ms.to_dict()}
elif tool_name == "update_milestone":
from fabledassistant.services.projects import get_project_by_title as _gpbt, list_projects as _lp
from fabledassistant.services.milestones import get_milestone_by_title as _gmbt, update_milestone as _um
project_name = arguments.get("project", "")
milestone_name = arguments.get("milestone", "")
if not project_name or not milestone_name:
return {"success": False, "error": "Both project and milestone are required"}
proj = await _gpbt(user_id, project_name)
if proj is None:
all_p = await _lp(user_id)
matches = [p for p in all_p if project_name.lower() in p.title.lower()]
proj = matches[0] if matches else None
proj = await _resolve_project(user_id, project_name)
if proj is None:
return {"success": False, "error": f"Project '{project_name}' not found"}
ms = await _gmbt(user_id, proj.id, milestone_name)
@@ -1474,14 +1462,9 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
return {"success": True, "type": "milestone", "data": updated.to_dict()}
elif tool_name == "list_milestones":
from fabledassistant.services.projects import get_project_by_title as _gpbt, list_projects as _lp
from fabledassistant.services.milestones import get_project_milestone_summary
project_name = arguments.get("project", "")
proj = await _gpbt(user_id, project_name)
if proj is None:
all_p = await _lp(user_id)
matches = [p for p in all_p if project_name.lower() in p.title.lower()]
proj = matches[0] if matches else None
proj = await _resolve_project(user_id, project_name)
if proj is None:
return {"success": False, "error": f"No project found matching '{project_name}'"}
summary = await get_project_milestone_summary(user_id, proj.id)
@@ -1520,15 +1503,17 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
elif tool_name == "create_project":
from fabledassistant.services.projects import create_project as _create_project, get_project_by_title as _gpbt, list_projects as _lp
proj_title = arguments.get("title", "New Project")
if not arguments.get("confirmed"):
return {"success": False, "requires_confirmation": True, "error": f"Project creation requires explicit user confirmation. Ask the user: 'Shall I create a new project titled \"{proj_title}\"?' Then retry with confirmed=true if they agree."}
# Always check for existing/similar projects first — even when confirmed=true
existing_proj = await _gpbt(user_id, proj_title)
if existing_proj is not None:
return {"success": False, "error": f"A project titled '{existing_proj.title}' already exists (id: {existing_proj.id}). Use update_project to modify it instead of creating a duplicate."}
return {"success": False, "error": f"A project titled '{existing_proj.title}' already exists (id: {existing_proj.id}). Use that project instead of creating a duplicate."}
all_projects = await _lp(user_id)
near_proj, ratio = _fuzzy_title_match(proj_title, all_projects)
# Broad fuzzy check (0.55) catches partial names and abbreviations
near_proj, ratio = _fuzzy_title_match(proj_title, all_projects, threshold=0.55)
if near_proj is not None:
return {"success": False, "error": f"A project with a very similar title '{near_proj.title}' already exists (id: {near_proj.id}, similarity: {ratio:.0%}). Use update_project to modify it instead of creating a duplicate."}
return {"success": False, "requires_confirmation": True, "error": f"A project with a similar name '{near_proj.title}' already exists (id: {near_proj.id}, similarity: {ratio:.0%}). This is likely the project you meant — use that project's name instead. Only call create_project again if you are certain this should be a completely new, separate project with a different confirmed=true."}
if not arguments.get("confirmed"):
return {"success": False, "requires_confirmation": True, "error": f"Project creation requires explicit user confirmation. Ask the user: 'Shall I create a new project titled \"{proj_title}\"?' Then retry with confirmed=true if they agree."}
project = await _create_project(
user_id,
title=proj_title,