tasks_tools.py

"""
Google Tasks MCP Tools

This module provides MCP tools for interacting with Google Tasks API.
"""

import asyncio
import logging
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional

from googleapiclient.errors import HttpError  # type: ignore
from mcp import Resource

from auth.service_decorator import require_google_service
from core.server import server
from core.utils import handle_http_errors

logger = logging.getLogger(__name__)

LIST_TASKS_MAX_RESULTS_DEFAULT = 20
LIST_TASKS_MAX_RESULTS_MAX = 10_000
LIST_TASKS_MAX_POSITION = "99999999999999999999"


class StructuredTask:
    def __init__(self, task: Dict[str, str], is_placeholder_parent: bool) -> None:
        self.id = task["id"]
        self.title = task.get("title", None)
        self.status = task.get("status", None)
        self.due = task.get("due", None)
        self.notes = task.get("notes", None)
        self.updated = task.get("updated", None)
        self.completed = task.get("completed", None)
        self.is_placeholder_parent = is_placeholder_parent
        self.subtasks: List["StructuredTask"] = []

    def add_subtask(self, subtask: "StructuredTask") -> None:
        self.subtasks.append(subtask)

    def __repr__(self) -> str:
        return f"StructuredTask(title={self.title}, {len(self.subtasks)} subtasks)"


def _adjust_due_max_for_tasks_api(due_max: str) -> str:
    """
    Compensate for the Google Tasks API treating dueMax as an exclusive bound.

    The API stores due dates at day resolution and compares using < dueMax, so to
    include tasks due on the requested date we bump the bound by one day.
    """
    try:
        parsed = datetime.fromisoformat(due_max.replace("Z", "+00:00"))
    except ValueError:
        logger.warning("[list_tasks] Unable to parse due_max '%s'; sending unmodified value", due_max)
        return due_max

    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)

    adjusted = parsed + timedelta(days=1)
    if adjusted.tzinfo == timezone.utc:
        return adjusted.isoformat().replace("+00:00", "Z")
    return adjusted.isoformat()


@server.tool()  # type: ignore
@require_google_service("tasks", "tasks_read")  # type: ignore
@handle_http_errors("list_task_lists", service_type="tasks")  # type: ignore
async def list_task_lists(
    service: Resource,
    user_google_email: str,
    max_results: int = 1000,
    page_token: Optional[str] = None
) -> str:
    """
    List all task lists for the user.

    Args:
        user_google_email (str): The user's Google email address. Required.
        max_results (int): Maximum number of task lists to return (default: 1000, max: 1000).
        page_token (Optional[str]): Token for pagination.

    Returns:
        str: List of task lists with their IDs, titles, and details.
    """
    logger.info(f"[list_task_lists] Invoked. Email: '{user_google_email}'")

    try:
        params: Dict[str, Any] = {}
        if max_results is not None:
            params["maxResults"] = max_results
        if page_token:
            params["pageToken"] = page_token

        result = await asyncio.to_thread(
            service.tasklists().list(**params).execute
        )

        task_lists = result.get("items", [])
        next_page_token = result.get("nextPageToken")

        if not task_lists:
            return f"No task lists found for {user_google_email}."

        response = f"Task Lists for {user_google_email}:\n"
        for task_list in task_lists:
            response += f"- {task_list['title']} (ID: {task_list['id']})\n"
            response += f"  Updated: {task_list.get('updated', 'N/A')}\n"

        if next_page_token:
            response += f"\nNext page token: {next_page_token}"

        logger.info(f"Found {len(task_lists)} task lists for {user_google_email}")
        return response

    except HttpError as error:
        message = f"API error: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with the user's email ({user_google_email}) and service_name='Google Tasks'."
        logger.error(message, exc_info=True)
        raise Exception(message)
    except Exception as e:
        message = f"Unexpected error: {e}."
        logger.exception(message)
        raise Exception(message)


@server.tool()  # type: ignore
@require_google_service("tasks", "tasks_read")  # type: ignore
@handle_http_errors("get_task_list", service_type="tasks")  # type: ignore
async def get_task_list(
    service: Resource,
    user_google_email: str,
    task_list_id: str
) -> str:
    """
    Get details of a specific task list.

    Args:
        user_google_email (str): The user's Google email address. Required.
        task_list_id (str): The ID of the task list to retrieve.

    Returns:
        str: Task list details including title, ID, and last updated time.
    """
    logger.info(f"[get_task_list] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}")

    try:
        task_list = await asyncio.to_thread(
            service.tasklists().get(tasklist=task_list_id).execute
        )

        response = f"""Task List Details for {user_google_email}:
- Title: {task_list['title']}
- ID: {task_list['id']}
- Updated: {task_list.get('updated', 'N/A')}
- Self Link: {task_list.get('selfLink', 'N/A')}"""

        logger.info(f"Retrieved task list '{task_list['title']}' for {user_google_email}")
        return response

    except HttpError as error:
        message = f"API error: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with the user's email ({user_google_email}) and service_name='Google Tasks'."
        logger.error(message, exc_info=True)
        raise Exception(message)
    except Exception as e:
        message = f"Unexpected error: {e}."
        logger.exception(message)
        raise Exception(message)


@server.tool()  # type: ignore
@require_google_service("tasks", "tasks")  # type: ignore
@handle_http_errors("create_task_list", service_type="tasks")  # type: ignore
async def create_task_list(
    service: Resource,
    user_google_email: str,
    title: str
) -> str:
    """
    Create a new task list.

    Args:
        user_google_email (str): The user's Google email address. Required.
        title (str): The title of the new task list.

    Returns:
        str: Confirmation message with the new task list ID and details.
    """
    logger.info(f"[create_task_list] Invoked. Email: '{user_google_email}', Title: '{title}'")

    try:
        body = {
            "title": title
        }

        result = await asyncio.to_thread(
            service.tasklists().insert(body=body).execute
        )

        response = f"""Task List Created for {user_google_email}:
- Title: {result['title']}
- ID: {result['id']}
- Created: {result.get('updated', 'N/A')}
- Self Link: {result.get('selfLink', 'N/A')}"""

        logger.info(f"Created task list '{title}' with ID {result['id']} for {user_google_email}")
        return response

    except HttpError as error:
        message = f"API error: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with the user's email ({user_google_email}) and service_name='Google Tasks'."
        logger.error(message, exc_info=True)
        raise Exception(message)
    except Exception as e:
        message = f"Unexpected error: {e}."
        logger.exception(message)
        raise Exception(message)


@server.tool()  # type: ignore
@require_google_service("tasks", "tasks")  # type: ignore
@handle_http_errors("update_task_list", service_type="tasks")  # type: ignore
async def update_task_list(
    service: Resource,
    user_google_email: str,
    task_list_id: str,
    title: str
) -> str:
    """
    Update an existing task list.

    Args:
        user_google_email (str): The user's Google email address. Required.
        task_list_id (str): The ID of the task list to update.
        title (str): The new title for the task list.

    Returns:
        str: Confirmation message with updated task list details.
    """
    logger.info(f"[update_task_list] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, New Title: '{title}'")

    try:
        body = {
            "id": task_list_id,
            "title": title
        }

        result = await asyncio.to_thread(
            service.tasklists().update(tasklist=task_list_id, body=body).execute
        )

        response = f"""Task List Updated for {user_google_email}:
- Title: {result['title']}
- ID: {result['id']}
- Updated: {result.get('updated', 'N/A')}"""

        logger.info(f"Updated task list {task_list_id} with new title '{title}' for {user_google_email}")
        return response

    except HttpError as error:
        message = f"API error: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with the user's email ({user_google_email}) and service_name='Google Tasks'."
        logger.error(message, exc_info=True)
        raise Exception(message)
    except Exception as e:
        message = f"Unexpected error: {e}."
        logger.exception(message)
        raise Exception(message)


@server.tool()  # type: ignore
@require_google_service("tasks", "tasks")  # type: ignore
@handle_http_errors("delete_task_list", service_type="tasks")  # type: ignore
async def delete_task_list(
    service: Resource,
    user_google_email: str,
    task_list_id: str
) -> str:
    """
    Delete a task list. Note: This will also delete all tasks in the list.

    Args:
        user_google_email (str): The user's Google email address. Required.
        task_list_id (str): The ID of the task list to delete.

    Returns:
        str: Confirmation message.
    """
    logger.info(f"[delete_task_list] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}")

    try:
        await asyncio.to_thread(
            service.tasklists().delete(tasklist=task_list_id).execute
        )

        response = f"Task list {task_list_id} has been deleted for {user_google_email}. All tasks in this list have also been deleted."

        logger.info(f"Deleted task list {task_list_id} for {user_google_email}")
        return response

    except HttpError as error:
        message = f"API error: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with the user's email ({user_google_email}) and service_name='Google Tasks'."
        logger.error(message, exc_info=True)
        raise Exception(message)
    except Exception as e:
        message = f"Unexpected error: {e}."
        logger.exception(message)
        raise Exception(message)


