feat: assertion node component with pass/fail routing (#180)

platform/components/__init__.py

@@ -31,6 +31,7 @@ def get_component_factory(component_type: str):
 from components import (  # noqa: E402, F401
     agent,
     ai_model,
+    assertion,
     categorizer,
     code,
     control_flow,

platform/components/assertion.py

@@ -0,0 +1,43 @@
+"""Assertion component — evaluates all rules and routes pass/fail."""
+
+from __future__ import annotations
+
+from components import register
+from components.operators import evaluate_rules
+
+
+@register("assertion")
+def assertion_factory(node):
+    """Build an assertion graph node that checks all rules against state."""
+    extra = node.component_config.extra_config
+    rules = extra.get("rules", [])
+
+    def assertion_node(state: dict) -> dict:
+        if not rules:
+            return {
+                "_route": "pass",
+                "output": {"passed": True, "results": []},
+            }
+
+        results = evaluate_rules(rules, state, mode="all")
+
+        results_list = []
+        for rule, result in zip(rules, results):
+            results_list.append({
+                "check": f"{rule.get('field', '')} {rule.get('operator', 'equals')} {rule.get('value', '')}",
+                "passed": result["passed"],
+                "actual": result["actual_value"],
+                "expected": rule.get("value", ""),
+            })
+
+        all_passed = all(r["passed"] for r in results)
+
+        return {
+            "_route": "pass" if all_passed else "fail",
+            "output": {
+                "passed": all_passed,
+                "results": results_list,
+            },
+        }
+
+    return assertion_node

platform/components/operators.py

@@ -101,3 +101,54 @@ def _resolve_field(path: str, state: dict):
         else:
             return None
     return current
+
+
+def evaluate_rules(
+    rules: list[dict],
+    state: dict,
+    mode: str = "first_match",
+) -> str | list[dict]:
+    """Evaluate a list of rules against workflow state.
+
+    Args:
+        rules: List of rule dicts with keys: id, field, operator, value.
+        state: The workflow state dict to evaluate against.
+        mode: "first_match" returns the first matching rule's id (str).
+              "all" evaluates every rule and returns a list of result dicts.
+
+    Returns:
+        mode="first_match": The matching rule's id (str), or "" if none match.
+        mode="all": List of {"rule_id": str, "passed": bool, "actual_value": any}.
+    """
+    if mode == "all":
+        results = []
+        for rule in rules:
+            field_path = rule.get("field", "")
+            operator = rule.get("operator", "equals")
+            value = rule.get("value", "")
+            rule_id = rule.get("id", "")
+
+            field_val = _resolve_field(field_path, state)
+            op_fn = OPERATORS.get(operator)
+            passed = bool(op_fn(field_val, value)) if op_fn else False
+
+            results.append({
+                "rule_id": rule_id,
+                "passed": passed,
+                "actual_value": field_val,
+            })
+        return results
+
+    # mode="first_match" (default)
+    for rule in rules:
+        field_path = rule.get("field", "")
+        operator = rule.get("operator", "equals")
+        value = rule.get("value", "")
+        rule_id = rule.get("id", "")
+
+        field_val = _resolve_field(field_path, state)
+        op_fn = OPERATORS.get(operator)
+        if op_fn and op_fn(field_val, value):
+            return rule_id
+
+    return ""

platform/components/switch.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from components import register
-from components.operators import OPERATORS, UNARY_OPERATORS, _resolve_field
+from components.operators import _resolve_field, evaluate_rules
 
 
 @register("switch")
@@ -17,18 +17,7 @@ def switch_factory(node):
         enable_fallback = extra.get("enable_fallback", False)
 
         def switch_node(state: dict) -> dict:
-            route = ""
-            for rule in rules:
-                field_path = rule.get("field", "")
-                operator = rule.get("operator", "equals")
-                value = rule.get("value", "")
-                rule_id = rule.get("id", "")
-
-                field_val = _resolve_field(field_path, state)
-                op_fn = OPERATORS.get(operator)
-                if op_fn and op_fn(field_val, value):
-                    route = rule_id
-                    break
+            route = evaluate_rules(rules, state, mode="first_match")
 
             if not route and enable_fallback:
                 route = "__other__"