Update to Tucana 0.0.82

.github/workflows/build.yml

@@ -36,7 +36,7 @@ jobs:
 
     env:
       # code0-definition release the built-in definitions are generated from.
-      DEFINITIONS_VERSION: ${{ vars.HERCULES_DEFINITIONS_VERSION || 'def-0.0.34' }}
+      DEFINITIONS_VERSION: ${{ vars.HERCULES_DEFINITIONS_VERSION || 'def-0.0.35' }}
 
     steps:
       - uses: actions/checkout@v6

.github/workflows/publish.yml

@@ -53,7 +53,7 @@ jobs:
 
     env:
       # code0-definition release the built-in definitions are generated from.
-      DEFINITIONS_VERSION: ${{ vars.HERCULES_DEFINITIONS_VERSION || 'def-0.0.34' }}
+      DEFINITIONS_VERSION: ${{ vars.HERCULES_DEFINITIONS_VERSION || 'def-0.0.35' }}
 
     steps:
       - uses: actions/checkout@v6

py/README.md

@@ -36,7 +36,7 @@ The built-in data types / functions in `hercules/definitions/` are **generated**
 release:
 
 ```bash
-uv run python scripts/build_definitions.py --version def-0.0.34
+uv run python scripts/build_definitions.py --version def-0.0.35
 ```
 
 ## Build & release
