(self, node: Node, inputs: List[Message])
| 12 | """Emit the configured literal message whenever triggered.""" |
| 13 | |
| 14 | def execute(self, node: Node, inputs: List[Message]) -> List[Message]: |
| 15 | if node.node_type != "literal": |
| 16 | raise ValueError(f"Node {node.id} is not a literal node") |
| 17 | |
| 18 | config = node.as_config(LiteralNodeConfig) |
| 19 | if config is None: |
| 20 | raise ValueError(f"Node {node.id} missing literal configuration") |
| 21 | |
| 22 | self._ensure_not_cancelled() |
| 23 | return [self._build_message( |
| 24 | role=config.role, |
| 25 | content=config.content, |
| 26 | source=node.id, |
| 27 | preserve_role=True, |
| 28 | )] |
| 29 |
nothing calls this directly
no test coverage detected