Feature/add template edge processor

.gitignore

@@ -27,3 +27,6 @@ node_modules/
 data/
 temp/
 WareHouse/
+
+# Git worktrees
+.worktrees/

entity/configs/edge/edge_processor.py

@@ -44,8 +44,6 @@ def to_external_value(self) -> Any:
         return _serialize_config(self)
 
 
-
-
 _NO_MATCH_DESCRIPTIONS = {
     "pass": "Leave the payload untouched when no match is found.",
     "default": "Apply default_value (or empty string) if nothing matches.",
@@ -117,7 +115,6 @@ class RegexEdgeProcessorConfig(EdgeProcessorTypeConfig):
             description="Whether to collect all matches instead of only the first.",
             advance=True,
         ),
-
         "template": ConfigFieldSpec(
             name="template",
             display_name="Output Template",
@@ -152,7 +149,9 @@ class RegexEdgeProcessorConfig(EdgeProcessorTypeConfig):
     }
 
     @classmethod
-    def from_dict(cls, data: Mapping[str, Any], *, path: str) -> "RegexEdgeProcessorConfig":
+    def from_dict(
+        cls, data: Mapping[str, Any], *, path: str
+    ) -> "RegexEdgeProcessorConfig":
         mapping = require_mapping(data, path)
         pattern = require_str(mapping, "pattern", path, allow_empty=False)
         group_value = mapping.get("group")
@@ -166,14 +165,19 @@ def from_dict(cls, data: Mapping[str, Any], *, path: str) -> "RegexEdgeProcessor
                 else:
                     group_normalized = group_value
             else:
-                raise ConfigError("group must be str or int", extend_path(path, "group"))
+                raise ConfigError(
+                    "group must be str or int", extend_path(path, "group")
+                )
         multiple = optional_bool(mapping, "multiple", path, default=False)
         case_sensitive = optional_bool(mapping, "case_sensitive", path, default=True)
         multiline = optional_bool(mapping, "multiline", path, default=False)
         dotall = optional_bool(mapping, "dotall", path, default=False)
         on_no_match = optional_str(mapping, "on_no_match", path) or "pass"
         if on_no_match not in {"pass", "default", "drop"}:
-            raise ConfigError("on_no_match must be pass, default or drop", extend_path(path, "on_no_match"))
+            raise ConfigError(
+                "on_no_match must be pass, default or drop",
+                extend_path(path, "on_no_match"),
+            )
 
         template = optional_str(mapping, "template", path)
         default_value = optional_str(mapping, "default_value", path)
@@ -230,18 +234,26 @@ def field_specs(cls) -> Dict[str, ConfigFieldSpec]:
         descriptions = {}
         for name in names:
             meta = metadata.get(name)
-            descriptions[name] = (meta.description if meta else None) or "No description provided."
+            descriptions[name] = (
+                meta.description if meta else None
+            ) or "No description provided."
 
         specs["name"] = replace(
             name_spec,
             enum=names or None,
-            enum_options=enum_options_from_values(names, descriptions, preserve_label_case=True) if names else None,
+            enum_options=enum_options_from_values(
+                names, descriptions, preserve_label_case=True
+            )
+            if names
+            else None,
             description=description,
         )
         return specs
 
     @classmethod
-    def from_dict(cls, data: Mapping[str, Any], *, path: str) -> "FunctionEdgeProcessorConfig":
+    def from_dict(
+        cls, data: Mapping[str, Any], *, path: str
+    ) -> "FunctionEdgeProcessorConfig":
         mapping = require_mapping(data, path)
         name = require_str(mapping, "name", path, allow_empty=False)
         return cls(name=name, path=path)
@@ -253,6 +265,37 @@ def to_external_value(self) -> Any:
         return {"name": self.name}
 
 
+@dataclass
+class TemplateEdgeProcessorConfig(EdgeProcessorTypeConfig):
+    """Configuration for Jinja2 template-based payload processors."""
+
+    template: str = ""
+
+    FIELD_SPECS = {
+        "template": ConfigFieldSpec(
+            name="template",
+            display_name="Template String",
+            type_hint="str",
+            required=True,
+            description="Jinja2 template string for transforming edge payloads.",
+        )
+    }
+
+    @classmethod
+    def from_dict(
+        cls, data: Mapping[str, Any], *, path: str
+    ) -> "TemplateEdgeProcessorConfig":
+        mapping = require_mapping(data, path)
+        template = require_str(mapping, "template", path, allow_empty=False)
+        return cls(template=template, path=path)
+
+    def display_label(self) -> str:
+        return "template"
+
+    def to_external_value(self) -> Any:
+        return {"template": self.template}
+
+
 TProcessorConfig = TypeVar("TProcessorConfig", bound=EdgeProcessorTypeConfig)
 
 