@@ -55,7 +55,7 @@ git tag 0.1.0 && git push origin 0.1.0
 ```
 
 The `code0-definition` release used for the built-in definitions is controlled by
-the `HERCULES_DEFINITIONS_VERSION` repository variable (default `def-0.0.34`).
+the `HERCULES_DEFINITIONS_VERSION` repository variable (default `def-0.0.35`).
 
 Build locally:
 

py/examples/simple-example-py/functions/for_each_consumers_runtime_function.py

@@ -0,0 +1,45 @@
+from hercules import (
+    DisplayMessage,
+    FunctionContext,
+    Identifier,
+    Name,
+    Parameter,
+    RuntimeFunctionRunnable,
+    Signature,
+)
+
+
+@Identifier("for_each_consumers_runtime")
+@Signature("<T>(list: LIST<T>, consumers: LIST<CONSUMER<T>>): void")
+@Name({"code": "en-US", "content": "For Each (Multiple Consumers)"})
+@DisplayMessage(
+    {"code": "en-US", "content": "For each element of ${list} run every consumer in ${consumers}"}
+)
+@Parameter(
+    {
+        "runtime_name": "list",
+        "name": [{"code": "en-US", "content": "List"}],
+        "description": [
+            {"code": "en-US", "content": "The list whose elements are iterated over"}
+        ],
+    }
+)
+@Parameter(
+    {
+        "runtime_name": "consumers",
+        "name": [{"code": "en-US", "content": "Consumers"}],
+        "description": [
+            {
+                "code": "en-US",
+                "content": "A list of sub flows (item) => void; every consumer is run once per element",
+            }
+        ],
+    }
+)
+class ForEachConsumersRuntimeFunction(RuntimeFunctionRunnable):
+    async def run(self, context: FunctionContext, items, consumers):
+        print(f"[for_each_consumers] received {len(consumers)} consumer(s)")
+        for element in items:
+            for index, consumer in enumerate(consumers):
+                result = await consumer(element)
+                print(f"[for_each_consumers] consumer #{index} result:", result)

py/examples/simple-example-py/functions/for_each_runtime_function.py

@@ -36,6 +36,8 @@
 )
 class ForEachRuntimeFunction(RuntimeFunctionRunnable):
     async def run(self, context: FunctionContext, items, consumer):
+        print("[for_each] consumer input schema:", consumer.input_schema)
+        print("[for_each] consumer output schema:", consumer.output_schema)
         for element in items:
             result = await consumer(element)
             print("[for_each] sub flow result:", result)

py/examples/simple-example-py/index.py

@@ -8,6 +8,7 @@
 from functions.fibonacci_function import FibonacciFunction
 from functions.fibonacci_runtime_function import FibonacciRuntimeFunction
 from functions.for_each_runtime_function import ForEachRuntimeFunction
+from functions.for_each_consumers_runtime_function import ForEachConsumersRuntimeFunction
 
 action = Action(
     os.environ.get("ACTION_ID", "testing-action"),
@@ -35,6 +36,9 @@
 # Runtime function that executes a sub flow parameter for each element of a list.
 action.register_runtime_function(ForEachRuntimeFunction)
 
+# Runtime function that takes a list of consumers (each an inline ${signature} sub flow reference).
+action.register_runtime_function(ForEachConsumersRuntimeFunction)
+
 # Data type: derived from a schema.
 action.register_data_type_class(EmailDataType)
 

py/hercules/action.py

@@ -103,10 +103,12 @@ def __init__(
         self._channel = None
         self._stream = None
         self._actions = {a.packet_type: a.handle for a in _packet_handlers}
-        # Pending sub flow / flow execution requests awaiting a response, keyed
-        # by execution identifier. A queue matches responses FIFO because a sub
-        # flow can be executed repeatedly under the same execution identifier.
-        self._pending_executions: Dict[str, List[asyncio.Future]] = {}
+        # Pending sub flow / flow execution requests awaiting a response, keyed by
+        # a per-invocation identifier: sub flow executions use the correlation
+        # identifier generated for each invocation (echoed back on the response),
+        # flow executions use their execution identifier. Both are unique per call,
+        # so each key maps to a single pending request.
+        self._pending_executions: Dict[str, asyncio.Future] = {}
 
         self.configs = ConfigManager()
         self.flows = FlowManager()
@@ -199,34 +201,46 @@ async def send(self, request) -> None:
             raise RuntimeError("STREAM_NOT_CONNECTED", "Not connected. Call connect() first.")
         await self._stream.write(request)
 
-    async def fire(self, event_class: type, project_id: int, payload: PlainValue) -> None:
+    async def fire(self, target, payload: PlainValue = None):
+        """Fire the flow(s) bound to an event, or a single flow by id.
+
+        ``fire(event_class, payload)`` executes every registered flow whose
+        ``type`` matches the event class' identifier, resolving the results
+        together as a list. ``fire(flow_id, payload)`` executes just that flow and
+        returns its result. Flows are tracked from the ActionFlowUpdate messages
+        Aquila streams (see :attr:`flows`); each is executed via
+        :meth:`execute_flow`.
+        """
         if self._stream is None:
             raise Exception("Not connected. Call connect() first.")
-        event_type = get_metadata("hercules:identifier", event_class)
-        if not event_type:
-            raise Exception(f"{event_class.__name__} is missing an @Identifier decorator.")
-        request = action_pb2.ActionTransferRequest(
-            event=action_pb2.ActionEvent(
-                project_id=int(project_id),
-                event_type=event_type,
-                payload=construct_value(payload if payload is not None else None),
-            )
-        )
-        await self.send(request)
-        self.emit(event_type, project_id, payload)
-        self.emit(CodeZeroEvent.stream_message_sent, request)
+
+        # Single flow by id.
+        if isinstance(target, int) and not isinstance(target, bool):
+            return await self.execute_flow(target, payload)
+
+        # All flows bound to the event's flow type.
+        flow_type = get_metadata("hercules:identifier", target)
+        if not flow_type:
+            raise Exception(f"{target.__name__} is missing an @Identifier decorator.")
+        flows = self.flows.filter(lambda flow, _key: flow.type == flow_type)
+        return await asyncio.gather(*(self.execute_flow(flow.flow_id, payload) for flow in flows))
 
     async def execute_sub_flow(self, sub_flow, *params: PlainValue) -> PlainValue:
         if self._stream is None:
             raise Exception("Not connected. Call connect() first.")
         execution_identifier = sub_flow.execution_identifier
-        result = self._await_execution_response(execution_identifier)
+        # Scope this individual invocation with a fresh correlation identifier so
+        # its response can be matched even when the same sub flow is executed
+        # repeatedly and responses arrive out of order.
+        correlation_identifier = str(uuid.uuid4())
+        result = self._await_execution_response(correlation_identifier)
         request = action_pb2.ActionTransferRequest(
             sub_flow_execution=action_pb2.ActionSubFlowExecutionRequest(
                 execution_identifier=execution_identifier,
                 parameters=[
                     construct_value(p if p is not None else None) for p in params
                 ],
+                correlation_identifier=correlation_identifier,
             )
         )
         await self.send(request)
@@ -249,22 +263,19 @@ async def execute_flow(self, flow_id, payload: Optional[PlainValue] = None) -> P
         self.emit(CodeZeroEvent.stream_message_sent, request)
         return await result
 
-    def _await_execution_response(self, execution_identifier: str) -> asyncio.Future:
+    def _await_execution_response(self, identifier: str) -> asyncio.Future:
         future: asyncio.Future = asyncio.get_event_loop().create_future()
-        self._pending_executions.setdefault(execution_identifier, []).append(future)
+        self._pending_executions[identifier] = future
         return future
 
-    def resolve_execution_response(self, execution_identifier: str, response) -> None:
-        queue = self._pending_executions.get(execution_identifier)
-        pending = queue.pop(0) if queue else None
-        if queue is not None and len(queue) == 0:
-            del self._pending_executions[execution_identifier]
+    def resolve_execution_response(self, identifier: str, response) -> None:
+        pending = self._pending_executions.pop(identifier, None)
         if pending is None:
             self.emit(
                 CodeZeroEvent.error,
                 Exception(
                     f"Received execution response for unknown execution identifier: "
-                    f"{execution_identifier}"
+                    f"{identifier}"
                 ),
             )
             return

py/hercules/actions/execution.py

@@ -5,10 +5,11 @@
 import inspect
 import time
 
-from hercules._tucana.helpers import construct_value, to_allowed_value
+from hercules._tucana.helpers import construct_value
 from tucana.generated.aquila import action_pb2
 from tucana.generated.shared import errors_pb2, execution_result_pb2
 from hercules.events import CodeZeroEvent
+from hercules.internal.literal import resolve_node_value
 from hercules.types import ProjectConfiguration, RuntimeError
 
 packet_type = "execution"
@@ -23,21 +24,7 @@ def _build_params(action, execution, func):
     parameters = func.parameters or []
     for index, _param in enumerate(parameters):
         field = execution.parameters[index] if index < len(execution.parameters) else None
-        which = field.WhichOneof("value") if field is not None else None
-        if which == "literal_value":
-            params.append(to_allowed_value(field.literal_value.value))
-        elif which == "sub_flow":
-            sub_flow = field.sub_flow
-
-            def make_caller(sub_flow):
-                async def caller(*args):
-                    return await action.execute_sub_flow(sub_flow, *args)
-
-                return caller
-
-            params.append(make_caller(sub_flow))
-        else:
-            params.append(None)
+        params.append(resolve_node_value(action, field))
     return params
 
 

py/hercules/actions/sub_flow_execution.py

@@ -11,4 +11,4 @@
 
 def handle(action, response) -> None:
     action.emit(CodeZeroEvent.sub_flow_execution_response_received, response)
-    action.resolve_execution_response(response.execution_identifier, response)
+    action.resolve_execution_response(response.correlation_identifier, response)

py/hercules/internal/literal.py

@@ -0,0 +1,113 @@
+"""Inline ``${signature}`` reference resolution (port of ``src/internal/literal.ts``).
+
+Resolves an ``aquila.ActionLiteralValue`` into a plain Python value, substituting
+every ``${signature}`` placeholder inside a (possibly nested) string leaf with the
+value of the matching inline reference.
+"""
+from __future__ import annotations
+
+import json
+import re
+from typing import Optional
+
+from hercules._tucana.helpers import PlainValue, to_allowed_value
+from hercules.types import RuntimeError
+from tucana.generated.shared import struct_pb2
+
+__all__ = ["resolve_literal", "resolve_node_value", "to_sub_flow_caller"]
+
+# Matches every ``${signature}`` placeholder inside a string.
+_REFERENCE_PATTERN = re.compile(r"\$\{([^}]+)\}")
+# Matches a string that consists of exactly one ``${signature}`` placeholder.
+# ``\A``/``\Z`` (not ``^``/``$``) so a trailing newline is not treated as a sole match,
+# mirroring JavaScript's ``$`` anchor.
+_SOLE_REFERENCE_PATTERN = re.compile(r"\A\$\{([^}]+)\}\Z")
+
+
+def to_sub_flow_caller(action, sub_flow):
+    """Wrap a sub flow value in an awaitable caller exposing its declared I/O schema."""
+
+    async def caller(*args):
+        return await action.execute_sub_flow(sub_flow, *args)
+
+    # Expose the sub flow's declared I/O so the handler can inspect it.
+    caller.input_schema = sub_flow.input_schema
+    caller.output_schema = sub_flow.output_schema
+    return caller
+
+
+def resolve_node_value(action, node):
+    """Resolve a single parameter node into a concrete value.
+
+    Literal values have their inline ``${signature}`` references substituted; sub
+    flows become awaitable callers. A missing node resolves to ``None``.
+    """
+    which = node.WhichOneof("value") if node is not None else None
+    if which == "literal_value":
+        return resolve_literal(action, node.literal_value)
+    if which == "sub_flow":
+        return to_sub_flow_caller(action, node.sub_flow)
+    return None
+
+
+def resolve_literal(action, literal):
+    """Resolve an ``ActionLiteralValue`` into a plain value.
+
+    A string that is exactly ``${signature}`` adopts the referenced value verbatim
+    (preserving its type); mixed strings interpolate the referenced value as text.
+    Unknown signatures are left untouched.
+    """
+    references = {}
+    for reference in literal.references:
+        references[reference.signature] = resolve_node_value(action, reference.value)
+    if literal.HasField("value"):
+        return _resolve_value(literal.value, references)
+    return None
+
+
+def _resolve_value(value: struct_pb2.Value, references: dict):
+    kind = value.WhichOneof("kind")
+    if kind == "string_value":
+        return _resolve_string(value.string_value, references)
+    if kind == "struct_value":
+        return {k: _resolve_value(v, references) for k, v in value.struct_value.fields.items()}
+    if kind == "list_value":
+        return [_resolve_value(v, references) for v in value.list_value.values]
+    # Numbers, booleans and null cannot carry placeholders.
+    return to_allowed_value(value)
+
+
+def _resolve_string(raw: str, references: dict):
+    sole = _SOLE_REFERENCE_PATTERN.match(raw)
+    if sole is not None:
+        signature = sole.group(1)
+        # Adopt the referenced value verbatim so its type (number, dict, sub flow, ...) is preserved.
+        return references[signature] if signature in references else raw
+
+    def replace(match: "re.Match[str]") -> str:
+        signature = match.group(1)
+        if signature in references:
+            return _stringify_reference(signature, references[signature])
+        return match.group(0)
+
+    return _REFERENCE_PATTERN.sub(replace, raw)
+
+
+def _stringify_reference(signature: str, value: Optional[PlainValue]) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, str):
+        return value
+    if callable(value):
+        # A sub flow has no textual form, so it cannot be interpolated into a string.
+        raise RuntimeError(
+            "INLINE_REFERENCE_NOT_STRINGIFIABLE",
+            f"Inline reference ${{{signature}}} resolves to a sub flow and cannot be interpolated into a string",
+        )
+    if isinstance(value, bool):
+        # Match the cross-SDK textual form ("true"/"false"), not Python's "True"/"False".
+        return "true" if value else "false"
+    if isinstance(value, (int, float)):
+        return str(value)
+    # dict / list -> compact JSON, matching the TS ``JSON.stringify`` output.
+    return json.dumps(value, separators=(",", ":"))