feat(platform): add Human In The Loop block with review workflow

autogpt_platform/backend/backend/blocks/human_in_the_loop.py

@@ -0,0 +1,160 @@
+import logging
+from typing import Any, Literal
+
+from prisma.enums import ReviewStatus
+
+from backend.data.block import (
+    Block,
+    BlockCategory,
+    BlockOutput,
+    BlockSchemaInput,
+    BlockSchemaOutput,
+)
+from backend.data.execution import ExecutionStatus
+from backend.data.human_review import ReviewResult
+from backend.data.model import SchemaField
+from backend.executor.manager import async_update_node_execution_status
+from backend.util.clients import get_database_manager_async_client
+
+logger = logging.getLogger(__name__)
+
+
+class HumanInTheLoopBlock(Block):
+    """
+    This block pauses execution and waits for human approval or modification of the data.
+
+    When executed, it creates a pending review entry and sets the node execution status
+    to REVIEW. The execution will remain paused until a human user either:
+    - Approves the data (with or without modifications)
+    - Rejects the data
+
+    This is useful for workflows that require human validation or intervention before
+    proceeding to the next steps.
+    """
+
+    class Input(BlockSchemaInput):
+        data: Any = SchemaField(description="The data to be reviewed by a human user")
+        name: str = SchemaField(
+            description="A descriptive name for what this data represents",
+        )
+        editable: bool = SchemaField(
+            description="Whether the human reviewer can edit the data",
+            default=True,
+            advanced=True,
+        )
+
+    class Output(BlockSchemaOutput):
+        reviewed_data: Any = SchemaField(
+            description="The data after human review (may be modified)"
+        )
+        status: Literal["approved", "rejected"] = SchemaField(
+            description="Status of the review: 'approved' or 'rejected'"
+        )
+        review_message: str = SchemaField(
+            description="Any message provided by the reviewer", default=""
+        )
+
+    def __init__(self):
+        super().__init__(
+            id="8b2a7b3c-6e9d-4a5f-8c1b-2e3f4a5b6c7d",
+            description="Pause execution and wait for human approval or modification of data",
+            categories={BlockCategory.BASIC},
+            input_schema=HumanInTheLoopBlock.Input,
+            output_schema=HumanInTheLoopBlock.Output,
+            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", ""),
+            ],
+            test_mock={
+                "get_or_create_human_review": lambda *_args, **_kwargs: ReviewResult(
+                    data={"name": "John Doe", "age": 30},
+                    status=ReviewStatus.APPROVED,
+                    message="",
+                    processed=False,
+                    node_exec_id="test-node-exec-id",
+                ),
+                "update_node_execution_status": lambda *_args, **_kwargs: None,
+            },
+        )
+
+    async def run(
+        self,
+        input_data: Input,
+        *,
+        user_id: str,
+        node_exec_id: str,
+        graph_exec_id: str,
+        graph_id: str,
+        graph_version: int,
+        **kwargs,
+    ) -> BlockOutput:
+        """
+        Execute the Human In The Loop block.
+
+        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(
+                user_id=user_id,
+                node_exec_id=node_exec_id,
+                graph_exec_id=graph_exec_id,
+                graph_id=graph_id,
+                graph_version=graph_version,
+                input_data=input_data.data,
+                message=input_data.name,
+                editable=input_data.editable,
+            )
+        except Exception as e:
+            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,
+                    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(
+                node_exec_id=node_exec_id, processed=True
+            )
+
+            if result.status == ReviewStatus.APPROVED:
+                yield "status", "approved"
+                yield "reviewed_data", result.data
+                if result.message:
+                    yield "review_message", result.message
+
+            elif result.status == ReviewStatus.REJECTED:
+                yield "status", "rejected"
+                if result.message:
+                    yield "review_message", result.message

autogpt_platform/backend/backend/data/credit_test.py

@@ -73,6 +73,7 @@ async def test_block_credit_usage(server: SpinTestServer):
         NodeExecutionEntry(
             user_id=DEFAULT_USER_ID,
             graph_id="test_graph",
+            graph_version=1,
             node_id="test_node",
             graph_exec_id="test_graph_exec",
             node_exec_id="test_node_exec",
@@ -94,6 +95,7 @@ async def test_block_credit_usage(server: SpinTestServer):
         NodeExecutionEntry(
             user_id=DEFAULT_USER_ID,
             graph_id="test_graph",
+            graph_version=1,
             node_id="test_node",
             graph_exec_id="test_graph_exec",
             node_exec_id="test_node_exec",

autogpt_platform/backend/backend/data/execution.py

@@ -34,6 +34,7 @@
     AgentNodeExecutionKeyValueDataCreateInput,
     AgentNodeExecutionUpdateInput,
     AgentNodeExecutionWhereInput,
+    AgentNodeExecutionWhereUniqueInput,
 )
 from pydantic import BaseModel, ConfigDict, JsonValue, ValidationError
 from pydantic.fields import Field
@@ -96,11 +97,14 @@ def error_rate(self) -> float:
 VALID_STATUS_TRANSITIONS = {
     ExecutionStatus.QUEUED: [
         ExecutionStatus.INCOMPLETE,
+        ExecutionStatus.TERMINATED,  # For resuming halted execution
+        ExecutionStatus.REVIEW,  # For resuming after review
     ],
     ExecutionStatus.RUNNING: [
         ExecutionStatus.INCOMPLETE,
         ExecutionStatus.QUEUED,
         ExecutionStatus.TERMINATED,  # For resuming halted execution
+        ExecutionStatus.REVIEW,  # For resuming after review
     ],
     ExecutionStatus.COMPLETED: [
         ExecutionStatus.RUNNING,
@@ -109,11 +113,16 @@ def error_rate(self) -> float:
         ExecutionStatus.INCOMPLETE,
         ExecutionStatus.QUEUED,
         ExecutionStatus.RUNNING,
+        ExecutionStatus.REVIEW,
     ],
     ExecutionStatus.TERMINATED: [
         ExecutionStatus.INCOMPLETE,
         ExecutionStatus.QUEUED,
         ExecutionStatus.RUNNING,
+        ExecutionStatus.REVIEW,
+    ],
+    ExecutionStatus.REVIEW: [
+        ExecutionStatus.RUNNING,
     ],
 }
 
@@ -446,6 +455,7 @@ def to_node_execution_entry(
             user_id=self.user_id,
             graph_exec_id=self.graph_exec_id,
             graph_id=self.graph_id,
+            graph_version=self.graph_version,
             node_exec_id=self.node_exec_id,
             node_id=self.node_id,
             block_id=self.block_id,
@@ -728,7 +738,7 @@ async def upsert_execution_input(
     input_name: str,
     input_data: JsonValue,
     node_exec_id: str | None = None,
-) -> tuple[str, BlockInput]:
+) -> tuple[NodeExecutionResult, BlockInput]:
     """
     Insert AgentNodeExecutionInputOutput record for as one of AgentNodeExecution.Input.
     If there is no AgentNodeExecution that has no `input_name` as input, create new one.
