ABW <: WorkFlowType

Agent-Based Workflow: execution is driven by a readiness queue instead of a fixed topological order. Nodes without input ports seed the queue, and every executed node enqueues the children attached to its ready output ports.

AbstractDataNode

Supertype of workflow nodes. The package ships one concrete implementation, DataNode; generic functions dispatch on this abstract type.

AbstractNodeType

Supertype of node behaviour tags. A user-defined node type is a singleton subtype used as the first type parameter of DataNode; execution and validation are attached to it through multiple dispatch:

struct MyNode <: AbstractNodeType end
 
function MetidaFlows.execute_unsafe!(node::DataNode{MyNode})
    setdata!(node, :out, 42)
    return [:out]
end

AbstractPortType

Supertype of port arity tags: SinglePort and MultiPort.

DAW <: WorkFlowType

Data Analysis Workflow: a deterministic acyclic graph. Its scheduler rejects cyclic graphs, resets every node and executes the whole graph in topological order, each node exactly once per run.

DataNode(type::Type{T}, properties, spec, settings::SettingsT, data::DataT, state::StateT, input_buffer::Dict{Symbol, Dict{Int, BufferT}} ) where T <: AbstractNodeType where SettingsT where DataT  where StateT  where BufferT
DataNode(type::Type, id, status, position, spec; settings = Dict{Symbol, Any}(), data = Dict{Symbol, Any}(), state = NodeState(), input_buffer = Dict{Symbol, Dict{Int, Any}}())
DataNode(type::Type, spec; settings = Dict{Symbol, Any}(), data = Dict{Symbol, Any}(), state = NodeState(), input_buffer = Dict{Symbol, Dict{Int, Any}}())

Concrete workflow node.

The first type parameter is the user-defined behaviour tag (AbstractNodeType) on which execute_unsafe! and the validation hooks dispatch; the remaining parameters are the types of the containers, so a node can be backed by plain Dicts (the default) or by custom structs.

Fields:

• properties::NodeProperties - id, status, position.
• spec::NodeSpec - port and settings interface.
• settings - configuration values (Dict{Symbol, Any} by default).
• data - cached output values, keyed by output port label.
• state - execution state (NodeState by default).
• input_buffer - input_buffer[port label][connection id] = value.

The constructor creates an empty buffer entry for every input port declared in spec and rejects buffer keys that are not input ports.

Example

node = DataNode(MyNode, spec)
id   = add_node!(workflow, node)

ExecuteSettings(; execute_upstream = true, invalidate_downstream = true,
                  check_cyclic = true, check_input_buffer = true)
ExecuteSettings(execute_upstream, invalidate_downstream, check_cyclic, check_input_buffer)
ExecuteSettings(all_flags::Bool)

Execution flags consumed by execute!:

• execute_upstream - recursively execute parent nodes first;
• invalidate_downstream - invalidate child nodes after successful execution;
• check_cyclic - detect re-entrant execution of the same node (Ring detected);
• check_input_buffer - require data on every required input port.

The single-argument form sets all four flags to the same value.

LogMsg(id, timestamp, level, message)
LogMsg(level, message)

Log record kept in Workflow.log and in the node execution state. The two-argument form generates a random id and stamps the current time.

MultiPort

Port type that accepts any number of connections; getinputdata returns the whole buffer Dict(connection_id => value).

NodeConnection(output_id, output_port, input_id, input_port)

Directed edge from an output port of the parent node (output_id) to an input port of the child node (input_id).

Connections are stored under an integer identifier, and that identifier is also the key used inside the child input buffer - so several connections can feed one MultiPort without overwriting each other.

NodeProperties(id, status, position)

Mutable node identity and presentation data: the workflow-assigned id, the execution status (see getstatus) and the graph editor position. The zero-argument form creates (0, :idle, (0, 0)).

NodeSpec(name, input_ports, output_ports, settings)
NodeSpec(name, input_ports, output_ports)

