cloacina-computation-graph

Rust

Core types for Cloacina computation graph plugins.

This crate contains the types that packaged computation graph cdylibs need at compile time. It is the computation-graph equivalent of cloacina-workflow — a thin crate that avoids pulling in the full engine.

The #[computation_graph] macro expands into code that references types from this crate. Embedded-mode users get these types re-exported from cloacina.

Re-exports

serde_json (serde_json)

Structs

`struct SourceName`

pub

Derives: Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize

Identifies an accumulator source by name.

Fields

Name	Type	Description
`0`	`String`

Methods

`new`

pub

fn new(name: impl Into<String>) -> Self

Source

    pub fn new(name: impl Into<String>) -> Self {
        Self(name.into())
    }

`as_str`

pub

fn as_str(&self) -> &str

Source

    pub fn as_str(&self) -> &str {
        &self.0
    }

`struct InputCache`

pub

Derives: Debug, Clone

The input cache holds the last-seen serialized boundary per source.

The reactor’s receiver task updates this cache continuously. The executor takes a snapshot before calling the compiled graph function.

Serialization format: bincode (compact binary). The FFI packaging bridge converts bincode→JSON at the boundary for plugin compatibility.

Methods

`new`

pub

fn new() -> Self

Source

    pub fn new() -> Self {
        Self {
            entries: HashMap::new(),
        }
    }

`update`

pub

fn update(&mut self, source: SourceName, bytes: Vec<u8>)

Update the cached value for a source.

Source

    pub fn update(&mut self, source: SourceName, bytes: Vec<u8>) {
        self.entries.insert(source, bytes);
    }

`get`

pub

fn get<T: DeserializeOwned>(
    &self,
    name: &str,
) -> Option<Result<T, GraphError>>

Get and deserialize a cached value by source name.

Source

    pub fn get<T: DeserializeOwned>(&self, name: &str) -> Option<Result<T, GraphError>> {
        let bytes = self.entries.get(&SourceName::new(name))?;
        Some(deserialize::<T>(bytes))
    }

`has`

pub

fn has(&self, name: &str) -> bool

Check if a source has an entry in the cache.

Source

    pub fn has(&self, name: &str) -> bool {
        self.entries.contains_key(&SourceName::new(name))
    }

`get_raw`

pub

fn get_raw(&self, name: &str) -> Option<&[u8]>

Get the raw bytes for a source.

Source

    pub fn get_raw(&self, name: &str) -> Option<&[u8]> {
        self.entries
            .get(&SourceName::new(name))
            .map(|v| v.as_slice())
    }

`snapshot`

pub

fn snapshot(&self) -> InputCache

Create a snapshot (clone) of the cache.

Source

    pub fn snapshot(&self) -> InputCache {
        self.clone()
    }

`len`

pub

fn len(&self) -> usize

Number of sources in the cache.

Source

    pub fn len(&self) -> usize {
        self.entries.len()
    }

`is_empty`

pub

fn is_empty(&self) -> bool

Whether the cache is empty.

Source

    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

`replace_all`

pub

fn replace_all(&mut self, other: InputCache)

Replace all entries.

Source

    pub fn replace_all(&mut self, other: InputCache) {
        self.entries = other.entries;
    }

`sources`

pub

fn sources(&self) -> Vec<&SourceName>

List all source names in the cache.

Source

    pub fn sources(&self) -> Vec<&SourceName> {
        self.entries.keys().collect()
    }

`entries_raw`

pub

fn entries_raw(&self) -> &HashMap<SourceName, Vec<u8>>

Get a reference to the raw entries map.

Source

    pub fn entries_raw(&self) -> &HashMap<SourceName, Vec<u8>> {
        &self.entries
    }

`entries_as_json`

pub

fn entries_as_json(&self) -> HashMap<String, String>

Return entries as a JSON-friendly map.

Source

    pub fn entries_as_json(&self) -> HashMap<String, String> {
        self.entries
            .iter()
            .map(|(name, bytes)| {
                let value = if cfg!(debug_assertions) {
                    serde_json::from_slice::<serde_json::Value>(bytes)
                        .map(|v| v.to_string())
                        .unwrap_or_else(|_| hex_encode(bytes))
                } else {
                    hex_encode(bytes)
                };
                (name.as_str().to_string(), value)
            })
            .collect()
    }

`struct ComputationGraphRegistration`

pub

Metadata about a registered computation graph.

accumulator_names and reaction_mode are the canonical fields consumed by the packaging FFI and the reconciler. Bundled-form graphs populate these from the local declaration; split-form graphs mirror the referenced reactor’s declaration. Trigger-less graphs carry empty accumulator_names and trigger_reactor = None.

Fields

