Update to Tucana 0.80.0

ts/examples/simple-example-ts/src/functions/forEachRuntimeFunction.ts

@@ -0,0 +1,33 @@
+import {
+    Consumer,
+    DisplayMessage,
+    FunctionContext,
+    Identifier,
+    List,
+    Name,
+    Parameter,
+    Signature,
+} from "@code0-tech/hercules";
+
+@Identifier("for_each_runtime")
+@Signature("<T>(list: LIST<T>, consumer: CONSUMER<T>): void")
+@Name({code: "en-US", content: "For Each"})
+@DisplayMessage({code: "en-US", content: "For each element of ${list} do ${consumer}"})
+@Parameter({
+    runtimeName: "list",
+    name: [{code: "en-US", content: "List"}],
+    description: [{code: "en-US", content: "The list whose elements are iterated over"}],
+})
+@Parameter({
+    runtimeName: "consumer",
+    name: [{code: "en-US", content: "Consumer"}],
+    description: [{code: "en-US", content: "The sub flow (item) => void executed once per element"}],
+})
+export class ForEachRuntimeFunction {
+    async run<T>(_context: FunctionContext, list: List<T>, consumer: Consumer<T>): Promise<void> {
+        for (const element of list) {
+            const result = await consumer(element);
+            console.log(`[for_each] sub flow result:`, result);
+        }
+    }
+}

ts/examples/simple-example-ts/src/index.ts

@@ -1,11 +1,12 @@
 import {Action, CodeZeroEvent} from "@code0-tech/hercules";
 import {FibonacciRuntimeFunction} from "./functions/fibonacciRuntimeFunction.js";
 import {FibonacciFunction} from "./functions/fibonacciFunction.js";
+import {ForEachRuntimeFunction} from "./functions/forEachRuntimeFunction.js";
 import {UserCreatedRuntimeEvent} from "./events/userCreatedRuntimeEvent.js";
 import {EmailDataType} from "./data_types/emailDataType.js";
 
 const action = new Action(
-    process.env.ACTION_ID ?? "example-action",
+    process.env.ACTION_ID ?? "testing-action",
     process.env.VERSION ?? "0.0.0",
     process.env.AQUILA_URL ?? "127.0.0.1:8081",
     "code0-tech",
@@ -25,6 +26,9 @@ action.registerRuntimeFunction(FibonacciRuntimeFunction);
 // Function: named public variant that extends the runtime function
 action.registerFunction(FibonacciFunction);
 
+// Runtime function that executes a sub flow parameter for each element of a list
+action.registerRuntimeFunction(ForEachRuntimeFunction);
+
 // Data type: derived from Zod schema
 action.registerDataTypeClass(EmailDataType);
 
@@ -36,17 +40,23 @@ action.on(CodeZeroEvent.connected, () => {
     console.log("Connected to aquila");
 });
 
-action.on(CodeZeroEvent.streamMessageReceived, (message: any) => {
-    console.log(message);
-});
-
 action.on(CodeZeroEvent.error, (error: Error) => {
     console.error("Stream error:", error.message);
-    process.exit(0);
+    console.log("Attempting to reconnect in 5s...");
+    setTimeout(() => {
+        action.connect(process.env.AUTH_TOKEN ?? "your_auth_token_here").catch((err: Error) => {
+            action.emit(CodeZeroEvent.error, err);
+        })
+    }, 5000);
 });
 
-
-action.connect(process.env.AUTH_TOKEN ?? "token").catch((err: unknown) => {
-    console.error("Failed to connect:", err);
-    process.exit(1);
+action.connect(process.env.AUTH_TOKEN ?? "your_auth_token_here").catch((err: Error) => {
+    action.emit(CodeZeroEvent.error, err);
 });
+
+action.on(CodeZeroEvent.moduleUpdated, (message: any) => {
+    console.dir(message, {depth: null});
+    console.dir(action.configs.values(), {depth: null})
+})
+
+export {action};

ts/package.json

@@ -39,7 +39,7 @@
   "devDependencies": {
     "@emnapi/core": "^1.11.1",
     "@emnapi/runtime": "^1.11.1",
-    "@code0-tech/tucana": "0.0.74",
+    "@code0-tech/tucana": "0.0.80",
     "@grpc/grpc-js": "^1.14.3",
     "@protobuf-ts/grpc-backend": "^2.11.1",
     "@protobuf-ts/grpc-transport": "^2.11.1",
@@ -57,7 +57,7 @@
     "vitest": "^4.1.2"
   },
   "peerDependencies": {
-    "@code0-tech/tucana": "0.0.74",
+    "@code0-tech/tucana": "0.0.80",
     "@grpc/grpc-js": "^1.14.3",
     "@protobuf-ts/grpc-backend": "^2.11.1",
     "@protobuf-ts/grpc-transport": "^2.11.1",

ts/src/action.ts

@@ -1,7 +1,10 @@
 import {EventEmitter} from "node:events";
+import {randomUUID} from "node:crypto";
 import type {GrpcOptions, GrpcTransport} from "@protobuf-ts/grpc-transport";
-import {ActionTransferRequest, type ActionTransferResponse} from "@code0-tech/tucana/aquila";
-import {constructValue, type PlainValue} from "@code0-tech/tucana/helpers";
+import {ActionTransferRequest, type ActionNodeSubFlowValue, type ActionTransferResponse} from "@code0-tech/tucana/aquila";
+import {constructValue, toAllowedValue, type PlainValue} from "@code0-tech/tucana/helpers";
+import type {Value, Error as ProtoError} from "@code0-tech/tucana/shared";
+import {RuntimeError} from "./types";
 import type {DuplexStreamingCall} from "@protobuf-ts/runtime-rpc";
 import type {FunctionClass} from "./models/function.model";
 import type {RuntimeFunctionClass} from "./models/runtime_function.model";
@@ -18,6 +21,7 @@ import {CodeZeroEvent, type CodeZeroEventMap} from "./events";
 import {createConnection} from "./internal/connection";
 import {buildModule} from "./internal/module-builder";
 import {ConfigManager} from "./manager/config-manager";
+import {FlowManager} from "./manager/FlowManager";
 import {FunctionManager} from "./manager/FunctionManager";
 import {RuntimeFunctionManager} from "./manager/RuntimeFunctionManager";
 import {DataTypeManager} from "./manager/DataTypeManager";
@@ -41,8 +45,16 @@ export class Action extends EventEmitter<CodeZeroEventMap> {
     private _transport?: GrpcTransport;
     private _stream?: DuplexStreamingCall<ActionTransferRequest, ActionTransferResponse>;
     private readonly _actions = new Map(actions.map(a => [a.packetType, a.handle]));
+    // Pending sub flow / flow execution requests awaiting a response, keyed by
+    // execution identifier. A queue is used because a sub flow can be executed
+    // repeatedly under the same execution identifier; responses are matched FIFO.
+    private readonly _pendingExecutions = new Map<string, {
+        resolve: (value: PlainValue) => void;
+        reject: (error: unknown) => void;
+    }[]>();
 
     readonly configs = new ConfigManager();
+    readonly flows = new FlowManager();
     readonly functions = new FunctionManager();
     readonly runtimeFunctions = new RuntimeFunctionManager();
     readonly dataTypes = new DataTypeManager();
@@ -130,6 +142,88 @@ export class Action extends EventEmitter<CodeZeroEventMap> {
         this.emit(CodeZeroEvent.streamMessageSent, request);
     }
 
+    /**
+     * Execute the sub flow referenced by a {@link ActionNodeSubFlowValue} parameter
+     * with the given parameters and resolve with its result. May be called
+     * repeatedly for the same sub flow (e.g. once per iteration).
+     */
+    async executeSubFlow(subFlow: ActionNodeSubFlowValue, ...params: PlainValue[]): Promise<PlainValue> {
+        if (!this._stream) throw new Error("Not connected. Call connect() first.");
+        const {executionIdentifier} = subFlow;
+        const result = this._awaitExecutionResponse(executionIdentifier);
+        const request = ActionTransferRequest.create({
+            data: {
+                oneofKind: "subFlowExecution",
+                subFlowExecution: {
+                    executionIdentifier,
+                    parameters: params.map(p => constructValue(p ?? null)),
+                },
+            },
+        });
+        await this._stream.requests.send(request);
+        this.emit(CodeZeroEvent.streamMessageSent, request);
+        return result;
+    }
+
+    /**
+     * Execute one of the action's own flows by id and resolve with its result.
+     */
+    async executeFlow(flowId: string | bigint, payload?: PlainValue): Promise<PlainValue> {
+        if (!this._stream) throw new Error("Not connected. Call connect() first.");
+        const executionIdentifier = randomUUID();
+        const result = this._awaitExecutionResponse(executionIdentifier);
+        const request = ActionTransferRequest.create({
+            data: {
+                oneofKind: "flowExecution",
+                flowExecution: {
+                    executionIdentifier,
+                    flowId: String(flowId),
+                    payload: constructValue(payload ?? null),
+                },
+            },
+        });
+        await this._stream.requests.send(request);
+        this.emit(CodeZeroEvent.streamMessageSent, request);
+        return result;
+    }
+
+    private _awaitExecutionResponse(executionIdentifier: string): Promise<PlainValue> {
+        return new Promise<PlainValue>((resolve, reject) => {
+            const queue = this._pendingExecutions.get(executionIdentifier) ?? [];
+            queue.push({resolve, reject});
+            this._pendingExecutions.set(executionIdentifier, queue);
+        });
+    }
+
+    /**
+     * Resolve or reject a pending sub flow / flow execution request. Invoked by
+     * the response handlers when Aquila reports an execution's outcome.
+     */
+    resolveExecutionResponse(
+        executionIdentifier: string,
+        result:
+            | {oneofKind: "success"; success: Value}
+            | {oneofKind: "failure"; failure: ProtoError}
+            | {oneofKind: undefined},
+    ): void {
+        const queue = this._pendingExecutions.get(executionIdentifier);
+        const pending = queue?.shift();
+        if (queue && queue.length === 0) this._pendingExecutions.delete(executionIdentifier);
+        if (!pending) {
+            this.emit(CodeZeroEvent.error, new Error(
+                `Received execution response for unknown execution identifier: ${executionIdentifier}`,
+            ));
+            return;
+        }
+        if (result.oneofKind === "success") {
+            pending.resolve(toAllowedValue(result.success));
+        } else if (result.oneofKind === "failure") {
+            pending.reject(new RuntimeError(result.failure.code, result.failure.message));
+        } else {
+            pending.reject(new Error("Received execution response with no result"));
+        }
+    }
+
     async connect(authToken: string, aquilaUrl?: string, grpcOptions?: GrpcOptions) {
         if (isExportMode()) return;
         const url = aquilaUrl ?? this._aquilaUrl;

ts/src/actions/Execution.ts

@@ -1,7 +1,7 @@
 import {constructValue, PlainValue, toAllowedValue} from "@code0-tech/tucana/helpers";
 import {ActionExecutionRequest, ActionExecutionResponse, ActionTransferRequest} from "@code0-tech/tucana/aquila";
 import {NodeExecutionResult, Error as ProtoError} from "@code0-tech/tucana/shared";
-import {FunctionContext, RuntimeError} from "../types";
+import {FunctionContext, RuntimeError, SubFlow} from "../types";
 import {RuntimeFunctionProps} from "../models/runtime_function.model";
 import {CodeZeroEvent} from "../events";
 import type {Action} from "../action";
@@ -12,10 +12,15 @@ function nowMicros(): bigint {
     return BigInt(Math.floor((performance.timeOrigin + performance.now()) * 1000));
 }
 
-function buildParams(execution: ActionExecutionRequest, func: RuntimeFunctionProps): (PlainValue | undefined)[] {
-    return (func.parameters || []).map(param => {
-        const field = execution.parameters?.fields?.[param.runtimeName];
-        return field ? toAllowedValue(field) : undefined;
+function buildParams(action: Action, execution: ActionExecutionRequest, func: RuntimeFunctionProps): (PlainValue | SubFlow | undefined)[] {
+    return (func.parameters || []).map((param, index) => {
+        const field = execution.parameters?.[index];
+        if (field?.value.oneofKind === "literalValue") return toAllowedValue(field.value.literalValue);
+        if (field?.value.oneofKind === "subFlow") {
+            const subFlow = field.value.subFlow;
+            return (...args: PlainValue[]) => action.executeSubFlow(subFlow, ...args);
+        }
+        return undefined;
     });
 }
 
@@ -28,7 +33,7 @@ export function handle(action: Action, execution: ActionExecutionRequest): void
         return;
     }
 
-    const params = buildParams(execution, func);
+    const params = buildParams(action, execution, func);
 
     const conf = action.configs.get(execution.projectId) ?? {
         projectId: 0n,
@@ -40,6 +45,7 @@ export function handle(action: Action, execution: ActionExecutionRequest): void
         projectId: execution.projectId,
         executionId: execution.executionIdentifier,
         matchedConfig: conf,
+        executeFlow: (flowId, payload) => action.executeFlow(flowId, payload),
     };
 
     const startedAt = nowMicros();
@@ -59,7 +65,7 @@ export function handle(action: Action, execution: ActionExecutionRequest): void
                         startedAt,
                         finishedAt,
                         parameterResults: [],
-                        result: {oneofKind: "success", success: constructValue(value)},
+                        result: {oneofKind: "success", success: constructValue(value ?? null)},
                     }),
                 }),
             },

ts/src/actions/FlowExecution.ts

@@ -0,0 +1,10 @@
+import type {ActionFlowExecutionResponse} from "@code0-tech/tucana/aquila";
+import {CodeZeroEvent} from "../events";
+import type {Action} from "../action";
+
+export const packetType = "flowExecutionResponse";
+
+export function handle(action: Action, response: ActionFlowExecutionResponse): void {
+    action.emit(CodeZeroEvent.flowExecutionResponseReceived, response);
+    action.resolveExecutionResponse(response.executionIdentifier, response.result);
+}

ts/src/actions/FlowUpdate.ts

@@ -0,0 +1,17 @@
+import type {ActionFlowUpdate} from "@code0-tech/tucana/aquila";
+import {CodeZeroEvent} from "../events";
+import type {Action} from "../action";
+
+export const packetType = "flowUpdate";
+
+export function handle(action: Action, update: ActionFlowUpdate): void {
+    if (update.data.oneofKind === "updatedFlow") {
+        const flow = update.data.updatedFlow;
+        action.flows.set(flow.flowId, flow);
+        action.emit(CodeZeroEvent.flowUpdated, flow);
+    } else if (update.data.oneofKind === "deletedFlow") {
+        const flowId = update.data.deletedFlow;
+        action.flows.delete(flowId);
+        action.emit(CodeZeroEvent.flowDeleted, flowId);
+    }
+}

ts/src/actions/SubFlowExecution.ts

@@ -0,0 +1,10 @@
+import type {ActionSubFlowExecutionResponse} from "@code0-tech/tucana/aquila";
+import {CodeZeroEvent} from "../events";
+import type {Action} from "../action";
+
+export const packetType = "subFlowExecutionResponse";
+
+export function handle(action: Action, response: ActionSubFlowExecutionResponse): void {
+    action.emit(CodeZeroEvent.subFlowExecutionResponseReceived, response);
+    action.resolveExecutionResponse(response.executionIdentifier, response.result);
+}

ts/src/actions/index.ts

@@ -1,6 +1,9 @@
 import type {Action} from "../action";
 import * as ModuleConfigurations from "./ModuleConfigurations";
 import * as Execution from "./Execution";
+import * as SubFlowExecution from "./SubFlowExecution";
+import * as FlowExecution from "./FlowExecution";
+import * as FlowUpdate from "./FlowUpdate";
 
 export interface ActionHandler {
     packetType: string;
@@ -10,4 +13,7 @@ export interface ActionHandler {
 export const actions: ActionHandler[] = [
     ModuleConfigurations,
     Execution,
+    SubFlowExecution,
+    FlowExecution,
+    FlowUpdate,
 ];