bvandeusen
103db883ad
create_note tool: - New 'description' parameter accepted and forwarded to the service. - When status is set (creating a task), 'body' is dropped before the service call. Task bodies are owned by the consolidation pipeline. update_note tool: - New 'description' parameter; routed through update_fields. - When the resolved target has is_task=True and 'body' is in the arguments, the call errors with a message nudging toward log_work or description. Knowledge notes are unaffected. HTTP routes (POST/PATCH/PUT /api/notes) accept body freely — the restriction is only at the LLM tool layer.
451 lines
21 KiB
Python
451 lines
21 KiB
Python
"""Note tools: create, update, get, list, delete, search."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from fabledassistant.services.notes import create_note, delete_note, get_note_by_title, list_notes, update_note
|
|
from fabledassistant.services.tag_suggestions import suggest_tags
|
|
from fabledassistant.services.tools._helpers import (
|
|
check_duplicate,
|
|
parse_due_date,
|
|
resolve_project,
|
|
schedule_embedding,
|
|
)
|
|
from fabledassistant.services.tools._registry import tool
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@tool(
|
|
name="create_note",
|
|
description=(
|
|
"Create a new note or task. "
|
|
"For a knowledge note, omit the status field. "
|
|
"For an actionable task (todo, reminder, action item), set status to 'todo'. "
|
|
"Use this whenever the user asks to write down, save, record, or add a task/todo."
|
|
),
|
|
parameters={
|
|
"title": {"type": "string", "description": "The title"},
|
|
"body": {"type": "string", "description": "Content in markdown. NOTE: when status is set (creating a task), body is ignored — task bodies are auto-maintained from work logs. Use `description` to provide the goal/context for tasks."},
|
|
"description": {"type": "string", "description": "User-stated goal or initial context for a task. Read-only context for the auto-summary pipeline. Ignored when status is omitted (knowledge note)."},
|
|
"tags": {"type": "array", "items": {"type": "string"}, "description": 'Tags (without # prefix, hyphens for multi-word: ["science-fiction", "story/idea"]). Do NOT embed #tags in the body.'},
|
|
"project": {"type": "string", "description": "Optional project name. Only set this if the user explicitly named a project. Do NOT infer a project from the content or context."},
|
|
"status": {"type": "string", "enum": ["todo", "in_progress", "done", "cancelled"], "description": "Set to 'todo' to create a task. Omit entirely for a knowledge note."},
|
|
"due_date": {"type": "string", "description": "Due date in YYYY-MM-DD format (tasks only)"},
|
|
"priority": {"type": "string", "enum": ["none", "low", "medium", "high"], "description": "Priority level (tasks only, default: none)"},
|
|
"parent_task": {"type": "string", "description": "Title of a parent task to make this a sub-task of"},
|
|
"milestone": {"type": "string", "description": "Milestone title within the project to assign to"},
|
|
"recurrence_rule": {"type": "object", "description": 'Recurrence rule (tasks only). Interval form: {"type":"interval","every":3,"unit":"month"} (units: day/week/month/year). Calendar form: {"type":"calendar","unit":"month","day_of_month":1}. Annual: add "month":6.'},
|
|
"confirmed": {"type": "boolean", "description": "Set to true only when the user has explicitly confirmed creation after being warned about a similar existing item."},
|
|
},
|
|
required=["title"],
|
|
)
|
|
async def create_note_tool(*, user_id, arguments, **_ctx):
|
|
title = arguments.get("title", "Untitled")
|
|
body = arguments.get("body", "")
|
|
description = arguments.get("description")
|
|
tags = arguments.get("tags", [])
|
|
if not isinstance(title, str):
|
|
return {"success": False, "error": "title must be a string. Call create_note once per item."}
|
|
if not isinstance(body, str):
|
|
body = ""
|
|
if not isinstance(tags, list):
|
|
tags = []
|
|
|
|
is_task = "status" in arguments and arguments["status"] is not None
|
|
status = arguments.get("status", "todo") if is_task else None
|
|
|
|
# Task bodies are auto-maintained by the consolidation pipeline; drop any
|
|
# body argument arriving with a task creation so it never lands in the DB.
|
|
if is_task:
|
|
body = ""
|
|
|
|
project_name = arguments.get("project")
|
|
milestone_name = arguments.get("milestone")
|
|
parent_task_name = arguments.get("parent_task")
|
|
project_id = None
|
|
milestone_id = None
|
|
|
|
if project_name:
|
|
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."}
|
|
project_id = proj.id
|
|
if milestone_name:
|
|
from fabledassistant.services.milestones import get_milestone_by_title as _gmbt
|
|
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."}
|
|
milestone_id = ms.id
|
|
|
|
dup = await check_duplicate(user_id, title, body, is_task=is_task, confirmed=bool(arguments.get("confirmed")))
|
|
if dup is not None:
|
|
return dup
|
|
|
|
note = await create_note(
|
|
user_id=user_id,
|
|
title=title,
|
|
body=body,
|
|
description=description,
|
|
tags=tags,
|
|
status=status,
|
|
priority=arguments.get("priority", "none") if is_task else None,
|
|
due_date=parse_due_date(arguments.get("due_date")) if is_task else None,
|
|
recurrence_rule=arguments.get("recurrence_rule") if is_task else None,
|
|
project_id=project_id,
|
|
milestone_id=milestone_id,
|
|
)
|
|
|
|
if parent_task_name:
|
|
parent_notes, _ = await list_notes(user_id=user_id, q=parent_task_name, is_task=True, limit=1)
|
|
if parent_notes:
|
|
note = await update_note(user_id, note.id, parent_id=parent_notes[0].id)
|
|
|
|
suggested = await suggest_tags(user_id, title, body)
|
|
schedule_embedding(note.id, user_id, title, body)
|
|
|
|
if is_task:
|
|
return {
|
|
"success": True,
|
|
"type": "task",
|
|
"data": {
|
|
"id": note.id,
|
|
"title": note.title,
|
|
"status": note.status,
|
|
"priority": note.priority,
|
|
"due_date": str(note.due_date) if note.due_date else None,
|
|
"project_id": note.project_id,
|
|
"milestone_id": note.milestone_id,
|
|
"parent_id": note.parent_id,
|
|
},
|
|
"suggested_tags": suggested,
|
|
}
|
|
return {
|
|
"success": True,
|
|
"type": "note",
|
|
"data": {"id": note.id, "title": note.title, "project_id": note.project_id},
|
|
"suggested_tags": suggested,
|
|
}
|
|
|
|
|
|
@tool(
|
|
name="update_note",
|
|
description="Update an existing note or task — content, title, status, priority, or due date. Use for edits, marking tasks done, changing priority. Never use create_note for existing notes.",
|
|
parameters={
|
|
"query": {"type": "string", "description": "Title or keyword to find the note or task to update"},
|
|
"body": {"type": "string", "description": "New note content in markdown (omit if only updating task fields). REJECTED on tasks — task bodies are auto-maintained from work logs; use `log_work` to record progress or `description` to revise the goal."},
|
|
"description": {"type": "string", "description": "Update the user-stated goal/context (tasks). Distinct from `body` (machine-maintained on tasks)."},
|
|
"title": {"type": "string", "description": "Optional new title"},
|
|
"mode": {"type": "string", "enum": ["replace", "append"], "description": "How to apply the new body: 'replace' overwrites existing content (default), 'append' adds after existing content"},
|
|
"status": {"type": "string", "enum": ["todo", "in_progress", "done", "cancelled"], "description": "New task status. Use to mark a task done, start it, cancel it, etc."},
|
|
"priority": {"type": "string", "enum": ["none", "low", "medium", "high"], "description": "New task priority"},
|
|
"due_date": {"type": "string", "description": "New due date in YYYY-MM-DD format"},
|
|
"tags": {"type": "array", "items": {"type": "string"}, "description": "Tags to apply (see tag_mode for how they're applied)"},
|
|
"tag_mode": {"type": "string", "enum": ["add", "remove", "replace"], "description": "'add' appends tags without duplicates (default), 'remove' removes listed tags, 'replace' sets tags to exactly this list (destructive — avoid unless explicitly requested)"},
|
|
"project": {"type": "string", "description": "Optional project name to assign this note/task to (set to empty string to remove project)"},
|
|
"milestone": {"type": "string", "description": "Optional milestone title within the project to assign this task to (set to empty string to remove milestone)"},
|
|
"recurrence_rule": {"type": "object", "description": 'Set or update recurrence. Interval form: {"type":"interval","every":3,"unit":"month"}. Calendar form: {"type":"calendar","unit":"month","day_of_month":1}. Pass null to remove.'},
|
|
},
|
|
required=["query"],
|
|
)
|
|
async def update_note_tool(*, user_id, arguments, **_ctx):
|
|
query = arguments.get("query", "")
|
|
new_body = arguments.get("body", "")
|
|
new_title = arguments.get("title")
|
|
mode = arguments.get("mode", "replace")
|
|
|
|
note = await get_note_by_title(user_id, query)
|
|
if note is None:
|
|
notes, _ = await list_notes(user_id=user_id, q=query, limit=5)
|
|
note_only = [n for n in notes if n.status is None]
|
|
candidates = note_only or notes
|
|
if not candidates:
|
|
return {"success": False, "error": f"No note found matching '{query}'."}
|
|
note = candidates[0]
|
|
|
|
# Schema-level separation: task bodies are auto-maintained from work logs.
|
|
# Reject body writes here rather than silently dropping so the LLM gets
|
|
# nudged toward log_work / description.
|
|
if note.is_task and arguments.get("body"):
|
|
return {
|
|
"success": False,
|
|
"error": (
|
|
"Cannot write to `body` on a task — the body is auto-maintained "
|
|
"from work logs. Use the `log_work` tool to record progress, or "
|
|
"update `description` to revise the goal."
|
|
),
|
|
}
|
|
|
|
update_fields: dict = {}
|
|
if new_title:
|
|
update_fields["title"] = new_title
|
|
if new_body:
|
|
if mode == "append" and note.body:
|
|
update_fields["body"] = note.body + "\n\n" + new_body
|
|
else:
|
|
update_fields["body"] = new_body
|
|
if "description" in arguments:
|
|
update_fields["description"] = arguments["description"]
|
|
if "status" in arguments:
|
|
update_fields["status"] = arguments["status"]
|
|
if "priority" in arguments:
|
|
update_fields["priority"] = arguments["priority"]
|
|
if "due_date" in arguments:
|
|
update_fields["due_date"] = parse_due_date(arguments["due_date"])
|
|
if "tags" in arguments:
|
|
new_tags = arguments["tags"]
|
|
if isinstance(new_tags, list):
|
|
tag_mode = arguments.get("tag_mode", "add")
|
|
if tag_mode == "add":
|
|
existing = list(note.tags or [])
|
|
for t in new_tags:
|
|
if t not in existing:
|
|
existing.append(t)
|
|
update_fields["tags"] = existing
|
|
elif tag_mode == "remove":
|
|
existing = list(note.tags or [])
|
|
update_fields["tags"] = [t for t in existing if t not in new_tags]
|
|
else:
|
|
update_fields["tags"] = new_tags
|
|
|
|
if "project" in arguments:
|
|
project_name = arguments["project"]
|
|
if project_name:
|
|
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
|
|
else:
|
|
update_fields["project_id"] = None
|
|
update_fields["milestone_id"] = None
|
|
|
|
if "milestone" in arguments:
|
|
milestone_name = arguments["milestone"]
|
|
if milestone_name:
|
|
ref_project_id = update_fields.get("project_id") or note.project_id
|
|
if ref_project_id is None:
|
|
return {"success": False, "error": "Cannot assign a milestone without a project. Set project first."}
|
|
from fabledassistant.services.milestones import get_milestone_by_title as _gmbt_upd
|
|
ms = await _gmbt_upd(user_id, ref_project_id, milestone_name)
|
|
if ms is None:
|
|
return {"success": False, "error": f"Milestone '{milestone_name}' not found. Use list_milestones to see available milestones."}
|
|
update_fields["milestone_id"] = ms.id
|
|
else:
|
|
update_fields["milestone_id"] = None
|
|
|
|
updated = await update_note(user_id, note.id, **update_fields)
|
|
if updated is None:
|
|
return {"success": False, "error": "Failed to update note."}
|
|
|
|
suggested = await suggest_tags(user_id, updated.title, updated.body or "")
|
|
schedule_embedding(updated.id, user_id, updated.title, updated.body or "")
|
|
item_type = "task" if updated.status is not None else "note"
|
|
return {
|
|
"success": True,
|
|
"type": "note_updated",
|
|
"updated": f"{item_type} '{updated.title}' (id: {updated.id})",
|
|
"data": {
|
|
"id": updated.id,
|
|
"title": updated.title,
|
|
"item_type": item_type,
|
|
"status": updated.status,
|
|
"tags": updated.tags or [],
|
|
"project_id": updated.project_id,
|
|
},
|
|
"suggested_tags": suggested,
|
|
}
|
|
|
|
|
|
@tool(
|
|
name="search_notes",
|
|
description=(
|
|
"Find notes or tasks by meaning. Returns a ranked list of matches with "
|
|
"short previews. Use this when looking for items on a topic but you "
|
|
"don't know the exact title. For the full body of a known item, use "
|
|
"read_note instead. Do not include 'task', 'note', or 'project' in the "
|
|
"`query` — use the `type` and `project` parameters instead."
|
|
),
|
|
parameters={
|
|
"query": {"type": "string", "description": "A natural-language description of what you're looking for — concepts, themes, topics, or keywords"},
|
|
"type": {"type": "string", "enum": ["note", "task"], "description": "Restrict results to only notes or only tasks. Omit to search both."},
|
|
"project": {"type": "string", "description": "Optional project name to restrict search to notes in that project"},
|
|
},
|
|
required=["query"],
|
|
read_only=True,
|
|
briefing=True,
|
|
)
|
|
async def search_notes_tool(*, user_id, arguments, **_ctx):
|
|
query = arguments.get("query", "")
|
|
type_filter = arguments.get("type")
|
|
project_name = arguments.get("project")
|
|
|
|
is_task: bool | None = None
|
|
if type_filter == "note":
|
|
is_task = False
|
|
elif type_filter == "task":
|
|
is_task = True
|
|
|
|
search_project_id: int | None = None
|
|
if project_name:
|
|
proj = await resolve_project(user_id, project_name)
|
|
if proj:
|
|
search_project_id = proj.id
|
|
|
|
results_map: dict[int, dict] = {}
|
|
|
|
try:
|
|
from fabledassistant.services.embeddings import semantic_search_notes as _ssn
|
|
sem_results = await _ssn(
|
|
user_id, query, limit=8, threshold=0.40,
|
|
project_id=search_project_id,
|
|
)
|
|
for score, n in sem_results:
|
|
n_is_task = n.status is not None
|
|
if is_task is True and not n_is_task:
|
|
continue
|
|
if is_task is False and n_is_task:
|
|
continue
|
|
results_map[n.id] = {
|
|
"id": n.id,
|
|
"title": n.title,
|
|
"type": "task" if n_is_task else "note",
|
|
"status": n.status,
|
|
"relevance": f"{round(score * 100)}%",
|
|
"preview": (n.body[:300] + "\u2026") if n.body and len(n.body) > 300 else (n.body or ""),
|
|
}
|
|
logger.debug("search_notes semantic: %d results for %r", len(results_map), query)
|
|
except Exception:
|
|
logger.debug("Semantic search unavailable in search_notes, using keyword only", exc_info=True)
|
|
|
|
try:
|
|
kw_notes, _ = await list_notes(
|
|
user_id=user_id, q=query, is_task=is_task,
|
|
project_id=search_project_id, limit=8,
|
|
)
|
|
for n in kw_notes:
|
|
if n.id not in results_map:
|
|
results_map[n.id] = {
|
|
"id": n.id,
|
|
"title": n.title,
|
|
"type": "task" if n.status is not None else "note",
|
|
"status": n.status,
|
|
"relevance": "keyword",
|
|
"preview": (n.body[:300] + "\u2026") if n.body and len(n.body) > 300 else (n.body or ""),
|
|
}
|
|
except Exception:
|
|
logger.warning("Keyword fallback in search_notes failed", exc_info=True)
|
|
|
|
results = list(results_map.values())[:8]
|
|
return {
|
|
"success": True,
|
|
"type": "search",
|
|
"data": {"query": query, "count": len(results), "results": results},
|
|
}
|
|
|
|
|
|
@tool(
|
|
name="read_note",
|
|
description="Read the full content of one specific note or task. Use when you know which item you want and need its complete body text. For discovering items by topic, use search_notes instead.",
|
|
parameters={
|
|
"query": {"type": "string", "description": "Title or keyword to identify the note or task"},
|
|
},
|
|
required=["query"],
|
|
read_only=True,
|
|
briefing=True,
|
|
)
|
|
async def get_note_tool(*, user_id, arguments, **_ctx):
|
|
query = arguments.get("query", "")
|
|
note = await get_note_by_title(user_id, query)
|
|
if note is None:
|
|
notes, _ = await list_notes(user_id=user_id, q=query, limit=3)
|
|
if not notes:
|
|
return {"success": False, "error": f"No note found matching '{query}'."}
|
|
note = notes[0]
|
|
return {
|
|
"success": True,
|
|
"type": "note_content",
|
|
"data": {"id": note.id, "title": note.title, "body": note.body or "", "tags": note.tags or []},
|
|
}
|
|
|
|
|
|
@tool(
|
|
name="list_notes",
|
|
description="Browse recent notes (not tasks) with optional filters. Returns titles and short previews. Use when the user asks 'show my notes' or wants to browse by tag, project, or recency. For tasks use list_tasks; for topic search use search_notes.",
|
|
parameters={
|
|
"q": {"type": "string", "description": "Optional keyword filter"},
|
|
"tags": {"type": "array", "items": {"type": "string"}, "description": "Filter notes that have ALL of these tags"},
|
|
"project": {"type": "string", "description": "Filter notes by project name"},
|
|
"sort": {"type": "string", "enum": ["updated_at", "created_at", "title"], "description": "Sort field (default: updated_at)"},
|
|
"limit": {"type": "integer", "description": "Maximum number of notes to return (default 10)"},
|
|
},
|
|
read_only=True,
|
|
briefing=True,
|
|
)
|
|
async def list_notes_tool(*, user_id, arguments, **_ctx):
|
|
tags_raw = arguments.get("tags")
|
|
tags = tags_raw if isinstance(tags_raw, list) else None
|
|
project_id = None
|
|
project_name = arguments.get("project")
|
|
if project_name:
|
|
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
|
|
notes, total = await list_notes(
|
|
user_id=user_id,
|
|
q=arguments.get("q") or arguments.get("query"),
|
|
tags=tags,
|
|
is_task=False,
|
|
project_id=project_id,
|
|
sort=arguments.get("sort", "updated_at"),
|
|
order="desc",
|
|
limit=int(arguments.get("limit", 10)),
|
|
)
|
|
results = [
|
|
{
|
|
"id": n.id,
|
|
"title": n.title,
|
|
"tags": n.tags or [],
|
|
"updated_at": n.updated_at.isoformat(),
|
|
"preview": (n.body[:200] + "...") if n.body and len(n.body) > 200 else (n.body or ""),
|
|
}
|
|
for n in notes
|
|
]
|
|
return {
|
|
"success": True,
|
|
"type": "notes_list",
|
|
"data": {"total": total, "count": len(results), "results": results},
|
|
}
|
|
|
|
|
|
@tool(
|
|
name="delete_note",
|
|
description="Delete any item from the user's knowledge base permanently — notes, tasks, persons (created via save_person), places (created via save_place), and lists (created via create_list) are all stored as notes and use this single delete tool. Use ONLY when the user explicitly asks to delete or remove an item. Always confirm with the user first — this cannot be undone.",
|
|
parameters={
|
|
"query": {"type": "string", "description": "Title or keyword to find the item to delete (works for notes, tasks, persons, places, and lists)"},
|
|
"confirmed": {"type": "boolean", "description": "Must be true — only set after the user has explicitly confirmed they want this item deleted."},
|
|
},
|
|
required=["query"],
|
|
)
|
|
async def delete_note_tool(*, user_id, arguments, **_ctx):
|
|
query = arguments.get("query", "")
|
|
note = await get_note_by_title(user_id, query)
|
|
if note is None:
|
|
notes, _ = await list_notes(user_id=user_id, q=query, limit=3)
|
|
if not notes:
|
|
return {"success": False, "error": f"No note or task found matching '{query}'."}
|
|
note = notes[0]
|
|
if not arguments.get("confirmed"):
|
|
item_type = "task" if note.status is not None else "note"
|
|
return {
|
|
"success": False,
|
|
"requires_confirmation": True,
|
|
"error": f"Deleting {item_type} '{note.title}' is permanent. Ask the user to confirm, then retry with confirmed=true.",
|
|
}
|
|
deleted = await delete_note(user_id, note.id)
|
|
if not deleted:
|
|
return {"success": False, "error": "Failed to delete."}
|
|
item_type = "task" if note.status is not None else "note"
|
|
return {"success": True, "type": f"{item_type}_deleted", "data": {"id": note.id, "title": note.title}}
|