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.

AbstractNodeFields

Supertype of the node field containers: NodeState and the reserved AbstractNodeSettings, AbstractNodeData, AbstractNodeInputBuffer.

Subtypes get a Dict-like interface over their fields, so a node can be backed either by plain Dicts (the default) or by purpose-built structs:

state = NodeState()
state[:exec_n]         # getindex  -> getfield
state[:exec_n] = 1     # setindex! -> setfield! (mutable subtypes only)
keys(state)            # (:exec_n, :ready_ports, :execution_id, :log)

setindex! raises an error for immutable subtypes.

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::UInt64, timestamp::DateTime, level::Symbol, message::String)
LogMsg(level::Symbol, message::String)

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. Records are currently produced in one place only: execute! appends an :error record to workflow.log when execute_unsafe! raises.

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 - number of times execute_unsafe! was entered since the last full reset. Incremented by execute! right before the node body runs, so a node rejected by validation leaves it unchanged while a failed node does not.

Zeroed by reset!(model) in every reset mode, by reset!(node) and by empty!(state) — but not by mark_dirty!, so invalidating a subtree does not lose the count. In a cyclic ABW run the counter therefore holds the number of loop turns, and a node may read its own node.state[:exec_n] inside execute_unsafe! as the current turn number;

• 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,
         kind::Symbol = :normal) 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.
• kind::Symbol - how the schedulers treat connections attached to the port.

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

Port kinds

Constraints are checked at construction time. kind = :feedback on an output port is rejected by the NodeSpec constructor: a delay is a property of the consumer, not of the producer.

Only isready and execution_node_validation read the kind. Buffering, invalidation and serialization are identical for all kinds.

Example

PortSpec("value", Int, :val)                                  # required single port
PortSpec("items", Int, :vals, MultiPort())                    # multi-connection port
PortSpec("hint",  Int, :in; required = false)                 # optional port
PortSpec("state", Int, :prev; kind = :feedback)                # closes a cycle
PortSpec("report", DataFrame, :out; kind = :terminal)          # result slot

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. Identifiers come from a monotonically increasing counter and are never reused.

Notes

A node whose status is neither :idle nor :clean is reset on insertion, which clears its settings, input buffers and execution state. A freshly created node and an already computed node are inserted unchanged, so a node may be configured before or after it joins the workflow.

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(),
         throw_error::Bool = false)

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. Increment exec_n and 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.

Arguments

• settings: per-call execution flags, see ExecuteSettings.
• throw_error: rethrow an exception raised by execute_unsafe! after the node status and the log record have been written. With the default false the error is swallowed and reported through the status and model.log.

Returns

Vector of output port labels (Vector{Symbol}) produced during execution, or an empty vector when the node was skipped, rejected by validation or failed.

The vector is not copied: for a :clean node it is the stored ready_ports state, and otherwise it is whatever execute_unsafe! returned. Treat the result as read-only.

Status after the call

:clean on success, otherwise :invalid_node, :invalid_settings, :invalid_result or :failed.

Example

execute!(w, id)                                                    # parents pulled in
execute!(w, id; settings = ExecuteSettings(; execute_upstream = false))
execute!(w, id; throw_error = true)

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.

Read inputs with getinputdata, write results with setdata! and return the vector of output port labels that were produced. That vector is the contract with the engine: it is stored as the ready_ports state and only connections leaving these ports are refreshed by push_buffer!. Returning a subset of the output ports is allowed.

Example

struct Doubler <: AbstractNodeType end
 
function MetidaFlows.execute_unsafe!(node::DataNode{Doubler})
    x = getinputdata(node, :in)
    setdata!(node, :out, 2x)
    return [:out]
end

Notes

Errors raised here are caught by execute!, which marks the node :failed and appends an :error record to workflow.log.

execution_node_validation(node::AbstractDataNode, check_input_buffer::Bool = true)

Internal function. Validate node readiness before execution.

With check_input_buffer enabled, checks that:

• every input port that is required and of kind :normal has a value in node.input_buffer,
• validate_node succeeds.

Optional ports and ports of kind :feedback or :error may be empty: a feedback buffer is empty on the first pass of a loop by definition.

With check_input_buffer disabled only validate_node is consulted. The ABW scheduler runs nodes that way.

Returns

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