Static description of a node interface: its ports and the settings keys it understands. The spec is the single source of truth - a port label that is not listed here is rejected by getdata, setdata!, getinputdata and by connection validation.

Fields:

• name::String - human-readable node name.
• input_ports::Vector{PortSpec}, output_ports::Vector{PortSpec} - port specifications.
• settings::Vector{Symbol} - settings keys advertised through the node schema.
• portmap::Dict{Tuple{Symbol,Symbol}, Int} - (direction, label) => port index, built by the constructor.

Example

spec = NodeSpec("Filter",
    [PortSpec("Input table", DataFrame, :table)],
    [PortSpec("Filtered table", DataFrame, :table)],
    [:column, :threshold])

NodeState()

Per-node execution state with dict-like field access:

• exec_n::Int - execution counter (reserved, not incremented yet);
• ready_ports::Vector{Symbol} - output ports produced by the last successful execution (used by push_buffer!);
• execution_id::UInt64 - id of the run that last touched the node;
• log::Vector{LogMsg} - per-run node log (cleared at the start of a run).

Reset in place with empty!(state).

PortSpec(name, datatype, label, ::T = SinglePort(); required::Bool = true) where T <: AbstractPortType

Specification of a node port.

Fields:

• name::String - human-readable name of the port.
• datatype::Type - Julia type of the port data (connection type checking).
• label::Symbol - unique label used for referencing the port in code.
• required::Bool - whether the port must have buffered data for execution.

The optional positional argument selects the port arity: SinglePort (default) or MultiPort.

Example

PortSpec("value", Int, :val)                          # required single port
PortSpec("items", Int, :vals, MultiPort())            # multi-connection port
PortSpec("hint",  Int, :in; required = false)         # optional port

SinglePort

Port type that accepts at most one connection (the default); getinputdata returns the single buffered value or nothing.

WorkFlowType

Supertype of workflow execution models. Concrete subtypes (DAW, ABW) parameterise Workflow and select the scheduler! implementation.

Workflow{T <: WorkFlowType}

Container holding the nodes, the connections and the connection indices of a workflow.

Fields:

• id, name - workflow identity.
• nodes::Dict{Int, DataNode} - nodes by identifier.
• connections::Dict{Int, NodeConnection} - connections by identifier.
• incoming, outgoing - node id => [connection ids] indices.
• n_iter, c_iter - monotonically increasing id counters; identifiers of deleted nodes and connections are never reused.
• run_id - identifier of the current scheduler run.
• log, audit_log - reserved for engine-level logging and audit events.

Workflow(id::Int; type::Symbol = :DAW)

Create a new workflow model with the specified identifier and type.

type selects the execution model and the concrete type parameter of the result: :DAW gives a Workflow{DAW}, :ABW gives a Workflow{ABW}. Any other value raises an error.

Base.empty!(ns::NodeState)

Empty node execution state.

add_connection!(model::Workflow, c::NodeConnection)
add_connection!(model::Workflow, id_out::Int, port_out::Symbol, id_in::Int, port_in::Symbol)

Add connection to workflow.

Performs:

• connection validation,
• connection registration,
• incoming/outgoing index updates.

If the source node already has status :clean, its output data is immediately propagated into the target node input buffer.

Returns

Assigned connection identifier (Int).

add_node!(model::Workflow, node::AbstractDataNode)

Add node to workflow.

Assigns a new unique node identifier and registers the node in workflow storage.

Returns

Assigned node identifier (Int).

check_connection_validity(model, c::NodeConnection)

Validate a connection before it is registered. Raises an error unless:

• both nodes exist in the workflow;
• both ports exist in the corresponding node specifications;
• the target input port is free, or declared as a MultiPort;
• the output port datatype is a subtype of the input port datatype.

A rejected connection does not consume a connection identifier.

connection_to_dict(conn::NodeConnection) -> Dict

Convert connection to dictionary representation.

delete_connection!(model::Workflow, id::Int)

