Mission M-D — Flow Builder Revamp

- Unlocks CONDITION and WAIT_DELAY nodes (no longer blocked from publish)
- Adds PARALLEL, AGENT_HANDOFF, QUALIFICATION_GATE, INTENT_ROUTER node types
- builder_translator.translate() now returns dual output: (definition_json, adk_pipeline_config)
- New translate_to_adk_pipeline() function maps builder graph → ADK orchestrator config
- FlowVersion.adk_pipeline_config column (JSON) populated on publish
- ExecutionInstance.resume_at + resume_node_id columns for WAIT_DELAY resumption
- Alembic migration s9t0u1v2w3x4
- build_orchestrator() accepts pipeline_config; applies builder tone/routing overrides
- build_reply_agent() accepts tone + max_length params
- Celery: resume_waiting_instances_task (every 60s) resumes WAITING instances
- Frontend: ActionNode multi-handle (CONDITION: true/false, QUALIFICATION_GATE: qualified/unqualified)
- Frontend: NodePalette groups (Triggers, Agent Nodes, Flow Control, Routing)
- Frontend: full config forms for all new node types
- 352 tests passing (8 new)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

backend/alembic/versions/s9t0u1v2w3x4_mission_md_flow_builder_revamp.py

@@ -0,0 +1,48 @@
+"""Mission M-D: flow builder revamp — add adk_pipeline_config to flowversion,
+add resume_at + resume_node_id to executioninstance
+
+Revision ID: s9t0u1v2w3x4
+Revises: r8s9t0u1v2w3
+Create Date: 2026-03-24
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers
+revision = "s9t0u1v2w3x4"
+down_revision = "r8s9t0u1v2w3"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+
+    # --- flowversion: add adk_pipeline_config ---
+    fv_cols = {c["name"] for c in inspector.get_columns("flowversion")}
+    if "adk_pipeline_config" not in fv_cols:
+        op.add_column(
+            "flowversion",
+            sa.Column("adk_pipeline_config", sa.JSON(), nullable=True),
+        )
+
+    # --- executioninstance: add resume_at + resume_node_id ---
+    ei_cols = {c["name"] for c in inspector.get_columns("executioninstance")}
+    if "resume_at" not in ei_cols:
+        op.add_column(
+            "executioninstance",
+            sa.Column("resume_at", sa.DateTime(), nullable=True),
+        )
+    if "resume_node_id" not in ei_cols:
+        op.add_column(
+            "executioninstance",
+            sa.Column("resume_node_id", sa.String(), nullable=True),
+        )
+
+
+def downgrade() -> None:
+    op.drop_column("executioninstance", "resume_node_id")
+    op.drop_column("executioninstance", "resume_at")
+    op.drop_column("flowversion", "adk_pipeline_config")

backend/app/api/v1/automations.py

@@ -472,8 +472,8 @@ async def publish_flow(
             "errors": errors,
         })
 
-    # Translate to runtime contract
-    definition_json = translate(draft.builder_graph_json)
+    # Translate to runtime contract (dual output: legacy + ADK pipeline)
+    definition_json, adk_pipeline_config = translate(draft.builder_graph_json)
 
     # Get next version number
     version_result = await db.execute(
@@ -488,6 +488,7 @@ async def publish_flow(
         flow_id=flow_id,
         version_number=new_version_number,
         definition_json=definition_json,
+        adk_pipeline_config=adk_pipeline_config,
         is_published=True,
         created_at=now,
         updated_at=now,

backend/app/core/adk/agents/orchestrator.py

@@ -37,6 +37,7 @@ async def build_orchestrator(
     conversation_id: UUID,
     session: AsyncSession,
     instance: ExecutionInstance,
+    pipeline_config: dict | None = None,
 ) -> LlmAgent:
     """
     Build the full multi-agent graph for one conversation turn.
@@ -88,7 +89,12 @@ async def build_orchestrator(
         for c in criteria_result.scalars().all()
     ]
 
-    # 5. Build sub-agents
+    # 5. Extract builder-configured policies (Mission M-D)
+    _pipeline = pipeline_config or {}
+    _reply_cfg = _pipeline.get("agents", {}).get("reply", {})
+    _routing_rules = _pipeline.get("agents", {}).get("orchestrator", {}).get("routing_rules", [])
+
+    # Build sub-agents (apply builder tone/max_length overrides to reply agent)
     qualification_agent = build_qualification_agent(
         qualification_questions=qualification_questions,
         qualification_criteria=qualification_criteria,
@@ -96,9 +102,41 @@ async def build_orchestrator(
         instance=instance,
     )
     crm_agent = build_crm_agent(session=session, instance=instance)
-    reply_agent = build_reply_agent(platform=platform, session=session, instance=instance)
+    reply_agent = build_reply_agent(
+        platform=platform,
+        session=session,
+        instance=instance,
+        tone=_reply_cfg.get("tone", "professional"),
+        max_length=_reply_cfg.get("max_length", 160),
+    )
     handover_agent = build_handover_agent(session=session, instance=instance)
 
+    # Build routing rules addendum from builder config
+    _routing_addendum = ""
+    if _routing_rules:
+        rule_lines = []
+        for rule in _routing_rules:
+            rt = rule.get("type")
+            if rt == "qualification_gate":
+                rule_lines.append(
+                    "- Qualification Gate: route qualified leads to the reply path, "
+                    "unqualified leads to the qualification path."
+                )
+            elif rt == "intent_router":
+                for r in rule.get("routes", []):
+                    rule_lines.append(
+                        f"- If detected intent is '{r.get('intent')}', delegate to {r.get('agent')} agent."
+                    )
+            elif rt == "parallel":
+                agents = ", ".join(rule.get("parallel_agents", []))
+                rule_lines.append(f"- Run these agents in parallel when appropriate: {agents}.")
+            elif rt == "agent_handoff":
+                rule_lines.append(
+                    f"- Explicit handoff configured: delegate to {rule.get('target_agent')} agent."
+                )
+        if rule_lines:
+            _routing_addendum = "\nBuilder-configured routing rules (apply these in addition to defaults):\n" + "\n".join(rule_lines)
+
     # 6. Orchestrator instruction with injected state
     orchestrator_instruction = f"""You are the Conversation Orchestrator for this business's AI system.
 
@@ -115,7 +153,7 @@ Routing rules (follow in order):
 5. After any sub-agent completes its task, always end the turn by delegating to reply_agent to send a response.
 
 Business context and tone:
-{compiled.system_instruction}
+{compiled.system_instruction}{_routing_addendum}
 """
 
     return LlmAgent(

backend/app/core/adk/agents/reply.py

@@ -19,6 +19,8 @@ def build_reply_agent(
     platform: str,
     session: AsyncSession,
     instance: ExecutionInstance,
+    tone: str = "professional",
+    max_length: int = 160,
 ) -> LlmAgent:
     """
     Build the reply agent for a specific platform.
@@ -27,12 +29,14 @@ def build_reply_agent(
         platform: Messaging platform ("whatsapp" | "messenger" | "instagram")
         session: DB session (bound via closure)
         instance: Current execution instance (bound via closure)
+        tone: Desired reply tone from builder config (default: professional)
+        max_length: Max message character length from builder config (default: 160)
     """
     instruction = f"""You are the Reply Agent. Your ONLY job is to craft and send messages to the lead on {platform}.
 
 Rules:
-- Keep WhatsApp messages under 160 characters unless detail is genuinely required.
-- Match the business tone from the conversation context.
+- Keep messages under {max_length} characters unless detail is genuinely required.
+- Use a {tone} tone in all messages.
 - Never send multiple messages when one will do.
 - Use send_reply() for all text messages.
 - Use send_media_message() only when an image or document is explicitly needed.

backend/app/core/adk/runner.py

@@ -19,6 +19,7 @@ from app.models.models import (
     ExecutionInstance,
     ExecutionStatus,
     ExecutionStepLog,
+    FlowVersion,
     Message,
 )
 from app.services.runtime_event_service import log_event
@@ -44,12 +45,17 @@ async def run_for_contact(
     marks the ExecutionInstance COMPLETED or FAILED.
     """
     try:
+        # Load ADK pipeline config from the published FlowVersion
+        flow_version = await session.get(FlowVersion, execution_instance.flow_version_id)
+        pipeline_config = (flow_version.adk_pipeline_config or {}) if flow_version else {}
+
         agent = await build_orchestrator(
             workspace_id=workspace_id,
             contact_id=contact_id,
             conversation_id=conversation_id,
             session=session,
             instance=execution_instance,
+            pipeline_config=pipeline_config,
         )
 
         session_service = LeadPilotSessionService(db_engine=engine)

backend/app/core/celery_app.py

@@ -39,6 +39,11 @@ celery_app.conf.beat_schedule = {
         "task": "app.workers.tasks.export_to_hf_task",
         "schedule": crontab(hour=2, minute=0),
     },
+    # Mission M-D: resume WAITING ExecutionInstances after WAIT_DELAY expires
+    "resume_waiting_instances_every_minute": {
+        "task": "app.workers.tasks.resume_waiting_instances_task",
+        "schedule": 60.0,
+    },
 }
 
 celery_app.autodiscover_tasks(["app.workers"])

backend/app/domain/builder_translator.py

@@ -1,9 +1,9 @@
 """
-Builder Translator — Mission 27
+Builder Translator — Mission 27 / M-D
 
-Converts builder_graph_json (React Flow native format) into the runtime
-definition_json contract, validates graphs before publish, and simulates
-traversal without side effects.
+Converts builder_graph_json (React Flow native format) into:
+  1. runtime definition_json (legacy format, stored in FlowVersion for audit/display)
+  2. adk_pipeline_config (new format, drives the ADK orchestrator at runtime)
 
 Builder graph format (stored in FlowDraft.builder_graph_json):
 {
@@ -22,6 +22,19 @@ Runtime definition_json format (stored in FlowVersion.definition_json):
     "nodes": [{"id": "...", "type": "TRIGGER", "config": {...}}, ...],
     "edges": [{"id": "...", "source_node_id": "...", "target_node_id": "...", "source_handle": null}]
 }
+
+ADK pipeline config format (stored in FlowVersion.adk_pipeline_config):
+{
+    "pipeline_type": "orchestrated",
+    "agents": {
+        "orchestrator": {"routing_rules": [...], "sub_agents": []},
+        "reply": {"tone": "...", "max_length": 160},
+        ...
+    },
+    "triggers": [...],
+    "condition_branches": [...],
+    "wait_delays": [...]
+}
 """
 from __future__ import annotations
 
@@ -33,14 +46,17 @@ logger = logging.getLogger(__name__)
 # Trigger node types that map to runtime TRIGGER node type
 TRIGGER_NODE_TYPES = {"MESSAGE_INBOUND", "LEAD_AD_SUBMIT"}
 
-# Node types supported by the runtime engine
+# All node types supported by the runtime engine (Mission M-D: unlocked + new)
 RUNTIME_SUPPORTED_NODE_TYPES = {
-    "AI_REPLY", "SEND_MESSAGE", "HUMAN_HANDOVER",
+    "AI_REPLY", "AGENT_REPLY", "SEND_MESSAGE", "HUMAN_HANDOVER",
     "TAG_CONTACT", "ZOHO_UPSERT_LEAD",
+    "CONDITION", "WAIT_DELAY",              # unlocked in M-D
+    "PARALLEL", "AGENT_HANDOFF",            # new in M-D
+    "QUALIFICATION_GATE", "INTENT_ROUTER",  # new in M-D
 }
 
-# Builder-only node types (visual palette only; blocked from publishing)
-BUILDER_ONLY_NODE_TYPES = {"CONDITION", "WAIT_DELAY"}
+# Builder-only node types blocked from publishing (now empty — all types publishable)
+BUILDER_ONLY_NODE_TYPES: set[str] = set()
 
 
 def validate_graph(builder_graph: dict[str, Any]) -> list[dict[str, Any]]:
@@ -76,16 +92,6 @@ def validate_graph(builder_graph: dict[str, Any]) -> list[dict[str, Any]]:
                 "message": "Flow can only have one trigger node."
             })
 
