toggle implementation

Author: majdyz

autogpt_platform/backend/backend/blocks/agent.py

@@ -11,7 +11,7 @@
     BlockType,
     get_block,
 )
-from backend.data.execution import ExecutionStatus, NodesInputMasks
+from backend.data.execution import ExecutionContext, ExecutionStatus, NodesInputMasks
 from backend.data.model import NodeExecutionStats, SchemaField
 from backend.util.json import validate_with_jsonschema
 from backend.util.retry import func_retry
@@ -72,9 +72,9 @@ async def run(
         input_data: Input,
         *,
         graph_exec_id: str,
+        execution_context: ExecutionContext,
         **kwargs,
     ) -> BlockOutput:
-
         from backend.executor import utils as execution_utils
 
         graph_exec = await execution_utils.add_graph_execution(
@@ -83,8 +83,9 @@ async def run(
             user_id=input_data.user_id,
             inputs=input_data.inputs,
             nodes_input_masks=input_data.nodes_input_masks,
-            parent_graph_exec_id=graph_exec_id,
-            is_sub_graph=True,  # AgentExecutorBlock executions are always sub-graphs
+            execution_context=execution_context.model_copy(
+                update={"parent_execution_id": graph_exec_id},
+            ),
         )
 
         logger = execution_utils.LogMetadata(

autogpt_platform/backend/backend/blocks/human_in_the_loop.py

@@ -9,8 +9,9 @@
     BlockOutput,
     BlockSchemaInput,
     BlockSchemaOutput,
+    BlockType,
 )
-from backend.data.execution import ExecutionStatus
+from backend.data.execution import ExecutionContext, ExecutionStatus
 from backend.data.human_review import ReviewResult
 from backend.data.model import SchemaField
 from backend.executor.manager import async_update_node_execution_status
@@ -61,15 +62,15 @@ def __init__(self):
             categories={BlockCategory.BASIC},
             input_schema=HumanInTheLoopBlock.Input,
             output_schema=HumanInTheLoopBlock.Output,
+            block_type=BlockType.HUMAN_IN_THE_LOOP,
             test_input={
                 "data": {"name": "John Doe", "age": 30},
                 "name": "User profile data",
                 "editable": True,
             },
             test_output=[
-                ("reviewed_data", {"name": "John Doe", "age": 30}),
                 ("status", "approved"),
-                ("review_message", ""),
+                ("reviewed_data", {"name": "John Doe", "age": 30}),
             ],
             test_mock={
                 "get_or_create_human_review": lambda *_args, **_kwargs: ReviewResult(
@@ -80,9 +81,25 @@ def __init__(self):
                     node_exec_id="test-node-exec-id",
                 ),
                 "update_node_execution_status": lambda *_args, **_kwargs: None,
+                "update_review_processed_status": lambda *_args, **_kwargs: None,
             },
         )
 
