AI Workflow engine implementation with WAL persistence, Tracing, Tools, Monitoring, and Budgeting. More...

#include "csilk/app/workflow.h"
#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <uv.h>
#include "cJSON.h"
#include "csilk/app/workflow_wal.h"
#include "csilk/core/internal.h"
#include <sys/stat.h>
#include <unistd.h>

Include dependency graph for workflow.c:

Data Structures
struct	csilk_ai_meta_t
	AI metadata attached to workflow node outputs for token tracking and budget enforcement. More...

struct	csilk_wf_tool_entry_t
	A registered workflow tool (function-calling capability exposed to AI nodes). Tools are invoked in parallel via the libuv thread pool during AI node execution. More...

struct	csilk_wf_edge_t

struct	csilk_wf_node_t
	A single node in a workflow DAG. Each node wraps a handler function with optional edges, error handling, dynamic routing, and timeout support. More...

struct	csilk_wf_t
	Workflow definition: a DAG of processing nodes connected by conditional or unconditional edges. Each node runs a handler function on the libuv thread pool. Supports persistence via WAL, real-time monitoring via WebSocket, tool registration for AI nodes, and budget (token/TTL) limits. More...

struct	csilk_wf_ctx_t
	Workflow execution context — per-run state tracking all node progress, scheduling, memory, and budget enforcement. Created by csilk_wf_run_ext_internal() and freed by cleanup_ctx(). More...

struct	node_work_t
	Per-node-execution state passed through libuv work requests. Allocated in execute_node(), freed in after_worker_cb(). More...

struct	sub_tool_work_t
	Per-tool-call context for parallel tool execution within an AI node. Each tool call runs on its own libuv thread-pool worker. More...

struct	stream_ctx_t

Macros
#define	MAX_WORKFLOW_STEPS 1000

Functions
static void	execute_node (csilk_wf_ctx_t ctx, csilk_wf_node_t node, csilk_data_t *input)
	Internal: enqueue a workflow node for execution on the libuv thread pool.

static void	cleanup_ctx (csilk_wf_ctx_t *ctx)
	Internal: free a workflow execution context and all resources.

static void	after_worker_cb (uv_work_t *req, int status)

static void	register_active_ctx (csilk_wf_t wf, csilk_wf_ctx_t ctx)

static void	unregister_active_ctx (csilk_wf_t wf, csilk_wf_ctx_t ctx)

static csilk_wf_ctx_t *	find_active_ctx (csilk_wf_t wf, const char exec_id)

static const char *	csilk_wf_run_ext_internal (csilk_wf_t wf, csilk_data_t input, void(callback)(csilk_data_t ), void(trace_cb)(csilk_data_t , csilk_wf_trace_t *))

static void	serve_ui_handler (csilk_ctx_t *c)

void	csilk_wf_serve_ui (csilk_app_t app, const char path)
	Register a default route to serve the workflow dashboard.

csilk_wf_t *	csilk_wf_new (const char *name)
	Create a new AI workflow instance.

static void	ai_config_free (void *ptr)

static void	node_free (csilk_wf_node_t *node)

void	csilk_wf_free (csilk_wf_t *wf)
	Deallocate a workflow and all its nodes.

csilk_wf_node_t *	csilk_wf_add (csilk_wf_t wf, const char id, csilk_wf_handler_t handler, void *user_data)
	Add a node to the workflow.

void	csilk_wf_node_set_entry (csilk_wf_node_t *node, int is_entry)
	Mark a node as an entry point for the workflow.

static void	node_add_edge (csilk_wf_node_t from, const char condition, csilk_wf_node_t *to, int is_loop)

void	csilk_wf_bind (csilk_wf_node_t from, csilk_wf_node_t to)
	Bind two nodes sequentially (default routing).

void	csilk_wf_on (csilk_wf_node_t from, const char condition, csilk_wf_node_t *to)
	Add a conditional route between nodes.

void	csilk_wf_on_loop (csilk_wf_node_t from, const char condition, csilk_wf_node_t *to)
	Add a loop-back / feedback route between nodes.

void	csilk_wf_on_error (csilk_wf_node_t from, csilk_wf_node_t target)
	Add an error fallback route.

void	csilk_wf_route (csilk_wf_node_t *node, csilk_wf_router_t router)
	Set a dynamic router for a node.

void	csilk_wf_node_set_join (csilk_wf_node_t *node, csilk_wf_join_policy_t policy)
	Set the join policy for a node.

void	csilk_wf_set_persistence (csilk_wf_t wf, const char wal_dir)
	Enable WAL persistence for a workflow definition.

void	csilk_wf_register_monitor (csilk_wf_t wf, csilk_ctx_t c)
	Register a WebSocket connection to receive live workflow updates.

static void	_wf_broadcast (csilk_wf_t wf, const char event, const char node_id, const char payload)