-    # Check builder-only (unsupported runtime) node types
-    for node in nodes:
-        nt = _get_node_type(node)
-        if nt in BUILDER_ONLY_NODE_TYPES:
-            errors.append({
-                "node_id": node.get("id"),
-                "field": "type",
-                "message": f"Node type '{nt}' is not yet supported by the runtime engine — cannot publish. Remove it or replace it with a supported action."
-            })
-
     # Build reachability set from trigger (BFS/DFS)
     if trigger_nodes:
         trigger_id = trigger_nodes[0].get("id")
@@ -116,7 +122,7 @@ def validate_graph(builder_graph: dict[str, Any]) -> list[dict[str, Any]]:
         config = node.get("data", {}).get("config", {})
         nid = node.get("id")
 
-        if nt == "AI_REPLY":
+        if nt in ("AI_REPLY", "AGENT_REPLY"):
             goal = (config.get("goal") or "").strip()
             if not goal:
                 errors.append({
@@ -143,61 +149,160 @@ def validate_graph(builder_graph: dict[str, Any]) -> list[dict[str, Any]]:
                     "message": "Tag Contact node requires a tag name."
                 })
 
+        elif nt == "CONDITION":
+            if not (config.get("condition_type") or "").strip():
+                errors.append({
+                    "node_id": nid,
+                    "field": "config.condition_type",
+                    "message": "Condition node requires a condition type (e.g. qualification_status, intent, tag)."
+                })
+            if not (config.get("operator") or "").strip():
+                errors.append({
+                    "node_id": nid,
+                    "field": "config.operator",
+                    "message": "Condition node requires an operator (e.g. equals, contains)."
+                })
+
+        elif nt == "WAIT_DELAY":
+            delay = config.get("delay_seconds")
+            try:
+                delay_val = int(delay) if delay is not None else 0
+            except (ValueError, TypeError):
+                delay_val = 0
+            if delay_val < 60:
+                errors.append({
+                    "node_id": nid,
+                    "field": "config.delay_seconds",
+                    "message": "Wait / Delay node requires a minimum delay of 60 seconds (1 minute)."
+                })
+
+        elif nt == "AGENT_HANDOFF":
+            if not (config.get("target_agent") or "").strip():
+                errors.append({
+                    "node_id": nid,
+                    "field": "config.target_agent",
+                    "message": "Agent Handoff node requires a target agent name."
+                })
+
+        elif nt == "INTENT_ROUTER":
+            routes = config.get("routes") or []
+            if not routes:
+                errors.append({
+                    "node_id": nid,
+                    "field": "config.routes",
+                    "message": "Intent Router node requires at least one intent-to-agent route."
+                })
+
+        # QUALIFICATION_GATE: no required config — reads workspace config at runtime
+        # PARALLEL: no required config — agents field is optional
+
     return errors
 
 