+    async def get_or_create_human_review(self, **kwargs):
+        return await get_database_manager_async_client().get_or_create_human_review(
+            **kwargs
+        )
+
+    async def update_node_execution_status(self, **kwargs):
+        return await async_update_node_execution_status(
+            db_client=get_database_manager_async_client(), **kwargs
+        )
+
+    async def update_review_processed_status(self, node_exec_id: str, processed: bool):
+        return await get_database_manager_async_client().update_review_processed_status(
+            node_exec_id, processed
+        )
+
     async def run(
         self,
         input_data: Input,
@@ -92,20 +109,20 @@ async def run(
         graph_exec_id: str,
         graph_id: str,
         graph_version: int,
+        execution_context: ExecutionContext,
         **kwargs,
     ) -> BlockOutput:
-        """
-        Execute the Human In The Loop block.
+        if not execution_context.safe_mode:
+            logger.info(
+                f"HITL block skipping review for node {node_exec_id} - safe mode disabled"
+            )
+            yield "status", "approved"
+            yield "reviewed_data", input_data.data
+            yield "review_message", "Auto-approved (safe mode disabled)"
+            return
 
-        This method uses one function to handle the complete workflow - checking existing reviews
-        and creating pending ones as needed.
-        """
         try:
-            logger.debug(f"HITL block executing for node {node_exec_id}")
-
-            # Use the data layer to handle the complete workflow
-            db_client = get_database_manager_async_client()
-            result = await db_client.get_or_create_human_review(
+            result = await self.get_or_create_human_review(
                 user_id=user_id,
                 node_exec_id=node_exec_id,
                 graph_exec_id=graph_exec_id,
@@ -119,32 +136,24 @@ async def run(
             logger.error(f"Error in HITL block for node {node_exec_id}: {str(e)}")
             raise
 
-        # Check if we're waiting for human input
         if result is None:
             logger.info(
                 f"HITL block pausing execution for node {node_exec_id} - awaiting human review"
             )
             try:
-                # Set node status to REVIEW so execution manager can't mark it as COMPLETED
-                # The VALID_STATUS_TRANSITIONS will then prevent any unwanted status changes
-                # Use the proper wrapper function to ensure websocket events are published
-                await async_update_node_execution_status(
-                    db_client=db_client,
+                await self.update_node_execution_status(
                     exec_id=node_exec_id,
                     status=ExecutionStatus.REVIEW,
                 )
-                # Execution pauses here until API routes process the review
                 return
             except Exception as e:
                 logger.error(
                     f"Failed to update node status for HITL block {node_exec_id}: {str(e)}"
                 )
                 raise
 
-        # Review is complete (approved or rejected) - check if unprocessed
         if not result.processed:
-            # Mark as processed before yielding
-            await db_client.update_review_processed_status(
+            await self.update_review_processed_status(
                 node_exec_id=node_exec_id, processed=True
             )
 

autogpt_platform/backend/backend/blocks/time_blocks.py

@@ -14,7 +14,7 @@
     BlockSchemaInput,
     BlockSchemaOutput,
 )
-from backend.data.execution import UserContext
+from backend.data.execution import ExecutionContext
 from backend.data.model import SchemaField
 
 # Shared timezone literal type for all time/date blocks
@@ -188,10 +188,9 @@ def __init__(self):
         )
 
     async def run(
-        self, input_data: Input, *, user_context: UserContext, **kwargs
+        self, input_data: Input, *, execution_context: ExecutionContext, **kwargs
     ) -> BlockOutput:
-        # Extract timezone from user_context (always present)
-        effective_timezone = user_context.timezone
+        effective_timezone = execution_context.user_timezone
 
         # Get the appropriate timezone
         tz = _get_timezone(input_data.format_type, effective_timezone)
@@ -298,10 +297,10 @@ def __init__(self):
             ],
         )
 
-    async def run(self, input_data: Input, **kwargs) -> BlockOutput:
-        # Extract timezone from user_context (required keyword argument)
-        user_context: UserContext = kwargs["user_context"]
-        effective_timezone = user_context.timezone
+    async def run(
+        self, input_data: Input, *, execution_context: ExecutionContext, **kwargs
+    ) -> BlockOutput:
+        effective_timezone = execution_context.user_timezone
 
         try:
             offset = int(input_data.offset)
@@ -404,10 +403,10 @@ def __init__(self):
             ],
         )
 
-    async def run(self, input_data: Input, **kwargs) -> BlockOutput:
-        # Extract timezone from user_context (required keyword argument)
-        user_context: UserContext = kwargs["user_context"]
-        effective_timezone = user_context.timezone
+    async def run(
+        self, input_data: Input, *, execution_context: ExecutionContext, **kwargs
+    ) -> BlockOutput:
+        effective_timezone = execution_context.user_timezone
 
         # Get the appropriate timezone
         tz = _get_timezone(input_data.format_type, effective_timezone)

autogpt_platform/backend/backend/data/block.py

@@ -71,6 +71,7 @@ class BlockType(Enum):
     AGENT = "Agent"
     AI = "AI"
     AYRSHARE = "Ayrshare"
+    HUMAN_IN_THE_LOOP = "Human In The Loop"
 
 
 class BlockCategory(Enum):
@@ -796,3 +797,12 @@ def get_io_block_ids() -> Sequence[str]:
         for id, B in get_blocks().items()
         if B().block_type in (BlockType.INPUT, BlockType.OUTPUT)
     ]
+
+
+@cached(ttl_seconds=3600)
+def get_human_in_the_loop_block_ids() -> Sequence[str]:
+    return [
+        id
+        for id, B in get_blocks().items()
+        if B().block_type == BlockType.HUMAN_IN_THE_LOOP
+    ]

autogpt_platform/backend/backend/data/credit_test.py

@@ -7,7 +7,7 @@
 from backend.blocks.llm import AITextGeneratorBlock
 from backend.data.block import get_block
 from backend.data.credit import BetaUserCredit, UsageTransactionMetadata
-from backend.data.execution import NodeExecutionEntry, UserContext
+from backend.data.execution import ExecutionContext, NodeExecutionEntry
 from backend.data.user import DEFAULT_USER_ID
 from backend.executor.utils import block_usage_cost
 from backend.integrations.credentials_store import openai_credentials
@@ -86,7 +86,7 @@ async def test_block_credit_usage(server: SpinTestServer):
                     "type": openai_credentials.type,
                 },
             },
-            user_context=UserContext(timezone="UTC"),
+            execution_context=ExecutionContext(user_timezone="UTC"),
         ),
     )
     assert spending_amount_1 > 0
@@ -101,7 +101,7 @@ async def test_block_credit_usage(server: SpinTestServer):
             node_exec_id="test_node_exec",
             block_id=AITextGeneratorBlock().id,
             inputs={"model": "gpt-4-turbo", "api_key": "owned_api_key"},