@server.tool()  # type: ignore
@require_google_service("tasks", "tasks_read")  # type: ignore
@handle_http_errors("list_tasks", service_type="tasks")  # type: ignore
async def list_tasks(
    service: Resource,
    user_google_email: str,
    task_list_id: str,
    max_results: int = LIST_TASKS_MAX_RESULTS_DEFAULT,
    page_token: Optional[str] = None,
    show_completed: bool = True,
    show_deleted: bool = False,
    show_hidden: bool = False,
    show_assigned: bool = False,
    completed_max: Optional[str] = None,
    completed_min: Optional[str] = None,
    due_max: Optional[str] = None,
    due_min: Optional[str] = None,
    updated_min: Optional[str] = None,
) -> str:
    """
    List all tasks in a specific task list.

    Args:
        user_google_email (str): The user's Google email address. Required.
        task_list_id (str): The ID of the task list to retrieve tasks from.
        max_results (int): Maximum number of tasks to return. (default: 20, max: 10000).
        page_token (Optional[str]): Token for pagination.
        show_completed (bool): Whether to include completed tasks (default: True). Note that show_hidden must also be true to show tasks completed in first party clients, such as the web UI and Google's mobile apps.
        show_deleted (bool): Whether to include deleted tasks (default: False).
        show_hidden (bool): Whether to include hidden tasks (default: False).
        show_assigned (bool): Whether to include assigned tasks (default: False).
        completed_max (Optional[str]): Upper bound for completion date (RFC 3339 timestamp).
        completed_min (Optional[str]): Lower bound for completion date (RFC 3339 timestamp).
        due_max (Optional[str]): Upper bound for due date (RFC 3339 timestamp).
        due_min (Optional[str]): Lower bound for due date (RFC 3339 timestamp).
        updated_min (Optional[str]): Lower bound for last modification time (RFC 3339 timestamp).

    Returns:
        str: List of tasks with their details.
    """
    logger.info(f"[list_tasks] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}")

    try:
        params: Dict[str, Any] = {"tasklist": task_list_id}
        if max_results is not None:
            params["maxResults"] = max_results
        if page_token:
            params["pageToken"] = page_token
        if show_completed is not None:
            params["showCompleted"] = show_completed
        if show_deleted is not None:
            params["showDeleted"] = show_deleted
        if show_hidden is not None:
            params["showHidden"] = show_hidden
        if show_assigned is not None:
            params["showAssigned"] = show_assigned
        if completed_max:
            params["completedMax"] = completed_max
        if completed_min:
            params["completedMin"] = completed_min
        if due_max:
            adjusted_due_max = _adjust_due_max_for_tasks_api(due_max)
            if adjusted_due_max != due_max:
                logger.info(
                    "[list_tasks] Adjusted due_max from '%s' to '%s' to include due date boundary",
                    due_max,
                    adjusted_due_max,
                )
            params["dueMax"] = adjusted_due_max
        if due_min:
            params["dueMin"] = due_min
        if updated_min:
            params["updatedMin"] = updated_min

        result = await asyncio.to_thread(
            service.tasks().list(**params).execute
        )

        tasks = result.get("items", [])
        next_page_token = result.get("nextPageToken")

        # In order to return a sorted and organized list of tasks all at once, we support retrieving more than a single
        # page from the Google tasks API.
        results_remaining = (
            min(max_results, LIST_TASKS_MAX_RESULTS_MAX) if max_results else LIST_TASKS_MAX_RESULTS_DEFAULT
        )
        results_remaining -= len(tasks)
        while results_remaining > 0 and next_page_token:
            params["pageToken"] = next_page_token
            params["maxResults"] = str(results_remaining)
            result = await asyncio.to_thread(
                service.tasks().list(**params).execute
            )
            more_tasks = result.get("items", [])
            next_page_token = result.get("nextPageToken")
            if len(more_tasks) == 0:
                # For some unexpected reason, no more tasks were returned. Break to avoid an infinite loop.
                break
            tasks.extend(more_tasks)
            results_remaining -= len(more_tasks)

        if not tasks:
            return f"No tasks found in task list {task_list_id} for {user_google_email}."

        structured_tasks = get_structured_tasks(tasks)

        response = f"Tasks in list {task_list_id} for {user_google_email}:\n"
        response += serialize_tasks(structured_tasks, 0)

        if next_page_token:
            response += f"Next page token: {next_page_token}\n"

        logger.info(f"Found {len(tasks)} tasks in list {task_list_id} for {user_google_email}")
        return response

    except HttpError as error:
        message = f"API error: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with the user's email ({user_google_email}) and service_name='Google Tasks'."
        logger.error(message, exc_info=True)
        raise Exception(message)
    except Exception as e:
        message = f"Unexpected error: {e}."
        logger.exception(message)
        raise Exception(message)