void	csilk_wf_set_budget (csilk_wf_t *wf, int max_tokens)
	Set a maximum token budget for the workflow.

void	csilk_wf_node_set_timeout (csilk_wf_node_t *node, int timeout_ms)
	Set a timeout for a specific node.

void	csilk_wf_set_ttl (csilk_wf_t *wf, int ttl_sec)
	Set a global TTL for the workflow execution.

void	csilk_wf_node_set_interactive (csilk_wf_node_t *node, int is_interactive)
	Mark a node as interactive (requires human signal to proceed).

void	csilk_wf_node_set_schema (csilk_wf_node_t node, const char schema)
	Set an expected JSON Schema for a node's output.

void	csilk_wf_node_set_retry (csilk_wf_node_t *node, int max_retries, int retry_delay_ms)
	Set automatic retry policy for a specific node.

static csilk_data_t *	remote_pass_handler (csilk_wf_ctx_t ctx, csilk_data_t input, void *user_data)

void	csilk_wf_node_set_remote (csilk_wf_node_t *node, int is_remote)
	Mark a node for remote execution via MQ.

static void	on_remote_result (csilk_mq_ctx_t *m_ctx)

void	csilk_wf_enable_distributed (csilk_wf_t wf, csilk_mq_t mq)
	Enable distributed execution by bridging workflow with an MQ.

void *	csilk_wf_alloc (csilk_wf_ctx_t *ctx, size_t size)
	Allocate memory from the workflow arena.

char *	csilk_wf_strdup (csilk_wf_ctx_t ctx, const char s)
	Duplicate a string using the workflow arena.

csilk_data_t *	csilk_wf_data_new (csilk_wf_ctx_t ctx, const char type, void *value)
	Allocate a new data container managed by the workflow arena.

static char *	_csilk_json_get_path (csilk_wf_ctx_t ctx, cJSON root, const char *path)
	Internal: traverse a cJSON tree following a dot-separated path.

static char *	apply_filter (csilk_wf_ctx_t ctx, const char filter, char *val)
	Internal: resolve template expressions in a prompt string.

static char *	resolve_templates (csilk_wf_ctx_t ctx, const char template)

static void	sub_worker_cb (uv_work_t *req)
	libuv thread-pool work callback for tool execution. Looks up the tool by name in the workflow's tool registry and calls its function with the arguments from the AI tool call.

static void	after_sub_worker_cb (uv_work_t *req, int status)
	libuv after-work callback for tool execution. Decrements the shared pending counter and signals the condition variable to wake the main AI node handler thread.

static void	on_ai_stream (const char chunk, void user_data)

static csilk_data_t *	ai_node_handler (csilk_wf_ctx_t ctx, csilk_data_t input, void *user_data)
	Built-in handler for AI workflow nodes.

csilk_wf_node_t *	csilk_wf_add_ai (csilk_wf_t wf, const char id, const csilk_ai_config_t *config)
	Add a built-in AI node with template support.

void	csilk_wf_register_tool (csilk_wf_t wf, const char name, const char description, const char parameters_json, csilk_wf_tool_fn fn, void *user_data)
	Register a tool that AI nodes can call.

csilk_wf_node_t *	csilk_wf_get_node (csilk_wf_t wf, const char id)
	Look up a workflow node by its string ID.

char *	csilk_wf_to_mermaid (csilk_wf_t *wf)
	Export the workflow graph as a Mermaid string.

char *	csilk_wf_trace_to_json (const csilk_wf_trace_t *trace)
	Convert a trace object to a JSON string.

void	csilk_wf_trace_free (csilk_wf_trace_t *trace)
	Free a trace object.

static void	cleanup_ctx_now (csilk_wf_ctx_t *ctx)

static void	on_ttl_timer_close (uv_handle_t *handle)

static void	wal_log_event (csilk_wf_ctx_t ctx, csilk_wf_event_type_t type, const char node_id, csilk_data_t *data)
	Internal: persist a workflow event to the Write-Ahead Log.

static void	worker_cb (uv_work_t *req)
	libuv thread-pool work callback — executes a workflow node's handler on a background thread.

static void	on_retry_timer (uv_timer_t *handle)
	libuv after-work callback — processes node completion on the main loop thread.

static void	on_work_timer_close (uv_handle_t *handle)

static void	free_work (node_work_t *work)

static void	on_node_timeout (uv_timer_t *handle)
	libuv timer callback — marks a node as timed out. Sets the is_timed_out flag on the node_work_t, which causes after_worker_cb to treat the output as NULL even if the handler eventually completes.

const char *	csilk_wf_run (csilk_wf_t wf, csilk_data_t input, void(callback)(csilk_data_t result))
	Run the workflow asynchronously.