Remove connection from workflow.

Performs:

• deletion of corresponding child input buffer entry,
• removal from incoming/outgoing indices,
• deletion from workflow connection storage.

Returns

• true if connection existed and was removed.
• false otherwise.

delete_node!(model::Workflow, id::Int)

Remove node from workflow.

Performs:

• deletion of all incoming and outgoing connections,
• cleanup of connection indices,
• removal of node from workflow storage.

Returns

• true if node existed and was removed.
• false otherwise.

execute!(model::Workflow, id::Int; settings::ExecuteSettings = ExecuteSettings())

Execute workflow node.

Main workflow execution entry point.

Execution Stages

1. Initialize per-run execution state and logs.
2. Optionally detect recursive cyclic execution.
3. Skip execution for nodes already marked :clean.
4. Mark node as :executing.
5. Optionally execute upstream dependencies recursively.
6. Validate node structure and execution readiness.
7. Validate node settings.
8. Execute node implementation via execute_unsafe!.
9. Validate execution result.
10. Store execution state (ready_ports).
11. Propagate outputs downstream through input buffers.
12. Optionally invalidate downstream nodes.
13. Mark node as :clean.

Returns

Vector of output port labels (Vector{Symbol}) produced during execution.

execute_unsafe!(node::AbstractDataNode)

Low-level node execution interface.

This function contains node-specific execution logic.

The default implementation throws an error and must be specialized for every executable node type.

execution_node_validation(node::AbstractDataNode)

Internal function. Validate node readiness before execution.

Checks that:

• every declared input port has a corresponding value in node.input_buffer,
• validate_node succeeds.

This validation is intended for runtime execution safety, ensuring that all required inputs are available before calling execute_unsafe!.

Returns

• true if node is ready for execution.
• false otherwise.

find_connections(model::Workflow, id::Int)

Return all connection IDs associated with a node (both incoming and outgoing).

get_children(model::Workflow, id::Int)

Get children. Returns Vector of Tuple (outputport, childid, input_port) for each child connection.

get_parents(model::Workflow, id::Int)

Return parents - id vector

getconnection(model::Workflow, id::Int)

Return connection by identifier id.

getdata(node::AbstractDataNode, l::Symbol)

Return output data stored under output port label.

getdata(model::Workflow, id::Int, l::Symbol)

Return output data stored under output port label.

getid(node::AbstractDataNode) -> Int

Return node unique identifier.

getinputdata(node::AbstractDataNode, l::Symbol)
getinputdata(node::AbstractDataNode, l::Symbol, con::Int)

Read value from node input buffer.

Does NOT consume or clear buffer.

getnode(model::Workflow, id::Int)

Return node by identifier id.

getportconnections(model::Workflow, id::Int, label::Symbol; direction = :both)

Return all connections attached to a specific port.

Direction:

• :input
• :output
• :both

getportnumber(node::AbstractDataNode, l::Symbol, direction::Symbol)

Return index of port by label and direction (:input or :output).

getportspec(node::AbstractDataNode, l::Symbol, direction::Symbol)

Return the PortSpec of the port labelled l (direction is :input or :output).

getporttype(node::AbstractDataNode, i::Int, direction::Symbol)

Return Julia datatype of port by index and direction.

getporttype(node, label, direction) -> Type

Return Julia datatype of port by label.

getposition(node::AbstractDataNode) -> Tuple{Int,Int}

Return node UI/graph position.

getstate(node::AbstractDataNode, s::Symbol)

Get value from node execution state.

getstatus(node::AbstractDataNode) -> Symbol

Return execution status of node.

Possible values:

• :idle
• :dirty
• :clean
• :executing
• :failed
• :invalid_node
• :invalid_settings
• :invalid_result

haveinputs(node::AbstractDataNode)

Returns true if node has at least one input port defined in its spec.

invalidate_buffer!(node::AbstractDataNode, l::Symbol, con::Int)

Delete input buffer entry for a specific port and connection (id).

