From c303f5bde6b8c0b77bec8fe8a8c4fcb62db3f81a Mon Sep 17 00:00:00 2001 From: nicosammito Date: Sat, 15 Aug 2026 01:54:57 +0200 Subject: [PATCH 1/4] feat: enhance sub flow execution with correlation identifiers and input/output schema exposure --- .../functions/for_each_runtime_function.py | 2 + py/hercules/action.py | 65 +++++++++------ py/hercules/actions/execution.py | 3 + py/hercules/actions/sub_flow_execution.py | 2 +- py/hercules/types.py | 19 ++++- py/pyproject.toml | 2 +- py/uv.lock | 8 +- .../src/functions/forEachRuntimeFunction.ts | 5 +- ts/package-lock.json | 10 +-- ts/package.json | 4 +- ts/src/action.ts | 82 +++++++++++-------- ts/src/actions/Execution.ts | 13 ++- ts/src/actions/SubFlowExecution.ts | 2 +- ts/src/types.ts | 15 +++- 14 files changed, 151 insertions(+), 81 deletions(-) diff --git a/py/examples/simple-example-py/functions/for_each_runtime_function.py b/py/examples/simple-example-py/functions/for_each_runtime_function.py index b521f28..a90695d 100644 --- a/py/examples/simple-example-py/functions/for_each_runtime_function.py +++ b/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) diff --git a/py/hercules/action.py b/py/hercules/action.py index e63ff1d..50bfd71 100644 --- a/py/hercules/action.py +++ b/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 diff --git a/py/hercules/actions/execution.py b/py/hercules/actions/execution.py index 82134c3..f8cc656 100644 --- a/py/hercules/actions/execution.py +++ b/py/hercules/actions/execution.py @@ -33,6 +33,9 @@ def make_caller(sub_flow): 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 params.append(make_caller(sub_flow)) diff --git a/py/hercules/actions/sub_flow_execution.py b/py/hercules/actions/sub_flow_execution.py index 98cb533..13019dc 100644 --- a/py/hercules/actions/sub_flow_execution.py +++ b/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) diff --git a/py/hercules/types.py b/py/hercules/types.py index 1ce5bc1..1b5017f 100644 --- a/py/hercules/types.py +++ b/py/hercules/types.py @@ -2,11 +2,12 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import Any, Awaitable, Callable, List, Optional, Union +from typing import Any, Awaitable, Callable, List, Optional, Protocol, Union from hercules._tucana.helpers import PlainValue from tucana.generated.shared.flow_type_pb2 import FlowTypeSetting from tucana.generated.shared.runtime_flow_type_pb2 import RuntimeFlowTypeSetting +from tucana.generated.shared.struct_pb2 import Struct # Re-export the uniqueness scope enums (mirror the TS re-exports). FlowTypeSetting_UniquenessScope = FlowTypeSetting.UniquenessScope @@ -14,6 +15,7 @@ __all__ = [ "PlainValue", + "Struct", "FlowTypeSetting_UniquenessScope", "RuntimeFlowTypeSetting_UniquenessScope", "Translation", @@ -31,8 +33,19 @@ class Translation: content: str -# A sub flow is an awaitable callable resolving to a plain value. -SubFlow = Callable[..., Awaitable[PlainValue]] +class SubFlow(Protocol): + """A sub flow passed as a parameter (e.g. a CONSUMER). + + Call it with the sub flow's parameters to execute it and await its result. + The input/output schema declared for the sub flow by the caller are exposed + via :attr:`input_schema` / :attr:`output_schema` (either may be ``None`` if + the caller omitted it). + """ + + input_schema: Optional[Struct] + output_schema: Optional[Struct] + + def __call__(self, *args: PlainValue) -> Awaitable[PlainValue]: ... @dataclass diff --git a/py/pyproject.toml b/py/pyproject.toml index 0039bec..f4159e9 100644 --- a/py/pyproject.toml +++ b/py/pyproject.toml @@ -12,7 +12,7 @@ license = "MIT" dependencies = [ "grpcio>=1.64", "protobuf>=5.0", - "tucana>=0.0.80", + "tucana>=0.0.82", "pydantic>=2", ] diff --git a/py/uv.lock b/py/uv.lock index a6a7fed..24dd373 100644 --- a/py/uv.lock +++ b/py/uv.lock @@ -118,7 +118,7 @@ requires-dist = [ { name = "grpcio", specifier = ">=1.64" }, { name = "protobuf", specifier = ">=5.0" }, { name = "pydantic", specifier = ">=2" }, - { name = "tucana", specifier = ">=0.0.80" }, + { name = "tucana", specifier = ">=0.0.82" }, ] provides-extras = ["codegen"] @@ -768,15 +768,15 @@ wheels = [ [[package]] name = "tucana" -version = "0.0.80" +version = "0.0.82" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "grpcio" }, { name = "protobuf" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/41/56/a409c44d53bf4f63a77d6c0d9e5dadaac3a59cfa02e89f3b00cfb85849a5/tucana-0.0.80.tar.gz", hash = "sha256:b249fbb505306d1ffd77fbdc6ac769bae25a82a9c9e2776d5062ea1b7bce1b2f", size = 21834, upload-time = "2026-08-04T14:09:37.11Z" } +sdist = { url = "https://files.pythonhosted.org/packages/b2/7e/b483593c7b0480a73515848819d6a55c9937c7772dde2f22217b682d18eb/tucana-0.0.82.tar.gz", hash = "sha256:1714911505bc9755e4e80f85dcc8fc755f6e03ad742a57c7de384bb6966e7ad2", size = 22076, upload-time = "2026-08-14T23:18:41.235Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/d7/8f/cdcf8e5d6c520ceca0816a83df7dfb9b8d3dcf5cb697c269f88c21b9e373/tucana-0.0.80-py3-none-any.whl", hash = "sha256:6c6331d31cb59455e7c0839720273eb6347daed093f502dbd73e0b3b81e40fda", size = 66111, upload-time = "2026-08-04T14:09:37.98Z" }, + { url = "https://files.pythonhosted.org/packages/98/84/a3436373627d5d78488b9a3797c40053a69cfdb51cc0490617ba8efe4f04/tucana-0.0.82-py3-none-any.whl", hash = "sha256:4d87dbe8987889fe8071907850a2d37e5c65d2160e9a8d1bc75f9aac1af06480", size = 66322, upload-time = "2026-08-14T23:18:40.214Z" }, ] [[package]] diff --git a/ts/examples/simple-example-ts/src/functions/forEachRuntimeFunction.ts b/ts/examples/simple-example-ts/src/functions/forEachRuntimeFunction.ts index ae5f827..df9e169 100644 --- a/ts/examples/simple-example-ts/src/functions/forEachRuntimeFunction.ts +++ b/ts/examples/simple-example-ts/src/functions/forEachRuntimeFunction.ts @@ -7,6 +7,7 @@ import { Name, Parameter, Signature, + SubFlow, } from "@code0-tech/hercules"; @Identifier("for_each_runtime") @@ -24,7 +25,9 @@ import { description: [{code: "en-US", content: "The sub flow (item) => void executed once per element"}], }) export class ForEachRuntimeFunction { - async run(_context: FunctionContext, list: List, consumer: Consumer): Promise { + async run(_context: FunctionContext, list: List, consumer: Consumer & SubFlow): Promise { + console.log(`[for_each] consumer input schema:`, consumer.inputSchema); + console.log(`[for_each] consumer output schema:`, consumer.outputSchema); for (const element of list) { const result = await consumer(element); console.log(`[for_each] sub flow result:`, result); diff --git a/ts/package-lock.json b/ts/package-lock.json index 60ad0c9..3bed6c6 100644 --- a/ts/package-lock.json +++ b/ts/package-lock.json @@ -12,7 +12,7 @@ "hercules": "bin/hercules.js" }, "devDependencies": { - "@code0-tech/tucana": "0.0.80", + "@code0-tech/tucana": "0.0.82", "@emnapi/core": "^1.11.1", "@emnapi/runtime": "^1.11.1", "@grpc/grpc-js": "^1.14.3", @@ -32,7 +32,7 @@ "zod-to-ts": "^2.1.0" }, "peerDependencies": { - "@code0-tech/tucana": "0.0.74", + "@code0-tech/tucana": "0.0.82", "@grpc/grpc-js": "^1.14.3", "@protobuf-ts/grpc-backend": "^2.11.1", "@protobuf-ts/grpc-transport": "^2.11.1", @@ -119,9 +119,9 @@ } }, "node_modules/@code0-tech/tucana": { - "version": "0.0.80", - "resolved": "https://registry.npmjs.org/@code0-tech/tucana/-/tucana-0.0.80.tgz", - "integrity": "sha512-ocpG/Mg3F5HoVrx8UERwton+3dsFocj50HLPRIbb6XYnqD/wv6Pj2w9rjEnx9CHnIWUkayVgHCLrr4zmKa/cBw==", + "version": "0.0.82", + "resolved": "https://registry.npmjs.org/@code0-tech/tucana/-/tucana-0.0.82.tgz", + "integrity": "sha512-9ImPNeGV09BeLT0M8/F08DHDe10MO6UI/VfE7i2aEYtRsB9DS3jIAGrcMEV67RXdmAj5ZWTtvxWVGOkem0CljQ==", "dev": true, "license": "MIT" }, diff --git a/ts/package.json b/ts/package.json index 96b88d6..f809858 100644 --- a/ts/package.json +++ b/ts/package.json @@ -39,7 +39,7 @@ "devDependencies": { "@emnapi/core": "^1.11.1", "@emnapi/runtime": "^1.11.1", - "@code0-tech/tucana": "0.0.80", + "@code0-tech/tucana": "0.0.82", "@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.80", + "@code0-tech/tucana": "0.0.82", "@grpc/grpc-js": "^1.14.3", "@protobuf-ts/grpc-backend": "^2.11.1", "@protobuf-ts/grpc-transport": "^2.11.1", diff --git a/ts/src/action.ts b/ts/src/action.ts index 4a82571..ba63e15 100644 --- a/ts/src/action.ts +++ b/ts/src/action.ts @@ -45,13 +45,15 @@ export class Action extends EventEmitter { private _transport?: GrpcTransport; private _stream?: DuplexStreamingCall; 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. + // 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. private readonly _pendingExecutions = new Map void; reject: (error: unknown) => void; - }[]>(); + }>(); readonly configs = new ConfigManager(); readonly flows = new FlowManager(); @@ -123,23 +125,35 @@ export class Action extends EventEmitter { } } - async fire(eventClass: EventClass | RuntimeEventClass, projectId: number | bigint, payload: PlainValue) { + /** + * Fire the flow(s) bound to an event, or a single flow by id. + * + * Given an event class, every registered flow whose {@link ActionFlow.type} + * matches the event's identifier is executed; the results are resolved + * together as an array. Given a flow id, only that flow is executed and its + * result is returned. + * + * Flows are tracked from the ActionFlowUpdate messages Aquila streams (see + * {@link Action.flows}). Each flow is executed via {@link Action.executeFlow}. + */ + fire(eventClass: EventClass | RuntimeEventClass, payload?: PlainValue): Promise; + fire(flowId: bigint | number, payload?: PlainValue): Promise; + async fire( + target: EventClass | RuntimeEventClass | bigint | number, + payload?: PlainValue, + ): Promise { if (!this._stream) throw new Error("Not connected. Call connect() first."); - const eventType: string = Reflect.getMetadata('hercules:identifier', eventClass); - if (!eventType) throw new Error(`${eventClass.name} is missing an @Identifier decorator.`); - const request = ActionTransferRequest.create({ - data: { - oneofKind: "event", - event: { - projectId: typeof projectId === "bigint" ? projectId : BigInt(projectId), - eventType, - payload: constructValue(payload ?? null), - }, - }, - }); - await this._stream.requests.send(request); - this.emit(eventType as Extract, projectId, payload); - this.emit(CodeZeroEvent.streamMessageSent, request); + + // Single flow by id. + if (typeof target === "bigint" || typeof target === "number") { + return this.executeFlow(target, payload); + } + + // All flows bound to the event's flow type. + const flowType = Reflect.getMetadata("hercules:identifier", target) as string | undefined; + if (!flowType) throw new Error(`${target.name} is missing an @Identifier decorator.`); + const flows = this.flows.filter(flow => flow.type === flowType); + return Promise.all(flows.map(flow => this.executeFlow(flow.flowId, payload ?? null))); } /** @@ -150,13 +164,18 @@ export class Action extends EventEmitter { async executeSubFlow(subFlow: ActionNodeSubFlowValue, ...params: PlainValue[]): Promise { if (!this._stream) throw new Error("Not connected. Call connect() first."); const {executionIdentifier} = subFlow; - const result = this._awaitExecutionResponse(executionIdentifier); + // 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. + const correlationIdentifier = randomUUID(); + const result = this._awaitExecutionResponse(correlationIdentifier); const request = ActionTransferRequest.create({ data: { oneofKind: "subFlowExecution", subFlowExecution: { executionIdentifier, parameters: params.map(p => constructValue(p ?? null)), + correlationIdentifier, }, }, }); @@ -168,7 +187,7 @@ export class Action extends EventEmitter { /** * Execute one of the action's own flows by id and resolve with its result. */ - async executeFlow(flowId: string | bigint, payload?: PlainValue): Promise { + async executeFlow(flowId: string | bigint | number, payload?: PlainValue): Promise { if (!this._stream) throw new Error("Not connected. Call connect() first."); const executionIdentifier = randomUUID(); const result = this._awaitExecutionResponse(executionIdentifier); @@ -187,31 +206,30 @@ export class Action extends EventEmitter { return result; } - private _awaitExecutionResponse(executionIdentifier: string): Promise { + private _awaitExecutionResponse(identifier: string): Promise { return new Promise((resolve, reject) => { - const queue = this._pendingExecutions.get(executionIdentifier) ?? []; - queue.push({resolve, reject}); - this._pendingExecutions.set(executionIdentifier, queue); + this._pendingExecutions.set(identifier, {resolve, reject}); }); } /** * Resolve or reject a pending sub flow / flow execution request. Invoked by - * the response handlers when Aquila reports an execution's outcome. + * the response handlers when Aquila reports an execution's outcome. The + * identifier is a sub flow's correlation identifier or a flow execution's + * execution identifier. */ resolveExecutionResponse( - executionIdentifier: string, + identifier: 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); + const pending = this._pendingExecutions.get(identifier); + this._pendingExecutions.delete(identifier); if (!pending) { this.emit(CodeZeroEvent.error, new Error( - `Received execution response for unknown execution identifier: ${executionIdentifier}`, + `Received execution response for unknown execution identifier: ${identifier}`, )); return; } diff --git a/ts/src/actions/Execution.ts b/ts/src/actions/Execution.ts index 740da1d..ab17064 100644 --- a/ts/src/actions/Execution.ts +++ b/ts/src/actions/Execution.ts @@ -15,10 +15,19 @@ function nowMicros(): bigint { 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); + // TODO(#358): resolve inline `${signature}` references on the literal value. + if (field?.value.oneofKind === "literalValue") { + const literal = field.value.literalValue.value; + return literal != null ? toAllowedValue(literal) : null; + } if (field?.value.oneofKind === "subFlow") { const subFlow = field.value.subFlow; - return (...args: PlainValue[]) => action.executeSubFlow(subFlow, ...args); + const caller = (...args: PlainValue[]) => action.executeSubFlow(subFlow, ...args); + // Expose the sub flow's declared I/O so the handler can inspect it. + return Object.assign(caller, { + inputSchema: subFlow.inputSchema, + outputSchema: subFlow.outputSchema, + }); } return undefined; }); diff --git a/ts/src/actions/SubFlowExecution.ts b/ts/src/actions/SubFlowExecution.ts index bc06646..e53b3e9 100644 --- a/ts/src/actions/SubFlowExecution.ts +++ b/ts/src/actions/SubFlowExecution.ts @@ -6,5 +6,5 @@ export const packetType = "subFlowExecutionResponse"; export function handle(action: Action, response: ActionSubFlowExecutionResponse): void { action.emit(CodeZeroEvent.subFlowExecutionResponseReceived, response); - action.resolveExecutionResponse(response.executionIdentifier, response.result); + action.resolveExecutionResponse(response.correlationIdentifier, response.result); } diff --git a/ts/src/types.ts b/ts/src/types.ts index 76d2ce5..b2f82c6 100644 --- a/ts/src/types.ts +++ b/ts/src/types.ts @@ -1,17 +1,28 @@ import {FlowTypeSetting_UniquenessScope, RuntimeFlowTypeSetting_UniquenessScope} from "@code0-tech/tucana/shared"; +import type {Struct} from "@code0-tech/tucana/shared"; import type {ActionNodeSubFlowValue} from "@code0-tech/tucana/aquila"; import {PlainValue} from "@code0-tech/tucana/helpers"; import 'reflect-metadata'; export {FlowTypeSetting_UniquenessScope, RuntimeFlowTypeSetting_UniquenessScope}; -export type {ActionNodeSubFlowValue, PlainValue}; +export type {ActionNodeSubFlowValue, PlainValue, Struct}; export interface Translation { code: "en-US" | "de-DE" | string, content: string } -export type SubFlow = (...args: PlainValue[]) => Promise; +/** + * A sub flow passed as a parameter (e.g. a CONSUMER). Call it with the sub flow's + * parameters to execute it and await its result. The input/output schema declared + * for the sub flow by the caller are exposed via {@link SubFlow.inputSchema} and + * {@link SubFlow.outputSchema} (either may be undefined if the caller omitted it). + */ +export interface SubFlow { + (...args: PlainValue[]): Promise; + readonly inputSchema?: Struct; + readonly outputSchema?: Struct; +} export interface FunctionContext { projectId: number | bigint, From f99cb9b6a6268d1d62cce52d002c640e56b2063b Mon Sep 17 00:00:00 2001 From: nicosammito Date: Tue, 18 Aug 2026 14:11:31 +0200 Subject: [PATCH 2/4] feat: add inline reference support --- .../forEachConsumersRuntimeFunction.ts | 37 ++++++ ts/examples/simple-example-ts/src/index.ts | 4 + ts/src/actions/Execution.ts | 28 ++--- ts/src/internal/literal.ts | 101 ++++++++++++++++ ts/test/literal.test.ts | 111 ++++++++++++++++++ 5 files changed, 260 insertions(+), 21 deletions(-) create mode 100644 ts/examples/simple-example-ts/src/functions/forEachConsumersRuntimeFunction.ts create mode 100644 ts/src/internal/literal.ts create mode 100644 ts/test/literal.test.ts diff --git a/ts/examples/simple-example-ts/src/functions/forEachConsumersRuntimeFunction.ts b/ts/examples/simple-example-ts/src/functions/forEachConsumersRuntimeFunction.ts new file mode 100644 index 0000000..5763dc6 --- /dev/null +++ b/ts/examples/simple-example-ts/src/functions/forEachConsumersRuntimeFunction.ts @@ -0,0 +1,37 @@ +import { + Consumer, + DisplayMessage, + FunctionContext, + Identifier, + List, + Name, + Parameter, + Signature, + SubFlow, +} from "@code0-tech/hercules"; + +@Identifier("for_each_consumers_runtime") +@Signature("(list: LIST, consumers: LIST>): 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({ + runtimeName: "list", + name: [{code: "en-US", content: "List"}], + description: [{code: "en-US", content: "The list whose elements are iterated over"}], +}) +@Parameter({ + runtimeName: "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"}], +}) +export class ForEachConsumersRuntimeFunction { + async run(_context: FunctionContext, list: List, consumers: (Consumer & SubFlow)[]): Promise { + console.log(`[for_each_consumers] received ${consumers.length} consumer(s)`); + for (const element of list) { + for (const [index, consumer] of consumers.entries()) { + const result = await consumer(element); + console.log(`[for_each_consumers] consumer #${index} result:`, result); + } + } + } +} diff --git a/ts/examples/simple-example-ts/src/index.ts b/ts/examples/simple-example-ts/src/index.ts index 07a2698..898fcdc 100644 --- a/ts/examples/simple-example-ts/src/index.ts +++ b/ts/examples/simple-example-ts/src/index.ts @@ -2,6 +2,7 @@ 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 {ForEachConsumersRuntimeFunction} from "./functions/forEachConsumersRuntimeFunction.js"; import {UserCreatedRuntimeEvent} from "./events/userCreatedRuntimeEvent.js"; import {EmailDataType} from "./data_types/emailDataType.js"; @@ -29,6 +30,9 @@ action.registerFunction(FibonacciFunction); // Runtime function that executes a sub flow parameter for each element of a list action.registerRuntimeFunction(ForEachRuntimeFunction); +// Runtime function that takes a list of consumers (each an inline ${signature} sub flow reference) +action.registerRuntimeFunction(ForEachConsumersRuntimeFunction); + // Data type: derived from Zod schema action.registerDataTypeClass(EmailDataType); diff --git a/ts/src/actions/Execution.ts b/ts/src/actions/Execution.ts index ab17064..2accd9a 100644 --- a/ts/src/actions/Execution.ts +++ b/ts/src/actions/Execution.ts @@ -1,8 +1,9 @@ -import {constructValue, PlainValue, toAllowedValue} from "@code0-tech/tucana/helpers"; +import {constructValue, PlainValue} 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, SubFlow} from "../types"; +import {FunctionContext, RuntimeError} from "../types"; import {RuntimeFunctionProps} from "../models/runtime_function.model"; +import {ResolvedValue, resolveNodeValue} from "../internal/literal"; import {CodeZeroEvent} from "../events"; import type {Action} from "../action"; @@ -12,25 +13,10 @@ function nowMicros(): bigint { return BigInt(Math.floor((performance.timeOrigin + performance.now()) * 1000)); } -function buildParams(action: Action, execution: ActionExecutionRequest, func: RuntimeFunctionProps): (PlainValue | SubFlow | undefined)[] { - return (func.parameters || []).map((param, index) => { - const field = execution.parameters?.[index]; - // TODO(#358): resolve inline `${signature}` references on the literal value. - if (field?.value.oneofKind === "literalValue") { - const literal = field.value.literalValue.value; - return literal != null ? toAllowedValue(literal) : null; - } - if (field?.value.oneofKind === "subFlow") { - const subFlow = field.value.subFlow; - const caller = (...args: PlainValue[]) => action.executeSubFlow(subFlow, ...args); - // Expose the sub flow's declared I/O so the handler can inspect it. - return Object.assign(caller, { - inputSchema: subFlow.inputSchema, - outputSchema: subFlow.outputSchema, - }); - } - return undefined; - }); +function buildParams(action: Action, execution: ActionExecutionRequest, func: RuntimeFunctionProps): (ResolvedValue | undefined)[] { + return (func.parameters || []).map((_param, index) => + resolveNodeValue(action, execution.parameters?.[index]), + ); } export function handle(action: Action, execution: ActionExecutionRequest): void { diff --git a/ts/src/internal/literal.ts b/ts/src/internal/literal.ts new file mode 100644 index 0000000..b9f5796 --- /dev/null +++ b/ts/src/internal/literal.ts @@ -0,0 +1,101 @@ +import {PlainValue, toAllowedValue} from "@code0-tech/tucana/helpers"; +import type {Value} from "@code0-tech/tucana/shared"; +import type {ActionLiteralValue, ActionNodeSubFlowValue, ActionNodeValue} from "@code0-tech/tucana/aquila"; +import {RuntimeError, SubFlow} from "../types"; +import type {Action} from "../action"; + +/** A resolved parameter/reference value: either a plain value or a callable sub flow. */ +export type ResolvedValue = PlainValue | SubFlow; + +/** Matches every `${signature}` placeholder inside a string. */ +const REFERENCE_PATTERN = /\$\{([^}]+)\}/g; +/** Matches a string that consists of exactly one `${signature}` placeholder. */ +const SOLE_REFERENCE_PATTERN = /^\$\{([^}]+)\}$/; + +/** + * Wrap a sub flow value in a caller that executes it and exposes its declared I/O + * schema (either may be undefined if the caller omitted it). + */ +export function toSubFlowCaller(action: Action, subFlow: ActionNodeSubFlowValue): SubFlow { + const caller = (...args: PlainValue[]) => action.executeSubFlow(subFlow, ...args); + return Object.assign(caller, { + inputSchema: subFlow.inputSchema, + outputSchema: subFlow.outputSchema, + }); +} + +/** + * Resolve a single parameter node into a concrete value. Literal values have their + * inline `${signature}` references substituted; sub flows become callable. + */ +export function resolveNodeValue(action: Action, node: ActionNodeValue | undefined): ResolvedValue | undefined { + if (node?.value.oneofKind === "literalValue") { + return resolveLiteral(action, node.value.literalValue); + } + if (node?.value.oneofKind === "subFlow") { + return toSubFlowCaller(action, node.value.subFlow); + } + return undefined; +} + +/** + * Resolve an {@link ActionLiteralValue} into a plain value. Any `${signature}` + * placeholder inside a (possibly nested) string leaf is substituted with the value + * of the matching inline reference. 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. + */ +export function resolveLiteral(action: Action, literal: ActionLiteralValue): ResolvedValue { + const references = new Map(); + for (const reference of literal.references) { + references.set(reference.signature, resolveNodeValue(action, reference.value)); + } + return literal.value != null ? resolveValue(literal.value, references) : null; +} + +function resolveValue(value: Value, references: Map): ResolvedValue { + switch (value.kind.oneofKind) { + case "stringValue": + return resolveString(value.kind.stringValue, references); + case "structValue": { + const result: Record = {}; + for (const [key, field] of Object.entries(value.kind.structValue.fields)) { + result[key] = resolveValue(field, references); + } + return result; + } + case "listValue": + return value.kind.listValue.values.map((element) => resolveValue(element, references)); + default: + // Numbers, booleans and null cannot carry placeholders. + return toAllowedValue(value); + } +} + +function resolveString(raw: string, references: Map): ResolvedValue { + const sole = raw.match(SOLE_REFERENCE_PATTERN); + if (sole) { + const signature = sole[1]; + // Adopt the referenced value verbatim so its type (number, object, sub flow, …) is preserved. + return references.has(signature) ? (references.get(signature) as ResolvedValue) : raw; + } + return raw.replace(REFERENCE_PATTERN, (placeholder, signature: string) => + references.has(signature) ? stringifyReference(signature, references.get(signature)) : placeholder, + ); +} + +function stringifyReference(signature: string, value: ResolvedValue | undefined): string { + if (value == null) return ""; + if (typeof value === "string") return value; + if (typeof value === "function") { + // A sub flow has no textual form, so it cannot be interpolated into a string. + throw new RuntimeError( + "INLINE_REFERENCE_NOT_STRINGIFIABLE", + `Inline reference \${${signature}} resolves to a sub flow and cannot be interpolated into a string`, + ); + } + if (typeof value === "bigint" || typeof value === "number" || typeof value === "boolean") { + return String(value); + } + return JSON.stringify(value); +} diff --git a/ts/test/literal.test.ts b/ts/test/literal.test.ts new file mode 100644 index 0000000..c76d70e --- /dev/null +++ b/ts/test/literal.test.ts @@ -0,0 +1,111 @@ +import {describe, expect, it} from "vitest"; +import type {Value} from "@code0-tech/tucana/shared"; +import type {ActionLiteralValue, ActionNodeValue} from "@code0-tech/tucana/aquila"; +import {resolveLiteral, resolveNodeValue} from "../src/internal/literal"; +import type {Action} from "../src/action"; + +// Minimal Value builders so the tests read like the data they describe. +const str = (stringValue: string): Value => ({kind: {oneofKind: "stringValue", stringValue}}); +const int = (n: bigint): Value => ({kind: {oneofKind: "numberValue", numberValue: {number: {oneofKind: "integer", integer: n}}}}); +const bool = (boolValue: boolean): Value => ({kind: {oneofKind: "boolValue", boolValue}}); +const list = (...values: Value[]): Value => ({kind: {oneofKind: "listValue", listValue: {values}}}); +const struct = (fields: Record): Value => ({kind: {oneofKind: "structValue", structValue: {fields}}}); + +const literalRef = (signature: string, value: Value, references: ActionLiteralValue["references"] = []): ActionLiteralValue["references"][number] => ({ + signature, + value: {value: {oneofKind: "literalValue", literalValue: {value, references}}}, +}); + +const literal = (value: Value | undefined, references: ActionLiteralValue["references"] = []): ActionLiteralValue => ({value, references}); + +// resolveLiteral only touches the Action for sub flow references. +const noopAction = {} as Action; + +describe("resolveLiteral", () => { + it("returns a plain literal untouched when there are no references", () => { + expect(resolveLiteral(noopAction, literal(str("hello")))).toBe("hello"); + expect(resolveLiteral(noopAction, literal(int(42n)))).toBe(42); + expect(resolveLiteral(noopAction, literal(undefined))).toBeNull(); + }); + + it("adopts the referenced value verbatim when the string is a sole placeholder", () => { + const result = resolveLiteral(noopAction, literal(str("${count}"), [literalRef("count", int(7n))])); + expect(result).toBe(7); + }); + + it("preserves non-string reference types on full replacement", () => { + const result = resolveLiteral(noopAction, literal(str("${enabled}"), [literalRef("enabled", bool(true))])); + expect(result).toBe(true); + }); + + it("interpolates references into surrounding text", () => { + const result = resolveLiteral(noopAction, literal(str("Hello ${name}, you are ${age}"), [ + literalRef("name", str("Ada")), + literalRef("age", int(36n)), + ])); + expect(result).toBe("Hello Ada, you are 36"); + }); + + it("leaves unknown signatures untouched", () => { + expect(resolveLiteral(noopAction, literal(str("${missing}")))).toBe("${missing}"); + expect(resolveLiteral(noopAction, literal(str("a ${missing} b")))).toBe("a ${missing} b"); + }); + + it("resolves placeholders inside nested structs and lists", () => { + const value = struct({ + greeting: str("Hi ${name}"), + tags: list(str("${primary}"), str("static")), + }); + const result = resolveLiteral(noopAction, literal(value, [ + literalRef("name", str("Grace")), + literalRef("primary", int(1n)), + ])); + expect(result).toEqual({greeting: "Hi Grace", tags: [1, "static"]}); + }); + + it("stringifies structured references during interpolation", () => { + const result = resolveLiteral(noopAction, literal(str("payload=${obj}"), [ + literalRef("obj", struct({a: int(1n)})), + ])); + expect(result).toBe('payload={"a":1}'); + }); + + it("resolves references that themselves contain references", () => { + const nested = literalRef("outer", str("<${inner}>"), [literalRef("inner", str("deep"))]); + const result = resolveLiteral(noopAction, literal(str("${outer}"), [nested])); + expect(result).toBe(""); + }); + + it("throws when a sub flow reference is interpolated into a string", () => { + const action = {executeSubFlow: () => Promise.resolve(null)} as unknown as Action; + const subFlowRef: ActionLiteralValue["references"][number] = { + signature: "run", + value: {value: {oneofKind: "subFlow", subFlow: {executionIdentifier: "x"}}}, + }; + expect(() => resolveLiteral(action, literal(str("result: ${run}"), [subFlowRef]))).toThrow(/cannot be interpolated/); + }); +}); + +describe("resolveNodeValue", () => { + it("returns undefined for an absent node", () => { + expect(resolveNodeValue(noopAction, undefined)).toBeUndefined(); + }); + + it("wraps a sub flow node in a caller exposing its schema", async () => { + const calls: unknown[][] = []; + const action = { + executeSubFlow: (subFlow: unknown, ...args: unknown[]) => { + calls.push([subFlow, ...args]); + return Promise.resolve("done"); + }, + } as unknown as Action; + const node: ActionNodeValue = { + value: {oneofKind: "subFlow", subFlow: {executionIdentifier: "sf", inputSchema: {fields: {}}}}, + }; + const caller = resolveNodeValue(action, node) as ((...args: unknown[]) => Promise) & {inputSchema?: unknown}; + expect(typeof caller).toBe("function"); + expect(caller.inputSchema).toEqual({fields: {}}); + await expect(caller("a")).resolves.toBe("done"); + expect(calls).toEqual([[{executionIdentifier: "sf", inputSchema: {fields: {}}}, "a"]]); + }); +}); From b808b774c2137324d15ea6fe4fca1ac981d0f5d6 Mon Sep 17 00:00:00 2001 From: nicosammito Date: Tue, 18 Aug 2026 14:31:08 +0200 Subject: [PATCH 3/4] feat: add inline reference support in py --- .../for_each_consumers_runtime_function.py | 45 ++++++ py/examples/simple-example-py/index.py | 4 + py/hercules/actions/execution.py | 22 +-- py/hercules/internal/literal.py | 113 ++++++++++++++ py/tests/test_literal.py | 139 ++++++++++++++++++ 5 files changed, 304 insertions(+), 19 deletions(-) create mode 100644 py/examples/simple-example-py/functions/for_each_consumers_runtime_function.py create mode 100644 py/hercules/internal/literal.py create mode 100644 py/tests/test_literal.py diff --git a/py/examples/simple-example-py/functions/for_each_consumers_runtime_function.py b/py/examples/simple-example-py/functions/for_each_consumers_runtime_function.py new file mode 100644 index 0000000..aef1ece --- /dev/null +++ b/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("(list: LIST, consumers: LIST>): 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) diff --git a/py/examples/simple-example-py/index.py b/py/examples/simple-example-py/index.py index 847175f..f00b10a 100644 --- a/py/examples/simple-example-py/index.py +++ b/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) diff --git a/py/hercules/actions/execution.py b/py/hercules/actions/execution.py index f8cc656..2066a26 100644 --- a/py/hercules/actions/execution.py +++ b/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,24 +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) - - # 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 - - params.append(make_caller(sub_flow)) - else: - params.append(None) + params.append(resolve_node_value(action, field)) return params diff --git a/py/hercules/internal/literal.py b/py/hercules/internal/literal.py new file mode 100644 index 0000000..5537872 --- /dev/null +++ b/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=(",", ":")) diff --git a/py/tests/test_literal.py b/py/tests/test_literal.py new file mode 100644 index 0000000..0307c75 --- /dev/null +++ b/py/tests/test_literal.py @@ -0,0 +1,139 @@ +"""Port of ``ts/test/literal.test.ts``.""" +import asyncio + +import pytest + +from hercules.internal.literal import resolve_literal, resolve_node_value +from tucana.generated.aquila import action_pb2 +from tucana.generated.shared import struct_pb2 + + +# --- Minimal Value builders so the tests read like the data they describe. --- +def _str(string_value): + return struct_pb2.Value(string_value=string_value) + + +def _int(n): + return struct_pb2.Value(number_value=struct_pb2.NumberValue(integer=n)) + + +def _bool(b): + return struct_pb2.Value(bool_value=b) + + +def _list(*values): + return struct_pb2.Value(list_value=struct_pb2.ListValue(values=list(values))) + + +def _struct(fields): + return struct_pb2.Value(struct_value=struct_pb2.Struct(fields=fields)) + + +def _literal_ref(signature, value, references=()): + return action_pb2.ActionInlineReferenceValue( + signature=signature, + value=action_pb2.ActionNodeValue( + literal_value=action_pb2.ActionLiteralValue(value=value, references=list(references)) + ), + ) + + +def _literal(value=None, references=()): + kwargs = {"references": list(references)} + if value is not None: + kwargs["value"] = value + return action_pb2.ActionLiteralValue(**kwargs) + + +# resolve_literal only touches the action for sub flow references. +_NOOP_ACTION = object() + + +class TestResolveLiteral: + def test_returns_plain_literal_untouched_without_references(self): + assert resolve_literal(_NOOP_ACTION, _literal(_str("hello"))) == "hello" + assert resolve_literal(_NOOP_ACTION, _literal(_int(42))) == 42 + assert resolve_literal(_NOOP_ACTION, _literal(None)) is None + + def test_adopts_referenced_value_on_sole_placeholder(self): + result = resolve_literal(_NOOP_ACTION, _literal(_str("${count}"), [_literal_ref("count", _int(7))])) + assert result == 7 + + def test_preserves_non_string_reference_types_on_full_replacement(self): + result = resolve_literal(_NOOP_ACTION, _literal(_str("${enabled}"), [_literal_ref("enabled", _bool(True))])) + assert result is True + + def test_interpolates_references_into_text(self): + result = resolve_literal( + _NOOP_ACTION, + _literal( + _str("Hello ${name}, you are ${age}"), + [_literal_ref("name", _str("Ada")), _literal_ref("age", _int(36))], + ), + ) + assert result == "Hello Ada, you are 36" + + def test_leaves_unknown_signatures_untouched(self): + assert resolve_literal(_NOOP_ACTION, _literal(_str("${missing}"))) == "${missing}" + assert resolve_literal(_NOOP_ACTION, _literal(_str("a ${missing} b"))) == "a ${missing} b" + + def test_resolves_placeholders_in_nested_structs_and_lists(self): + value = _struct( + { + "greeting": _str("Hi ${name}"), + "tags": _list(_str("${primary}"), _str("static")), + } + ) + result = resolve_literal( + _NOOP_ACTION, + _literal(value, [_literal_ref("name", _str("Grace")), _literal_ref("primary", _int(1))]), + ) + assert result == {"greeting": "Hi Grace", "tags": [1, "static"]} + + def test_stringifies_structured_references_during_interpolation(self): + result = resolve_literal( + _NOOP_ACTION, + _literal(_str("payload=${obj}"), [_literal_ref("obj", _struct({"a": _int(1)}))]), + ) + assert result == 'payload={"a":1}' + + def test_resolves_references_that_contain_references(self): + nested = _literal_ref("outer", _str("<${inner}>"), [_literal_ref("inner", _str("deep"))]) + result = resolve_literal(_NOOP_ACTION, _literal(_str("${outer}"), [nested])) + assert result == "" + + def test_raises_when_sub_flow_reference_is_interpolated(self): + class _Action: + async def execute_sub_flow(self, *args): + return None + + sub_flow_ref = action_pb2.ActionInlineReferenceValue( + signature="run", + value=action_pb2.ActionNodeValue( + sub_flow=action_pb2.ActionNodeSubFlowValue(execution_identifier="x") + ), + ) + with pytest.raises(Exception, match="cannot be interpolated"): + resolve_literal(_Action(), _literal(_str("result: ${run}"), [sub_flow_ref])) + + +class TestResolveNodeValue: + def test_returns_none_for_absent_node(self): + assert resolve_node_value(_NOOP_ACTION, None) is None + + def test_wraps_sub_flow_node_in_caller_exposing_schema(self): + calls = [] + + class _Action: + async def execute_sub_flow(self, sub_flow, *args): + calls.append((sub_flow, *args)) + return "done" + + node = action_pb2.ActionNodeValue( + sub_flow=action_pb2.ActionNodeSubFlowValue(execution_identifier="sf") + ) + caller = resolve_node_value(_Action(), node) + assert callable(caller) + assert caller.input_schema == node.sub_flow.input_schema + assert asyncio.get_event_loop().run_until_complete(caller("a")) == "done" + assert calls == [(node.sub_flow, "a")] From d758a45570e3a0e921a02f527006d0b7adf7373c Mon Sep 17 00:00:00 2001 From: nicosammito Date: Tue, 18 Aug 2026 14:34:22 +0200 Subject: [PATCH 4/4] feat: update definitions version --- .github/workflows/build.yml | 2 +- .github/workflows/publish.yml | 2 +- py/README.md | 4 ++-- py/scripts/build_definitions.py | 2 +- ts/package.json | 2 +- 5 files changed, 6 insertions(+), 6 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index abe571f..e4c3210 100644 --- a/.github/workflows/build.yml +++ b/.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 diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 87983d6..43e25a8 100644 --- a/.github/workflows/publish.yml +++ b/.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 diff --git a/py/README.md b/py/README.md index 83757d2..7c8ffc7 100644 --- a/py/README.md +++ b/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: diff --git a/py/scripts/build_definitions.py b/py/scripts/build_definitions.py index 3636c4f..816ead1 100644 --- a/py/scripts/build_definitions.py +++ b/py/scripts/build_definitions.py @@ -9,7 +9,7 @@ Usage:: - python scripts/build_definitions.py --version def-0.0.34 + python scripts/build_definitions.py --version def-0.0.35 """ from __future__ import annotations diff --git a/ts/package.json b/ts/package.json index f809858..d91560d 100644 --- a/ts/package.json +++ b/ts/package.json @@ -28,7 +28,7 @@ "clean": "rm -rf dist", "typecheck": "tsc --noEmit", "test": "vitest run", - "build:definitions": "npx tsx scripts/build-definitions.ts -- --version def-0.0.34" + "build:definitions": "npx tsx scripts/build-definitions.ts -- --version def-0.0.35" }, "keywords": [], "author": "",