void	csilk_wf_run_traced (csilk_wf_t wf, csilk_data_t input, void(callback)(csilk_data_t result, csilk_wf_trace_t *trace))
	Run workflow and generate a trace.

static void	on_workflow_ttl (uv_timer_t *handle)

void	csilk_wf_resume (csilk_wf_t wf, const char exec_id, void(callback)(csilk_data_t result))
	Resume an interrupted workflow execution from a WAL file.

void	csilk_wf_signal_continue (csilk_wf_t wf, const char exec_id, csilk_data_t input, void(callback)(csilk_data_t *result))
	Signal a paused workflow to continue.

Variables
static csilk_wf_t *	g_distributed_wfs [32]

static size_t	g_distributed_wf_count = 0

Detailed Description

AI Workflow engine implementation with WAL persistence, Tracing, Tools, Monitoring, and Budgeting.

Data Structure Documentation

◆ csilk_ai_meta_t

struct csilk_ai_meta_t

AI metadata attached to workflow node outputs for token tracking and budget enforcement.

Data Fields
int	completion_tokens	Tokens consumed by the completion.
char *	model	Model name (e.g., "gpt-3.5-turbo").
int	prompt_tokens	Tokens consumed by the prompt.

◆ csilk_wf_tool_entry_t

struct csilk_wf_tool_entry_t

A registered workflow tool (function-calling capability exposed to AI nodes). Tools are invoked in parallel via the libuv thread pool during AI node execution.

Data Fields
char *	description	Description for the AI model's tool schema.
csilk_wf_tool_fn	fn	Tool implementation callback.
char *	name	Tool name (e.g., "get_weather").
char *	parameters_json	JSON schema string for tool parameters.
void *	user_data	Opaque context for the callback.

◆ csilk_wf_edge_t

struct csilk_wf_edge_t

Data Fields
char *	condition	NULL for default/bind.
csilk_wf_node_t *	target	Destination node.

◆ csilk_wf_s

struct csilk_wf_s

Workflow definition: a DAG of processing nodes connected by conditional or unconditional edges. Each node runs a handler function on the libuv thread pool. Supports persistence via WAL, real-time monitoring via WebSocket, tool registration for AI nodes, and budget (token/TTL) limits.

Opaque handle for a workflow instance.

Collaboration diagram for csilk_wf_t:

[legend]

Data Fields
size_t	active_context_capacity
size_t	active_context_count
csilk_wf_ctx_t **	active_contexts
uv_mutex_t	ctx_mutex
uv_loop_t *	loop	libuv event loop for thread-pool scheduling.
int	max_tokens	Maximum total tokens across all AI calls (0 = unlimited).
size_t	monitor_capacity	Allocated monitor array capacity.
size_t	monitor_count	Number of active monitors.
uv_mutex_t	monitor_mutex	Protects the monitor array.
csilk_ctx_t **	monitors	WebSocket monitoring connections.
csilk_mq_t *	mq	Optional MQ for distributed execution.
char *	name	Human-readable workflow name.
size_t	node_capacity	Allocated node array capacity.
size_t	node_count	Number of registered nodes.
csilk_wf_node_t **	nodes	Array of node pointers.
size_t	tool_capacity	Allocated tool array capacity.
size_t	tool_count	Number of registered tools.
csilk_wf_tool_entry_t *	tools	Registered tool definitions.
int	ttl_sec	Workflow Time-To-Live in seconds (0 = no limit).
char *	wal_dir	WAL directory path (NULL = no persistence).

◆ node_work_t

struct node_work_t

Per-node-execution state passed through libuv work requests. Allocated in execute_node(), freed in after_worker_cb().

Collaboration diagram for node_work_t:

[legend]

Data Fields
csilk_wf_ctx_t *	ctx	Workflow execution context.
csilk_data_t *	input	Input data to the node's handler.
int	is_timed_out	Flag set by timer if node exceeds timeout_ms.
csilk_wf_node_t *	node	The node being executed.
uv_timer_t	node_timer	Per-node timeout or retry delay timer.
csilk_data_t *	output	Output data from the handler (set by worker_cb).
uv_work_t	req	libuv work request (must be first for cast).
int	retry_count	Current retry attempt.
int	timer_closing	Non-zero once uv_close is called on node_timer.
csilk_wf_trace_node_t *	trace_node	Trace record for this node (NULL if not tracing).

◆ sub_tool_work_t

struct sub_tool_work_t

Per-tool-call context for parallel tool execution within an AI node. Each tool call runs on its own libuv thread-pool worker.

Collaboration diagram for sub_tool_work_t:

[legend]

Data Fields
uv_cond_t *	cond	Shared condition variable for completion.
csilk_wf_ctx_t *	ctx	Workflow context (for tool registry lookup).
uv_mutex_t *	mutex	Shared mutex for the pending counter.
int *	pending	Shared atomic-like pending count.
char *	result	Tool output string (allocated by tool fn).
csilk_ai_tool_call_t *	tc	Tool call arguments from the AI response.