@@ -761,7 +771,7 @@ async def upsert_execution_input(
     existing_execution = await AgentNodeExecution.prisma().find_first(
         where=existing_exec_query_filter,
         order={"addedTime": "asc"},
-        include={"Input": True},
+        include={"Input": True, "GraphExecution": True},
     )
     json_input_data = SafeJson(input_data)
 
@@ -773,7 +783,7 @@ async def upsert_execution_input(
                 referencedByInputExecId=existing_execution.id,
             )
         )
-        return existing_execution.id, {
+        return NodeExecutionResult.from_db(existing_execution), {
             **{
                 input_data.name: type_utils.convert(input_data.data, JsonValue)
                 for input_data in existing_execution.Input or []
@@ -788,9 +798,10 @@ async def upsert_execution_input(
                 agentGraphExecutionId=graph_exec_id,
                 executionStatus=ExecutionStatus.INCOMPLETE,
                 Input={"create": {"name": input_name, "data": json_input_data}},
-            )
+            ),
+            include={"GraphExecution": True},
         )
-        return result.id, {input_name: input_data}
+        return NodeExecutionResult.from_db(result), {input_name: input_data}
 
     else:
         raise ValueError(
@@ -886,9 +897,25 @@ async def update_node_execution_status_batch(
     node_exec_ids: list[str],
     status: ExecutionStatus,
     stats: dict[str, Any] | None = None,
-):
-    await AgentNodeExecution.prisma().update_many(
-        where={"id": {"in": node_exec_ids}},
+) -> int:
+    # Validate status transitions - allowed_from should never be empty for valid statuses
+    allowed_from = VALID_STATUS_TRANSITIONS.get(status, [])
+    if not allowed_from:
+        raise ValueError(
+            f"Invalid status transition: {status} has no valid source statuses"
+        )
+
+    # For batch updates, we filter to only update nodes with valid current statuses
+    where_clause = cast(
+        AgentNodeExecutionWhereInput,
+        {
+            "id": {"in": node_exec_ids},
+            "executionStatus": {"in": [s.value for s in allowed_from]},
+        },
+    )
+
+    return await AgentNodeExecution.prisma().update_many(
+        where=where_clause,
         data=_get_update_status_data(status, None, stats),
     )
 
@@ -902,15 +929,32 @@ async def update_node_execution_status(
     if status == ExecutionStatus.QUEUED and execution_data is None:
         raise ValueError("Execution data must be provided when queuing an execution.")
 
-    res = await AgentNodeExecution.prisma().update(
-        where={"id": node_exec_id},
+    # Validate status transitions - allowed_from should never be empty for valid statuses
+    allowed_from = VALID_STATUS_TRANSITIONS.get(status, [])
+    if not allowed_from:
+        raise ValueError(
+            f"Invalid status transition: {status} has no valid source statuses"
+        )
+
+    if res := await AgentNodeExecution.prisma().update(
+        where=cast(
+            AgentNodeExecutionWhereUniqueInput,
+            {
+                "id": node_exec_id,
+                "executionStatus": {"in": [s.value for s in allowed_from]},
+            },
+        ),
         data=_get_update_status_data(status, execution_data, stats),
         include=EXECUTION_RESULT_INCLUDE,
-    )
-    if not res:
-        raise ValueError(f"Execution {node_exec_id} not found.")
+    ):
+        return NodeExecutionResult.from_db(res)
+
+    if res := await AgentNodeExecution.prisma().find_unique(
+        where={"id": node_exec_id}, include=EXECUTION_RESULT_INCLUDE
+    ):
+        return NodeExecutionResult.from_db(res)
 
-    return NodeExecutionResult.from_db(res)
+    raise ValueError(f"Execution {node_exec_id} not found.")
 
 
 def _get_update_status_data(
@@ -964,17 +1008,17 @@ async def get_node_execution(node_exec_id: str) -> NodeExecutionResult | None:
     return NodeExecutionResult.from_db(execution)
 
 
-async def get_node_executions(
+def _build_node_execution_where_clause(
     graph_exec_id: str | None = None,
     node_id: str | None = None,
     block_ids: list[str] | None = None,
     statuses: list[ExecutionStatus] | None = None,
-    limit: int | None = None,
     created_time_gte: datetime | None = None,
     created_time_lte: datetime | None = None,
-    include_exec_data: bool = True,
-) -> list[NodeExecutionResult]:
-    """⚠️ No `user_id` check: DO NOT USE without check in user-facing endpoints."""
+) -> AgentNodeExecutionWhereInput:
+    """
+    Build where clause for node execution queries.
+    """
     where_clause: AgentNodeExecutionWhereInput = {}
     if graph_exec_id:
         where_clause["agentGraphExecutionId"] = graph_exec_id
@@ -991,6 +1035,29 @@ async def get_node_executions(
             "lte": created_time_lte or datetime.max.replace(tzinfo=timezone.utc),
         }
 
+    return where_clause
+
+
+async def get_node_executions(
+    graph_exec_id: str | None = None,
+    node_id: str | None = None,
+    block_ids: list[str] | None = None,
+    statuses: list[ExecutionStatus] | None = None,
+    limit: int | None = None,
+    created_time_gte: datetime | None = None,
+    created_time_lte: datetime | None = None,
+    include_exec_data: bool = True,
+) -> list[NodeExecutionResult]:
+    """⚠️ No `user_id` check: DO NOT USE without check in user-facing endpoints."""
+    where_clause = _build_node_execution_where_clause(
+        graph_exec_id=graph_exec_id,
+        node_id=node_id,
+        block_ids=block_ids,
+        statuses=statuses,
+        created_time_gte=created_time_gte,
+        created_time_lte=created_time_lte,
+    )
+
     executions = await AgentNodeExecution.prisma().find_many(
         where=where_clause,
         include=(
@@ -1052,6 +1119,7 @@ class NodeExecutionEntry(BaseModel):
     user_id: str
     graph_exec_id: str
     graph_id: str
+    graph_version: int
     node_exec_id: str
     node_id: str
     block_id: str