diff --git a/AGENTS.md b/AGENTS.md index 269ca93e..ef7c7bfb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -2084,6 +2084,10 @@ Existing Agent Run with additional flow linkage fields. Edges are sorted by priority (descending); first match wins. +**Approval routing**: On `human.approval`, the engine follows the edge whose `meta.outcome` matches the decision (`approved`/`rejected`). An approval falls back to the success path when no matching edge exists; a rejection without an explicit `meta.outcome=rejected` edge FAILS the run instead of falling through the success path. + +**Loop routing**: `loop` nodes route to their body or done node via the `next_node_id` returned by the loop executor; without a done node the flow completes gracefully when the loop finishes. + ### Execution Modes **Normal Mode**: Engine follows edges deterministically. Agents run only when `agent.run` nodes are hit. @@ -2126,7 +2130,8 @@ Registered via `huf_tools` hook so agents can interact with flows: ### Security - **Expression edges**: AST-based restricted evaluator; no imports, no function calls, no attribute access -- **Human approval**: User/role verification before approve/reject +- **Human approval**: User/role verification before approve/reject (`approval_type` `user` and `users` are treated as synonyms) +- **Permissions**: Huf Manager has `create` on Flow Run (requires `bench migrate` on deploy to apply); agent tool handlers (`run_flow`, `get_flow_run`, `resume_flow_run`, `approve_flow_run`) enforce the same permission checks and approver-identity verification as the REST endpoints - **Hop limit**: Safety guard against infinite loops (default 100) ## Subsystems and Advanced Architectures diff --git a/frontend/src/components/ApprovalsBell.tsx b/frontend/src/components/ApprovalsBell.tsx new file mode 100644 index 00000000..8080d7d2 --- /dev/null +++ b/frontend/src/components/ApprovalsBell.tsx @@ -0,0 +1,105 @@ +import { useCallback, useEffect, useState } from 'react'; +import { Link } from 'react-router-dom'; +import { Bell } from 'lucide-react'; +import { Popover, PopoverContent, PopoverTrigger } from './ui/popover'; +import { Button } from './ui/button'; +import { Badge } from './ui/badge'; +import { approveFlowRun, getPendingApprovals, rejectFlowRun } from '../services/flowApi'; +import type { PendingApproval } from '../services/flowApi'; + +const REFRESH_INTERVAL_MS = 60000; +const FLOW_EVENTS = ['frappe:flow_paused', 'frappe:flow_completed', 'frappe:flow_error']; + +export function ApprovalsBell() { + const [approvals, setApprovals] = useState([]); + + const fetchApprovals = useCallback(async () => { + try { + const list = await getPendingApprovals(); + setApprovals(list || []); + } catch (error) { + console.warn('ApprovalsBell: failed to fetch pending approvals', error); + } + }, []); + + useEffect(() => { + fetchApprovals(); + const interval = setInterval(fetchApprovals, REFRESH_INTERVAL_MS); + FLOW_EVENTS.forEach((event) => window.addEventListener(event, fetchApprovals)); + return () => { + clearInterval(interval); + FLOW_EVENTS.forEach((event) => window.removeEventListener(event, fetchApprovals)); + }; + }, [fetchApprovals]); + + const handleApprove = async (flowRunId: string) => { + try { + await approveFlowRun(flowRunId); + } catch (error) { + console.warn('ApprovalsBell: failed to approve flow run', error); + } + await fetchApprovals(); + }; + + const handleReject = async (flowRunId: string) => { + try { + await rejectFlowRun(flowRunId); + } catch (error) { + console.warn('ApprovalsBell: failed to reject flow run', error); + } + await fetchApprovals(); + }; + + const count = approvals.length; + + return ( + + + + + +
Pending approvals
+ {count === 0 ? ( +
+ No approvals waiting +
+ ) : ( +
+ {approvals.map((approval) => ( +
+
+
{approval.title || approval.flow_id}
+
+ {approval.flow_id} + {approval.waiting_since ? ` · waiting since ${approval.waiting_since}` : ''} +
+
+
+ + + +
+
+ ))} +
+ )} + + + ); +} diff --git a/frontend/src/components/RightSidebar.tsx b/frontend/src/components/RightSidebar.tsx index 8399ff45..2e82fe5e 100644 --- a/frontend/src/components/RightSidebar.tsx +++ b/frontend/src/components/RightSidebar.tsx @@ -413,12 +413,12 @@ export function RightSidebar({ onToggle, variant = 'panel' }: RightSidebarProps)
- {selectedEdge.data?.type === 'expression' && ( + {selectedEdge.data?.edgeType === 'expression' && (
{ if (!activeFlow) return; updateEdges( activeFlow.edges.map((edge) => - edge.id === selectedEdge.id ? { ...edge, data: { ...edge.data, expression: e.target.value } } : edge + edge.id === selectedEdge.id ? { ...edge, data: { ...edge.data, condition: e.target.value } } : edge ) ); }} - placeholder="e.g., {{context.status}} == 'approved'" + placeholder='e.g., context["status"] == "approved"' />
)} + +
+ + { + if (!activeFlow) return; + const priority = e.target.value === '' ? 0 : Number(e.target.value); + updateEdges( + activeFlow.edges.map((edge) => + edge.id === selectedEdge.id ? { ...edge, data: { ...edge.data, priority } } : edge + ) + ); + }} + /> +

