| """ |
| Maintenance and State Submodule for Task Service (Phase 4.6.12 Hardening) |
| |
| This module handles task validation, AI agent notifications, |
| archiving, and maintenance tasks like pruning. |
| """ |
|
|
| from datetime import datetime, timedelta |
| from typing import Any, Literal, cast |
|
|
| from src.server.config.logfire_config import get_logger |
| from src.server.schemas.agent_outputs import AgentOutputSchema |
|
|
| logger = get_logger(__name__) |
|
|
|
|
| def validate_status_logic(status: str, valid_statuses: list[str]) -> tuple[bool, str]: |
| """Validate task status against allowed list""" |
| if status not in valid_statuses: |
| return ( |
| False, |
| f"Invalid status '{status}'. Must be one of: {', '.join(valid_statuses)}", |
| ) |
| return True, "" |
|
|
|
|
| def validate_assignee_logic(assignee: str) -> tuple[bool, str]: |
| """Validate task assignee consistency""" |
| if not assignee or not isinstance(assignee, str) or len(assignee.strip()) == 0: |
| return False, "Assignee must be a non-empty string" |
| return True, "" |
|
|
|
|
| async def notify_ai_agent_logic(task_id: str, agent_id: str): |
| """ |
| Triggers the agent service call in a non-blocking way. |
| """ |
| try: |
| from src.server.services.agent_service import agent_service |
|
|
| logger.info(f"Scheduling notification for AI agent {agent_id} for task {task_id}") |
| await agent_service.run_agent_task(task_id=task_id, agent_id=agent_id) |
| except Exception as e: |
| logger.error(f"Failed to notify agent {agent_id}: {e}") |
|
|
|
|
| async def archive_task_logic( |
| task_service_instance, task_id: str, archived_by: str = "mcp" |
| ) -> tuple[bool, dict[str, Any]]: |
| """ |
| Archive a task and all its subtasks (soft delete). |
| """ |
| try: |
| success_get, get_result = await task_service_instance.get_task(task_id) |
| if not success_get: |
| return False, cast(dict[str, Any], get_result) |
|
|
| task = get_result["task"] |
| if task.get("archived") is True: |
| return False, {"error": f"Task with ID {task_id} is already archived"} |
|
|
| archive_data = { |
| "archived": True, |
| "archived_at": datetime.now().isoformat(), |
| "archived_by": archived_by, |
| "updated_at": datetime.now().isoformat(), |
| } |
|
|
| def _archive_query(): |
| return ( |
| task_service_instance.supabase_client.table("archon_tasks") |
| .update(archive_data) |
| .eq("id", task_id) |
| .execute() |
| ) |
|
|
| success_archive, archive_result = task_service_instance.execute_query( |
| query_func=_archive_query, error_context=f"Failed to archive task {task_id}" |
| ) |
|
|
| if success_archive: |
| return True, {"task_id": task_id, "message": "Task archived successfully"} |
| return False, cast(dict[str, Any], archive_result) |
|
|
| except Exception as e: |
| logger.error(f"Error archiving task logic: {e}") |
| return False, {"error": str(e)} |
|
|
|
|
| async def prune_archived_tasks_logic(task_service_instance, days_old: int = 30) -> tuple[bool, dict[str, Any]]: |
| """ |
| Permanently delete archived tasks older than X days. |
| """ |
| try: |
| cutoff_date = (datetime.now() - timedelta(days=days_old)).isoformat() |
| logger.info(f"Pruning archived tasks older than {days_old} days (cutoff: {cutoff_date})") |
|
|
| tasks_to_prune = ( |
| task_service_instance.supabase_client.table("archon_tasks") |
| .select("id") |
| .eq("archived", True) |
| .lt("archived_at", cutoff_date) |
| .execute() |
| ) |
|
|
| count = len(tasks_to_prune.data) if tasks_to_prune.data else 0 |
|
|
| if count > 0: |
| task_service_instance.supabase_client.table("archon_tasks").delete().eq("archived", True).lt( |
| "archived_at", cutoff_date |
| ).execute() |
| logger.info(f"Successfully pruned {count} tasks") |
|
|
| return True, {"pruned_count": count, "cutoff_date": cutoff_date} |
| except Exception as e: |
| logger.error(f"Error pruning tasks logic: {e}") |
| return False, {"error": str(e)} |
|
|
|
|
| async def update_task_status_from_agent_logic( |
| task_service_instance, task_id: str, new_status: str, agent_id: str |
| ) -> tuple[bool, dict[str, Any]]: |
| """ |
| Update a task's status from an agent with authority validation. |
| """ |
| try: |
| success, result = await task_service_instance.get_task(task_id) |
| if not success: |
| return False, cast(dict[str, Any], result) |
| current_task = result["task"] |
|
|
| if current_task.get("assignee") != agent_id: |
| error_msg = f"Agent '{agent_id}' is not authorized to update task '{task_id}'. Assigned to '{current_task.get('assignee')}'." |
| logger.warning(error_msg) |
| return False, {"error": error_msg} |
|
|
| return cast( |
| tuple[bool, dict[str, Any]], |
| await task_service_instance.update_task(task_id, {"status": new_status, "assignee": agent_id}), |
| ) |
| except Exception as e: |
| logger.error(f"Error updating task status from agent logic: {e}") |
| return False, {"error": str(e)} |
|
|
|
|
| async def save_agent_output_logic( |
| task_service_instance, task_id: str, output: dict[str, Any], agent_id: str |
| ) -> tuple[bool, dict[str, Any]]: |
| """ |
| Save the output from an AI agent to the task's attachments. |
| """ |
| try: |
| success, result = await task_service_instance.get_task(task_id) |
| if not success: |
| return False, cast(dict[str, Any], result) |
| current_task = result["task"] |
|
|
| if current_task.get("assignee") != agent_id: |
| error_msg = f"Agent '{agent_id}' is not authorized to save output for task '{task_id}'." |
| logger.warning(error_msg) |
| return False, {"error": error_msg} |
|
|
| |
| try: |
| |
| output_type = "text" |
| if isinstance(output, dict): |
| output_type = "group_chat" if "summary" in output and "decisions" in output else "structured" |
|
|
| validated_output = AgentOutputSchema( |
| agent_id=agent_id, |
| timestamp=datetime.now(), |
| output_type=cast(Literal["text", "structured", "group_chat"], output_type), |
| output=output, |
| ) |
| |
| agent_output_data = validated_output.model_dump(mode="json") |
| except Exception as ve: |
| logger.error(f"Validation failed for agent output: {ve}") |
| |
| agent_output_data = {"agent_id": agent_id, "timestamp": datetime.now().isoformat(), "output": output} |
|
|
| current_attachments = current_task.get("attachments") or [] |
| if isinstance(current_attachments, list): |
| current_attachments.append(agent_output_data) |
| new_attachments = current_attachments |
| else: |
| new_attachments = [current_attachments, agent_output_data] |
|
|
| return cast( |
| tuple[bool, dict[str, Any]], |
| await task_service_instance.update_task(task_id, {"attachments": new_attachments}), |
| ) |
| except Exception as e: |
| logger.error(f"Error saving agent output logic: {e}") |
| return False, {"error": str(e)} |
|
|