@@ -288,12 +331,18 @@ def from_dict(cls, data: Any, *, path: str) -> "EdgeProcessorConfig":
         processor_type = require_str(mapping, "type", path)
         config_payload = mapping.get("config")
         if config_payload is None:
-            raise ConfigError("processor config is required", extend_path(path, "config"))
+            raise ConfigError(
+                "processor config is required", extend_path(path, "config")
+            )
         try:
             schema = get_edge_processor_schema(processor_type)
         except SchemaLookupError as exc:
-            raise ConfigError(f"unknown processor type '{processor_type}'", extend_path(path, "type")) from exc
-        processor_config = schema.config_cls.from_dict(config_payload, path=extend_path(path, "config"))
+            raise ConfigError(
+                f"unknown processor type '{processor_type}'", extend_path(path, "type")
+            ) from exc
+        processor_config = schema.config_cls.from_dict(
+            config_payload, path=extend_path(path, "config")
+        )
         return cls(type=processor_type, config=processor_config, path=path)
 
     @classmethod
@@ -310,11 +359,15 @@ def field_specs(cls) -> Dict[str, ConfigFieldSpec]:
         if type_spec:
             registrations = iter_edge_processor_schemas()
             names = list(registrations.keys())
-            descriptions = {name: schema.summary for name, schema in registrations.items()}
+            descriptions = {
+                name: schema.summary for name, schema in registrations.items()
+            }
             specs["type"] = replace(
                 type_spec,
                 enum=names,
-                enum_options=enum_options_from_values(names, descriptions, preserve_label_case=True),
+                enum_options=enum_options_from_values(
+                    names, descriptions, preserve_label_case=True
+                ),
             )
         return specs
 
@@ -327,7 +380,9 @@ def to_external_value(self) -> Any:
             "config": self.config.to_external_value(),
         }
 
-    def as_config(self, expected_type: Type[TProcessorConfig]) -> TProcessorConfig | None:
+    def as_config(
+        self, expected_type: Type[TProcessorConfig]
+    ) -> TProcessorConfig | None:
         config = self.config
         if isinstance(config, expected_type):
             return cast(TProcessorConfig, config)

entity/configs/node/template.py

@@ -0,0 +1,42 @@
+"""Configuration for template nodes."""
+
+from dataclasses import dataclass
+from typing import Mapping, Any
+
+from entity.configs.base import (
+    BaseConfig,
+    ConfigError,
+    ConfigFieldSpec,
+    require_mapping,
+    require_str,
+)
+
+
+@dataclass
+class TemplateNodeConfig(BaseConfig):
+    """Config describing the Jinja2 template used to format output."""
+
+    template: str = ""
+
+    @classmethod
+    def from_dict(cls, data: Mapping[str, Any], *, path: str) -> "TemplateNodeConfig":
+        mapping = require_mapping(data, path)
+        template = require_str(mapping, "template", path)
+        if not template:
+            raise ConfigError("template cannot be empty", f"{path}.template")
+
+        return cls(template=template, path=path)
+
+    def validate(self) -> None:
+        if not self.template:
+            raise ConfigError("template cannot be empty", f"{self.path}.template")
+
+    FIELD_SPECS = {
+        "template": ConfigFieldSpec(
+            name="template",
+            display_name="Jinja2 Template",
+            type_hint="text",
+            required=True,
+            description="Jinja2 template string for formatting output. Available context: {{ input }} (latest message content), {{ environment }} (execution environment variables).",
+        ),
+    }

pyproject.toml

@@ -39,6 +39,7 @@ dependencies = [
     "filelock>=3.20.1",
     "markdown>=3.10",
     "xhtml2pdf>=0.2.17",
+    "jinja2>=3.1.0",
 ]
 
 [build-system]

runtime/edge/processors/builtin_types.py

@@ -3,10 +3,12 @@
 from entity.configs.edge.edge_processor import (
     RegexEdgeProcessorConfig,
     FunctionEdgeProcessorConfig,
+    TemplateEdgeProcessorConfig,
 )
 from .registry import register_edge_processor
 from .regex_processor import RegexEdgePayloadProcessor
 from .function_processor import FunctionEdgePayloadProcessor
+from .template_processor import TemplateEdgePayloadProcessor
 
 register_edge_processor(
     "regex_extract",
@@ -21,3 +23,10 @@
     summary="Delegate message transformation to Python functions in functions/edge_processor.",
     config_cls=FunctionEdgeProcessorConfig,
 )
+
+register_edge_processor(
+    "template",
+    processor_cls=TemplateEdgePayloadProcessor,
+    summary="Transform payloads using Jinja2 templates.",
+    config_cls=TemplateEdgeProcessorConfig,
+)

runtime/edge/processors/template_processor.py