◆ stream_ctx_t

struct stream_ctx_t

Data Fields
csilk_wf_ctx_t *	ctx
const char *	node_id

Macro Definition Documentation

◆ MAX_WORKFLOW_STEPS

#define MAX_WORKFLOW_STEPS 1000

Function Documentation

◆ _csilk_json_get_path()

static char * _csilk_json_get_path	(	csilk_wf_ctx_t *	ctx,
		cJSON *	root,
		const char *	path
	)

static

Internal: traverse a cJSON tree following a dot-separated path.

Algorithm:

Split the path on "." using strtok_r.
For each token, if the current cJSON node is an array, index into it using atoi(); otherwise, use cJSON_GetObjectItemCaseSensitive().
If the final value is a string/number/bool, return it as a string allocated from the workflow arena. For objects/arrays, return a stringified JSON representation.

Parameters

ctx	Workflow context (for arena allocation).
root	Root cJSON node to start traversal from.
path	Dot-separated path (e.g., "user.address.city").

Returns: A string allocated in ctx->arena, or NULL if the path does not exist. The returned string is valid for the workflow's lifetime.

◆ _wf_broadcast()

static void _wf_broadcast	(	csilk_wf_t *	wf,
		const char *	event,
		const char *	node_id,
		const char *	payload
	)

static

◆ after_sub_worker_cb()

static void after_sub_worker_cb	(	uv_work_t *	req,
		int	status
	)

static

libuv after-work callback for tool execution. Decrements the shared pending counter and signals the condition variable to wake the main AI node handler thread.

◆ after_worker_cb()

static void after_worker_cb	(	uv_work_t *	req,
		int	status
	)

static

◆ ai_config_free()

static void ai_config_free ( void * ptr )

static

◆ ai_node_handler()

static csilk_data_t * ai_node_handler	(	csilk_wf_ctx_t *	ctx,
		csilk_data_t *	input,
		void *	user_data
	)

static

Built-in handler for AI workflow nodes.

Algorithm:

Resolve template expressions in the prompt string.
Create an AI engine instance (OpenAI driver by default) using AGENT_API_KEY and AGENT_API_BASE environment variables.
Build tool definitions from the workflow's registered tools.
Construct a message array (optional system message + user prompt).
Enter a request loop (max 10 iterations): a. Send a chat completion request with the current message array. b. If the response contains tool calls, execute each tool in parallel on the libuv thread pool, wait for all to complete via a condition variable, append tool results as new messages. c. If the response is a direct text completion, extract the content and AI metadata (model, token counts), return as csilk_data_t.
Clean up all temporary allocations (messages, tools, AI handle).

Parameters

ctx	Workflow execution context.
input	Ignored (AI prompts come from config templates).
user_data	Pointer to csilk_ai_config_t with model, prompt, etc.

Returns: Output data with type "text/plain" and AI metadata, or NULL on failure (missing API key, driver init failure, all retries exhausted).

◆ apply_filter()

static char * apply_filter	(	csilk_wf_ctx_t *	ctx,
		const char *	filter,
		char *	val
	)

static

Internal: resolve template expressions in a prompt string.

Template syntax: {{node_id.value}} -> raw value output from a node {{node_id.value.path.to}} -> JSONPath into a node's JSON output {{input.value}} -> the workflow's initial input {{input.value.path.to}} -> JSONPath into the initial input

Algorithm:

Iterate over all workflow nodes, searching for {{node_id.value}} patterns in the template string.
For each match, look up the node's output. If followed by ".path", parse the node output as JSON and traverse with _csilk_json_get_path(). Otherwise, use the raw output value.
Replace the {{...}} placeholder with the resolved value using arena memory.
Repeat for {{input.value}} patterns using the workflow's initial input.

Parameters

ctx	Workflow execution context.
template	Template string with {{...}} placeholders.

Returns: Resolved string allocated in ctx->arena.

Note: Unresolvable patterns (missing node output, bad path) are replaced with "(null)".

◆ cleanup_ctx()

static void cleanup_ctx ( csilk_wf_ctx_t * ctx )

static

Internal: free a workflow execution context and all resources.

Stops and closes the TTL timer if active, destroys all mutexes, frees the memory arena, node tracking arrays, WAL path, and the context struct itself.

Parameters

ctx	The execution context to clean up (may be NULL).

◆ cleanup_ctx_now()

static void cleanup_ctx_now ( csilk_wf_ctx_t * ctx )

static

◆ csilk_wf_add()

csilk_wf_node_t * csilk_wf_add	(	csilk_wf_t *	wf,
		const char *	id,
		csilk_wf_handler_t	handler,
		void *	user_data
	)