-def translate(builder_graph: dict[str, Any]) -> dict[str, Any]:
+def translate(builder_graph: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
     """
-    Translate builder_graph_json to runtime definition_json.
+    Translate builder_graph_json to dual output.
     Must only be called after validate_graph() returns an empty list.
-    Returns the definition_json ready for FlowVersion.definition_json.
+
+    Returns:
+        (definition_json, adk_pipeline_config)
+        definition_json: legacy runtime format, stored in FlowVersion.definition_json
+        adk_pipeline_config: new ADK format, stored in FlowVersion.adk_pipeline_config
     """
-    nodes_in = builder_graph.get("nodes", [])
-    edges_in = builder_graph.get("edges", [])
+    definition_json = _translate_legacy(builder_graph)
+    adk_pipeline = translate_to_adk_pipeline(builder_graph)
+    return definition_json, adk_pipeline
 
-    runtime_nodes = []
-    start_node_id: str | None = None
 
-    for node in nodes_in:
-        nid = node.get("id")
+def translate_to_adk_pipeline(builder_graph: dict[str, Any]) -> dict[str, Any]:
+    """
+    Translate builder_graph_json to an ADK pipeline config dict.
+    This drives how the OrchestratorAgent is configured at runtime.
+    """
+    nodes = builder_graph.get("nodes", [])
+    edges = builder_graph.get("edges", [])
+
+    pipeline: dict[str, Any] = {
+        "pipeline_type": "orchestrated",
+        "agents": {
+            "orchestrator": {"routing_rules": [], "sub_agents": []},
+            "qualification": {},
+            "crm": {},
+            "reply": {},
+            "handover": {},
+        },
+        "triggers": [],
+        "condition_branches": [],
+        "wait_delays": [],
+    }
+
+    for node in nodes:
         node_type = _get_node_type(node)
-        data = node.get("data", {})
-        config = data.get("config", {})
+        config = node.get("data", {}).get("config", {})
+        nid = node.get("id")
 
         if node_type in TRIGGER_NODE_TYPES:
-            # Trigger node → runtime TRIGGER type
-            runtime_nodes.append({
-                "id": nid,
-                "type": "TRIGGER",
-                "config": {
-                    "trigger_type": node_type,
-                    "platform": data.get("platform", config.get("platform", "whatsapp")),
-                    "keywords": data.get("keywords", config.get("keywords", [])),
-                },
-            })
-            start_node_id = nid
-        else:
-            # Action node → preserve type and config
-            runtime_nodes.append({
-                "id": nid,
+            pipeline["triggers"].append({
                 "type": node_type,
-                "config": config,
+                "platform": node.get("data", {}).get("platform", config.get("platform", "whatsapp")),
             })
 
-    runtime_edges = []
-    for edge in edges_in:
-        runtime_edges.append({
-            "id": edge.get("id"),
-            "source_node_id": edge.get("source"),
-            "target_node_id": edge.get("target"),
-            "source_handle": edge.get("sourceHandle"),
-        })
+        elif node_type in ("AI_REPLY", "AGENT_REPLY"):
+            pipeline["agents"]["reply"].update({
+                "goal": config.get("goal"),
+                "tone": config.get("tone", "professional"),
+                "max_length": config.get("max_length", 160),
+                "extra_instructions": config.get("extra_instructions"),
+            })
 
-    return {
-        "start_node_id": start_node_id,
-        "nodes": runtime_nodes,
-        "edges": runtime_edges,
-    }
+        elif node_type == "CONDITION":
+            pipeline["condition_branches"].append({
+                "node_id": nid,
+                "condition_type": config.get("condition_type"),
+                "operator": config.get("operator"),
+                "value": config.get("value"),
+                "true_branch": _get_next_node(nid, edges, handle="true"),
+                "false_branch": _get_next_node(nid, edges, handle="false"),
+            })
+
+        elif node_type == "WAIT_DELAY":
+            pipeline["wait_delays"].append({
+                "node_id": nid,
+                "delay_seconds": config.get("delay_seconds", 3600),
+                "delay_unit": config.get("delay_unit", "hours"),
+                "resume_node_id": _get_next_node(nid, edges),
+            })
+
+        elif node_type == "QUALIFICATION_GATE":
+            pipeline["agents"]["orchestrator"]["routing_rules"].append({
+                "type": "qualification_gate",
+                "node_id": nid,
+                "qualified_branch": _get_next_node(nid, edges, handle="qualified"),
+                "unqualified_branch": _get_next_node(nid, edges, handle="unqualified"),
+            })
+
+        elif node_type == "INTENT_ROUTER":
+            pipeline["agents"]["orchestrator"]["routing_rules"].append({
+                "type": "intent_router",
+                "node_id": nid,
+                "routes": config.get("routes", []),
+            })
+
+        elif node_type == "PARALLEL":
+            pipeline["agents"]["orchestrator"]["routing_rules"].append({
+                "type": "parallel",
+                "node_id": nid,
+                "parallel_agents": config.get("agents", []),
+            })
+
+        elif node_type == "AGENT_HANDOFF":
+            pipeline["agents"]["orchestrator"]["routing_rules"].append({
+                "type": "agent_handoff",
+                "node_id": nid,
+                "target_agent": config.get("target_agent"),
+            })
+
+    return pipeline
 
 
 def simulate(
@@ -252,8 +357,8 @@ def simulate(
             step["description"] = f"Trigger: {node_type}"
             if mock_payload:
                 step["mock_payload"] = mock_payload
-        elif node_type == "AI_REPLY":
-            step["description"] = "AI Reply — would generate a response using your workspace prompt configuration."
+        elif node_type in ("AI_REPLY", "AGENT_REPLY"):
+            step["description"] = "Agent Reply — would generate a response using your workspace prompt configuration."
             step["goal"] = config.get("goal", "(no goal set)")
             step["would_dispatch"] = False
         elif node_type == "SEND_MESSAGE":
@@ -270,9 +375,27 @@ def simulate(
             step["description"] = f"Tag Contact — would apply tag '{config.get('tag', '')}' to the contact."
         elif node_type == "ZOHO_UPSERT_LEAD":
             step["description"] = "Zoho CRM — would upsert the contact as a lead in Zoho CRM."
-        elif node_type in BUILDER_ONLY_NODE_TYPES:
-            step["description"] = f"{node_type} — not yet supported by runtime. This node would be skipped."
-            step["warning"] = "Builder-only node type"
+        elif node_type == "CONDITION":
+            step["description"] = (
+                f"Condition — branches on '{config.get('condition_type', '?')}' "
+                f"{config.get('operator', '?')} '{config.get('value', '?')}'. "
+                "Simulation follows the 'true' branch."
+            )
+        elif node_type == "WAIT_DELAY":
+            delay = config.get("delay_seconds", 3600)
+            unit = config.get("delay_unit", "seconds")
+            step["description"] = f"Wait / Delay — would pause flow for {delay} {unit}."
+            step["note"] = "Simulation continues immediately; actual delay requires Celery beat."
+        elif node_type == "PARALLEL":
+            agents = config.get("agents", [])
+            step["description"] = f"Parallel — would run {len(agents)} agent(s) simultaneously: {', '.join(agents) or 'none configured'}."
+        elif node_type == "AGENT_HANDOFF":
+            step["description"] = f"Agent Handoff — would delegate to '{config.get('target_agent', '?')}' agent."
+        elif node_type == "QUALIFICATION_GATE":
+            step["description"] = "Qualification Gate — routes qualified leads to one branch, unqualified to another."
+        elif node_type == "INTENT_ROUTER":
+            routes = config.get("routes", [])
+            step["description"] = f"Intent Router — routes based on detected intent ({len(routes)} route(s) configured)."
         else:
             step["description"] = f"Unknown node type: {node_type}"
 
@@ -288,12 +411,79 @@ def simulate(
     }
 
 
+# ---------------------------------------------------------------------------
+# Private helpers
+# ---------------------------------------------------------------------------
+
+def _translate_legacy(builder_graph: dict[str, Any]) -> dict[str, Any]:
+    """
+    Translate builder_graph_json to the legacy runtime definition_json.
+    Stored in FlowVersion.definition_json for audit and display.
+    """
+    nodes_in = builder_graph.get("nodes", [])
+    edges_in = builder_graph.get("edges", [])
+
+    runtime_nodes = []
+    start_node_id: str | None = None
+
+    for node in nodes_in:
+        nid = node.get("id")
+        node_type = _get_node_type(node)
+        data = node.get("data", {})
+        config = data.get("config", {})
+
+        if node_type in TRIGGER_NODE_TYPES:
+            runtime_nodes.append({
+                "id": nid,
+                "type": "TRIGGER",
+                "config": {
+                    "trigger_type": node_type,
+                    "platform": data.get("platform", config.get("platform", "whatsapp")),
+                    "keywords": data.get("keywords", config.get("keywords", [])),
+                },
+            })
+            start_node_id = nid
+        else:
+            runtime_nodes.append({
+                "id": nid,
+                "type": node_type,
+                "config": config,
+            })
+
+    runtime_edges = []
+    for edge in edges_in:
+        runtime_edges.append({
+            "id": edge.get("id"),
+            "source_node_id": edge.get("source"),
+            "target_node_id": edge.get("target"),
+            "source_handle": edge.get("sourceHandle"),
+        })
+
+    return {
+        "start_node_id": start_node_id,
+        "nodes": runtime_nodes,
+        "edges": runtime_edges,
+    }
+
+
 def _get_node_type(node: dict) -> str:
     """Extract the business logic node type from a builder node."""
     data = node.get("data", {})
     return data.get("nodeType", "")
 
 
+def _get_next_node(node_id: str, edges: list[dict], handle: str | None = None) -> str | None:
+    """
+    Return the first target node ID connected from node_id.
+    If handle is specified, only consider edges with that sourceHandle.
+    """
+    for edge in edges:
+        if edge.get("source") == node_id:
+            if handle is None or edge.get("sourceHandle") == handle:
+                return edge.get("target")
+    return None
+
+
 def _reachable_nodes(start_id: str, edges: list[dict]) -> set[str]:
     """BFS to find all node IDs reachable from start_id via edges."""
     edge_map: dict[str, list[str]] = {}

backend/app/models/models.py

@@ -169,7 +169,11 @@ class FlowVersion(BaseIDModel, table=True):
     version_number: int
     definition_json: Dict[str, Any] = Field(sa_column=Column(JSON))
     is_published: bool = Field(default=False)
-    
+    # Mission M-D: ADK pipeline config generated at publish time
+    adk_pipeline_config: Optional[Dict[str, Any]] = Field(
+        default=None, sa_column=Column(JSON)
+    )
+
     flow: "Flow" = Relationship(back_populates="versions")
 
 # --- Execution ---
@@ -188,6 +192,10 @@ class ExecutionInstance(WorkspaceScopedModel, table=True):
     abort_reason: Optional[str] = None
     aborted_at: Optional[datetime] = None
 
+    # Mission M-D: WAIT_DELAY resume support
+    resume_at: Optional[datetime] = Field(default=None)
+    resume_node_id: Optional[str] = Field(default=None)
+
     __table_args__ = (
         Index("idx_exec_stats", "workspace_id", "created_at", "status"),
     )

backend/app/workers/tasks.py

@@ -672,3 +672,64 @@ def export_to_hf_task():
             )
 
     run_async(_run())
+
+
+@celery_app.task(name="app.workers.tasks.resume_waiting_instances_task")
+def resume_waiting_instances_task():
+    """
+    Every-minute task: find WAITING ExecutionInstances whose delay has elapsed
+    and re-trigger the ADK runner for them (Mission M-D: WAIT_DELAY support).
+    """
+    from sqlalchemy.ext.asyncio import AsyncSession
+    from app.models.models import Conversation
+
+    async def _run():
+        now = datetime.utcnow()
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            result = await session.execute(
+                select(ExecutionInstance).where(
+                    ExecutionInstance.status == ExecutionStatus.WAITING,
+                    ExecutionInstance.resume_at <= now,
+                )
+            )
+            instances = result.scalars().all()
+            if not instances:
+                return
+
+            for instance in instances:
+                try:
+                    instance.status = ExecutionStatus.RUNNING
+                    session.add(instance)
+                    await session.flush()
+
+                    # Resolve conversation_id for this contact in this workspace
+                    conv_result = await session.execute(
+                        select(Conversation).where(
+                            Conversation.workspace_id == instance.workspace_id,
+                            Conversation.contact_id == instance.contact_id,
+                        ).limit(1)
+                    )
+                    conversation = conv_result.scalars().first()
+                    if not conversation:
+                        logger.warning(
+                            f"[resume_waiting_instances_task] No conversation for instance {instance.id}"
+                        )
+                        continue
+
+                    await run_for_contact(
+                        workspace_id=instance.workspace_id,
+                        contact_id=instance.contact_id,
+                        conversation_id=conversation.id,
+                        inbound_message="[Resumed after delay]",
+                        execution_instance=instance,
+                        session=session,
+                    )
+                except Exception:
+                    logger.exception(
+                        f"[resume_waiting_instances_task] Failed to resume instance {instance.id}"
+                    )
+
+            await session.commit()
+            logger.info(f"[resume_waiting_instances_task] Resumed {len(instances)} waiting instance(s)")
+
+    run_async(_run())

backend/tests/test_builder_translator.py

@@ -121,8 +121,8 @@ def test_validate_graph_tag_contact_missing_tag():
     assert any(e["node_id"] == "node-1" for e in errors)
 
 
-def test_validate_graph_condition_node_blocked():
-    """CONDITION node should be blocked from publish with a clear message."""
+def test_validate_graph_condition_node_requires_config():
+    """CONDITION node without condition_type should return a config error (not a 'not supported' error)."""
     graph = {
         "nodes": [
             make_trigger_node(),
@@ -131,11 +131,31 @@ def test_validate_graph_condition_node_blocked():
         "edges": [make_edge("e1", "trigger-1", "node-1")],
     }
     errors = validate_graph(graph)
-    assert any(e["node_id"] == "node-1" and "not yet supported" in e["message"] for e in errors)
+    # Should have config errors, NOT a "not yet supported" error
+    assert any(e["node_id"] == "node-1" for e in errors)
+    assert not any("not yet supported" in e["message"] for e in errors)
+    assert any("condition_type" in e.get("field", "") for e in errors)
+
+
+def test_validate_graph_condition_node_valid():
+    """CONDITION node with required config should pass validation."""
+    graph = {
+        "nodes": [
+            make_trigger_node(),
+            make_action_node("node-1", "CONDITION", {
+                "condition_type": "qualification_status",
+                "operator": "equals",
+                "value": "qualified",
+            }),
+        ],
+        "edges": [make_edge("e1", "trigger-1", "node-1")],
+    }
+    errors = validate_graph(graph)
+    assert errors == []
 
 
-def test_validate_graph_wait_delay_blocked():
-    """WAIT_DELAY node should be blocked from publish."""
+def test_validate_graph_wait_delay_requires_delay_seconds():
+    """WAIT_DELAY node without delay_seconds should return a config error (not 'not supported')."""
     graph = {
         "nodes": [
             make_trigger_node(),
@@ -144,7 +164,22 @@ def test_validate_graph_wait_delay_blocked():
         "edges": [make_edge("e1", "trigger-1", "node-1")],
     }
     errors = validate_graph(graph)
-    assert any(e["node_id"] == "node-1" and "not yet supported" in e["message"] for e in errors)
+    assert any(e["node_id"] == "node-1" for e in errors)
+    assert not any("not yet supported" in e["message"] for e in errors)
+    assert any("delay_seconds" in e.get("field", "") for e in errors)
+
+
+def test_validate_graph_wait_delay_valid():
+    """WAIT_DELAY node with sufficient delay_seconds should pass."""
+    graph = {
+        "nodes": [
+            make_trigger_node(),
+            make_action_node("node-1", "WAIT_DELAY", {"delay_seconds": 3600}),
+        ],
+        "edges": [make_edge("e1", "trigger-1", "node-1")],
+    }
+    errors = validate_graph(graph)
+    assert errors == []
 
 
 def test_validate_graph_disconnected_node():
@@ -180,9 +215,9 @@ def test_validate_graph_lead_ad_submit_trigger():
 # ---------------------------------------------------------------------------
 
 def test_translate_produces_valid_contract():
-    """translate() should produce a valid runtime definition_json."""
+    """translate() should produce a valid runtime definition_json (first tuple element)."""
     graph = make_valid_graph()
-    definition = translate(graph)
+    definition, adk_pipeline = translate(graph)
 
     assert "nodes" in definition
     assert "edges" in definition
@@ -205,6 +240,10 @@ def test_translate_produces_valid_contract():
     assert edge["source_node_id"] == "trigger-1"
     assert edge["target_node_id"] == "node-1"
 
+    # ADK pipeline also returned
+    assert "pipeline_type" in adk_pipeline
+    assert adk_pipeline["pipeline_type"] == "orchestrated"
+
 
 def test_translate_sets_start_node_id():
     """start_node_id in output should match the trigger node id."""
@@ -215,14 +254,14 @@ def test_translate_sets_start_node_id():
         ],
         "edges": [make_edge("e1", "my-trigger", "step-1")],
     }
-    definition = translate(graph)
+    definition, _ = translate(graph)
     assert definition["start_node_id"] == "my-trigger"
 
 
 def test_translate_edge_format():
     """Edges in runtime format should use source_node_id and target_node_id."""
     graph = make_valid_graph()
-    definition = translate(graph)
+    definition, _ = translate(graph)
     edge = definition["edges"][0]
     assert "source_node_id" in edge
     assert "target_node_id" in edge
@@ -244,7 +283,7 @@ def test_translate_multiple_nodes():
             make_edge("e3", "n2", "n3"),
         ],
     }
-    definition = translate(graph)
+    definition, _ = translate(graph)
     assert len(definition["nodes"]) == 4
     assert len(definition["edges"]) == 3
     types = {n["type"] for n in definition["nodes"]}

backend/tests/test_builder_translator_v2.py

@@ -0,0 +1,235 @@
+"""
+Mission M-D: Builder translator v2 tests.
+Tests for new node types: CONDITION, WAIT_DELAY, PARALLEL, AGENT_HANDOFF,
+QUALIFICATION_GATE, INTENT_ROUTER, and the dual translate() output.
+No DB required — pure logic tests.
+"""
+import pytest
+from app.domain.builder_translator import validate_graph, translate, translate_to_adk_pipeline
+
+
+# ---------------------------------------------------------------------------
+# Helpers (duplicated from test_builder_translator to keep tests isolated)
+# ---------------------------------------------------------------------------
+
+def make_trigger_node(node_id="trigger-1", node_type="MESSAGE_INBOUND", platform="whatsapp"):
+    return {
+        "id": node_id,
+        "type": "triggerNode",
+        "position": {"x": 250, "y": 50},
+        "data": {"nodeType": node_type, "platform": platform, "config": {}},
+    }
+
+
+def make_action_node(node_id, node_type, config=None):
+    return {
+        "id": node_id,
+        "type": "actionNode",
+        "position": {"x": 250, "y": 220},
+        "data": {"nodeType": node_type, "config": config or {}},
+    }
+
+
+def make_edge(edge_id, source, target, source_handle=None):
+    e = {"id": edge_id, "source": source, "target": target}
+    if source_handle:
+        e["sourceHandle"] = source_handle
+    return e
+
+
+# ---------------------------------------------------------------------------
+# Test 1 — CONDITION node translates to condition_branches entry
+# ---------------------------------------------------------------------------
+
+def test_condition_node_translates_to_condition_branches():
+    """CONDITION node should populate pipeline.condition_branches with true/false branch IDs."""
+    graph = {
+        "nodes": [
+            make_trigger_node(),
+            make_action_node("cond-1", "CONDITION", {
+                "condition_type": "qualification_status",
+                "operator": "equals",
+                "value": "qualified",
+            }),
+            make_action_node("true-node", "ZOHO_UPSERT_LEAD", {}),
+            make_action_node("false-node", "WAIT_DELAY", {"delay_seconds": 3600}),
+        ],
+        "edges": [
+            make_edge("e1", "trigger-1", "cond-1"),
+            make_edge("e2", "cond-1", "true-node", source_handle="true"),
+            make_edge("e3", "cond-1", "false-node", source_handle="false"),
+        ],
+    }
+    pipeline = translate_to_adk_pipeline(graph)
+
+    assert len(pipeline["condition_branches"]) == 1
+    branch = pipeline["condition_branches"][0]
+    assert branch["node_id"] == "cond-1"
+    assert branch["condition_type"] == "qualification_status"
+    assert branch["operator"] == "equals"
+    assert branch["value"] == "qualified"
+    assert branch["true_branch"] == "true-node"
+    assert branch["false_branch"] == "false-node"
+
+
+# ---------------------------------------------------------------------------
+# Test 2 — WAIT_DELAY node translates to wait_delays entry
+# ---------------------------------------------------------------------------
+
+def test_wait_delay_node_translates_to_wait_delays():
+    """WAIT_DELAY node should populate pipeline.wait_delays with delay config."""
+    graph = {
+        "nodes": [
+            make_trigger_node(),
+            make_action_node("wait-1", "WAIT_DELAY", {
+                "delay_seconds": 3600,
+                "delay_unit": "hours",
+            }),
+            make_action_node("after-wait", "AI_REPLY", {"goal": "Follow up"}),
+        ],
+        "edges": [
+            make_edge("e1", "trigger-1", "wait-1"),
+            make_edge("e2", "wait-1", "after-wait"),
+        ],
+    }
+    pipeline = translate_to_adk_pipeline(graph)
+
+    assert len(pipeline["wait_delays"]) == 1
+    delay = pipeline["wait_delays"][0]
+    assert delay["node_id"] == "wait-1"
+    assert delay["delay_seconds"] == 3600
+    assert delay["delay_unit"] == "hours"
+    assert delay["resume_node_id"] == "after-wait"
+
+
+# ---------------------------------------------------------------------------
+# Test 3 — PARALLEL node adds parallel routing rule
+# ---------------------------------------------------------------------------
+
+def test_parallel_node_adds_routing_rule():
+    """PARALLEL node should add a parallel routing rule to orchestrator.routing_rules."""
+    graph = {
+        "nodes": [
+            make_trigger_node(),
+            make_action_node("par-1", "PARALLEL", {"agents": ["crm", "reply"]}),
+        ],
+        "edges": [make_edge("e1", "trigger-1", "par-1")],
+    }
+    pipeline = translate_to_adk_pipeline(graph)
+
+    routing_rules = pipeline["agents"]["orchestrator"]["routing_rules"]
+    parallel_rule = next((r for r in routing_rules if r["type"] == "parallel"), None)
+    assert parallel_rule is not None
+    assert parallel_rule["parallel_agents"] == ["crm", "reply"]
+
+
+# ---------------------------------------------------------------------------
+# Test 4 — All new node types pass validation with valid config
+# ---------------------------------------------------------------------------
+
+def test_all_new_node_types_pass_validation():
+    """A graph with CONDITION + WAIT_DELAY + PARALLEL + AGENT_HANDOFF + QUALIFICATION_GATE + INTENT_ROUTER
+    should all pass validation when properly configured."""
+    graph = {
+        "nodes": [
+            make_trigger_node(),
+            make_action_node("cond-1", "CONDITION", {
+                "condition_type": "qualification_status",
+                "operator": "equals",
+                "value": "qualified",
+            }),
+            make_action_node("wait-1", "WAIT_DELAY", {"delay_seconds": 3600}),
+            make_action_node("par-1", "PARALLEL", {"agents": ["crm", "reply"]}),
+            make_action_node("handoff-1", "AGENT_HANDOFF", {"target_agent": "handover"}),
+            make_action_node("gate-1", "QUALIFICATION_GATE", {}),
+            make_action_node("router-1", "INTENT_ROUTER", {
+                "routes": [{"intent": "complaint", "agent": "handover"}]
+            }),
+        ],
+        "edges": [
+            make_edge("e1", "trigger-1", "cond-1"),
+            make_edge("e2", "trigger-1", "wait-1"),
+            make_edge("e3", "trigger-1", "par-1"),
+            make_edge("e4", "trigger-1", "handoff-1"),
+            make_edge("e5", "trigger-1", "gate-1"),
+            make_edge("e6", "trigger-1", "router-1"),
+        ],
+    }
+    errors = validate_graph(graph)
+    # None of the new node types should produce "not yet supported" errors
+    assert not any("not yet supported" in e.get("message", "") for e in errors)
+    # No config errors for these properly-configured nodes
+    new_type_ids = {"cond-1", "wait-1", "par-1", "handoff-1", "gate-1", "router-1"}
+    config_errors = [e for e in errors if e.get("node_id") in new_type_ids]
+    assert config_errors == []
+
+
+# ---------------------------------------------------------------------------
+# Test 5 — translate() returns dual output (definition_json + adk_pipeline_config)
+# ---------------------------------------------------------------------------
+
+def test_translate_returns_dual_output():
+    """translate() must return a 2-tuple: (definition_json, adk_pipeline_config)."""
+    graph = {
+        "nodes": [
+            make_trigger_node(),
+            make_action_node("n1", "AI_REPLY", {"goal": "Greet lead", "tone": "friendly", "max_length": 120}),
+        ],
+        "edges": [make_edge("e1", "trigger-1", "n1")],
+    }
+    result = translate(graph)
+    assert isinstance(result, tuple)
+    assert len(result) == 2
+
+    definition_json, adk_pipeline = result
+
+    # definition_json has legacy shape
+    assert "start_node_id" in definition_json
+    assert "nodes" in definition_json
+    assert "edges" in definition_json
+
+    # adk_pipeline has new shape
+    assert adk_pipeline["pipeline_type"] == "orchestrated"
+    assert "agents" in adk_pipeline
+    assert "triggers" in adk_pipeline
+
+    # AI_REPLY config should be reflected in reply agent config
+    reply_cfg = adk_pipeline["agents"]["reply"]
+    assert reply_cfg.get("goal") == "Greet lead"
+    assert reply_cfg.get("tone") == "friendly"
+    assert reply_cfg.get("max_length") == 120
+
+
+# ---------------------------------------------------------------------------
+# Test 6 — QUALIFICATION_GATE + INTENT_ROUTER produce correct routing rules
+# ---------------------------------------------------------------------------
+
+def test_qualification_gate_and_intent_router_routing_rules():
+    """QUALIFICATION_GATE adds qualification_gate rule; INTENT_ROUTER adds intent_router rule."""
+    graph = {
+        "nodes": [
+            make_trigger_node(),
+            make_action_node("gate-1", "QUALIFICATION_GATE", {}),
+            make_action_node("router-1", "INTENT_ROUTER", {
+                "routes": [
+                    {"intent": "complaint", "agent": "handover"},
+                    {"intent": "pricing", "agent": "reply"},
+                ]
+            }),
+        ],
+        "edges": [
+            make_edge("e1", "trigger-1", "gate-1"),
+            make_edge("e2", "trigger-1", "router-1"),
+        ],
+    }
+    pipeline = translate_to_adk_pipeline(graph)
+    routing_rules = pipeline["agents"]["orchestrator"]["routing_rules"]
+
+    gate_rule = next((r for r in routing_rules if r["type"] == "qualification_gate"), None)
+    assert gate_rule is not None
+    assert gate_rule["node_id"] == "gate-1"
+
+    intent_rule = next((r for r in routing_rules if r["type"] == "intent_router"), None)
+    assert intent_rule is not None
+    assert len(intent_rule["routes"]) == 2
+    assert intent_rule["routes"][0]["intent"] == "complaint"

frontend/src/app/(dashboard)/automations/[id]/NodeConfigPanel.tsx

@@ -1,6 +1,9 @@
 "use client";
 
-import { Bot, Send, User, Tag, Database, GitBranch, Clock, Info, X } from "lucide-react";
+import {
+    Bot, Send, User, Tag, Database, GitBranch, Clock, Info, X,
+    Shuffle, ArrowRightCircle, Filter, Map,
+} from "lucide-react";
 import type { BuilderNode } from "@/lib/automations-api";
 import { cn } from "@/lib/utils";
 
@@ -204,18 +207,284 @@ function ZohoUpsertConfig() {
     );
 }
 
-function ComingSoonConfig({ nodeType }: { nodeType: string }) {
+function ConditionConfig({
+    config,
+    onChange,
+}: {
+    config: Record<string, any>;
+    onChange: (updated: Record<string, any>) => void;
+}) {
+    const CONDITION_TYPES = [
+        { value: "qualification_status", label: "Qualification Status" },
+        { value: "intent", label: "Detected Intent" },
+        { value: "tag", label: "Contact Tag" },
+        { value: "custom_field", label: "Custom Field" },
+    ];
+    const OPERATORS = [
+        { value: "equals", label: "Equals" },
+        { value: "not_equals", label: "Not Equals" },
+        { value: "contains", label: "Contains" },
+        { value: "greater_than", label: "Greater Than" },
+        { value: "less_than", label: "Less Than" },
+    ];
+
     return (
-        <div className="flex items-start gap-2 p-3 rounded-lg bg-gray-50 dark:bg-gray-900/40 border border-border text-sm text-muted-foreground">
-            <Info className="w-4 h-4 shrink-0 mt-0.5" />
-            <p>
-                <strong>{nodeType}</strong> is coming soon and cannot be used in published
-                automations yet. You can add it to the canvas for planning purposes.
+        <div className="space-y-4">
+            <LabeledField label="Condition Type" required>
+                <select
+                    className="w-full rounded-lg border border-border bg-background px-3 py-2 text-sm focus:outline-none focus:ring-2 focus:ring-teal-500"
+                    value={config.condition_type ?? ""}
+                    onChange={(e) => onChange({ ...config, condition_type: e.target.value })}
+                >
+                    <option value="">Select a condition type...</option>
+                    {CONDITION_TYPES.map((ct) => (
+                        <option key={ct.value} value={ct.value}>{ct.label}</option>
+                    ))}
+                </select>
+            </LabeledField>
+            <LabeledField label="Operator" required>
+                <select
+                    className="w-full rounded-lg border border-border bg-background px-3 py-2 text-sm focus:outline-none focus:ring-2 focus:ring-teal-500"
+                    value={config.operator ?? ""}
+                    onChange={(e) => onChange({ ...config, operator: e.target.value })}
+                >
+                    <option value="">Select an operator...</option>
+                    {OPERATORS.map((op) => (
+                        <option key={op.value} value={op.value}>{op.label}</option>
+                    ))}
+                </select>
+            </LabeledField>
+            <LabeledField label="Value">
+                <input
+                    className="w-full rounded-lg border border-border bg-background px-3 py-2 text-sm focus:outline-none focus:ring-2 focus:ring-teal-500"
+                    placeholder="e.g. qualified"
+                    value={config.value ?? ""}
+                    onChange={(e) => onChange({ ...config, value: e.target.value })}
+                />
+            </LabeledField>
+            <div className="flex items-start gap-2 p-2.5 rounded-lg bg-cyan-50 dark:bg-cyan-950/20 border border-cyan-200 text-xs text-cyan-700">
+                <Info className="w-3.5 h-3.5 shrink-0 mt-0.5" />
+                <p>Connect the <strong>TRUE</strong> handle to the path taken when the condition passes, and <strong>FALSE</strong> to the fallback path.</p>
+            </div>
+        </div>
+    );
+}
+
+function WaitDelayConfig({
+    config,
+    onChange,
+}: {
+    config: Record<string, any>;
+    onChange: (updated: Record<string, any>) => void;
+}) {
+    const UNITS = [
+        { value: "minutes", seconds: 60, label: "Minutes" },
+        { value: "hours", seconds: 3600, label: "Hours" },
+        { value: "days", seconds: 86400, label: "Days" },
+    ];
+
+    const unit = config.delay_unit ?? "hours";
+    const unitEntry = UNITS.find((u) => u.value === unit) ?? UNITS[1];
+    const displayValue = config.delay_seconds
+        ? Math.round(Number(config.delay_seconds) / unitEntry.seconds)
+        : "";
+
+    const handleChange = (rawValue: string, rawUnit: string) => {
+        const unitEntry2 = UNITS.find((u) => u.value === rawUnit) ?? UNITS[1];
+        const seconds = Math.max(60, (Number(rawValue) || 0) * unitEntry2.seconds);
+        onChange({ ...config, delay_seconds: seconds, delay_unit: rawUnit });
+    };
+
+    return (
+        <div className="space-y-4">
+            <LabeledField label="Delay Duration" required>
+                <div className="flex gap-2">
+                    <input
+                        type="number"
+                        min="1"
+                        className="flex-1 rounded-lg border border-border bg-background px-3 py-2 text-sm focus:outline-none focus:ring-2 focus:ring-teal-500"
+                        placeholder="e.g. 2"
+                        value={displayValue}
+                        onChange={(e) => handleChange(e.target.value, unit)}
+                    />
+                    <select
+                        className="w-28 rounded-lg border border-border bg-background px-3 py-2 text-sm focus:outline-none focus:ring-2 focus:ring-teal-500"
+                        value={unit}
+                        onChange={(e) => handleChange(String(displayValue), e.target.value)}
+                    >
+                        {UNITS.map((u) => (
+                            <option key={u.value} value={u.value}>{u.label}</option>
+                        ))}
+                    </select>
+                </div>
+                <p className="text-xs text-muted-foreground mt-1">Minimum: 1 minute.</p>
+            </LabeledField>
+            <div className="flex items-start gap-2 p-2.5 rounded-lg bg-indigo-50 dark:bg-indigo-950/20 border border-indigo-200 text-xs text-indigo-700">
+                <Info className="w-3.5 h-3.5 shrink-0 mt-0.5" />
+                <p>Execution pauses here. The flow resumes after the delay via a background scheduler.</p>
+            </div>
+        </div>
+    );
+}
+
+function ParallelConfig({
+    config,
+    onChange,
+}: {
+    config: Record<string, any>;
+    onChange: (updated: Record<string, any>) => void;
+}) {
+    const AVAILABLE_AGENTS = [
+        { value: "qualification", label: "Qualification Agent" },
+        { value: "crm", label: "CRM Agent" },
+        { value: "reply", label: "Reply Agent" },
+        { value: "handover", label: "Handover Agent" },
+    ];
+
+    const selected: string[] = Array.isArray(config.agents) ? config.agents : [];
+
+    const toggle = (value: string) => {
+        const next = selected.includes(value)
+            ? selected.filter((a) => a !== value)
+            : [...selected, value];
+        onChange({ ...config, agents: next });
+    };
+
+    return (
+        <div className="space-y-4">
+            <LabeledField label="Agents to run in parallel">
+                <div className="space-y-2">
+                    {AVAILABLE_AGENTS.map((agent) => (
+                        <label key={agent.value} className="flex items-center gap-2.5 cursor-pointer group">
+                            <input
+                                type="checkbox"
+                                className="w-4 h-4 rounded border-border text-teal-600 focus:ring-teal-500"
+                                checked={selected.includes(agent.value)}
+                                onChange={() => toggle(agent.value)}
+                            />
+                            <span className="text-sm text-slate-700 group-hover:text-slate-900">{agent.label}</span>
+                        </label>
+                    ))}
+                </div>
+            </LabeledField>
+            <div className="flex items-start gap-2 p-2.5 rounded-lg bg-teal-50 dark:bg-teal-950/20 border border-teal-200 text-xs text-teal-700">
+                <Info className="w-3.5 h-3.5 shrink-0 mt-0.5" />
+                <p>Selected agents will run simultaneously. Results are merged before continuing.</p>
+            </div>
+        </div>
+    );
+}
+
+function AgentHandoffConfig({
+    config,
+    onChange,
+}: {
+    config: Record<string, any>;
+    onChange: (updated: Record<string, any>) => void;
+}) {
+    const AGENTS = [
+        { value: "qualification", label: "Qualification Agent" },
+        { value: "crm", label: "CRM Agent" },
+        { value: "reply", label: "Reply Agent" },
+        { value: "handover", label: "Handover Agent" },
+    ];
+
+    return (
+        <LabeledField label="Target Agent" required>
+            <select
+                className="w-full rounded-lg border border-border bg-background px-3 py-2 text-sm focus:outline-none focus:ring-2 focus:ring-teal-500"
+                value={config.target_agent ?? ""}
+                onChange={(e) => onChange({ ...config, target_agent: e.target.value })}
+            >
+                <option value="">Select an agent...</option>
+                {AGENTS.map((a) => (
+                    <option key={a.value} value={a.value}>{a.label}</option>
+                ))}
+            </select>
+        </LabeledField>
+    );
+}
+
+function QualificationGateConfig() {
+    return (
+        <div className="space-y-3">
+            <div className="flex items-start gap-2 p-3 rounded-lg bg-lime-50 dark:bg-lime-950/20 border border-lime-200 text-sm text-lime-800">
+                <Info className="w-4 h-4 shrink-0 mt-0.5" />
+                <p>
+                    Routes leads based on their qualification status. Connect the{" "}
+                    <strong>QUALIFIED</strong> handle for qualified leads and the{" "}
+                    <strong>UNQUALIFIED</strong> handle for those still being evaluated.
+                </p>
+            </div>
+            <p className="text-xs text-muted-foreground">
+                Qualification criteria are configured in your workspace settings.
             </p>
         </div>
     );
 }
 
+function IntentRouterConfig({
+    config,
+    onChange,
+}: {
+    config: Record<string, any>;
+    onChange: (updated: Record<string, any>) => void;
+}) {
+    const routes: { intent: string; agent: string }[] = Array.isArray(config.routes) ? config.routes : [];
+
+    const AGENTS = ["qualification", "crm", "reply", "handover"];
+
+    const addRoute = () =>
+        onChange({ ...config, routes: [...routes, { intent: "", agent: "reply" }] });
+    const removeRoute = (i: number) =>
+        onChange({ ...config, routes: routes.filter((_, idx) => idx !== i) });
+    const updateRoute = (i: number, field: "intent" | "agent", val: string) =>
+        onChange({
+            ...config,
+            routes: routes.map((r, idx) => (idx === i ? { ...r, [field]: val } : r)),
+        });
+
+    return (
+        <div className="space-y-4">
+            <LabeledField label="Intent → Agent Routes" required>
+                <div className="space-y-2">
+                    {routes.map((route, i) => (
+                        <div key={i} className="flex gap-2 items-center">
+                            <input
+                                className="flex-1 rounded-lg border border-border bg-background px-2 py-1.5 text-sm focus:outline-none focus:ring-2 focus:ring-teal-500"
+                                placeholder="Intent (e.g. complaint)"
+                                value={route.intent}
+                                onChange={(e) => updateRoute(i, "intent", e.target.value)}
+                            />
+                            <select
+                                className="w-28 rounded-lg border border-border bg-background px-2 py-1.5 text-sm focus:outline-none focus:ring-2 focus:ring-teal-500"
+                                value={route.agent}
+                                onChange={(e) => updateRoute(i, "agent", e.target.value)}
+                            >
+                                {AGENTS.map((a) => (
+                                    <option key={a} value={a}>{a}</option>
+                                ))}
+                            </select>
+                            <button
+                                onClick={() => removeRoute(i)}
+                                className="p-1.5 rounded-md text-muted-foreground hover:text-red-500 hover:bg-red-50 transition-colors"
+                            >
+                                <X className="w-3.5 h-3.5" />
+                            </button>
+                        </div>
+                    ))}
+                    <button
+                        onClick={addRoute}
+                        className="text-xs text-teal-600 hover:text-teal-700 font-medium"
+                    >
+                        + Add route
+                    </button>
+                </div>
+            </LabeledField>
+        </div>
+    );
+}
+
 function TriggerConfig({
     config,
     onChange,
@@ -256,24 +525,34 @@ interface NodeConfigPanelProps {
 
 const NODE_TYPE_ICONS: Record<string, React.ReactNode> = {
     AI_REPLY: <Bot className="w-4 h-4" />,
+    AGENT_REPLY: <Bot className="w-4 h-4" />,
     SEND_MESSAGE: <Send className="w-4 h-4" />,
     HUMAN_HANDOVER: <User className="w-4 h-4" />,
     TAG_CONTACT: <Tag className="w-4 h-4" />,
     ZOHO_UPSERT_LEAD: <Database className="w-4 h-4" />,
     CONDITION: <GitBranch className="w-4 h-4" />,
     WAIT_DELAY: <Clock className="w-4 h-4" />,
+    PARALLEL: <Shuffle className="w-4 h-4" />,
+    AGENT_HANDOFF: <ArrowRightCircle className="w-4 h-4" />,
+    QUALIFICATION_GATE: <Filter className="w-4 h-4" />,
+    INTENT_ROUTER: <Map className="w-4 h-4" />,
     MESSAGE_INBOUND: <Bot className="w-4 h-4" />,
     LEAD_AD_SUBMIT: <Bot className="w-4 h-4" />,
 };
 
 const NODE_TYPE_LABELS: Record<string, string> = {
     AI_REPLY: "AI Reply",
+    AGENT_REPLY: "Agent Reply",
     SEND_MESSAGE: "Send Message",
     HUMAN_HANDOVER: "Human Handover",
     TAG_CONTACT: "Tag Contact",
     ZOHO_UPSERT_LEAD: "Zoho: Upsert Lead",
     CONDITION: "Condition",
     WAIT_DELAY: "Wait / Delay",
+    PARALLEL: "Parallel",
+    AGENT_HANDOFF: "Agent Handoff",
+    QUALIFICATION_GATE: "Qualification Gate",
+    INTENT_ROUTER: "Intent Router",
     MESSAGE_INBOUND: "Inbound Message",
     LEAD_AD_SUBMIT: "Lead Ad Submission",
 };
@@ -327,7 +606,7 @@ export default function NodeConfigPanel({ node, onClose, onUpdateConfig }: NodeC
 
             {/* Config body */}
             <div className="px-4 py-4 flex-1 space-y-4">
-                {nodeType === "AI_REPLY" && (
+                {(nodeType === "AI_REPLY" || nodeType === "AGENT_REPLY") && (
                     <AiReplyConfig config={config} onChange={handleChange} />
                 )}
                 {nodeType === "SEND_MESSAGE" && (
@@ -340,10 +619,23 @@ export default function NodeConfigPanel({ node, onClose, onUpdateConfig }: NodeC
                     <TagContactConfig config={config} onChange={handleChange} />
                 )}
                 {nodeType === "ZOHO_UPSERT_LEAD" && <ZohoUpsertConfig />}
-                {(nodeType === "CONDITION" || nodeType === "WAIT_DELAY") && (
-                    <ComingSoonConfig nodeType={NODE_TYPE_LABELS[nodeType] ?? nodeType} />
+                {nodeType === "CONDITION" && (
+                    <ConditionConfig config={config} onChange={handleChange} />
+                )}
+                {nodeType === "WAIT_DELAY" && (
+                    <WaitDelayConfig config={config} onChange={handleChange} />
+                )}
+                {nodeType === "PARALLEL" && (
+                    <ParallelConfig config={config} onChange={handleChange} />
+                )}
+                {nodeType === "AGENT_HANDOFF" && (
+                    <AgentHandoffConfig config={config} onChange={handleChange} />
+                )}
+                {nodeType === "QUALIFICATION_GATE" && <QualificationGateConfig />}
+                {nodeType === "INTENT_ROUTER" && (
+                    <IntentRouterConfig config={config} onChange={handleChange} />
                 )}
-                {isTrigger && (nodeType === "MESSAGE_INBOUND") && (
+                {isTrigger && nodeType === "MESSAGE_INBOUND" && (
                     <TriggerConfig config={config} onChange={handleChange} />
                 )}
                 {isTrigger && nodeType === "LEAD_AD_SUBMIT" && (

frontend/src/app/(dashboard)/automations/[id]/NodePalette.tsx

@@ -3,7 +3,10 @@
 import { useState, useEffect } from "react";
 import { useCatalog, type CatalogEntry } from "@/lib/catalog";
 import { apiClient } from "@/lib/api";
-import { Zap, Bot, Send, User, Tag, Database, GitBranch, Clock, GripVertical, Lock } from "lucide-react";
+import {
+    Zap, Bot, Send, User, Tag, Database, GitBranch, Clock,
+    GripVertical, Lock, Shuffle, ArrowRightCircle, Filter, Map,
+} from "lucide-react";
 import { cn } from "@/lib/utils";
 
 // ---------------------------------------------------------------------------
@@ -20,6 +23,10 @@ const ICON_MAP: Record<string, React.ReactNode> = {
     clock: <Clock className="w-4 h-4" />,
     message: <Zap className="w-4 h-4" />,
     zap: <Zap className="w-4 h-4" />,
+    shuffle: <Shuffle className="w-4 h-4" />,
+    "arrow-right-circle": <ArrowRightCircle className="w-4 h-4" />,
+    filter: <Filter className="w-4 h-4" />,
+    map: <Map className="w-4 h-4" />,
 };
 
 // ---------------------------------------------------------------------------
@@ -171,12 +178,15 @@ export default function NodePalette({ hasTrigger }: NodePaletteProps) {
             {/* Divider */}
             <div className="mx-3 border-t border-border" />
 
-            {/* Action Nodes */}
+            {/* Agent Nodes */}
             <div className="px-3 py-4 space-y-2">
                 <p className="text-[10px] font-bold uppercase tracking-widest text-slate-400 px-1 mb-3">
-                    Actions
+                    Agent Nodes
                 </p>
-                {nodeTypes?.map((n) => (
+                {nodeTypes?.filter((n) => [
+                    "AI_REPLY", "AGENT_REPLY", "SEND_MESSAGE", "HUMAN_HANDOVER",
+                    "TAG_CONTACT", "ZOHO_UPSERT_LEAD",
+                ].includes(n.key)).map((n) => (
                     <PaletteItem
                         key={n.key}
                         nodeType={n.key}
@@ -188,21 +198,77 @@ export default function NodePalette({ hasTrigger }: NodePaletteProps) {
                 )) ?? (
                         <div className="space-y-1.5">
                             {[
-                                { key: "AI_REPLY", label: "AI Reply", iconHint: "bot" },
+                                { key: "AI_REPLY", label: "Agent Reply", iconHint: "bot" },
                                 { key: "SEND_MESSAGE", label: "Send Message", iconHint: "send" },
                                 { key: "HUMAN_HANDOVER", label: "Human Handover", iconHint: "user" },
                                 { key: "TAG_CONTACT", label: "Tag Contact", iconHint: "tag" },
                                 { key: "ZOHO_UPSERT_LEAD", label: "Zoho: Upsert Lead", iconHint: "database" },
-                                { key: "CONDITION", label: "Condition", iconHint: "git-branch", comingSoon: true },
-                                { key: "WAIT_DELAY", label: "Wait / Delay", iconHint: "clock", comingSoon: true },
                             ].map((n) => (
-                                <PaletteItem
-                                    key={n.key}
-                                    nodeType={n.key}
-                                    label={n.label}
-                                    iconHint={n.iconHint}
-                                    comingSoon={"comingSoon" in n ? n.comingSoon : false}
-                                />
+                                <PaletteItem key={n.key} nodeType={n.key} label={n.label} iconHint={n.iconHint} />
+                            ))}
+                        </div>
+                    )}
+            </div>
+
+            {/* Divider */}
+            <div className="mx-3 border-t border-border" />
+
+            {/* Flow Control */}
+            <div className="px-3 py-4 space-y-2">
+                <p className="text-[10px] font-bold uppercase tracking-widest text-slate-400 px-1 mb-3">
+                    Flow Control
+                </p>
+                {nodeTypes?.filter((n) => [
+                    "CONDITION", "WAIT_DELAY", "PARALLEL", "AGENT_HANDOFF",
+                ].includes(n.key)).map((n) => (
+                    <PaletteItem
+                        key={n.key}
+                        nodeType={n.key}
+                        label={n.label}
+                        iconHint={n.icon_hint ?? "git-branch"}
+                        comingSoon={n.runtime_supported === false}
+                        locked={isLocked(n)}
+                    />
+                )) ?? (
+                        <div className="space-y-1.5">
+                            {[
+                                { key: "CONDITION", label: "Condition", iconHint: "git-branch" },
+                                { key: "WAIT_DELAY", label: "Wait / Delay", iconHint: "clock" },
+                                { key: "PARALLEL", label: "Parallel", iconHint: "shuffle" },
+                                { key: "AGENT_HANDOFF", label: "Agent Handoff", iconHint: "arrow-right-circle" },
+                            ].map((n) => (
+                                <PaletteItem key={n.key} nodeType={n.key} label={n.label} iconHint={n.iconHint} />
+                            ))}
+                        </div>
+                    )}
+            </div>
+
+            {/* Divider */}
+            <div className="mx-3 border-t border-border" />
+
+            {/* Routing */}
+            <div className="px-3 py-4 space-y-2">
+                <p className="text-[10px] font-bold uppercase tracking-widest text-slate-400 px-1 mb-3">
+                    Routing
+                </p>
+                {nodeTypes?.filter((n) => [
+                    "QUALIFICATION_GATE", "INTENT_ROUTER",
+                ].includes(n.key)).map((n) => (
+                    <PaletteItem
+                        key={n.key}
+                        nodeType={n.key}
+                        label={n.label}
+                        iconHint={n.icon_hint ?? "filter"}
+                        comingSoon={n.runtime_supported === false}
+                        locked={isLocked(n)}
+                    />
+                )) ?? (
+                        <div className="space-y-1.5">
+                            {[
+                                { key: "QUALIFICATION_GATE", label: "Qualification Gate", iconHint: "filter" },
+                                { key: "INTENT_ROUTER", label: "Intent Router", iconHint: "map" },
+                            ].map((n) => (
+                                <PaletteItem key={n.key} nodeType={n.key} label={n.label} iconHint={n.iconHint} />
                             ))}
                         </div>
                     )}

frontend/src/app/(dashboard)/automations/[id]/nodes/ActionNode.tsx

@@ -12,6 +12,10 @@ import {
     Clock,
     X,
     AlertCircle,
+    Shuffle,
+    ArrowRightCircle,
+    Filter,
+    Map,
 } from "lucide-react";
 import { cn } from "@/lib/utils";
 
@@ -26,6 +30,8 @@ interface NodeMeta {
     headerClass: string; // Tailwind header background
     iconClass: string; // Tailwind icon background + text color
     summary: (config: Record<string, any>) => string | null;
+    /** Nodes with conditional output handles (renders labeled true/false or qualified/unqualified handles) */
+    multiHandle?: "true-false" | "qualified-unqualified";
     comingSoon?: boolean;
 }
 
@@ -38,6 +44,14 @@ const NODE_META: Record<string, NodeMeta> = {
         iconClass: "text-violet-600 bg-violet-100/50 border-violet-200/50",
         summary: (c) => c.goal ? `Goal: ${String(c.goal).substring(0, 40)}` : null,
     },
+    AGENT_REPLY: {
+        label: "Agent Reply",
+        icon: <Bot className="w-4 h-4" />,
+        colorClass: "border-violet-300 hover:border-violet-400",
+        headerClass: "bg-violet-50/50",
+        iconClass: "text-violet-600 bg-violet-100/50 border-violet-200/50",
+        summary: (c) => c.goal ? `Goal: ${String(c.goal).substring(0, 40)}` : null,
+    },
     SEND_MESSAGE: {
         label: "Send Message",
         icon: <Send className="w-4 h-4" />,
@@ -73,20 +87,58 @@ const NODE_META: Record<string, NodeMeta> = {
     CONDITION: {
         label: "Condition",
         icon: <GitBranch className="w-4 h-4" />,
-        colorClass: "border-slate-300 hover:border-slate-400",
-        headerClass: "bg-slate-50/50",
-        iconClass: "text-slate-600 bg-slate-100/50 border-slate-200/50",
-        summary: () => "Branch logic",
-        comingSoon: true,
+        colorClass: "border-cyan-300 hover:border-cyan-400",
+        headerClass: "bg-cyan-50/50",
+        iconClass: "text-cyan-600 bg-cyan-100/50 border-cyan-200/50",
+        summary: (c) => c.condition_type ? `If ${c.condition_type} ${c.operator ?? ""} ${c.value ?? ""}`.trim() : "Configure condition",
+        multiHandle: "true-false",
     },
     WAIT_DELAY: {
         label: "Wait / Delay",
         icon: <Clock className="w-4 h-4" />,
-        colorClass: "border-slate-300 hover:border-slate-400",
-        headerClass: "bg-slate-50/50",
-        iconClass: "text-slate-600 bg-slate-100/50 border-slate-200/50",
-        summary: () => "Pause flow",
-        comingSoon: true,
+        colorClass: "border-indigo-300 hover:border-indigo-400",
+        headerClass: "bg-indigo-50/50",
+        iconClass: "text-indigo-600 bg-indigo-100/50 border-indigo-200/50",
+        summary: (c) => c.delay_seconds ? `Wait ${c.delay_seconds}s (${c.delay_unit ?? "seconds"})` : "Set delay",
+    },
+    PARALLEL: {
+        label: "Parallel",
+        icon: <Shuffle className="w-4 h-4" />,
+        colorClass: "border-teal-300 hover:border-teal-400",
+        headerClass: "bg-teal-50/50",
+        iconClass: "text-teal-600 bg-teal-100/50 border-teal-200/50",
+        summary: (c) => {
+            const agents: string[] = Array.isArray(c.agents) ? c.agents : [];
+            return agents.length ? `Run: ${agents.join(", ")}` : "Configure agents";
+        },
+    },
+    AGENT_HANDOFF: {
+        label: "Agent Handoff",
+        icon: <ArrowRightCircle className="w-4 h-4" />,
+        colorClass: "border-emerald-300 hover:border-emerald-400",
+        headerClass: "bg-emerald-50/50",
+        iconClass: "text-emerald-600 bg-emerald-100/50 border-emerald-200/50",
+        summary: (c) => c.target_agent ? `→ ${c.target_agent}` : "Select agent",
+    },
+    QUALIFICATION_GATE: {
+        label: "Qualification Gate",
+        icon: <Filter className="w-4 h-4" />,
+        colorClass: "border-lime-300 hover:border-lime-400",
+        headerClass: "bg-lime-50/50",
+        iconClass: "text-lime-600 bg-lime-100/50 border-lime-200/50",
+        summary: () => "Route qualified / unqualified",
+        multiHandle: "qualified-unqualified",
+    },
+    INTENT_ROUTER: {
+        label: "Intent Router",
+        icon: <Map className="w-4 h-4" />,
+        colorClass: "border-rose-300 hover:border-rose-400",
+        headerClass: "bg-rose-50/50",
+        iconClass: "text-rose-600 bg-rose-100/50 border-rose-200/50",
+        summary: (c) => {
+            const routes: any[] = Array.isArray(c.routes) ? c.routes : [];
+            return routes.length ? `${routes.length} route(s)` : "Configure routes";
+        },
     },
 };
 
@@ -174,12 +226,56 @@ function ActionNode({ id, data, selected }: NodeProps) {
                 )}
             </div>
 
-            {/* Output handle */}
-            <Handle
-                type="source"
-                position={Position.Bottom}
-                className="!w-3 !h-3 !bg-slate-400 !border-2 !border-white !-bottom-1.5"
-            />
+            {/* Output handle(s) — single by default, two for branching nodes */}
+            {meta.multiHandle === "true-false" ? (
+                <div className="relative h-4 flex justify-between px-6">
+                    <div className="relative flex flex-col items-center">
+                        <span className="text-[9px] font-bold text-emerald-600 mb-0.5">TRUE</span>
+                        <Handle
+                            type="source"
+                            position={Position.Bottom}
+                            id="true"
+                            className="!w-3 !h-3 !bg-emerald-400 !border-2 !border-white !static !transform-none"
+                        />
+                    </div>
+                    <div className="relative flex flex-col items-center">
+                        <span className="text-[9px] font-bold text-red-500 mb-0.5">FALSE</span>
+                        <Handle
+                            type="source"
+                            position={Position.Bottom}
+                            id="false"
+                            className="!w-3 !h-3 !bg-red-400 !border-2 !border-white !static !transform-none"
+                        />
+                    </div>
+                </div>
+            ) : meta.multiHandle === "qualified-unqualified" ? (
+                <div className="relative h-4 flex justify-between px-4">
+                    <div className="relative flex flex-col items-center">
+                        <span className="text-[9px] font-bold text-lime-600 mb-0.5">QUALIFIED</span>
+                        <Handle
+                            type="source"
+                            position={Position.Bottom}
+                            id="qualified"
+                            className="!w-3 !h-3 !bg-lime-500 !border-2 !border-white !static !transform-none"
+                        />
+                    </div>
+                    <div className="relative flex flex-col items-center">
+                        <span className="text-[9px] font-bold text-slate-500 mb-0.5">UNQUALIFIED</span>
+                        <Handle
+                            type="source"
+                            position={Position.Bottom}
+                            id="unqualified"
+                            className="!w-3 !h-3 !bg-slate-400 !border-2 !border-white !static !transform-none"
+                        />
+                    </div>
+                </div>
+            ) : (
+                <Handle
+                    type="source"
+                    position={Position.Bottom}
+                    className="!w-3 !h-3 !bg-slate-400 !border-2 !border-white !-bottom-1.5"
+                />
+            )}
         </div>
     );
 }