lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: resolve composite step inputs

Browse Source
This commit is contained in:
lda
2026-08-13 16:13:00 +07:00 Unverified
parent e661691e32
commit acda465d35
8 changed files with 657 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_core/runtime/input_bindings.py
+115
View File
@@ -0,0 +1,115 @@
from __future__ import annotations
from collections.abc import Mapping, Sequence
from typing import Any
from wf_core.conditions import safe_resolve_path
from wf_core.errors import WorkflowExecutionError
from wf_core.local_paths import LocalPathError, set_local_value
from wf_core.models.input_bindings import (
ArrayExpression,
InputExpression,
InputExpressionBinding,
InputPathBinding,
InputValueBinding,
LiteralExpression,
ObjectExpression,
PathExpression,
StepInputBinding,
)
from wf_core.models.json_values import JsonValue, validate_strict_json_value
def resolve_input_expression(
expression: InputExpression,
*,
state: Mapping[str, Any],
workflow_input: Mapping[str, Any],
context: Mapping[str, Any],
label: str,
location: str,
) -> JsonValue:
"""Resolve one composite expression while preserving its payload location."""
if isinstance(expression, LiteralExpression):
return expression.value
if isinstance(expression, PathExpression):
try:
value = safe_resolve_path(
str(expression.path),
state=state,
workflow_input=workflow_input,
context=context,
)
return validate_strict_json_value(value)
except (ValueError, WorkflowExecutionError) as exc:
raise WorkflowExecutionError(f"{label} {location}: {exc}") from exc
if isinstance(expression, ArrayExpression):
return [
resolve_input_expression(
item,
state=state,
workflow_input=workflow_input,
context=context,
label=label,
location=f"{location}[{index}]",
)
for index, item in enumerate(expression.items)
]
if isinstance(expression, ObjectExpression):
return {
field: resolve_input_expression(
value,
state=state,
workflow_input=workflow_input,
context=context,
label=label,
location=f"{location}.{field}",
)
for field, value in expression.fields.items()
}
raise WorkflowExecutionError(f"unsupported input expression for {label} {location}")
def resolve_step_input_bindings(
bindings: Sequence[StepInputBinding],
*,
state: Mapping[str, Any],
workflow_input: Mapping[str, Any],
context: Mapping[str, Any],
label: str,
) -> dict[str, Any]:
"""Build one node-local payload from simple or composite input bindings."""
payload: dict[str, Any] = {}
for binding in bindings:
location = str(binding.target)
if isinstance(binding, InputValueBinding):
value = binding.value
elif isinstance(binding, InputPathBinding):
try:
value = safe_resolve_path(
str(binding.path),
state=state,
workflow_input=workflow_input,
context=context,
)
value = validate_strict_json_value(value)
except (ValueError, WorkflowExecutionError) as exc:
raise WorkflowExecutionError(f"{label} {location}: {exc}") from exc
elif isinstance(binding, InputExpressionBinding):
value = resolve_input_expression(
binding.expression,
state=state,
workflow_input=workflow_input,
context=context,
label=label,
location=location,
)
else:
raise WorkflowExecutionError(f"unsupported input binding for {label}")
try:
set_local_value(payload, binding.target, value)
except LocalPathError as exc:
raise WorkflowExecutionError(f"{label} {location}: {exc}") from exc
return payload
src/wf_core/runtime/ops/interrupts.py
+9 -22
View File
@@ -3,10 +3,8 @@ from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from wf_core.conditions import safe_resolve_path
from wf_core.errors import WorkflowExecutionError
from wf_core.local_paths import LocalPathError, set_local_value
from wf_core.models.steps import InputPathBinding, InputValueBinding, InterruptNode
from wf_core.models.steps import InterruptNode
from wf_core.models.workflow import Workflow
from wf_core.run_state import (
FrameStatus,
@@ -15,6 +13,7 @@ from wf_core.run_state import (
RunState,
StepExecutionResult,
)
from wf_core.runtime.input_bindings import resolve_step_input_bindings
from wf_core.runtime.lineage import commit_patch_for_frame
from wf_core.runtime.ops.flow import advance_frame, append_step_result_trace
from wf_core.runtime.ops.index import WorkflowIndex
@@ -35,25 +34,13 @@ def build_interrupt_request(
public_node_id: str | None = None,
route: InterruptRoute | None = None,
) -> InterruptRequest:
payload: dict[str, Any] = {}
for binding in node.request:
if isinstance(binding, InputValueBinding):
value = binding.value
elif isinstance(binding, InputPathBinding):
value = safe_resolve_path(
str(binding.path),
state=state,
workflow_input=workflow_input,
context=context,
)
else:
raise WorkflowExecutionError(
f"unsupported request binding for interrupt {node.id!r}"
)
try:
set_local_value(payload, binding.target, value)
except LocalPathError as exc:
raise WorkflowExecutionError(str(exc)) from exc
payload = resolve_step_input_bindings(
node.request,
state=state,
workflow_input=workflow_input,
context=context,
label=f"interrupt {node.id!r} request",
)
validate_payload_against_schema(
node.request_schema,
payload,
src/wf_core/runtime/ops/nodes.py
+9 -22
View File
@@ -4,12 +4,10 @@ from collections.abc import Awaitable, Callable, Mapping
from dataclasses import dataclass
from typing import Any, cast
from wf_core.conditions import safe_resolve_path
from wf_core.errors import WorkflowExecutionError
from wf_core.local_paths import LocalPathError, set_local_value
from wf_core.models.results import NodeResult
from wf_core.models.schemas import NodeDef
from wf_core.models.steps import InputPathBinding, InputValueBinding, NodeUse
from wf_core.models.steps import NodeUse
from wf_core.models.workflow import Workflow
from wf_core.run_state import (
ExecutionFrame,
@@ -18,6 +16,7 @@ from wf_core.run_state import (
StepExecutionResult,
)
from wf_core.runtime.foreach_state import ForeachBarrierState, item_frame_owner
from wf_core.runtime.input_bindings import resolve_step_input_bindings
from wf_core.runtime.lineage import (
append_lineage_writes,
commit_patch_for_frame,
@@ -59,25 +58,13 @@ def _resolve_node_execution(
) -> tuple[dict[str, Any], RuntimeContext, dict[str, Any]]:
context_values = frame_context_values(frame)
state_view = state_view_for_frame(run, frame)
resolved_input: dict[str, Any] = {}
for binding in node.input:
if isinstance(binding, InputValueBinding):
value = binding.value
elif isinstance(binding, InputPathBinding):
value = safe_resolve_path(
str(binding.path),
state=state_view,
workflow_input=scope_input_for_frame(run, frame),
context=context_values,
)
else:
raise WorkflowExecutionError(
f"unsupported input binding for node {node.id!r}"
)
try:
set_local_value(resolved_input, binding.target, value)
except LocalPathError as exc:
raise WorkflowExecutionError(str(exc)) from exc
resolved_input = resolve_step_input_bindings(
node.input,
state=state_view,
workflow_input=scope_input_for_frame(run, frame),
context=context_values,
label=f"node {node.id!r} input",
)
validate_payload_against_schema(
node_def.input_schema, resolved_input, f"node input for {node.id}"
)
src/wf_core/runtime/subgraphs.py
+4 -39
View File
@@ -1,18 +1,11 @@
from __future__ import annotations
from collections.abc import Callable, Mapping, Sequence
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from typing import Any, Generic, TypeVar
from wf_core.conditions import safe_resolve_path
from wf_core.errors import WorkflowExecutionError
from wf_core.local_paths import LocalPathError, set_local_value
from wf_core.models.steps import (
InputBinding,
InputPathBinding,
InputValueBinding,
SubgraphNode,
)
from wf_core.models.steps import SubgraphNode
from wf_core.models.workflow import Workflow
from wf_core.models.workflow_refs import WorkflowRef
from wf_core.run_state import (
@@ -23,6 +16,7 @@ from wf_core.run_state import (
RuntimeScope,
StepExecutionResult,
)
from wf_core.runtime.input_bindings import resolve_step_input_bindings
from wf_core.runtime.lineage import commit_patch_for_frame
from wf_core.runtime.ops.frames import frame_context_values
from wf_core.runtime.ops.merges import ReducerDefinition
@@ -111,35 +105,6 @@ def resolve_prepared_subgraph(
return prepared
def resolve_input_bindings(
bindings: Sequence[InputBinding],
*,
state: Mapping[str, Any],
workflow_input: Mapping[str, Any],
context: Mapping[str, Any],
label: str,
) -> dict[str, Any]:
"""Build a local input payload from canonical value/path bindings."""
payload: dict[str, Any] = {}
for binding in bindings:
if isinstance(binding, InputValueBinding):
value = binding.value
elif isinstance(binding, InputPathBinding):
value = safe_resolve_path(
str(binding.path),
state=state,
workflow_input=workflow_input,
context=context,
)
else:
raise WorkflowExecutionError(f"unsupported input binding for {label}")
try:
set_local_value(payload, binding.target, value)
except LocalPathError as exc:
raise WorkflowExecutionError(str(exc)) from exc
return payload
def step_subgraph(
workflow: Workflow,
run: RunState,
@@ -171,7 +136,7 @@ def _start_subgraph(
) -> None:
prepared.workflow.validate_structure().raise_for_errors()
parent_scope = run.scopes[frame.scope_id]
child_input = resolve_input_bindings(
child_input = resolve_step_input_bindings(
step.input,
state=state_view_for_frame(run, frame),
workflow_input=parent_scope.workflow_input,