fix(platform): Remove un-runnable agents from schedule (#11374)

Currently when an agent fails validation during a scheduled run, we
raise an error then try again, regardless of why.

This change removed the agent schedule and notifies the user 

### Changes 🏗️

- add schedule_id to the GraphExecutionJobArgs
- add agent_name to the GraphExecutionJobArgs
- Delete schedule on GraphValidationError
- Notify the user with a message that include the agent name

### Checklist 📋

#### For code changes:
- [x] I have clearly listed my changes in the PR description
- [x] I have made a test plan
- [x] I have tested my changes according to the test plan:
  <!-- Put your test plan here: -->
  - [x] I have ensured the scheduler tests work with these changes

Author: Swiftyos

autogpt_platform/backend/backend/executor/scheduler.py

@@ -2,6 +2,7 @@
 import logging
 import os
 import threading
+import uuid
 from enum import Enum
 from typing import Optional
 from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
@@ -36,7 +37,9 @@
 from backend.util.clients import get_scheduler_client
 from backend.util.cloud_storage import cleanup_expired_files_async
 from backend.util.exceptions import (
+    GraphNotFoundError,
     GraphNotInLibraryError,
+    GraphValidationError,
     NotAuthorizedError,
     NotFoundError,
 )
@@ -160,14 +163,12 @@ async def _execute_graph(**kwargs):
                 f"Graph execution {graph_exec.id} took {elapsed:.2f}s to create/publish - "
                 f"this is unusually slow and may indicate resource contention"
             )
+    except GraphNotFoundError as e:
+        await _handle_graph_not_available(e, args, start_time)
     except GraphNotInLibraryError as e:
-        elapsed = asyncio.get_event_loop().time() - start_time
-        logger.warning(
-            f"Scheduled execution blocked for deleted/archived graph {args.graph_id} "
-            f"(user {args.user_id}) after {elapsed:.2f}s: {e}"
-        )
-        # Clean up orphaned schedules for this graph
-        await _cleanup_orphaned_schedules_for_graph(args.graph_id, args.user_id)
+        await _handle_graph_not_available(e, args, start_time)
+    except GraphValidationError:
+        await _handle_graph_validation_error(args)
     except Exception as e:
         elapsed = asyncio.get_event_loop().time() - start_time
         logger.error(
@@ -176,6 +177,34 @@ async def _execute_graph(**kwargs):
         )
 
 
+async def _handle_graph_validation_error(args: "GraphExecutionJobArgs") -> None:
+    logger.error(
+        f"Scheduled Graph {args.graph_id} failed validation. Unscheduling graph"
+    )
+    if args.schedule_id:
+        scheduler_client = get_scheduler_client()
+        await scheduler_client.delete_schedule(
+            schedule_id=args.schedule_id,
+            user_id=args.user_id,
+        )
+    else:
+        logger.error(
+            f"Unable to unschedule graph: {args.graph_id} as this is an old job with no associated schedule_id please remove manually"
+        )
+
+
+async def _handle_graph_not_available(
+    e: Exception, args: "GraphExecutionJobArgs", start_time: float
+) -> None:
+    elapsed = asyncio.get_event_loop().time() - start_time
+    logger.warning(
+        f"Scheduled execution blocked for deleted/archived graph {args.graph_id} "
+        f"(user {args.user_id}) after {elapsed:.2f}s: {e}"
+    )
+    # Clean up orphaned schedules for this graph
+    await _cleanup_orphaned_schedules_for_graph(args.graph_id, args.user_id)
+
+
 async def _cleanup_orphaned_schedules_for_graph(graph_id: str, user_id: str) -> None:
     """
     Clean up orphaned schedules for a specific graph when execution fails with GraphNotAccessibleError.
@@ -220,9 +249,11 @@ class Jobstores(Enum):
 
 
 class GraphExecutionJobArgs(BaseModel):
+    schedule_id: str | None = None
     user_id: str
     graph_id: str
     graph_version: int
+    agent_name: str | None = None
     cron: str
     input_data: BlockInput
     input_credentials: dict[str, CredentialsMetaInput] = Field(default_factory=dict)
@@ -468,11 +499,14 @@ def add_graph_execution_schedule(
         logger.info(
             f"Scheduling job for user {user_id} with timezone {user_timezone} (cron: {cron})"
         )
+        schedule_id = str(uuid.uuid4())
 
         job_args = GraphExecutionJobArgs(
+            schedule_id=schedule_id,
             user_id=user_id,
             graph_id=graph_id,
             graph_version=graph_version,
+            agent_name=name,
             cron=cron,
             input_data=input_data,
             input_credentials=input_credentials,
@@ -484,6 +518,7 @@ def add_graph_execution_schedule(
             trigger=CronTrigger.from_crontab(cron, timezone=user_timezone),
             jobstore=Jobstores.EXECUTION.value,
             replace_existing=True,
+            id=schedule_id,
         )
         logger.info(
             f"Added job {job.id} with cron schedule '{cron}' in timezone {user_timezone}, input data: {input_data}"

autogpt_platform/backend/backend/executor/utils.py

@@ -42,7 +42,11 @@
     get_database_manager_async_client,
     get_integration_credentials_store,
 )
-from backend.util.exceptions import GraphValidationError, NotFoundError
+from backend.util.exceptions import (
+    GraphNotFoundError,
+    GraphValidationError,
+    NotFoundError,
+)
 from backend.util.logging import TruncatedLogger, is_structured_logging_enabled
 from backend.util.settings import Config
 from backend.util.type import convert
@@ -516,7 +520,7 @@ async def validate_and_construct_node_execution_input(
         skip_access_check=True,
     )
     if not graph:
-        raise NotFoundError(f"Graph #{graph_id} not found.")
+        raise GraphNotFoundError(f"Graph #{graph_id} not found.")
 
     # Validate that the user has permission to execute this graph
     # This checks both library membership and execution permissions,

autogpt_platform/backend/backend/util/exceptions.py

@@ -38,6 +38,10 @@ class NotFoundError(ValueError):
     """The requested record was not found, resulting in an error condition"""
 
 
+class GraphNotFoundError(ValueError):
+    """The requested Agent Graph was not found, resulting in an error condition"""
+
+
 class NeedConfirmation(Exception):
     """The user must explicitly confirm that they want to proceed"""
 

autogpt_platform/frontend/src/app/api/openapi.json

@@ -6018,9 +6018,17 @@
       },
       "GraphExecutionJobInfo": {
         "properties": {
+          "schedule_id": {
+            "anyOf": [{ "type": "string" }, { "type": "null" }],
+            "title": "Schedule Id"
+          },
           "user_id": { "type": "string", "title": "User Id" },
           "graph_id": { "type": "string", "title": "Graph Id" },
           "graph_version": { "type": "integer", "title": "Graph Version" },
+          "agent_name": {
+            "anyOf": [{ "type": "string" }, { "type": "null" }],
+            "title": "Agent Name"
+          },
           "cron": { "type": "string", "title": "Cron" },
           "input_data": {
             "additionalProperties": true,