cold

src/langbot/pkg/api/http/controller/groups/pipelines/embed.py

@@ -62,16 +62,24 @@ class EmbedRouterGroup(group.RouterGroup):
         """Resolve *bot_uuid* to ``(runtime_bot, pipeline_uuid)``.
 
         Returns ``(None, None)`` when the bot does not exist, is not a
-        ``web_page_bot``, is disabled, or has no pipeline bound.
+        ``web_page_bot``, is disabled, or has no pipeline/workflow bound.
         """
         for bot in self.ap.platform_mgr.bots:
             if (
                 bot.bot_entity.uuid == bot_uuid
                 and bot.bot_entity.adapter == 'web_page_bot'
                 and bot.bot_entity.enable
-                and bot.bot_entity.use_pipeline_uuid
             ):
-                return bot, bot.bot_entity.use_pipeline_uuid
+                # Check for workflow binding first
+                binding_type = getattr(bot.bot_entity, 'binding_type', 'pipeline') or 'pipeline'
+                binding_uuid = getattr(bot.bot_entity, 'binding_uuid', None)
+
+                if binding_type == 'workflow' and binding_uuid:
+                    # For workflow binding, return workflow UUID
+                    return bot, binding_uuid
+                elif bot.bot_entity.use_pipeline_uuid:
+                    # For pipeline binding, return pipeline UUID
+                    return bot, bot.bot_entity.use_pipeline_uuid
         return None, None
 
     def _get_bot_config(self, bot_uuid: str) -> dict:

src/langbot/pkg/api/http/controller/groups/workflows/workflows.py

@@ -61,6 +61,7 @@ class WorkflowsRouterGroup(group.RouterGroup):
             elif quart.request.method == 'DELETE':
                 await self.ap.workflow_service.delete_workflow(workflow_uuid)
                 return self.success()