Higher priority edges are evaluated first

+
+ +
+ + +

For edges leaving a Human Approval node: route this edge when the decision matches.

+
) : selectedNode ? ( diff --git a/frontend/src/contexts/SocketContext.tsx b/frontend/src/contexts/SocketContext.tsx index 3ebee55b..743afceb 100644 --- a/frontend/src/contexts/SocketContext.tsx +++ b/frontend/src/contexts/SocketContext.tsx @@ -79,6 +79,7 @@ export function SocketProvider({ children }: { children: ReactNode }) { 'flow_node_end', 'flow_paused', 'flow_completed', + 'flow_failed', 'flow_error', ]; diff --git a/frontend/src/layouts/UnifiedHeader.tsx b/frontend/src/layouts/UnifiedHeader.tsx index 2b9ca6ca..b56d46bb 100644 --- a/frontend/src/layouts/UnifiedHeader.tsx +++ b/frontend/src/layouts/UnifiedHeader.tsx @@ -9,6 +9,7 @@ import { BreadcrumbSeparator, } from '../components/ui/breadcrumb'; import { BreadcrumbItem as BreadcrumbItemType } from './UnifiedLayout'; +import { ApprovalsBell } from '../components/ApprovalsBell'; interface UnifiedHeaderProps { actions?: ReactNode; @@ -85,6 +86,7 @@ export function UnifiedHeader({ actions, breadcrumbs }: UnifiedHeaderProps) { */}
+ {actions}
diff --git a/frontend/src/services/flowApi.ts b/frontend/src/services/flowApi.ts index 69c123d5..a386b234 100644 --- a/frontend/src/services/flowApi.ts +++ b/frontend/src/services/flowApi.ts @@ -39,7 +39,7 @@ export interface BackendFlowGraph { export interface BackendNode { id: string; - type: 'trigger.webhook' | 'agent.run' | 'tool.call' | 'router.llm' | 'human.approval' | 'condition' | 'http_request' | 'transform' | 'loop' | 'end'; + type: 'trigger.webhook' | 'trigger.schedule' | 'trigger.doc-event' | 'agent.run' | 'tool.call' | 'router.llm' | 'human.approval' | 'condition' | 'http_request' | 'transform' | 'loop' | 'end'; config: Record; /** Frontend-only: stored for visual layout, ignored by engine */ _position?: { x: number; y: number }; @@ -104,6 +104,19 @@ export interface FlowRunDetail { completed_at: string | null; } +/** Pending human approval (from get_pending_approvals endpoint) */ +export interface PendingApproval { + flow_run_id: string; + flow_id: string; + current_node_id: string; + title: string; + instructions: string; + approval_type: string; + started_at: string | null; + waiting_since: string | null; + view_link: string; +} + // ─── Flow Definition APIs ──────────────────────────────────────────── const FLOW_LIST_FIELDS = [ @@ -281,3 +294,13 @@ export async function resumeFlowRun( handleFrappeError(error, `Error resuming flow run ${flowRunId}`); } } + +/** List flow runs waiting for human approval */ +export async function getPendingApprovals(): Promise { + try { + const result = await call.get('huf.ai.flow_api.get_pending_approvals'); + return result.message as PendingApproval[]; + } catch (error) { + handleFrappeError(error, 'Error fetching pending approvals'); + } +} diff --git a/frontend/src/services/flowSerializer.ts b/frontend/src/services/flowSerializer.ts index 5e7a2a0f..4db7fb12 100644 --- a/frontend/src/services/flowSerializer.ts +++ b/frontend/src/services/flowSerializer.ts @@ -60,6 +60,9 @@ function mapFrontendNodeTypeToBackend(node: FlowNode): BackendNode['type'] { if (nodeType === 'end') return 'end'; if (nodeType === 'trigger') { + const triggerType = node.data?.triggerConfig?.type; + if (triggerType === 'schedule') return 'trigger.schedule'; + if (triggerType === 'doc-event') return 'trigger.doc-event'; return 'trigger.webhook'; } @@ -178,10 +181,16 @@ function buildNodeData(node: BackendNode): FlowNodeData { }; if (frontendType === 'trigger') { + const triggerType = + node.type === 'trigger.schedule' + ? ('schedule' as const) + : node.type === 'trigger.doc-event' + ? ('doc-event' as const) + : ('webhook' as const); base.triggerConfig = { - type: 'webhook' as const, + type: triggerType, ...node.config, - }; + } as FlowNodeData['triggerConfig']; } else if (frontendType === 'action') { base.actionConfig = { type: mapBackendActionType(node.type), @@ -195,6 +204,8 @@ function buildNodeData(node: BackendNode): FlowNodeData { function getDefaultLabel(backendType: string): string { const labels: Record = { 'trigger.webhook': 'Webhook Trigger', + 'trigger.schedule': 'Schedule Trigger', + 'trigger.doc-event': 'Document Event Trigger', 'agent.run': 'Run Agent', 'tool.call': 'Call Tool', 'router.llm': 'LLM Router', @@ -211,6 +222,8 @@ function getDefaultLabel(backendType: string): string { function getDefaultIcon(backendType: string): string { const icons: Record = { 'trigger.webhook': 'Webhook', + 'trigger.schedule': 'Clock', + 'trigger.doc-event': 'Database', 'agent.run': 'Bot', 'tool.call': 'Play', 'router.llm': 'GitBranch', diff --git a/huf/ai/flow_api.py b/huf/ai/flow_api.py index 399d4c90..563c75ed 100644 --- a/huf/ai/flow_api.py +++ b/huf/ai/flow_api.py @@ -282,7 +282,7 @@ def approve_flow_run(flow_run_id: str, comment: str | None = None) -> dict: Returns: dict with status and current_node_id """ - if not frappe.has_permission("Flow Run", "write"): + if not frappe.has_permission("Flow Run", "read"): frappe.throw(_("Not permitted"), frappe.PermissionError) from huf.ai.flow_engine import approve_flow_run as engine_approve @@ -380,6 +380,9 @@ def flow_webhook(flow_id: str, webhook_key: str | None = None) -> dict: if not _webhook_key_is_valid(defn, webhook_key): frappe.throw(_("Invalid webhook key"), frappe.AuthenticationError) + # Switch execution identity to the flow owner so the run does not execute as Guest + frappe.set_user(defn_doc.owner or "Administrator") + # Get payload from request payload = {} if frappe.request: @@ -801,6 +804,9 @@ def handle_run_flow(flow_id: str, payload: str | dict | None = None, mode: str | Returns: dict with flow_run_id, status, message """ + if not frappe.has_permission("Flow Run", "create"): + return {"error": "Insufficient permissions to start flows"} + try: if isinstance(payload, str): try: @@ -838,6 +844,9 @@ def handle_get_flow_run(flow_run_id: str, **kwargs) -> dict: Returns: dict with status, context summary, waiting state """ + if not frappe.has_permission("Flow Run", "read"): + return {"error": "Insufficient permissions to view flow runs"} + try: doc = frappe.get_doc("Flow Run", flow_run_id) ctx = {} @@ -875,6 +884,9 @@ def handle_resume_flow_run(flow_run_id: str, input: dict | None = None, **kwargs Returns: dict with updated status """ + if not frappe.has_permission("Flow Run", "write"): + return {"error": "Insufficient permissions to resume flow runs"} + try: from huf.ai.flow_engine import resume_flow_run as engine_resume @@ -903,11 +915,18 @@ def handle_approve_flow_run(flow_run_id: str, decision: str = "approved", commen Returns: dict with updated status """ + if not frappe.has_permission("Flow Run", "read"): + return {"error": "Insufficient permissions to approve flow runs"} + try: - from huf.ai.flow_engine import approve_flow_run as engine_approve + from huf.ai.flow_engine import approve_flow_run as engine_approve, _verify_approval_permission - engine_approve(flow_run_id, decision=decision, comment=comment) doc = frappe.get_doc("Flow Run", flow_run_id) + waiting = json.loads(doc.waiting) if doc.waiting else {} + _verify_approval_permission(waiting) + + engine_approve(flow_run_id, decision=decision, comment=comment) + doc.reload() return { "success": True, @@ -977,7 +996,7 @@ def get_pending_approvals(limit: int = 50) -> list: approver_role = waiting.get("approver_role") if approver_role and approver_role in user_roles: can_approve = True - elif approval_type == "users": + elif approval_type in ("user", "users"): approver_users = waiting.get("approver_users", []) if isinstance(approver_users, str): approver_users = [u.strip() for u in approver_users.split(",") if u.strip()] diff --git a/huf/ai/flow_engine.py b/huf/ai/flow_engine.py index 2863203f..4d884a5d 100644 --- a/huf/ai/flow_engine.py +++ b/huf/ai/flow_engine.py @@ -218,10 +218,19 @@ def approve_flow_run(flow_run_name: str, decision: str, comment: str | None = No next_node = edge.get("to") break - if not next_node: + if not next_node and decision == "approved": next_node = _evaluate_edges(flow_run, current_node, {"status": "success"}, edges_list) if not next_node: + if decision == "rejected": + # Rejection without an explicit 'rejected' edge must not route like approval + flow_run.db_set("waiting", None) + _fail_flow_run( + flow_run, + f"Approval rejected by {frappe.session.user}" + (f": {comment}" if comment else ""), + ) + _clear_flow_notifications(flow_run) + return # No outgoing edges — approval was the final step, complete the flow gracefully outgoing = [e for e in edges_list if e.get("from") == current_node] if not outgoing: @@ -327,6 +336,15 @@ def _execute_loop(flow_run, nodes_map: dict, edges_list: list, settings: dict): _fail_flow_run(flow_run, "Condition node did not resolve a branch") return + elif node.get("type") == "loop": + # Loop executor returns next_node_id (body node or done node) + next_node_id = node_result.get("next_node_id") if isinstance(node_result, dict) else None + if not next_node_id: + # No done_node configured — loop finished, complete gracefully + _complete_flow_run(flow_run) + _publish_flow_event(flow_run, "flow_completed", {"status": "Success"}) + return + elif node.get("type") == "human.approval": # Already paused above; this shouldn't be reached return @@ -755,7 +773,7 @@ def _send_approval_notifications(flow_run, node: dict, config: dict, waiting_dat }, pluck="name" ) - elif approval_type == "users": + elif approval_type in ("user", "users"): approver_users = waiting_data.get("approver_users", []) if isinstance(approver_users, str): # Handle comma-separated string @@ -1303,7 +1321,7 @@ def _verify_approval_permission(waiting: dict): approval_type = waiting.get("approval_type", "role") user = frappe.session.user - if approval_type == "user": + if approval_type in ("user", "users"): approver_users = waiting.get("approver_users", []) if approver_users and user not in approver_users: frappe.throw( diff --git a/huf/huf/doctype/flow_definition/flow_definition.py b/huf/huf/doctype/flow_definition/flow_definition.py index 16c4f786..2d069888 100644 --- a/huf/huf/doctype/flow_definition/flow_definition.py +++ b/huf/huf/doctype/flow_definition/flow_definition.py @@ -8,6 +8,8 @@ ALLOWED_NODE_TYPES = { "trigger.webhook", + "trigger.schedule", + "trigger.doc-event", "agent.run", "tool.call", "router.llm", diff --git a/huf/huf/doctype/flow_run/flow_run.json b/huf/huf/doctype/flow_run/flow_run.json index 17570b85..4105e057 100644 --- a/huf/huf/doctype/flow_run/flow_run.json +++ b/huf/huf/doctype/flow_run/flow_run.json @@ -211,7 +211,7 @@ "write": 1 }, { - "create": 0, + "create": 1, "delete": 0, "email": 1, "export": 1,