refactor(backend): extract shared auto-credential parsing to utils.py

Addresses review feedback on #12004:
- Added AutoCredentialFieldInfo dataclass and parse_auto_credential_field()
  helper to executor/utils.py
- Updated _acquire_auto_credentials in manager.py to use shared helper
- Updated _validate_node_input_credentials in utils.py to use shared helper

This consolidates the duplicate logic for parsing GoogleDriveFileField-style
auto-credential fields, making manager.py less cluttered while ensuring
consistent validation/acquisition behavior.

autogpt_platform/backend/backend/executor/manager.py

@@ -92,6 +92,7 @@ from .utils import (
     block_usage_cost,
     create_execution_queue_config,
     execution_usage_cost,
+    parse_auto_credential_field,
     validate_exec,
 )
 
@@ -196,53 +197,33 @@ async def _acquire_auto_credentials(
         field_name = info["field_name"]
         field_data = input_data.get(field_name)
 
-        if field_data and isinstance(field_data, dict):
-            # Check if _credentials_id key exists in the field data
-            if "_credentials_id" in field_data:
-                cred_id = field_data["_credentials_id"]
-                if cred_id:
-                    # Credential ID provided - acquire credentials
-                    provider = info.get("config", {}).get(
-                        "provider", "external service"
-                    )
-                    file_name = field_data.get("name", "selected file")
-                    try:
-                        credentials, lock = await creds_manager.acquire(
-                            user_id, cred_id
-                        )
-                        locks.append(lock)
-                        extra_exec_kwargs[kwarg_name] = credentials
-                    except ValueError:
-                        raise ValueError(
-                            f"{provider.capitalize()} credentials for "
-                            f"'{file_name}' in field '{field_name}' are not "
-                            f"available in your account. "
-                            f"This can happen if the agent was created by another "
-                            f"user or the credentials were deleted. "
-                            f"Please open the agent in the builder and re-select "
-                            f"the file to authenticate with your own account."
-                        )
-                # else: _credentials_id is explicitly None, skip (chained data)
-            else:
-                # _credentials_id key missing entirely - this is an error
-                provider = info.get("config", {}).get("provider", "external service")
-                file_name = field_data.get("name", "selected file")
+        # Use shared helper to parse the field
+        parsed = parse_auto_credential_field(
+            field_name=field_name,
+            info=info,
+            field_data=field_data,
+            field_present_in_input=field_name in input_data,
+        )
+
+        if parsed.error:
+            raise ValueError(parsed.error)
+
+        if parsed.cred_id:
+            # Credential ID provided - acquire credentials
+            try:
+                credentials, lock = await creds_manager.acquire(user_id, parsed.cred_id)
+                locks.append(lock)
+                extra_exec_kwargs[kwarg_name] = credentials
+            except ValueError:
                 raise ValueError(
-                    f"Authentication missing for '{file_name}' in field "
-                    f"'{field_name}'. Please re-select the file to authenticate "
-                    f"with {provider.capitalize()}."
+                    f"{parsed.provider.capitalize()} credentials for "
+                    f"'{parsed.file_name}' in field '{parsed.field_name}' are not "
+                    f"available in your account. "
+                    f"This can happen if the agent was created by another "
+                    f"user or the credentials were deleted. "
+                    f"Please open the agent in the builder and re-select "
+                    f"the file to authenticate with your own account."
                 )
-        elif field_data is None and field_name not in input_data:
-            # Field not in input_data at all = connected from upstream block, skip
-            pass
-        else:
-            # field_data is None/empty but key IS in input_data = user didn't select
-            provider = info.get("config", {}).get("provider", "external service")
-            raise ValueError(
-                f"No file selected for '{field_name}'. "
-                f"Please select a file to provide "
-                f"{provider.capitalize()} authentication."
-            )
 
     return extra_exec_kwargs, locks
 

autogpt_platform/backend/backend/executor/utils.py

@@ -4,7 +4,7 @@ import threading
 import time
 from collections import defaultdict
 from concurrent.futures import Future
-from typing import Mapping, Optional, cast
+from typing import Any, Mapping, Optional, cast
 
 from pydantic import BaseModel, JsonValue, ValidationError
 
@@ -55,6 +55,87 @@ from backend.util.type import convert
 config = Config()
 logger = TruncatedLogger(logging.getLogger(__name__), prefix="[GraphExecutorUtil]")
 
+# ============ Auto-Credentials Helpers ============ #
+
+
+class AutoCredentialFieldInfo(BaseModel):
+    """Parsed info from an auto-credential field (e.g., GoogleDriveFileField)."""
+
+    cred_id: str | None
+    """The credential ID to use, or None if not provided."""
+    provider: str
+    """The provider name (e.g., 'google')."""
+    file_name: str
+    """The display name for error messages."""
+    field_name: str
+    """The original field name in the schema."""
+    error: str | None = None
+    """Validation error message, if any."""
+
+
+def parse_auto_credential_field(
+    field_name: str,
+    info: dict,
+    field_data: Any,
+    *,
+    field_present_in_input: bool = True,
+) -> AutoCredentialFieldInfo:
+    """
+    Parse auto-credential field data and extract credential info.
+
+    This is shared logic used by both credential acquisition (manager.py)
+    and credential validation (utils.py).
+
+    Args:
+        field_name: The name of the field in the schema
+        info: The auto_credentials field info from get_auto_credentials_fields()
+        field_data: The actual field data from input
+        field_present_in_input: Whether the field key exists in input_data
+
+    Returns:
+        AutoCredentialFieldInfo with parsed data and any validation errors
+    """
+    provider = info.get("config", {}).get("provider", "external service")
+    file_name = (
+        field_data.get("name", "selected file")
+        if isinstance(field_data, dict)
+        else "selected file"
+    )
+
+    result = AutoCredentialFieldInfo(
+        cred_id=None,
+        provider=provider,
+        file_name=file_name,
+        field_name=field_name,
+    )
+
+    if field_data and isinstance(field_data, dict):
+        if "_credentials_id" not in field_data:
+            # Key removed (e.g., on fork) — needs re-auth
+            result.error = (
+                f"Authentication missing for '{file_name}' in field "
+                f"'{field_name}'. Please re-select the file to authenticate "
+                f"with {provider.capitalize()}."
+            )
+        else:
+            cred_id = field_data.get("_credentials_id")
+            if cred_id:
+                result.cred_id = cred_id
+            # else: _credentials_id is explicitly None, skip (chained data)
+    elif field_data is None and not field_present_in_input:
+        # Field not in input_data at all = connected from upstream block, skip
+        pass
+    elif field_present_in_input:
+        # field_data is None/empty but key IS in input_data = user didn't select
+        result.error = (
+            f"No file selected for '{field_name}'. "
+            f"Please select a file to provide "
+            f"{provider.capitalize()} authentication."
+        )
+
+    return result
+
+
 # ============ Resource Helpers ============ #
 
 
@@ -352,34 +433,39 @@ async def _validate_node_input_credentials(
                         field_name, field_value
                     )
 
-                if field_value and isinstance(field_value, dict):
-                    if "_credentials_id" not in field_value:
-                        # Key removed (e.g., on fork) — needs re-auth
+                # Use shared helper to parse the field
+                parsed = parse_auto_credential_field(
+                    field_name=field_name,
+                    info=info,
+                    field_data=field_value,
+                    field_present_in_input=True,  # For validation, assume present
+                )
+
+                if parsed.error:
+                    has_missing_credentials = True
+                    credential_errors[node.id][field_name] = parsed.error
+                    continue
+
+                if parsed.cred_id:
+                    # Validate that credentials exist and are accessible
+                    try:
+                        creds_store = get_integration_credentials_store()
+                        creds = await creds_store.get_creds_by_id(
+                            user_id, parsed.cred_id
+                        )
+                    except Exception as e:
+                        has_missing_credentials = True
+                        credential_errors[node.id][
+                            field_name
+                        ] = f"Credentials not available: {e}"
+                        continue
+                    if not creds:
                         has_missing_credentials = True
                         credential_errors[node.id][field_name] = (
-                            "Authentication missing for the selected file. "
-                            "Please re-select the file to authenticate with "
-                            "your own account."
+                            "The saved credentials are not available "
+                            "for your account. Please re-select the file to "
+                            "authenticate with your own account."
                         )
-                        continue
-                    cred_id = field_value.get("_credentials_id")
-                    if cred_id and isinstance(cred_id, str):
-                        try:
-                            creds_store = get_integration_credentials_store()
-                            creds = await creds_store.get_creds_by_id(user_id, cred_id)
-                        except Exception as e:
-                            has_missing_credentials = True
-                            credential_errors[node.id][
-                                field_name
-                            ] = f"Credentials not available: {e}"
-                            continue
-                        if not creds:
-                            has_missing_credentials = True
-                            credential_errors[node.id][field_name] = (
-                                "The saved credentials are not available "
-                                "for your account. Please re-select the file to "
-                                "authenticate with your own account."
-                            )
 
         # If node has optional credentials and any are missing, mark for skipping
         # But only if there are no other errors for this node