invalidate_downstream!(model::Workflow, id::Int)

Recursively invalidate all downstream nodes.

Marks the specified node and all descendant nodes as :dirty using mark_dirty!.

The traversal follows all outgoing connections recursively.

ismultiport(ps::PortSpec{MultiPort})
ismultiport(ps::PortSpec{SinglePort})

Check whether a port specification is a multiport.

isnodeexist(model::Workflow, id::Int)

isportexist(node::AbstractDataNode, port::Symbol, direction::Symbol = :any)

Check whether a port exists in node specification.

Direction:

• :input
• :output
• :any

isportinspec(p::Symbol, spec::NodeSpec, direction::Symbol)

Check whether label p is declared in spec. direction is :input, :output or :both; unlike isportexist, an unknown direction simply yields false instead of raising.

isready(model::Workflow, id::Int)

Check whether node is ready for execution.

A node is considered ready when:

• all parent nodes connected through incoming edges have status :clean.

Notes

• Current node status itself is not checked.
• Input buffer completeness is validated separately via execution_node_validation.

makegraph(model::Workflow)

Build directed graph representation of workflow.

mark_dirty!(node::AbstractDataNode)

Invalidate node execution result.

Performs the following operations:

• sets node status to :dirty,
• clears ready_ports,
• clears cached output data stored in node.data.

This function intentionally does not clear:

• node settings,
• input buffer,
• execution logs,
• execution counters/state metadata.

node_properties_to_dict(np::NodeProperties) -> Dict

Convert node properties to JSON-serializable dictionary.

node_schema(node::AbstractDataNode) -> Dict

Default node schema.

node_schema_usermod!(d, node::AbstractDataNode) -> Dict

Modify node schema for user-defined node types.

Possible user modifications:

node_schema_usermod!(d, node::AbstractDataNode)
    d["section"]   = "Section 1"
    d["groupname"] = "Group 1"
    d["color"]     = "#8b5cf6"
end

node_to_dict(node::AbstractDataNode; specs::Bool = true) -> Dict

Convert node to JSON-serializable dictionary.

nodetypestr(node::DataNode{T}) where T

Return string representation of node type T.

portspec_to_dict(ps::PortSpec) -> Dict

Convert PortSpec to dictionary representation.

push_buffer!(model::Workflow, id::Int)
push_buffer!(model::Workflow, id::Int, port::Symbol)
push_buffer!(model::Workflow, id::Int, ready_ports::Vector{Symbol})

Propagate output data from a node to its downstream children.

reset!(node::AbstractDataNode)

Reset node.

Performs the following operations:

• sets node status to :idle,
• clears ready_ports,
• clears cached output data stored in node.data.
• node settings,
• input buffer,
• state.

reset!(model::Workflow)

Reset workflow execution state.

Marks every node in the workflow as :dirty using mark_dirty!, clears node execution state, and removes all cached output data.

reset_status!(model::Workflow)

Reset only node statuses.

Sets status of every node in the workflow to :dirty.

scheduler!(model::Workflow{ABW}; maxiter = 1000)

Execute workflow using queue-based agent/event scheduling.

This scheduler is intended for dynamic or agent-based workflows (ABW), where execution readiness is determined during runtime.

Arguments

• maxiter: maximum number of scheduler iterations before aborting execution.

scheduler!(model::Workflow{DAW})

Execute entire data analysis workflow (DAW) using topological ordering.

This scheduler is designed for deterministic acyclic data-analysis workflows.

Execution Steps

1. Build workflow graph.
2. Validate graph acyclicity.
3. Generate new workflow run_id.
4. Reset node execution states.
5. Execute nodes in topological order.

Notes

• Nodes are executed exactly once per scheduler run.
• Upstream execution and downstream invalidation are disabled because execution order is already guaranteed by topology.
• Cyclic workflows are rejected before execution starts.

setdata!(node::AbstractDataNode, l::Symbol, d)