def get_structured_tasks(tasks: List[Dict[str, str]]) -> List[StructuredTask]:
    """
    Convert a flat list of task dictionaries into StructuredTask objects based on parent-child relationships sorted by position.

    Args:
        tasks: List of task dictionaries.

    Returns:
        list: Sorted list of top-level StructuredTask objects with nested subtasks.
    """
    tasks_by_id = {
        task["id"]: StructuredTask(task, is_placeholder_parent=False) for task in tasks
    }
    positions_by_id = {
        task["id"]: int(task["position"]) for task in tasks if "position" in task
    }

    # Placeholder virtual root as parent for top-level tasks
    root_task = StructuredTask(
        {"id": "root", "title": "Root"}, is_placeholder_parent=False
    )

    for task in tasks:
        structured_task = tasks_by_id[task["id"]]
        parent_id = task.get("parent")
        parent = None

        if not parent_id:
            # Task without parent: parent to the virtual root
            parent = root_task
        elif parent_id in tasks_by_id:
            # Subtask: parent to its actual parent
            parent = tasks_by_id[parent_id]
        else:
            # Orphaned subtask: create placeholder parent
            # Due to paging or filtering, a subtask may have a parent that is not present in the list of tasks.
            # We will create placeholder StructuredTask objects for these missing parents to maintain the hierarchy.
            parent = StructuredTask({"id": parent_id}, is_placeholder_parent=True)
            tasks_by_id[parent_id] = parent
            root_task.add_subtask(parent)

        parent.add_subtask(structured_task)

    sort_structured_tasks(root_task, positions_by_id)
    return root_task.subtasks


def sort_structured_tasks(
    root_task: StructuredTask, positions_by_id: Dict[str, int]
) -> None:
    """
    Recursively sort--in place--StructuredTask objects and their subtasks based on position.

    Args:
        root_task: The root StructuredTask object.
        positions_by_id: Dictionary mapping task IDs to their positions.
    """

    def get_position(task: StructuredTask) -> int | float:
        # Tasks without position go to the end (infinity)
        result = positions_by_id.get(task.id, float("inf"))
        return result

    root_task.subtasks.sort(key=get_position)
    for subtask in root_task.subtasks:
        sort_structured_tasks(subtask, positions_by_id)


