"""Project and milestone tools.""" from __future__ import annotations from fabledassistant.services.tools._helpers import fuzzy_title_match, resolve_project from fabledassistant.services.tools._registry import tool @tool( name="create_project", description="Create a new project. Use list_projects first to check for duplicates. Only call after user has explicitly confirmed.", parameters={ "title": {"type": "string", "description": "Project title"}, "description": {"type": "string", "description": "Brief description"}, "goal": {"type": "string", "description": "The goal or desired outcome"}, "color": {"type": "string", "description": "Optional hex color for the project (e.g. '#6366f1')"}, "confirmed": {"type": "boolean", "description": "Must be true — only set after the user has explicitly confirmed they want this project created."}, }, required=["title", "confirmed"], ) async def create_project_tool(*, user_id, arguments, **_ctx): 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") 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 that project instead of creating a duplicate."} all_projects = await _lp(user_id) near_proj, ratio = fuzzy_title_match(proj_title, all_projects, threshold=0.55) if near_proj is not None: 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, description=arguments.get("description", ""), goal=arguments.get("goal", ""), color=arguments.get("color") or None, ) return {"success": True, "type": "project", "data": project.to_dict()} @tool( name="list_projects", description="List the user's projects. Use when asked about projects, initiatives, or campaigns.", parameters={ "status": {"type": "string", "enum": ["active", "completed", "archived"], "description": "Filter by status (omit for all)"}, }, read_only=True, briefing=True, ) async def list_projects_tool(*, user_id, arguments, **_ctx): from fabledassistant.services.projects import list_projects as _list_projects projects = await _list_projects(user_id, status=arguments.get("status")) return { "success": True, "type": "projects_list", "data": {"count": len(projects), "projects": [p.to_dict() for p in projects]}, } @tool( name="get_project", description="Get details and summary of a specific project.", parameters={ "query": {"type": "string", "description": "Project name or keyword to find it"}, }, required=["query"], read_only=True, briefing=True, ) async def get_project_tool(*, user_id, arguments, **_ctx): from fabledassistant.services.projects import get_project_by_title as _gpbt, get_project_summary as _gps, list_projects as _lp query = arguments.get("query", "") project = await _gpbt(user_id, query) if project is None: all_projects = await _lp(user_id) matches = [p for p in all_projects if query.lower() in p.title.lower()] if matches: project = matches[0] if project is None: return {"success": False, "error": f"No project found matching '{query}'"} summary = await _gps(user_id, project.id) data = project.to_dict() data["summary"] = summary return {"success": True, "type": "project_detail", "data": data} @tool( name="update_project", description="Update a project's title, status, description, goal, or color.", parameters={ "query": {"type": "string", "description": "Project name to find"}, "title": {"type": "string", "description": "New project title"}, "status": {"type": "string", "enum": ["active", "completed", "archived"]}, "description": {"type": "string"}, "goal": {"type": "string"}, "color": {"type": "string", "description": "Hex color (e.g. '#6366f1'), or empty string to clear"}, }, required=["query"], ) async def update_project_tool(*, user_id, arguments, **_ctx): from fabledassistant.services.projects import get_project_by_title as _gpbt, update_project as _up, list_projects as _lp query = arguments.get("query", "") project = await _gpbt(user_id, query) if project is None: all_projects = await _lp(user_id) matches = [p for p in all_projects if query.lower() in p.title.lower()] if matches: project = matches[0] if project is None: return {"success": False, "error": f"No project found matching '{query}'"} fields = {} for k in ("title", "status", "description", "goal"): if k in arguments: fields[k] = arguments[k] if "color" in arguments: fields["color"] = arguments["color"] or None updated = await _up(user_id, project.id, **fields) return {"success": True, "type": "project_updated", "data": updated.to_dict()} @tool( name="search_projects", description="Search for projects by name, description, or theme. Use this to find which project a user is referring to and get its ID for set_rag_scope.", parameters={ "query": {"type": "string", "description": "Search query — project name, topic, or theme"}, }, required=["query"], read_only=True, briefing=True, ) async def search_projects_tool(*, user_id, arguments, **_ctx): from difflib import SequenceMatcher from fabledassistant.services.projects import list_projects query = str(arguments.get("query", "")).lower() projects = await list_projects(user_id) scored: list[tuple[float, object]] = [] for p in projects: combined = f"{p.title} {p.description or ''} {p.auto_summary or ''}".lower() base_score = SequenceMatcher(None, query, combined).ratio() query_words = set(query.split()) overlap = sum(1 for w in query_words if w in combined) score = base_score + overlap * 0.05 scored.append((score, p)) scored.sort(key=lambda x: x[0], reverse=True) results = [] for score, p in scored[:5]: results.append({ "id": p.id, "title": p.title, "summary_snippet": (p.auto_summary or p.description or "")[:200], "score": round(score, 3), }) return {"type": "projects_list", "data": {"projects": results}} @tool( name="create_milestone", description="Create a milestone inside a project. Confirm name and project with user before calling.", parameters={ "project": {"type": "string", "description": "Project title"}, "title": {"type": "string", "description": "Milestone title"}, "description": {"type": "string", "description": "Optional description"}, "confirmed": {"type": "boolean", "description": "Must be true — only set after the user has explicitly confirmed they want this milestone created."}, }, required=["project", "title", "confirmed"], ) async def create_milestone_tool(*, user_id, arguments, **_ctx): from fabledassistant.services.milestones import create_milestone as _cm, get_milestone_by_title as _gmbt2, list_milestones as _lms project_name = arguments.get("project", "") ms_title = arguments.get("title", "Untitled Milestone") if not project_name: 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}\" in project \"{project_name}\"?' Then retry with confirmed=true if they agree."} 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 find the correct project name and retry with the exact name."} existing_ms = await _gmbt2(user_id, proj.id, ms_title) if existing_ms is not None: return {"success": False, "error": f"A milestone titled '{existing_ms.title}' already exists in '{proj.title}' (id: {existing_ms.id}). Use update_milestone to modify it instead."} all_ms = await _lms(user_id, proj.id) near_ms, ratio = fuzzy_title_match(ms_title, all_ms) if near_ms is not None: return {"success": False, "error": f"A milestone with a very similar title '{near_ms.title}' already exists in '{proj.title}' (id: {near_ms.id}, similarity: {ratio:.0%}). Use update_milestone to modify it instead."} ms = await _cm(user_id, proj.id, title=ms_title, description=arguments.get("description")) return {"success": True, "type": "milestone", "data": ms.to_dict()} @tool( name="update_milestone", description="Update the title, description, or status of an existing milestone.", parameters={ "project": {"type": "string", "description": "Project title the milestone belongs to"}, "milestone": {"type": "string", "description": "Current milestone title to look up"}, "title": {"type": "string", "description": "New title (omit to keep current)"}, "description": {"type": "string", "description": "New description (omit to keep current)"}, "status": {"type": "string", "enum": ["active", "done"], "description": "New status: 'active' (in progress) or 'done' (completed)"}, }, required=["project", "milestone"], ) async def update_milestone_tool(*, user_id, arguments, **_ctx): 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 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) if ms is None: return {"success": False, "error": f"Milestone '{milestone_name}' not found in project '{proj.title}'. Use list_milestones to see available milestones."} fields: dict[str, object] = {} if "title" in arguments: fields["title"] = arguments["title"] if "description" in arguments: fields["description"] = arguments["description"] if "status" in arguments: fields["status"] = arguments["status"] if not fields: return {"success": False, "error": "No fields to update — provide at least one of: title, description, status"} updated = await _um(user_id, ms.id, **fields) return {"success": True, "type": "milestone", "data": updated.to_dict()} @tool( name="list_milestones", description="List milestones for a project with their progress percentages.", parameters={ "project": {"type": "string", "description": "Project name to look up"}, }, required=["project"], read_only=True, briefing=True, ) async def list_milestones_tool(*, user_id, arguments, **_ctx): from fabledassistant.services.milestones import get_project_milestone_summary project_name = arguments.get("project", "") 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) return { "success": True, "type": "milestones_list", "data": {"project": proj.title, "count": len(summary), "milestones": summary}, }