exportmeta(model::Workflow, id::Int, port::Symbol)

Describe what node id will produce on its output port port, without executing anything.

Walks up the graph: the node's producers are asked first, their answers are handed to exportmeta_unsafe as inmeta. The walk stops at nodes without incoming connections, and at any node for which ismetasource returns true.

Returns nothing when the answer is unknown — because the node did not implement exportmeta_unsafe, because a producer up the chain did not, or because the required settings are not filled in yet.

Arguments

• maxdepth: longest chain of nodes the walk may descend into before raising. Guards against a pathological graph exhausting the stack; a node declaring ismetasource ends the walk without consuming depth.

Notes

• Within one call each (node, output port) pair is described once: a diamond does not compute its common ancestor twice. The memo lives for the duration of the call only, so nothing can go stale between calls — a file may change on disk, and the next call sees it.
• Nothing is cached between calls. Memoize inside the node if a source is expensive to inspect.
• A node declaring ismetasource cuts the walk short, so the branches above it are never visited.
• In a graph containing a cycle built from :normal edges the answer for a node inside the cycle depends on where the walk started. Such a graph is rejected by the DAW scheduler and never produced by ABW, where cycles are closed by :feedback edges that this walk does not follow.
• :feedback and :error edges are not followed, so cyclic ABW workflows are safe. A cycle built from :normal edges only is detected as a back edge and yields nothing for the repeated node instead of recursing forever.

Example

portmeta(w, csv_id, :table)      # (columns = [:Subject, :Formulation, :Time, :Concentration],)

exportmeta_unsafe(node::AbstractDataNode, port::Symbol, inmeta)

Describe what the node will produce on output port port, without executing it.

This is the extension point of metadata propagation, the configuration-time counterpart of execute_unsafe!. Like it, the method sees only the node: inmeta carries the descriptions of the node's own input ports, keyed by the node's own port labels, in exactly the shape getinputdata uses:

• SinglePort: the description, or nothing when the port is unconnected or the producer does not know;
• MultiPort: a Dict{connection id, description}.

The default implementation returns nothing, meaning "unknown". The shape of a description is a contract between nodes sharing a port datatype, not something the engine defines.

Rules

• Never pass a description through unchanged unless the node really preserves it. A node that renames or drops columns must describe its own result; returning the input description would propagate a lie downstream.
• This is not validation. nothing means "cannot tell", not "misconfigured" — checking that a configured column actually exists belongs in validate_settings and in the node body.
• Reading a little is fine, computing is not: parse a header, not a file.

Example

# источник: читает только заголовок
function MetidaFlows.exportmeta_unsafe(node::DataNode{ReadCSV}, ::Symbol, inmeta)
    path = get(node.settings, :file, nothing)
    (path === nothing || !isfile(path)) && return nothing
    return (columns = propertynames(CSV.File(path; limit = 0)),)
end

# прозрачная нода: строки фильтруются, схема сохраняется
MetidaFlows.exportmeta_unsafe(::DataNode{FilterRows}, ::Symbol, inmeta) = inmeta[:table]

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 the output data stored under output port label l.

Returns nothing when the port is declared but holds no value yet - for instance after mark_dirty!. Raises an error when l is not an output port of the node specification.

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

Return the output data stored under output port label l of node id.

Raises a KeyError for an unknown node id; see the two-argument method for the port lookup rules.

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.

The two-argument form depends on the port arity:

• SinglePort: returns the buffered value, or nothing when the buffer is empty; more than one buffered value is an error.
• MultiPort: returns the whole Dict{connection id, value} buffer.

The three-argument form returns the value written by connection con, or nothing when that connection has nothing buffered.

Both forms raise an error when l is not an input port of the node specification, and neither consumes or clears the buffer.

getinputmeta(model::Workflow, id::Int) -> Dict{Symbol, Any}

Collect descriptions of every input port of node id, asking its producers recursively.

Keys are the input port labels of node id — taken from the connections, so a producer never needs to know which port of which consumer it feeds. Every :normal input port is present in the result: an unconnected SinglePort maps to nothing, an unconnected MultiPort to an empty Dict.

Ports of kind :feedback and :error are skipped: a feedback edge closes a cycle, and following it would not terminate.

done and maxdepth are threaded through to exportmeta: the first keeps a common ancestor from being described twice within one call, the second bounds the depth of the walk.