Add a node to the workflow.

Parameters

wf	Workflow handle.
id	Unique ID for the node.
handler	Function to execute.
user_data	Opaque pointer passed to the handler.

Returns: Handle to the created node, or NULL on failure.

◆ csilk_wf_add_ai()

csilk_wf_node_t * csilk_wf_add_ai	(	csilk_wf_t *	wf,
		const char *	id,
		const csilk_ai_config_t *	config
	)

Add a built-in AI node with template support.

Parameters

wf	Workflow handle.
id	Unique ID.
config	AI configuration (copied internally).

Returns: Node handle.

◆ csilk_wf_alloc()

void * csilk_wf_alloc	(	csilk_wf_ctx_t *	ctx,
		size_t	size
	)

Allocate memory from the workflow arena.

Parameters

ctx	Execution context.
size	Number of bytes.

Returns: Pointer to allocated memory.

◆ csilk_wf_bind()

void csilk_wf_bind	(	csilk_wf_node_t *	from,
		csilk_wf_node_t *	to
	)

Bind two nodes sequentially (default routing).

Parameters

from	Source node.
to	Destination node.

◆ csilk_wf_data_new()

csilk_data_t * csilk_wf_data_new	(	csilk_wf_ctx_t *	ctx,
		const char *	type,
		void *	value
	)

Allocate a new data container managed by the workflow arena.

Parameters

ctx	Execution context.
type	Data type identifier (copied).
value	Pointer to the data.

Returns: New data container.

◆ csilk_wf_enable_distributed()

void csilk_wf_enable_distributed	(	csilk_wf_t *	wf,
		csilk_mq_t *	mq
	)

Enable distributed execution by bridging workflow with an MQ.

Parameters

wf	Workflow handle.
mq	MQ handle for task distribution.

◆ csilk_wf_free()

void csilk_wf_free ( csilk_wf_t * wf )

Deallocate a workflow and all its nodes.

Parameters

wf	Workflow handle.

◆ csilk_wf_get_node()

csilk_wf_node_t * csilk_wf_get_node	(	csilk_wf_t *	wf,
		const char *	id
	)

Look up a workflow node by its string ID.

Get a node by ID.

Parameters

wf	The workflow instance.
id	Node identifier (set in csilk_wf_add()).

Returns: Node pointer, or NULL if not found.

Note: Linear search of the node array — O(n).

◆ csilk_wf_new()

csilk_wf_t * csilk_wf_new ( const char * name )

Create a new AI workflow instance.

Parameters

name	Descriptive name for the workflow.

Returns: New workflow handle, or NULL on failure.

◆ csilk_wf_node_set_entry()

void csilk_wf_node_set_entry	(	csilk_wf_node_t *	node,
		int	is_entry
	)

Mark a node as an entry point for the workflow.

Parameters

node	Node handle.
is_entry	Non-zero to mark as entry, 0 to unmark.

Note: By default, nodes with 0 incoming edges are entries. Use this to explicitly start a node that is also part of a loop.

◆ csilk_wf_node_set_interactive()

void csilk_wf_node_set_interactive	(	csilk_wf_node_t *	node,
		int	is_interactive
	)

Mark a node as interactive (requires human signal to proceed).

Parameters

node	Node handle.
is_interactive	Non-zero to enable interactive mode.

◆ csilk_wf_node_set_join()

void csilk_wf_node_set_join	(	csilk_wf_node_t *	node,
		csilk_wf_join_policy_t	policy
	)

Set the join policy for a node.

Parameters

node	Node handle.
policy	AND (default) or OR.

◆ csilk_wf_node_set_remote()

void csilk_wf_node_set_remote	(	csilk_wf_node_t *	node,
		int	is_remote
	)

Mark a node for remote execution via MQ.

Parameters

node	Node handle.
is_remote	Non-zero to offload to remote worker.

◆ csilk_wf_node_set_retry()

void csilk_wf_node_set_retry	(	csilk_wf_node_t *	node,
		int	max_retries,
		int	retry_delay_ms
	)

Set automatic retry policy for a specific node.

Parameters

node	Node handle.
max_retries	Maximum number of retry attempts.
retry_delay_ms	Delay before each retry.

◆ csilk_wf_node_set_schema()

void csilk_wf_node_set_schema	(	csilk_wf_node_t *	node,
		const char *	schema
	)

Set an expected JSON Schema for a node's output.

Parameters

node	Node handle.
schema	JSON Schema string (NULL to disable).

◆ csilk_wf_node_set_timeout()

void csilk_wf_node_set_timeout	(	csilk_wf_node_t *	node,
		int	timeout_ms
	)

Set a timeout for a specific node.

Parameters

node	Node handle.
timeout_ms	Timeout in milliseconds (0 for no timeout).