def serialize_tasks(structured_tasks: List[StructuredTask], subtask_level: int) -> str:
    """
    Serialize a list of StructuredTask objects into a formatted string with indentation for subtasks.
    Args:
        structured_tasks (list): List of StructuredTask objects.
        subtask_level (int): Current level of indentation for subtasks.

    Returns:
        str: Formatted string representation of the tasks.
    """
    response = ""
    placeholder_parent_count = 0
    placeholder_parent_title = "Unknown parent"
    for task in structured_tasks:
        indent = "  " * subtask_level
        bullet = "-" if subtask_level == 0 else "*"
        if task.title is not None:
            title = task.title
        elif task.is_placeholder_parent:
            title = placeholder_parent_title
            placeholder_parent_count += 1
        else:
            title = "Untitled"
        response += f"{indent}{bullet} {title} (ID: {task.id})\n"
        response += f"{indent}  Status: {task.status or 'N/A'}\n"
        response += f"{indent}  Due: {task.due}\n" if task.due else ""
        if task.notes:
            response += f"{indent}  Notes: {task.notes[:100]}{'...' if len(task.notes) > 100 else ''}\n"
        response += f"{indent}  Completed: {task.completed}\n" if task.completed else ""
        response += f"{indent}  Updated: {task.updated or 'N/A'}\n"
        response += "\n"

        response += serialize_tasks(task.subtasks, subtask_level + 1)

    if placeholder_parent_count > 0:
        # Placeholder parents should only appear at the top level
        assert subtask_level == 0
        response += f"""
{placeholder_parent_count} tasks with title {placeholder_parent_title} are included as placeholders.
These placeholders contain subtasks whose parents were not present in the task list.
This can occur due to pagination. Callers can often avoid this problem if max_results is large enough to contain all tasks (subtasks and their parents) without paging.
This can also occur due to filtering that excludes parent tasks while including their subtasks or due to deleted or hidden parent tasks.
"""

    return response


@server.tool()  # type: ignore
@require_google_service("tasks", "tasks_read")  # type: ignore
@handle_http_errors("get_task", service_type="tasks")  # type: ignore
async def get_task(
    service: Resource,
    user_google_email: str,
    task_list_id: str,
    task_id: str
) -> str:
    """
    Get details of a specific task.

    Args:
        user_google_email (str): The user's Google email address. Required.
        task_list_id (str): The ID of the task list containing the task.
        task_id (str): The ID of the task to retrieve.

    Returns:
        str: Task details including title, notes, status, due date, etc.
    """
    logger.info(f"[get_task] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, Task ID: {task_id}")

    try:
        task = await asyncio.to_thread(
            service.tasks().get(tasklist=task_list_id, task=task_id).execute
        )

        response = f"""Task Details for {user_google_email}:
- Title: {task.get('title', 'Untitled')}
- ID: {task['id']}
- Status: {task.get('status', 'N/A')}
- Updated: {task.get('updated', 'N/A')}"""

        if task.get('due'):
            response += f"\n- Due Date: {task['due']}"
        if task.get('completed'):
            response += f"\n- Completed: {task['completed']}"
        if task.get('notes'):
            response += f"\n- Notes: {task['notes']}"
        if task.get('parent'):
            response += f"\n- Parent Task ID: {task['parent']}"
        if task.get('position'):
            response += f"\n- Position: {task['position']}"
        if task.get('selfLink'):
            response += f"\n- Self Link: {task['selfLink']}"
        if task.get('webViewLink'):
            response += f"\n- Web View Link: {task['webViewLink']}"

        logger.info(f"Retrieved task '{task.get('title', 'Untitled')}' for {user_google_email}")
        return response

    except HttpError as error:
        message = f"API error: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with the user's email ({user_google_email}) and service_name='Google Tasks'."
        logger.error(message, exc_info=True)
        raise Exception(message)
    except Exception as e:
        message = f"Unexpected error: {e}."
        logger.exception(message)
        raise Exception(message)


@server.tool()  # type: ignore
@require_google_service("tasks", "tasks")  # type: ignore
@handle_http_errors("create_task", service_type="tasks")  # type: ignore
async def create_task(
    service: Resource,
    user_google_email: str,
    task_list_id: str,
    title: str,
    notes: Optional[str] = None,
    due: Optional[str] = None,
    parent: Optional[str] = None,
    previous: Optional[str] = None
) -> str:
    """
    Create a new task in a task list.

    Args:
        user_google_email (str): The user's Google email address. Required.
        task_list_id (str): The ID of the task list to create the task in.
        title (str): The title of the task.
        notes (Optional[str]): Notes/description for the task.
        due (Optional[str]): Due date in RFC 3339 format (e.g., "2024-12-31T23:59:59Z").
        parent (Optional[str]): Parent task ID (for subtasks).
        previous (Optional[str]): Previous sibling task ID (for positioning).

    Returns:
        str: Confirmation message with the new task ID and details.
    """
    logger.info(f"[create_task] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, Title: '{title}'")

    try:
        body = {
            "title": title
        }
        if notes:
            body["notes"] = notes
        if due:
            body["due"] = due

        params = {"tasklist": task_list_id, "body": body}
        if parent:
            params["parent"] = parent
        if previous:
            params["previous"] = previous

        result = await asyncio.to_thread(
            service.tasks().insert(**params).execute
        )

        response = f"""Task Created for {user_google_email}:
- Title: {result['title']}
- ID: {result['id']}
- Status: {result.get('status', 'N/A')}
- Updated: {result.get('updated', 'N/A')}"""

        if result.get('due'):
            response += f"\n- Due Date: {result['due']}"
        if result.get('notes'):
            response += f"\n- Notes: {result['notes']}"
        if result.get('webViewLink'):
            response += f"\n- Web View Link: {result['webViewLink']}"

        logger.info(f"Created task '{title}' with ID {result['id']} for {user_google_email}")
        return response

    except HttpError as error:
        message = f"API error: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with the user's email ({user_google_email}) and service_name='Google Tasks'."
        logger.error(message, exc_info=True)
        raise Exception(message)
    except Exception as e:
        message = f"Unexpected error: {e}."
        logger.exception(message)
        raise Exception(message)