Example

collect_input_meta(w, mean_id)
# Dict{Symbol, Any}(:input_data => (columns = [:Subject, :Time, :Concentration],))

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.

Notes

A :feedback port is still an input port, so a node whose only inputs are feedback ports also returns true. The ABW scheduler seeds its queue with nodes for which this function returns false, which is why every cyclic workflow needs at least one node with no input ports at all - see scheduler!.

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.

ismetasource(node::AbstractDataNode, port::Symbol) -> Bool

Declare that the node can describe port on its own, without asking its producers. When it returns true, exportmeta stops the upward walk at this node and calls exportmeta_unsafe with an empty inmeta.

The default is false, so a node describes its output from its inputs unless it says otherwise.

The predicate may depend on the node state, which is the point: a node is often a source of metadata only once it has been configured.

Consistency

ismetasource and exportmeta_unsafe are written together and must agree. Reaching for inmeta[:label] in a branch declared as a metadata source raises a KeyError, because the dictionary passed there is empty — the mismatch is loud on purpose, as with validate_settings and execute_unsafe!.

Example

# схему знаю, только если задан путь к файлу
MetidaFlows.ismetasource(node::DataNode{ReadCSV}, ::Symbol) = haskey(node.settings, :file)

# нода с необязательным входом «схема из файла»
MetidaFlows.ismetasource(node::DataNode{Import}, ::Symbol) =
    !isempty(get(node.settings, :schema_file, ""))

function MetidaFlows.exportmeta_unsafe(node::DataNode{Import}, ::Symbol, inmeta)
    haskey(node.settings, :schema_file) && return read_schema(node.settings[:schema_file])
    return inmeta[:table]                  # ветка, где нода НЕ источник описания
end

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 in ABW.

A node is ready when every producer connected to it through a :normal input port has status :clean. Connections entering ports of any other kind are not waited for.

This asymmetry is what makes cycles possible: a :feedback edge carries the value of the previous iteration, so requiring its producer to be :clean would deadlock - the node would wait for a producer that waits for the node.

Notes

• Current node status itself is not checked.
• Input buffer completeness is validated separately via execution_node_validation.
• A node with :normal input ports but no incoming connection is ready immediately; it just never gets enqueued unless something feeds it.

makegraph(model::Workflow)

Build directed graph representation of workflow: node identifiers become vertex names and every connection becomes an edge.

Notes

All connections are included regardless of the kind of the ports they attach to. A :feedback edge is therefore an ordinary edge here, so a graph with a feedback loop is cyclic and scheduler! for a DAW workflow rejects it. That is by design: cycles are an ABW feature.

mark_dirty!(node::AbstractDataNode; soft::Bool = false)

Invalidate node execution result.

Performs the following operations:

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

With soft = true the last two steps are skipped: the status and the counter are reset, but ready_ports and node.data are left in place. That is what reset_mode = :soft of scheduler! uses to keep results between runs.

This function intentionally does not clear:

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

Returns

The node.

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, settings::Bool = true) -> Dict

Convert node to JSON-serializable dictionary.

Keys: "id", "properties", "status", plus "spec" when specs is set and "settings" when settings is set.

Notes

The "settings" entry holds the settings schema produced by settings_schema, not the current setting values.

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

Return string representation of node type T.

portspec_to_dict(ps::PortSpec)

Convert port specification to JSON-serializable dictionary.

Keys: "name", "label", "datatype", "required", "kind", "type".

"label", "datatype" and "kind" are stringified; "type" holds the port arity as "SinglePort" or "MultiPort".

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 to its initial state.

Performs the following operations:

• sets node status to :idle,
• clears node settings,
• clears the contents of every input buffer, keeping the port keys,
• resets execution state (exec_n, ready_ports, execution_id, node log),
• clears cached output data stored in node.data.

Unlike mark_dirty!, this drops the node configuration as well, so it is meant for reusing a node object rather than for invalidation.

Returns

The node.

reset!(model::Workflow; soft::Bool = false)

Invalidate the whole workflow.

Zeroes the execution counter exec_n of every node, then applies mark_dirty! to each of them, forwarding soft:

• soft = false (default): statuses become :dirty, cached output data and ready_ports are dropped;
• soft = true: only statuses are reset, cached output data survives.