◆ csilk_wf_on()

void csilk_wf_on	(	csilk_wf_node_t *	from,
		const char *	condition,
		csilk_wf_node_t *	to
	)

Add a conditional route between nodes.

Parameters

from	Source node.
condition	Result string that triggers this route (e.g., "fail").
to	Destination node.

◆ csilk_wf_on_error()

void csilk_wf_on_error	(	csilk_wf_node_t *	from,
		csilk_wf_node_t *	target
	)

Add an error fallback route.

Parameters

from	Source node.
target	Destination node triggered if the source handler fails.

◆ csilk_wf_on_loop()

void csilk_wf_on_loop	(	csilk_wf_node_t *	from,
		const char *	condition,
		csilk_wf_node_t *	to
	)

Add a loop-back / feedback route between nodes.

Parameters

from	Source node.
condition	Result string that triggers this route.
to	Destination node (typically an earlier node).

Note: Unlike csilk_wf_on, this does NOT increment the 'to' node's incoming dependency count, preventing deadlocks in join logic.

◆ csilk_wf_register_monitor()

void csilk_wf_register_monitor	(	csilk_wf_t *	wf,
		csilk_ctx_t *	c
	)

Register a WebSocket connection to receive live workflow updates.

Parameters

wf	Workflow handle.
c	Request context (must be upgraded to WebSocket).

◆ csilk_wf_register_tool()

void csilk_wf_register_tool	(	csilk_wf_t *	wf,
		const char *	name,
		const char *	description,
		const char *	parameters_json,
		csilk_wf_tool_fn	fn,
		void *	user_data
	)

Register a tool that AI nodes can call.

Parameters

wf	Workflow handle.
name	Function name exposed to the LLM.
description	Description of what the tool does.
parameters_json	JSON Schema string for the arguments.
fn	C function to execute.
user_data	Context for the function.

◆ csilk_wf_resume()

void csilk_wf_resume	(	csilk_wf_t *	wf,
		const char *	exec_id,
		void()(csilk_data_t result)	callback
	)

Resume an interrupted workflow execution from a WAL file.

Parameters

wf	Workflow definition.
exec_id	The execution ID to resume.
callback	Callback for when the resumed workflow finishes.

◆ csilk_wf_route()

void csilk_wf_route	(	csilk_wf_node_t *	node,
		csilk_wf_router_t	router
	)

Set a dynamic router for a node.

Parameters

node	Node handle.
router	Function that determines the next node based on output.

◆ csilk_wf_run()

const char * csilk_wf_run	(	csilk_wf_t *	wf,
		csilk_data_t *	input,
		void()(csilk_data_t result)	callback
	)

Run the workflow asynchronously.

Parameters

wf	Workflow handle.
input	Initial input data.
callback	Callback invoked when the workflow completes or exits.

Returns: Unique Execution ID (string). Caller must not free.

◆ csilk_wf_run_ext_internal()

static const char * csilk_wf_run_ext_internal	(	csilk_wf_t *	wf,
		csilk_data_t *	input,
		void()(csilk_data_t )	callback,
		void()(csilk_data_t , csilk_wf_trace_t *)	trace_cb
	)

static

◆ csilk_wf_run_traced()

void csilk_wf_run_traced	(	csilk_wf_t *	wf,
		csilk_data_t *	input,
		void()(csilk_data_t result, csilk_wf_trace_t *trace)	callback
	)

Run workflow and generate a trace.

Parameters

wf	Workflow handle.
input	Initial input.
callback	Callback receiving final result and the full trace.

◆ csilk_wf_serve_ui()

void csilk_wf_serve_ui	(	csilk_app_t *	app,
		const char *	path
	)

Register a default route to serve the workflow dashboard.

Parameters

app	Application handle.
path	URL path (e.g., "/admin/workflow").

◆ csilk_wf_set_budget()

void csilk_wf_set_budget	(	csilk_wf_t *	wf,
		int	max_tokens
	)

Set a maximum token budget for the workflow.

Parameters

wf	Workflow handle.
max_tokens	Maximum total tokens (prompt + completion) allowed.

◆ csilk_wf_set_persistence()

void csilk_wf_set_persistence	(	csilk_wf_t *	wf,
		const char *	wal_dir
	)

Enable WAL persistence for a workflow definition.

Parameters

wf	Workflow handle.
wal_dir	Directory to store execution logs.

◆ csilk_wf_set_ttl()

void csilk_wf_set_ttl	(	csilk_wf_t *	wf,
		int	ttl_sec
	)

Set a global TTL for the workflow execution.

Parameters

wf	Workflow handle.
ttl_sec	TTL in seconds (0 for no limit).

◆ csilk_wf_signal_continue()