Name	Type	Description
`graph_fn`	`CompiledGraphFn`	The compiled graph function.
`trigger_reactor`	`Option<String>`	Name of the reactor this graph is bound to, if any. `None` for trigger-less graphs (T-02/T-03 invoke these directly from workflow tasks or Python tasks).
`accumulator_names`	`Vec<String>`	Accumulator names. For split-form graphs this mirrors the reactor’s accumulators; for trigger-less graphs it is empty.
`reaction_mode`	`String`	Reaction mode: `"when_any"`, `"when_all"`, or `"none"` for trigger-less graphs.

`struct ReactorConstructorRef`

pub

Derives: Debug, Clone

A reference to a packaged WASM reactor constructor carried by a reactor declaration (CLOACI-T-0830).

When present, the reactor’s firing decision is delegated to the named WASM constructor’s evaluate (installed on the live Reactor via with_evaluator), REPLACING the built-in WhenAny/WhenAll dirty-flag criteria. The fields mirror the constructor!(...) consumer surface added for task/trigger in CLOACI-T-0829, so a reactor constructor is authored/resolved the same way:

from — the provider package, "name[@version]", resolved against the T-0829 provider search path (CLOACINA_PROVIDER_PATH / ./providers / the process override);
constructor — which constructor inside that provider (validated against the provider’s constructor.json name);
config — the author’s config = { name = value } kwargs as (name, value) pairs in WRITTEN order, bound BY NAME at load.

Resolution happens in the CG scheduler (load_reactor) behind the constructors-wasm feature; this type itself is always compiled (it carries only String/serde_json::Value, no wasmtime types).

Fields

Name	Type	Description
`from`	`String`	Provider package reference, `"name[@version]"`.
`constructor`	`String`	The constructor’s `constructor.json` name inside the provider.
`config`	`Vec<(String, serde_json::Value)>`	Author config as `(name, value)` pairs in written order; bound by name.
`grants`	`Vec<(String, Vec<String>)>`	Tenant capability grants as raw `(kind, patterns)` pairs (kinds: `http`, `tcp`, `fs`, `env`) — the `grants = { .. }` literal from the reactor’s consumer surface (CLOACI-T-0834). Carried opaquely (like `config`) and translated to a fidius `EgressPolicy` + capability allow-list at load. Empty ⇒ default-closed (the constructor reaches nothing).
`runtime`	`Option<String>`	CLOACI-T-0920: the author’s optional trust-tier PIN — the `runtime = "wasm"\|"native"` literal from the consumer surface, or `None` when unpinned. Carried as a `String` (validated at macro-expansion time, re-parsed fail-closed at load) so this crate stays free of any dependency on the constructor contract crate. When pinned, the load fails unless the RESOLVED provider’s runtime matches; `"native"` additionally acknowledges that `grants` are advisory-only on that tier.

`struct AccumulatorSpec`

pub

Derives: Debug, Clone

The authored KIND + config of one accumulator a reactor consumes (CLOACI-T-0839). Carried on ReactorRegistration so runtime-walking load paths (notably packaged Python, where @cloaca.state_accumulator is the only declaration) can spawn the right accumulator runtime instead of defaulting to passthrough. config is the same String-keyed map the manifest override channel uses (e.g. capacity for state accumulators).

Fields

Name	Type	Description
`name`	`String`
`accumulator_type`	`String`	“passthrough” \| “stream” \| “polling” \| “batch” \| “state”.
`config`	`std::collections::HashMap<String, String>`

`struct ReactorRegistration`

pub

Derives: Debug, Clone

Runtime-side description of a reactor.

Populated by the #[reactor] macro’s emitted inventory entry.

Fields

Name	Type	Description
`name`	`String`
`accumulator_names`	`Vec<String>`
`accumulator_specs`	`Vec<AccumulatorSpec>`	Authored accumulator kinds/configs for (a subset of) `accumulator_names` (CLOACI-T-0839). Empty on paths where the kind travels out-of-band (Rust packaged reactors carry it in the FFI metadata); populated by the Python `@cloaca.reactor` path from the module’s accumulator decorators. Consumers fall back to manifest overrides, then passthrough.
`reaction_mode`	`ReactionMode`
`constructor`	`Option<ReactorConstructorRef>`	Optional packaged WASM reactor-constructor reference (CLOACI-T-0830). `Some(..)` makes the WASM guest’s `evaluate` the reactor’s firing criteria; `None` is the native dirty-flag reactor. Defaults to `None` so every pre-existing construction site stays valid.

Enums

`enum GraphResult`

pub

Result of executing a compiled computation graph.

Variants

Completed - Graph executed to completion. Contains terminal node outputs.
Error - Graph execution failed.

Methods

`completed`

pub

fn completed(outputs: Vec<Box<dyn Any + Send>>) -> Self

Source

    pub fn completed(outputs: Vec<Box<dyn Any + Send>>) -> Self {
        Self::Completed {
            outputs,
            outputs_json: Vec::new(),
        }
    }

`completed_with_json`