+            return self.http_status(405, -1, 'method not allowed')
 
         # Publish workflow (enable)
         @self.route('/<workflow_uuid>/publish', methods=['POST'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
@@ -193,6 +194,7 @@ class WorkflowsRouterGroup(group.RouterGroup):
                     return self.success()
                 except ValueError as e:
                     return self.http_status(404, -1, str(e))
+            return self.http_status(405, -1, 'method not allowed')
 
         # Debug API - Start debug execution
         @self.route('/<workflow_uuid>/debug/start', methods=['POST'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)

src/langbot/pkg/api/http/service/workflow.py

@@ -6,7 +6,7 @@ import asyncio
 import logging
 import uuid
 from datetime import datetime, timedelta
-from typing import Optional, TYPE_CHECKING
+from typing import Optional
 
 import sqlalchemy
 

src/langbot/pkg/workflow/executor.py

@@ -10,10 +10,9 @@ Debug execution support has been moved to the ``debug`` module.
 
 from __future__ import annotations
 
-import ast
 import asyncio
 import logging
-import operator
+import re
 import uuid
 from datetime import datetime
 from typing import Any, Optional, TYPE_CHECKING
@@ -32,6 +31,7 @@ from .entities import (
 )
 from ..entity.persistence import workflow as persistence_workflow
 from .registry import NodeTypeRegistry
+from .safe_eval import safe_eval_with_vars
 
 if TYPE_CHECKING:
     from ..core import app
@@ -39,108 +39,6 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-# ─── Safe expression evaluator (replaces eval()) ─────────────────────
-# Uses Python's ast module to whitelist only comparison / boolean / arithmetic
-# operations.  No function calls, attribute access, or subscript injection.
-
-_SAFE_OPS = {
-    ast.Add: operator.add,
-    ast.Sub: operator.sub,
-    ast.Mult: operator.mul,
-    ast.Div: operator.truediv,
-    ast.FloorDiv: operator.floordiv,
-    ast.Mod: operator.mod,
-    ast.Pow: operator.pow,
-    ast.USub: operator.neg,
-    ast.UAdd: operator.pos,
-    ast.Not: operator.not_,
-    ast.Eq: operator.eq,
-    ast.NotEq: operator.ne,
-    ast.Lt: operator.lt,
-    ast.LtE: operator.le,
-    ast.Gt: operator.gt,
-    ast.GtE: operator.ge,
-    ast.Is: operator.is_,
-    ast.IsNot: operator.is_not,
-    ast.In: lambda a, b: a in b,
-    ast.NotIn: lambda a, b: a not in b,
-}
-
-
-def _safe_eval(expr: str) -> Any:
-    """Evaluate a simple expression safely via AST whitelist.
-
-    Supports: literals, comparisons (==, !=, <, >, <=, >=, in, not in, is, is not),
-    boolean logic (and, or, not), arithmetic (+, -, *, /, //, %, **), and
-    string operations (contains via ``in``).
-
-    Raises ``ValueError`` on any disallowed construct (function calls,
-    attribute access, imports, etc.).
-    """
-    tree = ast.parse(expr.strip(), mode='eval')
-    return _eval_node(tree.body)
-
-
-def _eval_node(node: ast.AST) -> Any:
-    # Literals: numbers, strings, True/False/None
-    if isinstance(node, ast.Constant):
-        return node.value
-
-    # Unary operators: -x, +x, not x
-    if isinstance(node, ast.UnaryOp):
-        op_fn = _SAFE_OPS.get(type(node.op))
-        if op_fn is None:
-            raise ValueError(f'Unsupported unary op: {type(node.op).__name__}')
-        return op_fn(_eval_node(node.operand))
-
-    # Binary operators: x + y, x * y, etc.
-    if isinstance(node, ast.BinOp):
-        op_fn = _SAFE_OPS.get(type(node.op))
-        if op_fn is None:
-            raise ValueError(f'Unsupported binary op: {type(node.op).__name__}')
-        return op_fn(_eval_node(node.left), _eval_node(node.right))
-
-    # Comparisons: x == y, x > y, x in y, etc.  (chained)
-    if isinstance(node, ast.Compare):
-        left = _eval_node(node.left)
-        for op, comparator in zip(node.ops, node.comparators):
-            op_fn = _SAFE_OPS.get(type(op))
-            if op_fn is None:
-                raise ValueError(f'Unsupported comparison: {type(op).__name__}')
-            right = _eval_node(comparator)
-            if not op_fn(left, right):
-                return False
-            left = right
-        return True
-
-    # Boolean operators: x and y, x or y
-    if isinstance(node, ast.BoolOp):
-        if isinstance(node.op, ast.And):
-            return all(_eval_node(v) for v in node.values)
-        if isinstance(node.op, ast.Or):
-            return any(_eval_node(v) for v in node.values)
-
-    # Ternary: x if cond else y
-    if isinstance(node, ast.IfExp):
-        return _eval_node(node.body) if _eval_node(node.test) else _eval_node(node.orelse)
-
-    # Tuples / Lists (used in "x in [1,2,3]")
-    if isinstance(node, (ast.Tuple, ast.List)):
-        return [_eval_node(e) for e in node.elts]
-
-    # Name lookup – only allow None, True, False
-    if isinstance(node, ast.Name):
-        if node.id == 'None':
-            return None
-        if node.id == 'True':
-            return True
-        if node.id == 'False':
-            return False
-        raise ValueError(f'Unsupported variable reference: {node.id}')
-
-    raise ValueError(f'Unsupported expression node: {type(node).__name__}')
-
-
 class WorkflowExecutor:
     """
     Workflow execution engine.
@@ -169,10 +67,6 @@ class WorkflowExecutor:
         context.status = ExecutionStatus.RUNNING
         context.start_time = datetime.now()
 
-        # Note: Frontend panel logging has been removed.
-        # A new solution will be implemented separately.
-        monitoring_message_id = ''
-
         try:
             # Build execution graph
             node_map = {node.id: node for node in workflow.nodes}
@@ -575,42 +469,40 @@ class WorkflowExecutor:
         return None
 
     async def _evaluate_condition(self, condition: str, context: ExecutionContext) -> bool:
-        """Evaluate a condition expression safely using AST whitelist"""
-        try:
-            # Resolve variable references in condition
-            if '{{' in condition:
-                import re
+        """Evaluate a condition expression safely.
 
+        Any ``{{ ... }}`` references are resolved against the execution context
+        and bound as **variables** that are passed to :func:`safe_eval_with_vars`.
+        Values are never string-concatenated into the expression, which avoids
+        broken parsing (e.g. values containing quotes) and any injection risk
+        from non-literal value types (lists, dicts, etc.).
+        """
+        variables: dict[str, Any] = {}
+        try:
+            # Resolve variable references in condition into bound variables.
+            if '{{' in condition:
                 pattern = r'\{\{([^}]+)\}\}'
 
-                # First pass: replace all variable references with placeholders
-                placeholders = {}
+                placeholders: dict[str, str] = {}
                 placeholder_idx = 0
 
-                def replace_with_placeholder(match):
+                def replace_with_placeholder(match: re.Match[str]) -> str:
                     nonlocal placeholder_idx
                     var_expr = match.group(1)
-                    placeholder = f'__PH{placeholder_idx}__'
+                    placeholder = f'__ph{placeholder_idx}__'
                     placeholders[placeholder] = var_expr
                     placeholder_idx += 1
                     return placeholder
 
-                condition_with_placeholders = re.sub(pattern, replace_with_placeholder, condition)
+                condition = re.sub(pattern, replace_with_placeholder, condition)
 
-                # Second pass: resolve each placeholder asynchronously
+                # Resolve each placeholder and bind it as a variable, so the
+                # actual value (of any type) is passed through unchanged.
                 for placeholder, var_expr in placeholders.items():
-                    value = await self._resolve_expression(var_expr, context)
-                    if isinstance(value, str):
-                        condition_with_placeholders = condition_with_placeholders.replace(placeholder, f'"{value}"')
-                    elif value is None:
-                        condition_with_placeholders = condition_with_placeholders.replace(placeholder, 'None')
-                    else:
-                        condition_with_placeholders = condition_with_placeholders.replace(placeholder, str(value))
+                    variables[placeholder] = await self._resolve_expression(var_expr, context)
 
-                condition = condition_with_placeholders
-
-            # Safe expression evaluation using AST whitelist
-            result = _safe_eval(condition)
+            # Safe expression evaluation with bound variables (AST whitelist).
+            result = safe_eval_with_vars(condition, variables)
             return bool(result)
 
         except Exception as e:
@@ -753,8 +645,13 @@ class ParallelExecutor:
         results = await asyncio.gather(*tasks, return_exceptions=True)
 
         processed_results = []
-        for result in results:
+        for index, result in enumerate(results):
             if isinstance(result, Exception):
+                logger.error(
+                    f'Parallel branch {index} failed: {result}',
+                    exc_info=result,
+                    extra={'branch_index': index, 'execution_id': context.execution_id},
+                )
                 processed_results.append({'error': str(result)})
             else:
                 processed_results.append(result)

src/langbot/pkg/workflow/nodes/plugin_call.py

@@ -1,41 +0,0 @@
-# """Plugin Call Node - invoke a plugin
-
-# Node metadata is loaded from: ../../templates/metadata/nodes/plugin_call.yaml
-# """
-
-# from __future__ import annotations
-
-# from typing import Any
-
-# from langbot_plugin.api.entities.builtin.workflow.entities import ExecutionContext
-# from ..node import WorkflowNode, workflow_node
-
-# @workflow_node('plugin_call')
-# class PluginCallNode(WorkflowNode):
-#     """Plugin call node - invoke a plugin"""
-
-#     type_name = "plugin_call"
-#     category = "action"
-#     icon = "🔌"
-#     name = "plugin_call"
-#     description = "plugin_call"
-
-#     inputs: ClassVar[list[NodePort]] = []
-#     outputs: ClassVar[list[NodePort]] = []
-#     config_schema: ClassVar[list[NodeConfig]] = []
-
-#     async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-#         plugin_name = self.get_config("plugin_name", "")
-#         method_name = self.get_config("method_name", "")
-#         arguments = inputs.get("arguments", {})
-
-#         return {
-#             "result": None,
-#             "success": False,
-#             "error": f"Plugin call '{plugin_name}/{method_name}' not implemented yet",
-#             "_debug": {
-#                 "plugin_name": plugin_name,
-#                 "method_name": method_name,
-#                 "arguments": arguments,
-#             },
-#         }