void csilk_wf_signal_continue	(	csilk_wf_t *	wf,
		const char *	exec_id,
		csilk_data_t *	input,
		void()(csilk_data_t result)	callback
	)

Signal a paused workflow to continue.

Parameters

wf	Workflow definition.
exec_id	Execution ID of the paused workflow.
input	Optional replacement input (e.g., human-edited prompt).
callback	Callback for when the resumed workflow finishes.

◆ csilk_wf_strdup()

char * csilk_wf_strdup	(	csilk_wf_ctx_t *	ctx,
		const char *	s
	)

Duplicate a string using the workflow arena.

Parameters

ctx	Execution context.
s	Source string.

Returns: Copied string.

◆ csilk_wf_to_mermaid()

char * csilk_wf_to_mermaid ( csilk_wf_t * wf )

Export the workflow graph as a Mermaid string.

Parameters

wf	Workflow handle.

Returns: Heap-allocated Mermaid code (caller must free).

◆ csilk_wf_trace_free()

void csilk_wf_trace_free ( csilk_wf_trace_t * trace )

Free a trace object.

◆ csilk_wf_trace_to_json()

char * csilk_wf_trace_to_json ( const csilk_wf_trace_t * trace )

Convert a trace object to a JSON string.

Returns: JSON string (caller must free).

◆ execute_node()

static void execute_node	(	csilk_wf_ctx_t *	ctx,
		csilk_wf_node_t *	node,
		csilk_data_t *	input
	)

static

Internal: enqueue a workflow node for execution on the libuv thread pool.

Algorithm:

Check termination flag (budget exceeded, TTL expired).
Handle Interactive Nodes: if a node is marked as interactive and has not yet been approved in this context, set is_paused flag, log WF_EV_PAUSE to WAL, broadcast to monitors, and return without executing.
Log WF_EV_NODE_START to WAL and broadcast "node_queued" to monitors.
Allocate a node_work_t struct. If tracing is active, create a trace node with start time and input dump.
Increment total_executions and nodes_active counters.
If the node has a per-node timeout, initialize and arm a uv_timer.
Queue the work via uv_queue_work(). The node handler runs on a background thread; after_worker_cb processes the result.

Parameters

ctx	Workflow execution context.
node	The node to execute.
input	Input data to pass to the node's handler.

◆ find_active_ctx()

static csilk_wf_ctx_t * find_active_ctx	(	csilk_wf_t *	wf,
		const char *	exec_id
	)

static

◆ free_work()

static void free_work ( node_work_t * work )

static

◆ node_add_edge()

static void node_add_edge	(	csilk_wf_node_t *	from,
		const char *	condition,
		csilk_wf_node_t *	to,
		int	is_loop
	)

static

◆ node_free()

static void node_free ( csilk_wf_node_t * node )

static

◆ on_ai_stream()

static void on_ai_stream	(	const char *	chunk,
		void *	user_data
	)

static

◆ on_node_timeout()

static void on_node_timeout ( uv_timer_t * handle )

static

libuv timer callback — marks a node as timed out. Sets the is_timed_out flag on the node_work_t, which causes after_worker_cb to treat the output as NULL even if the handler eventually completes.

◆ on_remote_result()

static void on_remote_result ( csilk_mq_ctx_t * m_ctx )

static

◆ on_retry_timer()

static void on_retry_timer ( uv_timer_t * handle )

static

libuv after-work callback — processes node completion on the main loop thread.

Algorithm (the central scheduler dispatch in the workflow engine):

Stop the per-node timeout timer if active.
Store the output in ctx->node_outputs and accumulate token usage from AI metadata.
Log to WAL (if enabled) and broadcast "node_finish" to monitors.
Record trace data (start/end time, input/output dump, model info).
Check budget (max_tokens): if exceeded, set is_terminated flag and terminate the workflow on next idle check.
If output is NULL and error_target is set, route to error node.
If the node has a dynamic router function, call it to determine the next node; otherwise, evaluate each outgoing edge:
- Unconditional edges (condition == NULL) always match.
- Conditional edges match if output type equals condition string.
For matching edges, check the target's join policy: AND join requires all incoming edges to fire before the target is ready; OR join fires on any single edge.
If no edges are triggered and no nodes are active, the workflow is complete: log WF_EV_END, deliver the final output via callback, and clean up the context.

◆ on_ttl_timer_close()

static void on_ttl_timer_close ( uv_handle_t * handle )

static

◆ on_work_timer_close()

static void on_work_timer_close ( uv_handle_t * handle )

static

◆ on_workflow_ttl()

static void on_workflow_ttl ( uv_timer_t * handle )

static

◆ register_active_ctx()

static void register_active_ctx	(	csilk_wf_t *	wf,
		csilk_wf_ctx_t *	ctx
	)

static

◆ remote_pass_handler()

