602 lines
24 KiB
Python
602 lines
24 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
Automation2 routes - node-types, execute, workflows, runs, tasks, connections, browse.
|
|
"""
|
|
|
|
import logging
|
|
from fastapi import APIRouter, Depends, Path, Query, Body, Request, HTTPException
|
|
from fastapi.responses import JSONResponse
|
|
from modules.auth import limiter, getRequestContext, RequestContext
|
|
|
|
from modules.features.automation2.mainAutomation2 import getAutomation2Services
|
|
from modules.features.automation2.nodeRegistry import getNodeTypesForApi
|
|
from modules.features.automation2.interfaceFeatureAutomation2 import getAutomation2Interface
|
|
from modules.workflows.automation2.executionEngine import executeGraph
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(
|
|
prefix="/api/automation2",
|
|
tags=["Automation2"],
|
|
responses={404: {"description": "Not found"}, 403: {"description": "Forbidden"}},
|
|
)
|
|
|
|
|
|
def _validateInstanceAccess(instanceId: str, context: RequestContext) -> str:
|
|
"""Validate user has access to the automation2 feature instance. Returns mandateId."""
|
|
from fastapi import HTTPException
|
|
from modules.interfaces.interfaceDbApp import getRootInterface
|
|
|
|
rootInterface = getRootInterface()
|
|
instance = rootInterface.getFeatureInstance(instanceId)
|
|
if not instance:
|
|
raise HTTPException(status_code=404, detail=f"Feature instance {instanceId} not found")
|
|
featureAccess = rootInterface.getFeatureAccess(str(context.user.id), instanceId)
|
|
if not featureAccess or not featureAccess.enabled:
|
|
raise HTTPException(status_code=403, detail="Access denied to this feature instance")
|
|
return str(instance.mandateId) if instance.mandateId else ""
|
|
|
|
|
|
@router.get("/{instanceId}/info")
|
|
@limiter.limit("60/minute")
|
|
def get_automation2_info(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Minimal info endpoint - proves the feature works."""
|
|
_validateInstanceAccess(instanceId, context)
|
|
return {
|
|
"featureCode": "automation2",
|
|
"instanceId": instanceId,
|
|
"status": "ok",
|
|
"message": "Automation2 feature ready. Build from here.",
|
|
}
|
|
|
|
|
|
@router.get("/{instanceId}/node-types")
|
|
@limiter.limit("60/minute")
|
|
def get_node_types(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
language: str = Query("en", description="Localization (en, de, fr)"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Return node types for the flow builder: static + I/O from methodDiscovery."""
|
|
logger.info("automation2 node-types request: instanceId=%s language=%s", instanceId, language)
|
|
mandateId = _validateInstanceAccess(instanceId, context)
|
|
services = getAutomation2Services(
|
|
context.user,
|
|
mandateId=mandateId,
|
|
featureInstanceId=instanceId,
|
|
)
|
|
result = getNodeTypesForApi(services, language=language)
|
|
logger.info(
|
|
"automation2 node-types response: %d nodeTypes %d categories",
|
|
len(result.get("nodeTypes", [])),
|
|
len(result.get("categories", [])),
|
|
)
|
|
return result
|
|
|
|
|
|
@router.post("/{instanceId}/execute")
|
|
@limiter.limit("30/minute")
|
|
async def post_execute(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
body: dict = Body(..., description="{ workflowId?, graph: { nodes, connections } }"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Execute automation2 graph. Body: { workflowId?, graph: { nodes, connections } }."""
|
|
userId = str(context.user.id) if context.user else None
|
|
logger.info(
|
|
"automation2 execute request: instanceId=%s userId=%s body_keys=%s",
|
|
instanceId,
|
|
userId,
|
|
list(body.keys()),
|
|
)
|
|
mandateId = _validateInstanceAccess(instanceId, context)
|
|
services = getAutomation2Services(
|
|
context.user,
|
|
mandateId=mandateId,
|
|
featureInstanceId=instanceId,
|
|
)
|
|
# Ensure workflow methods (outlook, ai, sharepoint, etc.) are discovered for ActionExecutor
|
|
from modules.workflows.processing.shared.methodDiscovery import discoverMethods
|
|
discoverMethods(services)
|
|
|
|
graph = body.get("graph") or body
|
|
workflowId = body.get("workflowId")
|
|
req_nodes = graph.get("nodes") or []
|
|
# When workflowId is set: prefer graph from request (current editor state) if it has nodes.
|
|
# Only fall back to stored workflow graph when request graph is empty (e.g. resume from email).
|
|
if workflowId and len(req_nodes) == 0:
|
|
a2 = getAutomation2Interface(context.user, mandateId, instanceId)
|
|
wf = a2.getWorkflow(workflowId)
|
|
if wf and wf.get("graph"):
|
|
graph = wf["graph"]
|
|
logger.info("automation2 execute: loaded graph from workflow %s", workflowId)
|
|
# Use transient workflowId when none provided (e.g. execute from editor without save)
|
|
# Required for email.checkEmail pause/resume - run must be created
|
|
if not workflowId:
|
|
import uuid
|
|
workflowId = f"transient-{uuid.uuid4().hex[:12]}"
|
|
logger.info("automation2 execute: using transient workflowId=%s", workflowId)
|
|
nodes_count = len(graph.get("nodes") or [])
|
|
connections_count = len(graph.get("connections") or [])
|
|
logger.info(
|
|
"automation2 execute: graph nodes=%d connections=%d workflowId=%s mandateId=%s",
|
|
nodes_count,
|
|
connections_count,
|
|
workflowId,
|
|
mandateId,
|
|
)
|
|
a2_interface = getAutomation2Interface(context.user, mandateId, instanceId)
|
|
result = await executeGraph(
|
|
graph=graph,
|
|
services=services,
|
|
workflowId=workflowId,
|
|
instanceId=instanceId,
|
|
userId=userId,
|
|
mandateId=mandateId,
|
|
automation2_interface=a2_interface,
|
|
)
|
|
logger.info(
|
|
"automation2 execute result: success=%s error=%s nodeOutputs_keys=%s failedNode=%s paused=%s",
|
|
result.get("success"),
|
|
result.get("error"),
|
|
list(result.get("nodeOutputs", {}).keys()) if result.get("nodeOutputs") else [],
|
|
result.get("failedNode"),
|
|
result.get("paused"),
|
|
)
|
|
return result
|
|
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Connections and Browse (for Email/SharePoint node config - like workspace)
|
|
# -------------------------------------------------------------------------
|
|
|
|
|
|
def _buildResolverDbInterface(chatService):
|
|
"""Build a DB adapter that ConnectorResolver can use to load UserConnections."""
|
|
class _ResolverDbAdapter:
|
|
def __init__(self, appInterface):
|
|
self._app = appInterface
|
|
|
|
def getUserConnection(self, connectionId: str):
|
|
if hasattr(self._app, "getUserConnectionById"):
|
|
return self._app.getUserConnectionById(connectionId)
|
|
return None
|
|
|
|
appIf = getattr(chatService, "interfaceDbApp", None)
|
|
if appIf:
|
|
return _ResolverDbAdapter(appIf)
|
|
return getattr(chatService, "interfaceDbComponent", None)
|
|
|
|
|
|
@router.get("/{instanceId}/connections")
|
|
@limiter.limit("300/minute")
|
|
def list_automation2_connections(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Return the user's active connections (UserConnections) for Email/SharePoint node config."""
|
|
mandateId = _validateInstanceAccess(instanceId, context)
|
|
from modules.serviceCenter import getService
|
|
from modules.serviceCenter.context import ServiceCenterContext
|
|
ctx = ServiceCenterContext(
|
|
user=context.user,
|
|
mandate_id=str(context.mandateId) if context.mandateId else mandateId,
|
|
feature_instance_id=instanceId,
|
|
)
|
|
chatService = getService("chat", ctx)
|
|
connections = chatService.getUserConnections()
|
|
items = []
|
|
for c in connections or []:
|
|
conn = c if isinstance(c, dict) else (c.model_dump() if hasattr(c, "model_dump") else {})
|
|
authority = conn.get("authority")
|
|
if hasattr(authority, "value"):
|
|
authority = authority.value
|
|
status = conn.get("status")
|
|
if hasattr(status, "value"):
|
|
status = status.value
|
|
items.append({
|
|
"id": conn.get("id"),
|
|
"authority": authority,
|
|
"externalUsername": conn.get("externalUsername"),
|
|
"externalEmail": conn.get("externalEmail"),
|
|
"status": status,
|
|
})
|
|
return {"connections": items}
|
|
|
|
|
|
@router.get("/{instanceId}/connections/{connectionId}/services")
|
|
@limiter.limit("120/minute")
|
|
async def list_connection_services(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
connectionId: str = Path(..., description="Connection ID"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Return the available services for a specific UserConnection."""
|
|
mandateId = _validateInstanceAccess(instanceId, context)
|
|
try:
|
|
from modules.connectors.connectorResolver import ConnectorResolver
|
|
from modules.serviceCenter import getService as getSvc
|
|
from modules.serviceCenter.context import ServiceCenterContext
|
|
ctx = ServiceCenterContext(
|
|
user=context.user,
|
|
mandate_id=str(context.mandateId) if context.mandateId else mandateId,
|
|
feature_instance_id=instanceId,
|
|
)
|
|
chatService = getSvc("chat", ctx)
|
|
securityService = getSvc("security", ctx)
|
|
dbInterface = _buildResolverDbInterface(chatService)
|
|
resolver = ConnectorResolver(securityService, dbInterface)
|
|
provider = await resolver.resolve(connectionId)
|
|
services = provider.getAvailableServices()
|
|
_serviceLabels = {
|
|
"sharepoint": "SharePoint",
|
|
"outlook": "Outlook",
|
|
"teams": "Teams",
|
|
"onedrive": "OneDrive",
|
|
"drive": "Google Drive",
|
|
"gmail": "Gmail",
|
|
"files": "Files (FTP)",
|
|
}
|
|
_serviceIcons = {
|
|
"sharepoint": "sharepoint",
|
|
"outlook": "mail",
|
|
"teams": "chat",
|
|
"onedrive": "cloud",
|
|
"drive": "cloud",
|
|
"gmail": "mail",
|
|
"files": "folder",
|
|
}
|
|
items = [
|
|
{"service": s, "label": _serviceLabels.get(s, s), "icon": _serviceIcons.get(s, "folder")}
|
|
for s in services
|
|
]
|
|
return {"services": items}
|
|
except Exception as e:
|
|
logger.error(f"Error listing services for connection {connectionId}: {e}")
|
|
return JSONResponse({"services": [], "error": str(e)}, status_code=400)
|
|
|
|
|
|
@router.get("/{instanceId}/connections/{connectionId}/browse")
|
|
@limiter.limit("300/minute")
|
|
async def browse_connection_service(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
connectionId: str = Path(..., description="Connection ID"),
|
|
service: str = Query(..., description="Service name (e.g. sharepoint, onedrive, outlook)"),
|
|
path: str = Query("/", description="Path within the service to browse"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Browse folders/items within a connection's service at a given path."""
|
|
mandateId = _validateInstanceAccess(instanceId, context)
|
|
try:
|
|
from modules.connectors.connectorResolver import ConnectorResolver
|
|
from modules.serviceCenter import getService as getSvc
|
|
from modules.serviceCenter.context import ServiceCenterContext
|
|
ctx = ServiceCenterContext(
|
|
user=context.user,
|
|
mandate_id=str(context.mandateId) if context.mandateId else mandateId,
|
|
feature_instance_id=instanceId,
|
|
)
|
|
chatService = getSvc("chat", ctx)
|
|
securityService = getSvc("security", ctx)
|
|
dbInterface = _buildResolverDbInterface(chatService)
|
|
resolver = ConnectorResolver(securityService, dbInterface)
|
|
adapter = await resolver.resolveService(connectionId, service)
|
|
entries = await adapter.browse(path, filter=None)
|
|
items = []
|
|
for entry in (entries or []):
|
|
items.append({
|
|
"name": entry.name,
|
|
"path": entry.path,
|
|
"isFolder": entry.isFolder,
|
|
"size": entry.size,
|
|
"mimeType": entry.mimeType,
|
|
"metadata": entry.metadata if hasattr(entry, "metadata") else {},
|
|
})
|
|
return {"items": items, "path": path, "service": service}
|
|
except Exception as e:
|
|
logger.error(f"Error browsing {service} for connection {connectionId} at '{path}': {e}")
|
|
return JSONResponse({"items": [], "error": str(e)}, status_code=400)
|
|
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Workflow CRUD
|
|
# -------------------------------------------------------------------------
|
|
|
|
|
|
def _get_node_label_from_graph(graph: dict, nodeId: str) -> str:
|
|
"""Extract human-readable label for a node from graph."""
|
|
if not graph or not nodeId:
|
|
return nodeId or ""
|
|
nodes = graph.get("nodes") or []
|
|
for n in nodes:
|
|
if n.get("id") == nodeId:
|
|
params = n.get("parameters") or {}
|
|
config = params.get("config") or {}
|
|
if isinstance(config, dict):
|
|
label = config.get("title") or config.get("label")
|
|
else:
|
|
label = None
|
|
return (
|
|
n.get("title")
|
|
or label
|
|
or params.get("title")
|
|
or params.get("label")
|
|
or n.get("type", "")
|
|
or nodeId
|
|
)
|
|
return nodeId or ""
|
|
|
|
|
|
@router.get("/{instanceId}/workflows")
|
|
@limiter.limit("60/minute")
|
|
def get_workflows(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""List all workflows for this feature instance.
|
|
Enriches each workflow with runCount, isRunning, stuckAtNodeId, stuckAtNodeLabel,
|
|
createdAt, lastStartedAt.
|
|
"""
|
|
mandateId = _validateInstanceAccess(instanceId, context)
|
|
a2 = getAutomation2Interface(context.user, mandateId, instanceId)
|
|
items = a2.getWorkflows()
|
|
enriched = []
|
|
for wf in items:
|
|
wf_id = wf.get("id")
|
|
runs = a2.getRunsByWorkflow(wf_id) if wf_id else []
|
|
run_count = len(runs)
|
|
active_run = None
|
|
last_started_at = None
|
|
for r in runs:
|
|
ts = r.get("_createdAt")
|
|
if ts and (last_started_at is None or ts > last_started_at):
|
|
last_started_at = ts
|
|
if r.get("status") in ("running", "paused"):
|
|
active_run = r
|
|
stuck_at_node_id = active_run.get("currentNodeId") if active_run else None
|
|
stuck_at_node_label = ""
|
|
if stuck_at_node_id and wf.get("graph"):
|
|
stuck_at_node_label = _get_node_label_from_graph(wf["graph"], stuck_at_node_id)
|
|
enriched.append({
|
|
**wf,
|
|
"runCount": run_count,
|
|
"isRunning": active_run is not None,
|
|
"runStatus": active_run.get("status") if active_run else None,
|
|
"stuckAtNodeId": stuck_at_node_id,
|
|
"stuckAtNodeLabel": stuck_at_node_label or stuck_at_node_id or "",
|
|
"createdAt": wf.get("_createdAt"),
|
|
"lastStartedAt": last_started_at,
|
|
})
|
|
return {"workflows": enriched}
|
|
|
|
|
|
@router.get("/{instanceId}/workflows/{workflowId}")
|
|
@limiter.limit("60/minute")
|
|
def get_workflow(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
workflowId: str = Path(..., description="Workflow ID"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Get a single workflow by ID."""
|
|
mandateId = _validateInstanceAccess(instanceId, context)
|
|
a2 = getAutomation2Interface(context.user, mandateId, instanceId)
|
|
wf = a2.getWorkflow(workflowId)
|
|
if not wf:
|
|
raise HTTPException(status_code=404, detail="Workflow not found")
|
|
return wf
|
|
|
|
|
|
@router.post("/{instanceId}/workflows")
|
|
@limiter.limit("30/minute")
|
|
def create_workflow(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
body: dict = Body(..., description="{ label, graph }"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Create a new workflow."""
|
|
mandateId = _validateInstanceAccess(instanceId, context)
|
|
a2 = getAutomation2Interface(context.user, mandateId, instanceId)
|
|
created = a2.createWorkflow(body)
|
|
return created
|
|
|
|
|
|
@router.put("/{instanceId}/workflows/{workflowId}")
|
|
@limiter.limit("30/minute")
|
|
def update_workflow(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
workflowId: str = Path(..., description="Workflow ID"),
|
|
body: dict = Body(..., description="{ label?, graph? }"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Update a workflow."""
|
|
mandateId = _validateInstanceAccess(instanceId, context)
|
|
a2 = getAutomation2Interface(context.user, mandateId, instanceId)
|
|
updated = a2.updateWorkflow(workflowId, body)
|
|
if not updated:
|
|
raise HTTPException(status_code=404, detail="Workflow not found")
|
|
return updated
|
|
|
|
|
|
@router.delete("/{instanceId}/workflows/{workflowId}")
|
|
@limiter.limit("30/minute")
|
|
def delete_workflow(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
workflowId: str = Path(..., description="Workflow ID"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Delete a workflow."""
|
|
mandateId = _validateInstanceAccess(instanceId, context)
|
|
a2 = getAutomation2Interface(context.user, mandateId, instanceId)
|
|
if not a2.deleteWorkflow(workflowId):
|
|
raise HTTPException(status_code=404, detail="Workflow not found")
|
|
return {"success": True}
|
|
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Runs and Resume
|
|
# -------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/{instanceId}/workflows/{workflowId}/runs")
|
|
@limiter.limit("60/minute")
|
|
def get_workflow_runs(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
workflowId: str = Path(..., description="Workflow ID"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Get runs for a workflow."""
|
|
mandateId = _validateInstanceAccess(instanceId, context)
|
|
a2 = getAutomation2Interface(context.user, mandateId, instanceId)
|
|
if not a2.getWorkflow(workflowId):
|
|
raise HTTPException(status_code=404, detail="Workflow not found")
|
|
runs = a2.getRunsByWorkflow(workflowId)
|
|
return {"runs": runs}
|
|
|
|
|
|
@router.post("/{instanceId}/runs/{runId}/resume")
|
|
@limiter.limit("30/minute")
|
|
async def resume_run(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
runId: str = Path(..., description="Run ID"),
|
|
body: dict = Body(..., description="{ taskId, result }"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Resume a paused run after task completion."""
|
|
mandateId = _validateInstanceAccess(instanceId, context)
|
|
a2 = getAutomation2Interface(context.user, mandateId, instanceId)
|
|
run = a2.getRun(runId)
|
|
if not run:
|
|
raise HTTPException(status_code=404, detail="Run not found")
|
|
taskId = body.get("taskId")
|
|
result = body.get("result")
|
|
if not taskId or result is None:
|
|
raise HTTPException(status_code=400, detail="taskId and result required")
|
|
task = a2.getTask(taskId)
|
|
if not task or task.get("runId") != runId:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
if task.get("status") != "pending":
|
|
raise HTTPException(status_code=400, detail="Task already completed")
|
|
a2.updateTask(taskId, status="completed", result=result)
|
|
nodeId = task.get("nodeId")
|
|
nodeOutputs = dict(run.get("nodeOutputs") or {})
|
|
nodeOutputs[nodeId] = result
|
|
runContext = run.get("context") or {}
|
|
connectionMap = runContext.get("connectionMap", {})
|
|
inputSources = runContext.get("inputSources", {})
|
|
workflowId = run.get("workflowId")
|
|
wf = a2.getWorkflow(workflowId) if workflowId else None
|
|
if not wf or not wf.get("graph"):
|
|
raise HTTPException(status_code=400, detail="Workflow graph not found")
|
|
graph = wf["graph"]
|
|
services = getAutomation2Services(context.user, mandateId=mandateId, featureInstanceId=instanceId)
|
|
resume_result = await executeGraph(
|
|
graph=graph,
|
|
services=services,
|
|
workflowId=workflowId,
|
|
instanceId=instanceId,
|
|
userId=str(context.user.id) if context.user else None,
|
|
mandateId=mandateId,
|
|
automation2_interface=a2,
|
|
initialNodeOutputs=nodeOutputs,
|
|
startAfterNodeId=nodeId,
|
|
runId=runId,
|
|
)
|
|
return resume_result
|
|
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Tasks
|
|
# -------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/{instanceId}/tasks")
|
|
@limiter.limit("60/minute")
|
|
def get_tasks(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
workflowId: str = Query(None, description="Filter by workflow ID"),
|
|
status: str = Query(None, description="Filter: pending, completed, rejected"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Get tasks - by default those assigned to current user, or all if no assignee filter.
|
|
Enriches each task with workflowLabel and createdAt (_createdAt).
|
|
"""
|
|
mandateId = _validateInstanceAccess(instanceId, context)
|
|
a2 = getAutomation2Interface(context.user, mandateId, instanceId)
|
|
assigneeId = str(context.user.id) if context.user else None
|
|
items = a2.getTasks(workflowId=workflowId, status=status, assigneeId=assigneeId)
|
|
workflows = {w["id"]: w for w in a2.getWorkflows()}
|
|
enriched = []
|
|
for t in items:
|
|
wf = workflows.get(t.get("workflowId") or "")
|
|
enriched.append({
|
|
**t,
|
|
"workflowLabel": wf.get("label", t.get("workflowId", "")) if wf else t.get("workflowId", ""),
|
|
"createdAt": t.get("_createdAt"),
|
|
})
|
|
return {"tasks": enriched}
|
|
|
|
|
|
@router.post("/{instanceId}/tasks/{taskId}/complete")
|
|
@limiter.limit("30/minute")
|
|
async def complete_task(
|
|
request: Request,
|
|
instanceId: str = Path(..., description="Feature instance ID"),
|
|
taskId: str = Path(..., description="Task ID"),
|
|
body: dict = Body(..., description="{ result }"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Complete a task and resume the run."""
|
|
mandateId = _validateInstanceAccess(instanceId, context)
|
|
a2 = getAutomation2Interface(context.user, mandateId, instanceId)
|
|
task = a2.getTask(taskId)
|
|
if not task:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
runId = task.get("runId")
|
|
result = body.get("result")
|
|
if result is None:
|
|
raise HTTPException(status_code=400, detail="result required")
|
|
run = a2.getRun(runId)
|
|
if not run:
|
|
raise HTTPException(status_code=404, detail="Run not found")
|
|
if task.get("status") != "pending":
|
|
raise HTTPException(status_code=400, detail="Task already completed")
|
|
a2.updateTask(taskId, status="completed", result=result)
|
|
nodeId = task.get("nodeId")
|
|
nodeOutputs = dict(run.get("nodeOutputs") or {})
|
|
nodeOutputs[nodeId] = result
|
|
workflowId = run.get("workflowId")
|
|
wf = a2.getWorkflow(workflowId) if workflowId else None
|
|
if not wf or not wf.get("graph"):
|
|
raise HTTPException(status_code=400, detail="Workflow graph not found")
|
|
graph = wf["graph"]
|
|
services = getAutomation2Services(context.user, mandateId=mandateId, featureInstanceId=instanceId)
|
|
return await executeGraph(
|
|
graph=graph,
|
|
services=services,
|
|
workflowId=workflowId,
|
|
instanceId=instanceId,
|
|
userId=str(context.user.id) if context.user else None,
|
|
mandateId=mandateId,
|
|
automation2_interface=a2,
|
|
initialNodeOutputs=nodeOutputs,
|
|
startAfterNodeId=nodeId,
|
|
runId=runId,
|
|
)
|