-            user_context=UserContext(timezone="UTC"),
+            execution_context=ExecutionContext(user_timezone="UTC"),
         ),
     )
     assert spending_amount_2 == 0

autogpt_platform/backend/backend/data/execution.py

@@ -71,6 +71,18 @@
 config = Config()
 
 
+class ExecutionContext(BaseModel):
+    """
+    Unified context that carries execution-level data throughout the entire execution flow.
+    This includes information needed by blocks, sub-graphs, and execution management.
+    """
+
+    safe_mode: bool = True
+    user_timezone: str = "UTC"
+    root_execution_id: Optional[str] = None
+    parent_execution_id: Optional[str] = None
+
+
 # -------------------------- Models -------------------------- #
 
 
@@ -365,18 +377,16 @@ def from_db(_graph_exec: AgentGraphExecution):
 
     def to_graph_execution_entry(
         self,
-        user_context: "UserContext",
+        execution_context: ExecutionContext,
         compiled_nodes_input_masks: Optional[NodesInputMasks] = None,
-        parent_graph_exec_id: Optional[str] = None,
     ):
         return GraphExecutionEntry(
             user_id=self.user_id,
             graph_id=self.graph_id,
             graph_version=self.graph_version or 0,
             graph_exec_id=self.id,
             nodes_input_masks=compiled_nodes_input_masks,
-            user_context=user_context,
-            parent_graph_exec_id=parent_graph_exec_id,
+            execution_context=execution_context,
         )
 
 
@@ -449,7 +459,7 @@ def from_db(_node_exec: AgentNodeExecution, user_id: Optional[str] = None):
         )
 
     def to_node_execution_entry(
-        self, user_context: "UserContext"
+        self, execution_context: ExecutionContext
     ) -> "NodeExecutionEntry":
         return NodeExecutionEntry(
             user_id=self.user_id,
@@ -460,7 +470,7 @@ def to_node_execution_entry(
             node_id=self.node_id,
             block_id=self.block_id,
             inputs=self.input_data,
-            user_context=user_context,
+            execution_context=execution_context,
         )
 
 
@@ -1099,20 +1109,13 @@ async def get_latest_node_execution(
 # ----------------- Execution Infrastructure ----------------- #
 
 
-class UserContext(BaseModel):
-    """Generic user context for graph execution containing user-specific settings."""
-
-    timezone: str
-
-
 class GraphExecutionEntry(BaseModel):
     user_id: str
     graph_exec_id: str
     graph_id: str
     graph_version: int
     nodes_input_masks: Optional[NodesInputMasks] = None
-    user_context: UserContext
-    parent_graph_exec_id: Optional[str] = None
+    execution_context: ExecutionContext
 
 
 class NodeExecutionEntry(BaseModel):
@@ -1124,7 +1127,7 @@ class NodeExecutionEntry(BaseModel):
     node_id: str
     block_id: str
     inputs: BlockInput
-    user_context: UserContext
+    execution_context: ExecutionContext
 
 
 class ExecutionQueue(Generic[T]):

autogpt_platform/backend/backend/data/graph.py

@@ -61,6 +61,10 @@
 logger = logging.getLogger(__name__)
 
 
+class GraphSettings(BaseModel):
+    human_in_the_loop_safe_mode: bool | None = None
+
+
 class Link(BaseDbModel):
     source_id: str
     sink_id: str
@@ -225,6 +229,15 @@ def output_schema(self) -> dict[str, Any]:
     def has_external_trigger(self) -> bool:
         return self.webhook_input_node is not None
 
+    @computed_field
+    @property
+    def has_human_in_the_loop(self) -> bool:
+        return any(
+            node.block_id
+            for node in self.nodes
+            if node.block.block_type == BlockType.HUMAN_IN_THE_LOOP
+        )
+
     @property
     def webhook_input_node(self) -> Node | None:
         return next(
@@ -1105,6 +1118,28 @@ async def delete_graph(graph_id: str, user_id: str) -> int:
     return entries_count
 
 
+async def get_graph_settings(user_id: str, graph_id: str) -> GraphSettings:
+    lib = await LibraryAgent.prisma().find_first(
+        where={
+            "userId": user_id,
+            "agentGraphId": graph_id,
+            "isDeleted": False,
+            "isArchived": False,
+        },
+        order={"agentGraphVersion": "desc"},
+    )
+    if not lib or not lib.settings:
+        return GraphSettings()
+
+    try:
+        return GraphSettings.model_validate(lib.settings)
+    except Exception:
+        logger.warning(
+            f"Malformed settings for LibraryAgent user={user_id} graph={graph_id}"
+        )
+        return GraphSettings()
+
+
 async def validate_graph_execution_permissions(
     user_id: str, graph_id: str, graph_version: int, is_sub_graph: bool = False
 ) -> None: