"""Timeline agent — project milestone management (list, create, update, delete).""" from __future__ import annotations import re from typing import Any from langchain_core.tools import tool from app.core.ws_context import execute_on_client _UUID_RE = re.compile( r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$" ) def _is_uuid(value: str) -> bool: return bool(_UUID_RE.match(value)) @tool async def list_timelines(project_id: str = "") -> str: """List timelines. Provide project_id to scope to a specific project.""" normalized_project_id = project_id if (project_id and _is_uuid(project_id)) else "" result = await execute_on_client( action="select", table="timelines", filters={"projectId": normalized_project_id or None}, ) rows = result.get("rows", []) if not rows: return "No timelines found." lines = [f"- {r['title']} (date: {r['date']}, id: {r['id']})" for r in rows] return f"Found {len(rows)} timeline(s):\n" + "\n".join(lines) @tool async def create_timeline( project_id: str, title: str, date: int, is_ai_suggested: int = 0, ) -> str: """Create a project timeline (milestone). project_id: REQUIRED UUID of the parent project title: descriptive name for the milestone date: Unix timestamp in milliseconds is_ai_suggested: 1 if proactively suggested, 0 if user-requested """ result = await execute_on_client( action="insert", table="timelines", data={ "projectId": project_id, "title": title, "date": date, "isAiSuggested": is_ai_suggested, }, ) row = result["row"] return f"Timeline created: '{row['title']}' (id: {row['id']}, date: {row['date']})" @tool async def update_timeline( timeline_id: str, title: str = "", date: int = -1, ) -> str: """Update a timeline. Only pass fields that should change. timeline_id: UUID of the timeline (required) date: -1 means unchanged; any other value sets the new date (ms timestamp) """ updates: dict[str, Any] = {} if title: updates["title"] = title if date != -1: updates["date"] = date result = await execute_on_client( action="update", table="timelines", data={"id": timeline_id, "updates": updates}, ) row = result["row"] return f"Timeline updated: '{row['title']}' (id: {row['id']})" @tool async def delete_timeline(timeline_id: str) -> str: """Delete a timeline permanently by its UUID.""" await execute_on_client(action="delete", table="timelines", data={"id": timeline_id}) return f"Timeline {timeline_id} deleted." TIMELINE_TOOLS: list[Any] = [ list_timelines, create_timeline, update_timeline, delete_timeline, ]