@server.tool()  # type: ignore
@require_google_service("tasks", "tasks")  # type: ignore
@handle_http_errors("update_task", service_type="tasks")  # type: ignore
async def update_task(
    service: Resource,
    user_google_email: str,
    task_list_id: str,
    task_id: str,
    title: Optional[str] = None,
    notes: Optional[str] = None,
    status: Optional[str] = None,
    due: Optional[str] = None
) -> str:
    """
    Update an existing task.

    Args:
        user_google_email (str): The user's Google email address. Required.
        task_list_id (str): The ID of the task list containing the task.
        task_id (str): The ID of the task to update.
        title (Optional[str]): New title for the task.
        notes (Optional[str]): New notes/description for the task.
        status (Optional[str]): New status ("needsAction" or "completed").
        due (Optional[str]): New due date in RFC 3339 format.

    Returns:
        str: Confirmation message with updated task details.
    """
    logger.info(f"[update_task] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, Task ID: {task_id}")

    try:
        # First get the current task to build the update body
        current_task = await asyncio.to_thread(
            service.tasks().get(tasklist=task_list_id, task=task_id).execute
        )

        body = {
            "id": task_id,
            "title": title if title is not None else current_task.get("title", ""),
            "status": status if status is not None else current_task.get("status", "needsAction")
        }

        if notes is not None:
            body["notes"] = notes
        elif current_task.get("notes"):
            body["notes"] = current_task["notes"]

        if due is not None:
            body["due"] = due
        elif current_task.get("due"):
            body["due"] = current_task["due"]

        result = await asyncio.to_thread(
            service.tasks().update(tasklist=task_list_id, task=task_id, body=body).execute
        )

        response = f"""Task Updated for {user_google_email}:
- Title: {result['title']}
- ID: {result['id']}
- Status: {result.get('status', 'N/A')}
- Updated: {result.get('updated', 'N/A')}"""

        if result.get('due'):
            response += f"\n- Due Date: {result['due']}"
        if result.get('notes'):
            response += f"\n- Notes: {result['notes']}"
        if result.get('completed'):
            response += f"\n- Completed: {result['completed']}"

        logger.info(f"Updated task {task_id} for {user_google_email}")
        return response

    except HttpError as error:
        message = f"API error: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with the user's email ({user_google_email}) and service_name='Google Tasks'."
        logger.error(message, exc_info=True)
        raise Exception(message)
    except Exception as e:
        message = f"Unexpected error: {e}."
        logger.exception(message)
        raise Exception(message)


@server.tool()  # type: ignore
@require_google_service("tasks", "tasks")  # type: ignore
@handle_http_errors("delete_task", service_type="tasks")  # type: ignore
async def delete_task(
    service: Resource,
    user_google_email: str,
    task_list_id: str,
    task_id: str
) -> str:
    """
    Delete a task from a task list.

    Args:
        user_google_email (str): The user's Google email address. Required.
        task_list_id (str): The ID of the task list containing the task.
        task_id (str): The ID of the task to delete.

    Returns:
        str: Confirmation message.
    """
    logger.info(f"[delete_task] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, Task ID: {task_id}")

    try:
        await asyncio.to_thread(
            service.tasks().delete(tasklist=task_list_id, task=task_id).execute
        )

        response = f"Task {task_id} has been deleted from task list {task_list_id} for {user_google_email}."

        logger.info(f"Deleted task {task_id} for {user_google_email}")
        return response

    except HttpError as error:
        message = f"API error: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with the user's email ({user_google_email}) and service_name='Google Tasks'."
        logger.error(message, exc_info=True)
        raise Exception(message)
    except Exception as e:
        message = f"Unexpected error: {e}."
        logger.exception(message)
        raise Exception(message)