pub

fn completed_with_json(
    outputs: Vec<Box<dyn Any + Send>>,
    outputs_json: Vec<serde_json::Value>,
) -> Self

Completed with the terminal outputs also captured as JSON (CLOACI-T-0775).

Source

    pub fn completed_with_json(
        outputs: Vec<Box<dyn Any + Send>>,
        outputs_json: Vec<serde_json::Value>,
    ) -> Self {
        Self::Completed {
            outputs,
            outputs_json,
        }
    }

`error`

pub

fn error(err: GraphError) -> Self

Source

    pub fn error(err: GraphError) -> Self {
        Self::Error(err)
    }

`is_completed`

pub

fn is_completed(&self) -> bool

Source

    pub fn is_completed(&self) -> bool {
        matches!(self, Self::Completed { .. })
    }

`is_error`

pub

fn is_error(&self) -> bool

Source

    pub fn is_error(&self) -> bool {
        matches!(self, Self::Error(_))
    }

`enum GraphError`

pub

Errors that can occur during graph execution.

Variants

Serialization
Deserialization
MissingInput
NodeExecution
Execution

`enum ReactionMode`

pub

How a reactor decides when to fire.

Variants

WhenAny - Fire as soon as any one accumulator has new input.
WhenAll - Fire only when every accumulator has new input since the last firing.

Methods

`as_str`

pub const

const fn as_str(&self) -> &'static str

Source

    pub const fn as_str(&self) -> &'static str {
        match self {
            ReactionMode::WhenAny => "when_any",
            ReactionMode::WhenAll => "when_all",
        }
    }

Traits

`trait Reactor`

pub

Compile-time handle for a reactor declaration.

Implemented by the unit struct emitted by #[reactor]. The constants are readable at const context, which the #[computation_graph] macro uses to const-check the graph’s entry accumulators against the reactor’s accumulator set.

`trait Graph`

pub

Compile-time handle for a computation graph declaration.

Implemented by the __CGHandle_<modname> unit struct emitted by #[computation_graph]. Lets other macros (notably #[task(invokes = computation_graph(H))]) reference a graph by type path and const-check invariants like trigger-less-ness at compile time.

Functions

`fn serialize`

pub

fn serialize<T: Serialize>(value: &T) -> Result<Vec<u8>, GraphError>

Serialize a value to bincode bytes.

Bincode is used for all internal wire formats (boundary channels, checkpoint persistence, accumulator-to-reactor messaging).

Source

pub fn serialize<T: Serialize>(value: &T) -> Result<Vec<u8>, GraphError> {
    bincode::serialize(value).map_err(|e| GraphError::Serialization(e.to_string()))
}

`fn deserialize`

pub

fn deserialize<T: DeserializeOwned>(
    bytes: &[u8],
) -> Result<T, GraphError>

Deserialize bincode bytes to a value.

Source

pub fn deserialize<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, GraphError> {
    bincode::deserialize(bytes).map_err(|e| GraphError::Deserialization(e.to_string()))
}

Type aliases

Name	Type	Description
`CompiledGraphFn`	`Arc<dyn Fn(InputCache) -> Pin<Box<dyn Future<Output = GraphResult> + Send>> + Send + Sync>`	Type alias for the compiled graph function.
`ComputationGraphConstructor`	`Box<dyn Fn() -> ComputationGraphRegistration + Send + Sync>`
`ReactorConstructor`	`Box<dyn Fn() -> ReactorRegistration + Send + Sync>`

cloacina-computation-graph

Re-exports

Structs

struct SourceName

Fields

Methods

new

as_​str

struct InputCache

Methods

new

update

get

has

get_​raw

snapshot

len

is_​empty

replace_​all

sources

entries_​raw

entries_​as_​json

struct ComputationGraphRegistration

Fields

struct ReactorConstructorRef

Fields

struct AccumulatorSpec

Fields

struct ReactorRegistration

Fields

Enums

enum GraphResult

Variants

Methods

completed

completed_​with_​json

error

is_​completed

is_​error

enum GraphError

Variants

enum ReactionMode

Variants

Methods

as_​str

Traits

trait Reactor

trait Graph

Functions

fn serialize

fn deserialize

Type aliases

`struct SourceName`

`new`

`as_str`

`struct InputCache`

`new`

`update`

`get`

`has`

`get_raw`

`snapshot`

`len`

`is_empty`

`replace_all`

`sources`

`entries_raw`

`entries_as_json`

`struct ComputationGraphRegistration`

`struct ReactorConstructorRef`

`struct AccumulatorSpec`

`struct ReactorRegistration`

`enum GraphResult`

`completed`

`completed_with_json`

`error`

`is_completed`

`is_error`

`enum GraphError`

`enum ReactionMode`

`as_str`

`trait Reactor`

`trait Graph`

`fn serialize`

`fn deserialize`