Runtime-independent asynchronous graph compiler and runtime
///|
test "workgraph-core compile usage" {
let entry = @core.NodeId::NodeId("plan")
let definition = @core.GraphDefinition::GraphDefinition(
@core.Reducer::Reducer(fn(state : Int, _patch : Unit) { state }),
)
definition.add_node(
@core.Node::Node(
entry,
@core.NodeMetadata::NodeMetadata(
name="Plan",
description=None,
kind=@core.Function,
tags=[],
),
async fn(_context, _state) {
@async.pause()
@core.NodeOutput::NodeOutput(None, None)
},
),
)
definition.set_router(
entry,
@core.router([], fn(_state, _completion) { @core.End }),
)
definition.set_entry(entry)
let compiled = definition.compile()
inspect(compiled.entry().to_string(), content="plan")
inspect(compiled.snapshot().nodes.length(), content="1")
}pub suberror GraphRuntimeError {
NodeTimedOut(node_id~ : NodeId, step~ : Int, timeout_ms~ : Int)
NodeFailed(node_id~ : NodeId, step~ : Int, cause~ : Error)
ReduceFailed(node_id~ : NodeId, step~ : Int, cause~ : Error)
RouteFailed(node_id~ : NodeId, step~ : Int, cause~ : Error)
RouteContractViolated(from~ : NodeId, to~ : NodeId)
StepLimitExceeded(limit~ : Int)
ExplicitFailure(node_id~ : NodeId, message~ : String)
ResourceCleanupFailed(failure~ : RunFailure)
NodeNotFound(NodeId)
RouterNotFound(NodeId)
} derive(Debug)pub suberror ResourceStoreError {
NodeOwnerRequired(ResourceKey)
InvalidCleanupTimeout(Int)
CleanupTimedOut(Int)
CloseFailed(errors~ : ReadOnlyArray[Error])
} derive(Debug)pub suberror RunConfigurationError {
MaxStepsMustBePositive(Int)
NodeTimeoutMustBePositive(Int)
CleanupTimeoutMustBePositive(Int)
} derive(Debug)fn[S, P] CompiledGraph::get_node(self : CompiledGraph[S, P], id : NodeId) -> Node[S, P] raise GraphRuntimeErrorfn[S, P] CompiledGraph::get_router(self : CompiledGraph[S, P], id : NodeId) -> Router[S] raise GraphRuntimeErrorpub struct CompiledGraphSnapshot {
entry : NodeId
nodes : ReadOnlyArray[CompiledNodeSnapshot]
} derive(Debug)pub struct CompiledNodeSnapshot {
id : NodeId
metadata : NodeMetadata
router_metadata : RouterMetadata
declared_routes : ReadOnlyArray[DeclaredRoute]
} derive(Debug)fn DeclaredRoute::DeclaredRoute(target : NodeId, metadata? : DeclaredRouteMetadata) -> DeclaredRoutefn[S, P] GraphDefinition::add_node(self : GraphDefinition[S, P], node : Node[S, P]) -> Unit raise GraphBuildErrorfn[S, P] GraphDefinition::compile(self : GraphDefinition[S, P]) -> CompiledGraph[S, P] raise GraphValidationErrorfn[S, P] GraphDefinition::set_entry(self : GraphDefinition[S, P], entry : NodeId) -> Unit raise GraphBuildErrorfn[S, P] GraphDefinition::set_router(self : GraphDefinition[S, P], from : NodeId, value : Router[S]) -> Unit raise GraphBuildErrorpub enum GraphEvent {
RunStarted(RunId)
RunCompleted(run_id~ : RunId, steps~ : Int)
RunFailed(run_id~ : RunId, cause~ : Error)
RunCancelled(RunId)
NodeStarted(run_id~ : RunId, node_id~ : NodeId, step~ : Int)
NodeCompleted(run_id~ : RunId, node_id~ : NodeId, step~ : Int, completion~ : NodeCompletion)
NodeFailed(run_id~ : RunId, node_id~ : NodeId, step~ : Int, cause~ : Error)
StateUpdated(run_id~ : RunId, node_id~ : NodeId, step~ : Int)
RouteSelected(run_id~ : RunId, from~ : NodeId, route~ : Route)
ResourceOpening(run_id~ : RunId, key~ : ResourceKey)
ResourceOpened(run_id~ : RunId, key~ : ResourceKey)
ResourceClosed(run_id~ : RunId, key~ : ResourceKey)
} derive(Debug)fn[S, P] GraphRuntime::GraphRuntime(graph : CompiledGraph[S, P], events? : EventSink) -> GraphRuntime[S, P]async fn[S, P] GraphRuntime::invoke(self : GraphRuntime[S, P], initial_state : S, options? : RunOptions) -> RunResult[S]pub struct Node[S, P] {
id : NodeId
metadata : NodeMetadata
execute : async (NodeContext, S) -> NodeOutput[P]
}fn[S, P] Node::Node(id : NodeId, metadata : NodeMetadata, execute : async (NodeContext, S) -> NodeOutput[P]) -> Node[S, P]pub struct NodeContext {
run_id : RunId
node_id : NodeId
step : Int
deadline_ms : Int64?
task_group : TaskGroup[Unit]
events : EventSink
resources : ResourceStore
}fn NodeContext::NodeContext(run_id : RunId, node_id : NodeId, step : Int, task_group : TaskGroup[Unit], events? : EventSink, resources? : ResourceStore, deadline_ms? : Int64) -> NodeContextfn NodeMetadata::NodeMetadata(name~ : String, description~ : String?, kind~ : NodeKind, tags~ : ReadOnlyArray[String]) -> NodeMetadatapub struct Reducer[S, P] {
apply : (S, P) -> S raise
}pub struct ResourceStore {
values : AnyMutableMap[ResourceKey]
cleanups : Array[StoredResourceCleanup]
}async fn[T] ResourceStore::acquire_resource(self : ResourceStore, reference : AnyRef[ResourceKey, T], scope : ResourceScope, open : async () -> T, cleanup : async (T) -> Unit, owner? : NodeId) -> Tfn[T] ResourceStore::set(self : ResourceStore, reference : AnyRef[ResourceKey, T], value : T) -> Unitpub struct Router[S] {
metadata : RouterMetadata
declared_routes : ReadOnlyArray[DeclaredRoute]
evaluate : (S, NodeCompletion) -> Route raise
}pub struct RunOptions {
max_steps : Int
node_timeout_ms : Int?
cleanup_timeout_ms : Int
} derive(Debug)fn RunOptions::RunOptions(max_steps? : Int, node_timeout_ms? : Int, cleanup_timeout_ms? : Int) -> RunOptions raise RunConfigurationErrorfn[S] router(declared_routes : ReadOnlyArray[DeclaredRoute], evaluate : (S, NodeCompletion) -> Route raise, metadata? : RouterMetadata) -> Router[S]Install
Download zipRuntime-independent asynchronous graph compiler and runtime