Store d as the value of output port l and return true.

Raises an error when l is not an output port of the node specification. This is what a node implementation calls from execute_unsafe! before returning the list of ready ports.

setid!(node::AbstractDataNode, id::Int) -> DataNode

Set node identifier. Mutates node in-place.

setinputbuffer!(node::AbstractDataNode, label::Symbol, connection_id::Int, value)

Write value into node input buffer.

Used by workflow engine to propagate outputs between nodes.

setposition!(node::AbstractDataNode, p::Tuple{Int,Int}) -> DataNode

Set UI/graph position of node.

setreadyports!(node::AbstractDataNode, v)

Set ready output ports in node execution state.

Overwrite the ready_ports execution state with v, reusing the existing vector. Called by execute! with the value returned by execute_unsafe!.

setsettings!(model::Workflow, id::Int, settings::Dict{Symbol, <: Any})

Apply new node settings and invalidate dependent nodes.

Settings are applied using setsettings_unsafe!, after which the target node and all downstream nodes are invalidated via invalidate_downstream!.

Notes

This function is the safe high-level entry point for mutating node configuration inside a workflow.

setsettings_unsafe!(node::AbstractDataNode, settings::Dict{Symbol, <: Any})

Direct mutation of node settings without invalidation. Can be re-implemented for every node type.

Default implementation copies all provided key-value pairs into node.settings.

Warning

This function does NOT invalidate cached execution results or downstream nodes.

Use setsettings! for normal workflow operation

setstate!(node::AbstractDataNode, s::Symbol, v) -> DataNode

Store value in node execution state.

setstatus!(node::AbstractDataNode, s::Symbol) -> DataNode

Set execution status of node.

This does NOT trigger validation or propagation. Pure mutation.

settings_schema(node::AbstractDataNode) -> Dict

Default settings schema.

settings_schema_usermod!(d, node::AbstractDataNode) -> Dict

Modify settings schema for user-defined node types.

Possible user modifications:

settings_schema_usermod!(d, node::DataNode{MyNodeType})
    settings_dict = Dict{String, Any}()
    settings_dict["my_setting1"] = Dict("type" => Int, 
        "default" => 0, 
        "description" => "My setting 1", 
        "required" => true, 
        "pinned" => true,
        "source"	=> "upstream",
        "validator" => (x -> x >= 0))
        
    settings_dict["my_setting2"] = Dict("type" => Array{Int}, 
        "default" => [0], 
        "description" => "My setting 2", 
        "required" => true, 
        "pinned" => false,
        "source"	=> "none",
        "validator" => (x -> x in [1,2,3]))
    d["schema"] = settings_dict
end

spec_to_dict(spec::NodeSpec) -> Dict

Convert NodeSpec to dictionary representation.

validate_node(node::AbstractDataNode)
validate_node(model::Workflow, node_id::Int)

Validate node structure and configuration.

Default implementation always returns true.

This function is intended for specialization by concrete node implementations.

Typical validation rules may include:

• internal consistency checks,
• structural constraints,
• node-specific invariants.

Returns

• true if node structure is valid.
• false otherwise.

validate_result(node::AbstractDataNode)
validate_result(model::Workflow, node_id::Int)

Validate node execution result.

Called after node execution completes.

Default implementation always returns true.

This function is intended for specialization by concrete node implementations.

Typical validation rules may include:

• output datatype verification,
• required output ports presence,
• shape or schema validation,
• domain-specific consistency checks.

Returns

• true if execution result is valid.
• false otherwise.

validate_settings(node::AbstractDataNode)
validate_settings(model::Workflow, node_id::Int)

Validate node settings before execution.

Default implementation always returns true.

This function is intended for specialization by concrete node implementations.

Typical validation rules may include:

• required setting presence,
• range checks,
• semantic validation of configuration values.

Returns

• true if settings are valid.
• false otherwise.

workflow_to_dict(w::Workflow) -> Dict

Convert workflow to dictionary representation.