This is the only place where exec_n is zeroed for the whole workflow, so the counter measures work done since the last reset! rather than since the last invalidation.

Node settings, input buffers and execution logs are kept in both cases - use reset!(node) for a full per-node reset.

Returns

The workflow.

reset_status!(model::Workflow)

Reset only node statuses.

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

scheduler!(model::Workflow{ABW}; reset_mode::Symbol = :full, maxiter = 1000,
           throw_error::Bool = false, throw_warn::Bool = true)

Execute workflow using queue-based agent/event scheduling.

Intended for dynamic, agent-based and iterative workflows. Unlike DAW, this scheduler accepts cyclic graphs, and a node may execute several times during one call.

Execution Steps

1. Generate a new workflow run_id and apply reset_mode.
2. Seed the queue with every node that has no input ports.
3. Pop a node and execute it when isready reports that every producer connected through a :normal port is :clean; otherwise drop it - a producer finishing later re-queues it.
4. For every port the execution produced, mark each child :dirty and enqueue it.

Arguments

• reset_mode: :full (default) invalidates every node via reset! and drops cached output data; :soft only marks nodes :dirty and keeps the cache, which is how state is carried between runs; :none leaves statuses untouched, so :clean nodes are not recomputed. Any other value raises an error.
• maxiter: maximum number of scheduler iterations before aborting execution. For a cyclic workflow this is the backstop against a loop that never converges.
• throw_error: forwarded to execute!.
• throw_warn: warn when the seed queue is empty. Nothing is executed in that case and the nodes stay :dirty, which is otherwise silent.

Cycles

A cycle must be closed by a connection entering a port declared with kind = :feedback; see PortSpec. Two properties make the loop run:

• isready does not wait for the producer of a feedback edge, so the cycle has a defined entry point;
• step 4 marks children :dirty unconditionally, so a node that already ran during this call runs again instead of short-circuiting on :clean inside execute!.

The loop ends when a node publishes nothing that has a consumer - by returning an empty vector of ready ports, or by publishing only unconnected or :terminal ports. The queue then drains.

Every cyclic workflow needs at least one node with no input ports: a :feedback port is still an input port, so a loop on its own is never seeded and never starts. See haveinputs.

Notes

• Nodes are executed with ExecuteSettings(false), so required input ports are not enforced here, unlike the DAW path.
• A node with input ports but without incoming connections never becomes ready and stays :dirty.

Returns

true when the queue is drained.

Example

scheduler!(w)                             # full reset, one pass or one loop
scheduler!(w; reset_mode = :soft)         # keep cached results between runs
scheduler!(w; maxiter = 10_000)           # long-running iteration
scheduler!(w; throw_warn = false)         # a deliberately unseeded graph

scheduler!(model::Workflow{DAW}; reset_mode::Symbol = :full,
           throw_error::Bool = false)

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. Apply reset_mode.
5. Execute nodes in topological order.

Arguments

• reset_mode: :full (default) invalidates every node via reset! and drops cached output data; :soft only marks nodes :dirty and keeps the cache; :none leaves statuses untouched, so nodes that are already :clean are not recomputed. Any other value raises an error.
• throw_error: forwarded to execute!; aborts the run on the first node that raises instead of recording the failure.

Notes

• Nodes are executed at most once per scheduler run.
• Upstream execution and downstream invalidation are disabled because execution order is already guaranteed by topology.
• Required input ports are enforced (check_input_buffer stays on).
• Cyclic workflows are rejected before execution starts, including cycles closed by a :feedback port - see makegraph. In a DAW workflow the only effect of kind = :feedback is that the port is exempt from the required-buffer check; there is no delay and no second pass.

Returns

true when the traversal completed. This is not a success flag: individual nodes may still end up :failed or :invalid_*, so check getstatus when it matters.

Example

scheduler!(workflow)
result = getdata(workflow, output_id, :result)

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.
• Settings are merged: only the keys present in settings are overwritten, the rest are kept.

Returns

The node.

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, ready to be serialized to JSON.

Keys: "id", "name", "n_iter", "c_iter", "nodes", "connections", "incoming", "outgoing".

Notes

• "nodes" and "connections" are keyed by stringified identifiers, while "incoming" and "outgoing" keep integer node ids.
• Execution state - run_id, logs, cached output data and settings values - is not included, so the result describes structure only.