fix(backend/executor): Improve graph execution permission check (#11323)

- Resolves #11316
- Durable fix to replace #11318

### Changes 🏗️

- Expand graph execution permissions check
  - Don't require library membership for execution as sub-graph

### 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:
  - [x] Can run sub-agent with non-latest graph version
- [x] Can run sub-agent that is available in Marketplace but not added
to Library

autogpt_platform/backend/backend/executor/scheduler.py

@@ -178,8 +178,9 @@ async def _execute_graph(**kwargs):
 
 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 GraphNotInLibraryError.
-    This happens when an agent is deleted but schedules still exist.
+    Clean up orphaned schedules for a specific graph when execution fails with GraphNotAccessibleError.
+    This happens when an agent is pulled from the Marketplace or deleted
+    but schedules still exist.
     """
     # Use scheduler client to access the scheduler service
     scheduler_client = get_scheduler_client()

autogpt_platform/backend/backend/executor/utils.py

@@ -477,6 +477,7 @@ async def validate_and_construct_node_execution_input(
     graph_version: Optional[int] = None,
     graph_credentials_inputs: Optional[Mapping[str, CredentialsMetaInput]] = None,
     nodes_input_masks: Optional[NodesInputMasks] = None,
+    is_sub_graph: bool = False,
 ) -> tuple[GraphModel, list[tuple[str, BlockInput]], NodesInputMasks]:
     """
     Public wrapper that handles graph fetching, credential mapping, and validation+construction.
@@ -489,6 +490,7 @@ async def validate_and_construct_node_execution_input(
         graph_version: The version of the graph to use.
         graph_credentials_inputs: Credentials inputs to use.
         nodes_input_masks: Node inputs to use.
+        is_sub_graph: Whether this is a sub-graph execution.
 
     Returns:
         GraphModel: Full graph object for the given `graph_id`.
@@ -510,19 +512,20 @@ async def validate_and_construct_node_execution_input(
         user_id=user_id,
         version=graph_version,
         include_subgraphs=True,
+        # Execution/access permission is checked by validate_graph_execution_permissions
+        skip_access_check=True,
     )
     if not graph:
         raise NotFoundError(f"Graph #{graph_id} not found.")
 
     # Validate that the user has permission to execute this graph
     # This checks both library membership and execution permissions,
-    # raising specific exceptions for appropriate error handling
-    # Note: Version-agnostic check to allow execution of graphs that reference
-    # older versions of sub-graphs that may no longer be in the library
+    # raising specific exceptions for appropriate error handling.
     await gdb.validate_graph_execution_permissions(
-        graph_id=graph_id,
         user_id=user_id,
-        # graph_version omitted for version-agnostic permission check
+        graph_id=graph.id,
+        graph_version=graph.version,
+        is_sub_graph=is_sub_graph,
     )
 
     nodes_input_masks = _merge_nodes_input_masks(
@@ -756,6 +759,7 @@ async def add_graph_execution(
     graph_credentials_inputs: Optional[Mapping[str, CredentialsMetaInput]] = None,
     nodes_input_masks: Optional[NodesInputMasks] = None,
     parent_graph_exec_id: Optional[str] = None,
+    is_sub_graph: bool = False,
 ) -> GraphExecutionWithNodes:
     """
     Adds a graph execution to the queue and returns the execution entry.
@@ -770,6 +774,7 @@ async def add_graph_execution(
             Keys should map to the keys generated by `GraphModel.aggregate_credentials_inputs`.
         nodes_input_masks: Node inputs to use in the execution.
         parent_graph_exec_id: The ID of the parent graph execution (for nested executions).
+        is_sub_graph: Whether this is a sub-graph execution.
     Returns:
         GraphExecutionEntry: The entry for the graph execution.
     Raises:
@@ -788,6 +793,7 @@ async def add_graph_execution(
             graph_version=graph_version,
             graph_credentials_inputs=graph_credentials_inputs,
             nodes_input_masks=nodes_input_masks,
+            is_sub_graph=is_sub_graph,
         )
     )
     graph_exec = None