static csilk_data_t * remote_pass_handler	(	csilk_wf_ctx_t *	ctx,
		csilk_data_t *	input,
		void *	user_data
	)

static

◆ resolve_templates()

static char * resolve_templates	(	csilk_wf_ctx_t *	ctx,
		const char *	template
	)

static

◆ serve_ui_handler()

static void serve_ui_handler ( csilk_ctx_t * c )

static

◆ sub_worker_cb()

static void sub_worker_cb ( uv_work_t * req )

static

libuv thread-pool work callback for tool execution. Looks up the tool by name in the workflow's tool registry and calls its function with the arguments from the AI tool call.

◆ unregister_active_ctx()

static void unregister_active_ctx	(	csilk_wf_t *	wf,
		csilk_wf_ctx_t *	ctx
	)

static

◆ wal_log_event()

static void wal_log_event	(	csilk_wf_ctx_t *	ctx,
		csilk_wf_event_type_t	type,
		const char *	node_id,
		csilk_data_t *	data
	)

static

Internal: persist a workflow event to the Write-Ahead Log.

Packs node_id, data type, and data value into a flat payload buffer and delegates to _wf_wal_append(). The payload format is: [node_id\0][data_type\0][data_value\0] Fields are NUL-terminated strings for simple parsing during recovery.

Parameters

ctx	Workflow execution context (must have wal_path set).
type	Event type (WF_EV_START, WF_EV_NODE_START, etc.).
node_id	Originating node ID, or NULL for workflow-level events.
data	Associated data (may be NULL for simple events).

Note: This is a no-op if ctx has no WAL path configured.

◆ worker_cb()

static void worker_cb ( uv_work_t * req )

static

libuv thread-pool work callback — executes a workflow node's handler on a background thread.

Broadcasts a "node_start" event to monitors, then calls the node's handler function. The output is stored in the work request struct for retrieval by after_worker_cb on the main loop thread.

Variable Documentation

◆ g_distributed_wf_count

size_t g_distributed_wf_count = 0

static

◆ g_distributed_wfs

csilk_wf_t* g_distributed_wfs[32]

static

Data Structures

Macros

Functions

Variables

Detailed Description

Data Structure Documentation

◆ csilk_ai_meta_t

◆ csilk_wf_tool_entry_t

◆ csilk_wf_edge_t

◆ csilk_wf_s

◆ node_work_t

◆ sub_tool_work_t

◆ stream_ctx_t

Macro Definition Documentation

◆ MAX_WORKFLOW_STEPS

Function Documentation

◆ _csilk_json_get_path()

◆ _wf_broadcast()

◆ after_sub_worker_cb()

◆ after_worker_cb()

◆ ai_config_free()

◆ ai_node_handler()

◆ apply_filter()

◆ cleanup_ctx()

◆ cleanup_ctx_now()

◆ csilk_wf_add()

◆ csilk_wf_add_ai()

◆ csilk_wf_alloc()

◆ csilk_wf_bind()

◆ csilk_wf_data_new()

◆ csilk_wf_enable_distributed()

◆ csilk_wf_free()

◆ csilk_wf_get_node()

◆ csilk_wf_new()

◆ csilk_wf_node_set_entry()

◆ csilk_wf_node_set_interactive()

◆ csilk_wf_node_set_join()

◆ csilk_wf_node_set_remote()

◆ csilk_wf_node_set_retry()

◆ csilk_wf_node_set_schema()

◆ csilk_wf_node_set_timeout()

◆ csilk_wf_on()

◆ csilk_wf_on_error()

◆ csilk_wf_on_loop()

◆ csilk_wf_register_monitor()

◆ csilk_wf_register_tool()

◆ csilk_wf_resume()

◆ csilk_wf_route()

◆ csilk_wf_run()

◆ csilk_wf_run_ext_internal()

◆ csilk_wf_run_traced()

◆ csilk_wf_serve_ui()

◆ csilk_wf_set_budget()

◆ csilk_wf_set_persistence()

◆ csilk_wf_set_ttl()

◆ csilk_wf_signal_continue()

◆ csilk_wf_strdup()

◆ csilk_wf_to_mermaid()

◆ csilk_wf_trace_free()

◆ csilk_wf_trace_to_json()

◆ execute_node()

◆ find_active_ctx()

◆ free_work()

◆ node_add_edge()

◆ node_free()

◆ on_ai_stream()

◆ on_node_timeout()

◆ on_remote_result()

◆ on_retry_timer()

◆ on_ttl_timer_close()

◆ on_work_timer_close()

◆ on_workflow_ttl()

◆ register_active_ctx()

◆ remote_pass_handler()

◆ resolve_templates()

◆ serve_ui_handler()

◆ sub_worker_cb()

◆ unregister_active_ctx()

◆ wal_log_event()

◆ worker_cb()