@@ -0,0 +1,99 @@
+"""Jinja2 template-based edge payload processor."""
+
+import json
+from typing import Any
+
+from jinja2 import Environment, TemplateSyntaxError, UndefinedError, StrictUndefined
+from jinja2.sandbox import SandboxedEnvironment
+
+from entity.configs.edge.edge_processor import TemplateEdgeProcessorConfig
+from entity.messages import Message
+from runtime.node.executor import ExecutionContext
+from utils.log_manager import LogManager
+
+from .base import EdgePayloadProcessor, ProcessorFactoryContext
+
+
+class TemplateRenderError(Exception):
+    """Raised when template rendering fails."""
+
+    pass
+
+
+def _fromjson_filter(value: str) -> Any:
+    """Parse JSON string into Python object."""
+    try:
+        return json.loads(value)
+    except json.JSONDecodeError as exc:
+        raise TemplateRenderError(f"JSON decode error: {exc}") from exc
+
+
+def _tojson_filter(value: Any) -> str:
+    """Serialize Python object to JSON string."""
+    try:
+        return json.dumps(value, ensure_ascii=False)
+    except (TypeError, ValueError) as exc:
+        raise TemplateRenderError(f"JSON encode error: {exc}") from exc
+
+
+class TemplateEdgePayloadProcessor(EdgePayloadProcessor[TemplateEdgeProcessorConfig]):
+    """Transform edge payloads using Jinja2 templates."""
+
+    def __init__(
+        self, config: TemplateEdgeProcessorConfig, ctx: ProcessorFactoryContext
+    ) -> None:
+        super().__init__(config, ctx)
+
+        # Create sandboxed Jinja2 environment
+        self.env = SandboxedEnvironment(
+            autoescape=False,
+            undefined=StrictUndefined,  # Strict mode - fail on undefined variables
+        )
+
+        # Register custom filters
+        self.env.filters["fromjson"] = _fromjson_filter
+        self.env.filters["tojson"] = _tojson_filter
+
+        # Compile template during initialization to catch syntax errors early
+        try:
+            self.template = self.env.from_string(config.template)
+        except TemplateSyntaxError as exc:
+            raise TemplateRenderError(f"Invalid template syntax: {exc}") from exc
+
+    def transform(
+        self,
+        payload: Message,
+        *,
+        source_result: Message,
+        from_node: Any,
+        edge_link: Any,
+        log_manager: LogManager,
+        context: ExecutionContext,
+    ) -> Message | None:
+        """Render template with payload and context variables."""
+        input_text = self._text(payload)
+
+        # Build template context
+        template_context = {
+            "input": input_text,
+            "environment": getattr(context, "environment", {}),
+            "extracted": input_text,  # Default to input if no prior extraction
+        }
+
+        # Render template
+        try:
+            output = self.template.render(template_context)
+        except UndefinedError as exc:
+            available_keys = ", ".join(sorted(template_context.keys()))
+            raise TemplateRenderError(
+                f"Undefined variable in template: {exc}. Available context keys: {available_keys}"
+            ) from exc
+        except TemplateRenderError:
+            raise
+        except Exception as exc:
+            raise TemplateRenderError(f"Template rendering failed: {exc}") from exc
+
+        # Return new message with rendered output
+        cloned = payload.clone()
+        cloned.content = output
+        return cloned

runtime/node/builtin_nodes.py

@@ -13,6 +13,7 @@
 from entity.configs.node.python_runner import PythonRunnerConfig
 from entity.configs.node.loop_counter import LoopCounterConfig
 from entity.configs.node.loop_timer import LoopTimerConfig
+from entity.configs.node.template import TemplateNodeConfig
 from runtime.node.executor.agent_executor import AgentNodeExecutor
 from runtime.node.executor.human_executor import HumanNodeExecutor
 from runtime.node.executor.passthrough_executor import PassthroughNodeExecutor
@@ -21,6 +22,7 @@
 from runtime.node.executor.subgraph_executor import SubgraphNodeExecutor
 from runtime.node.executor.loop_counter_executor import LoopCounterNodeExecutor
 from runtime.node.executor.loop_timer_executor import LoopTimerNodeExecutor
+from runtime.node.executor.template_executor import TemplateNodeExecutor
 from runtime.node.registry import NodeCapabilities, register_node_type
 
 
@@ -100,6 +102,14 @@
     summary="Blocks downstream edges until the configured time limit is reached, then emits a message to release the loop.",
 )
 
+register_node_type(
+    "template",
+    config_cls=TemplateNodeConfig,
+    executor_cls=TemplateNodeExecutor,
+    capabilities=NodeCapabilities(),
+    summary="Blocks downstream edges until the configured time limit is reached, then emits a message to release the loop.",
+)
+
 # Register subgraph source types (file-based and inline config)
 register_subgraph_source(
     "config",