@server.tool()  # type: ignore
@require_google_service("tasks", "tasks")  # type: ignore
@handle_http_errors("move_task", service_type="tasks")  # type: ignore
async def move_task(
    service: Resource,
    user_google_email: str,
    task_list_id: str,
    task_id: str,
    parent: Optional[str] = None,
    previous: Optional[str] = None,
    destination_task_list: Optional[str] = None
) -> str:
    """
    Move a task to a different position or parent within the same list, or to a different list.

    Args:
        user_google_email (str): The user's Google email address. Required.
        task_list_id (str): The ID of the current task list containing the task.
        task_id (str): The ID of the task to move.
        parent (Optional[str]): New parent task ID (for making it a subtask).
        previous (Optional[str]): Previous sibling task ID (for positioning).
        destination_task_list (Optional[str]): Destination task list ID (for moving between lists).

    Returns:
        str: Confirmation message with updated task details.
    """
    logger.info(f"[move_task] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}, Task ID: {task_id}")

    try:
        params = {
            "tasklist": task_list_id,
            "task": task_id
        }
        if parent:
            params["parent"] = parent
        if previous:
            params["previous"] = previous
        if destination_task_list:
            params["destinationTasklist"] = destination_task_list

        result = await asyncio.to_thread(
            service.tasks().move(**params).execute
        )

        response = f"""Task Moved for {user_google_email}:
- Title: {result['title']}
- ID: {result['id']}
- Status: {result.get('status', 'N/A')}
- Updated: {result.get('updated', 'N/A')}"""

        if result.get('parent'):
            response += f"\n- Parent Task ID: {result['parent']}"
        if result.get('position'):
            response += f"\n- Position: {result['position']}"

        move_details = []
        if destination_task_list:
            move_details.append(f"moved to task list {destination_task_list}")
        if parent:
            move_details.append(f"made a subtask of {parent}")
        if previous:
            move_details.append(f"positioned after {previous}")

        if move_details:
            response += f"\n- Move Details: {', '.join(move_details)}"

        logger.info(f"Moved task {task_id} for {user_google_email}")
        return response

    except HttpError as error:
        message = f"API error: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with the user's email ({user_google_email}) and service_name='Google Tasks'."
        logger.error(message, exc_info=True)
        raise Exception(message)
    except Exception as e:
        message = f"Unexpected error: {e}."
        logger.exception(message)
        raise Exception(message)


@server.tool()  # type: ignore
@require_google_service("tasks", "tasks")  # type: ignore
@handle_http_errors("clear_completed_tasks", service_type="tasks")  # type: ignore
async def clear_completed_tasks(
    service: Resource,
    user_google_email: str,
    task_list_id: str
) -> str:
    """
    Clear all completed tasks from a task list. The tasks will be marked as hidden.

    Args:
        user_google_email (str): The user's Google email address. Required.
        task_list_id (str): The ID of the task list to clear completed tasks from.

    Returns:
        str: Confirmation message.
    """
    logger.info(f"[clear_completed_tasks] Invoked. Email: '{user_google_email}', Task List ID: {task_list_id}")

    try:
        await asyncio.to_thread(
            service.tasks().clear(tasklist=task_list_id).execute
        )

        response = f"All completed tasks have been cleared from task list {task_list_id} for {user_google_email}. The tasks are now hidden and won't appear in default task list views."

        logger.info(f"Cleared completed tasks from list {task_list_id} for {user_google_email}")
        return response

    except HttpError as error:
        message = f"API error: {error}. You might need to re-authenticate. LLM: Try 'start_google_auth' with the user's email ({user_google_email}) and service_name='Google Tasks'."
        logger.error(message, exc_info=True)
        raise Exception(message)
    except Exception as e:
        message = f"Unexpected error: {e}."
        logger.exception(message)
        raise Exception(message)