cloacina-python::computation_graph

Rust

Python computation graph bindings.

Mirrors the WorkflowBuilder + @task pattern. Two flavors:

Reactor-triggered — the graph is bound to a @cloaca.reactor class. Entry nodes consume the reactor’s accumulator payloads.

@cloaca.reactor(name="market_maker",
                accumulators=["alpha", "beta"],
                mode="when_any")
class MarketMaker: pass

with cloaca.ComputationGraphBuilder("market_maker",
    reactor=MarketMaker,
    graph={...}) as builder: ...

Trigger-less — the graph is invoked by a workflow task via @cloaca.task(invokes=...). Entry nodes take only the task context; no accumulator/cache inputs are permitted.
```
with cloaca.ComputationGraphBuilder("score_transactions",
    graph={"score": {}}) as score_graph: ...
```

Pre-split (bundled) form:

with cloaca.ComputationGraphBuilder("market_maker",
    react={"mode": "when_any", "accumulators": ["alpha", "beta"]},
    graph={
        "decision": {"inputs": ["alpha", "beta"], "routes": {
            "Signal": "signal_handler",
            "NoAction": "audit_logger",
        }},
        "signal_handler": {},
        "audit_logger": {},
    }
) as builder:

    @cloaca.node
    def decision(alpha, beta):
        if alpha["value"] + beta["estimate"] > 10:
            return ("Signal", {"output": alpha["value"] + beta["estimate"]})
        return ("NoAction", {"reason": "below threshold"})

    @cloaca.node
    def signal_handler(signal):
        return {"published": True}

    @cloaca.node
    def audit_logger(reason):
        return {"logged": True}

Structs

`struct PyAccumulatorRegistration`

pub

Derives: Debug, Clone

Metadata for a registered Python accumulator.

Fields

Name	Type	Description
`name`	`String`
`accumulator_type`	`String`
`config`	`HashMap<String, String>`
`capacity`	`i32`	Capacity for `state` accumulators (bounded buffer). `> 0` bounded, `< 0` unbounded, `0` write-only sink. Defaults to `0` for the other accumulator types, which ignore it.

`struct GraphKey`

pub

Derives: Debug, Clone, PartialEq, Eq, Hash

CLOACI-T-0921: process-global Python registries are keyed by (tenant, package, name), not by bare name.

The @graph / @*_accumulator decorators run at import time; the loader installs a RegistrationScope around each package import (crate::registration_scope) so a registration is stamped with whose it is. Two tenants shipping a graph called pipeline used to share one executor slot — last import wins, cross-tenant.

Fields

Name	Type	Description
`tenant_id`	`Option<String>`
`package`	`Option<String>`
`name`	`String`

Methods

`new`

pub

fn new(scope: &RegistrationScope, name: &str) -> Self

Source

    pub fn new(scope: &RegistrationScope, name: &str) -> Self {
        Self {
            tenant_id: scope.tenant_id.clone(),
            package: scope.package.clone(),
            name: name.to_string(),
        }
    }

`current`

pub

fn current(name: &str) -> Self

The key this name would take under the thread’s current scope.

Source

    pub fn current(name: &str) -> Self {
        Self::new(&current_registration_scope(), name)
    }

`describe`

pub

fn describe(&self) -> String

Source

    pub fn describe(&self) -> String {
        format!(
            "{}::{}::{}",
            self.tenant_id.as_deref().unwrap_or("<untenanted>"),
            self.package.as_deref().unwrap_or("<unpackaged>"),
            self.name
        )
    }

`class ComputationGraphBuilder`

Binding

Python API: cloaca.ComputationGraphBuilder

Fields

Name	Type	Description
`name`	`String`
`reactor_binding`	`Option<ReactorBinding>`	Reactor binding for the split form; `None` means trigger-less.
`nodes_decl`	`Vec<PyNodeDecl>`

Methods

`new`

fn new(
    _py: Python<'_>,
    name: &str,
    graph: &Bound<'_, PyDict>,
    reactor: Option<&Bound<'_, PyAny>>,
    react: Option<&Bound<'_, PyAny>>,
) -> PyResult<Self>

Python API: cloaca.ComputationGraphBuilder.new

Source

    pub fn new(
        _py: Python<'_>,
        name: &str,
        graph: &Bound<'_, PyDict>,
        reactor: Option<&Bound<'_, PyAny>>,
        react: Option<&Bound<'_, PyAny>>,
    ) -> PyResult<Self> {
        // Hard-break migration: bundled `react={...}` form is gone — point users
        // at I-0101 with a clear error.
        if react.is_some() {
            return Err(PyValueError::new_err(
                "ComputationGraphBuilder no longer accepts the bundled `react={...}` kwarg. \
                 Declare a `@cloaca.reactor` class and pass `reactor=YourReactor`, \
                 or omit `reactor` entirely for a trigger-less graph invoked by \
                 `@cloaca.task(invokes=...)`. See initiative CLOACI-I-0101.",
            ));
        }

        let reactor_binding = match reactor {
            None => None,
            Some(r) => Some(extract_reactor_binding(r)?),
        };

        let nodes_decl = parse_graph_dict(graph)?;

        Ok(PyComputationGraphBuilder {
            name: name.to_string(),
            reactor_binding,
            nodes_decl,
        })
    }

`NAME`

fn NAME(&self) -> String

Python API: cloaca.ComputationGraphBuilder.NAME

The graph’s declared name. Mirrors Rust’s Graph::NAME so this instance can serve as the handle referenced by @cloaca.task(invokes=...).

Source

    pub fn NAME(&self) -> String {
        self.name.clone()
    }

`enter`

fn __enter__(slf: PyRef<Self>) -> PyRef<Self>

Python API: cloaca.ComputationGraphBuilder.__enter__

Context manager entry — establish graph context for @node decorators

Source

    pub fn __enter__(slf: PyRef<Self>) -> PyRef<Self> {
        push_graph_context(slf.name.clone());
        slf
    }

`exit`

fn __exit__(
    &self,
    py: Python,
    _exc_type: Option<&Bound<PyAny>>,
    _exc_value: Option<&Bound<PyAny>>,
    _traceback: Option<&Bound<PyAny>>,
) -> PyResult<bool>

Python API: cloaca.ComputationGraphBuilder.__exit__

Context manager exit — validate nodes against topology, build executor

Source

    pub fn __exit__(
        &self,
        py: Python,
        _exc_type: Option<&Bound<PyAny>>,
        _exc_value: Option<&Bound<PyAny>>,
        _traceback: Option<&Bound<PyAny>>,
    ) -> PyResult<bool> {
        pop_graph_context();

        let registered_nodes = drain_nodes();

        // Validate: every node in topology must have a registered function
        for decl in &self.nodes_decl {
            if !registered_nodes.contains_key(&decl.name) {
                return Err(PyAttributeError::new_err(format!(
                    "computation graph '{}' topology references node '{}' but no @cloaca.node function with that name was defined",
                    self.name, decl.name
                )));
            }
        }

        // Validate: every registered function must appear in topology
        for fn_name in registered_nodes.keys() {
            if !self.nodes_decl.iter().any(|d| d.name == *fn_name) {
                return Err(PyValueError::new_err(format!(
                    "function '{}' was decorated with @cloaca.node but does not appear in the graph topology",
                    fn_name
                )));
            }
        }

        // Trigger-less entry contract: in trigger-less form, no node may
        // declare cache `inputs=[...]` — those inputs only exist when a
        // reactor is feeding the graph. Trigger-less graphs receive their
        // input via the task `Context` passed by `@cloaca.task(invokes=...)`.
        if self.reactor_binding.is_none() {
            for decl in &self.nodes_decl {
                if !decl.cache_inputs.is_empty() {
                    return Err(PyValueError::new_err(format!(
                        "trigger-less computation graph '{}': node '{}' declares cache \
                         inputs={:?}, but trigger-less graphs have no accumulator inputs. \
                         Either remove the inputs (graph receives data through the task \
                         context) or bind a reactor with `reactor=YourReactor`.",
                        self.name, decl.name, decl.cache_inputs
                    )));
                }
            }
        } else if let Some(ref binding) = self.reactor_binding {
            // Split-form sanity: every cache input declared by a node must be
            // one of the reactor's accumulators. Catches typos at registration
            // time rather than producing silent `None`s at execution time.
            for decl in &self.nodes_decl {
                for input in &decl.cache_inputs {
                    if !binding.accumulators.contains(input) {
                        return Err(PyValueError::new_err(format!(
                            "computation graph '{}': node '{}' references cache input \
                             '{}' which is not in reactor '{}' accumulators {:?}",
                            self.name, decl.name, input, binding.name, binding.accumulators
                        )));
                    }
                }
            }
        }

        // Build the executor
        let node_map: HashMap<String, PyNodeDecl> = self
            .nodes_decl
            .iter()
            .cloned()
            .map(|n| (n.name.clone(), n))
            .collect();
        let execution_order = compute_execution_order(&self.nodes_decl);

        let (react_mode, accumulators, reactor_name) = match &self.reactor_binding {
            Some(b) => (b.mode.clone(), b.accumulators.clone(), Some(b.name.clone())),
            None => (String::new(), Vec::new(), None),
        };

        let executor = PythonGraphExecutor {
            name: self.name.clone(),
            node_functions: registered_nodes,
            node_map,
            execution_order,
            react_mode,
            accumulators,
            has_reactor: self.reactor_binding.is_some(),
            reactor_name,
        };

        // Register the executor globally (similar to workflow registration)
        register_graph_executor(self.name.clone(), executor, py)?;

        Ok(false)
    }

`repr`

fn __repr__(&self) -> String

Python API: cloaca.ComputationGraphBuilder.__repr__

Source

    pub fn __repr__(&self) -> String {
        format!(
            "ComputationGraphBuilder(name='{}', nodes={})",
            self.name,
            self.nodes_decl.len()
        )
    }

`execute`

fn execute(
    &self,
    py: Python<'_>,
    inputs: &Bound<'_, PyDict>,
) -> PyResult<Py<PyAny>>

Python API: cloaca.ComputationGraphBuilder.execute

Execute the computation graph with the given input cache.

inputs is a dict mapping source names to their data dicts. Returns the terminal node’s output dict.

Source

    pub fn execute(&self, py: Python<'_>, inputs: &Bound<'_, PyDict>) -> PyResult<Py<PyAny>> {
        let executor = get_graph_executor(&self.name).ok_or_else(|| {
            PyValueError::new_err(format!(
                "graph '{}' not built yet — call execute after the 'with' block exits",
                self.name
            ))
        })?;

        // Convert Python dict inputs → HashMap<String, PyObject>
        let mut cache: HashMap<String, Py<PyAny>> = HashMap::new();
        for (key, value) in inputs.iter() {
            let key_str: String = key.extract()?;
            cache.insert(key_str, value.unbind());
        }

        // Execute synchronously (this is a simplified path for tutorials)
        let result = executor.execute_sync(py, &cache)?;
        Ok(result)
    }

`class PythonGraphExecutor`

Binding

Fields

Name	Type	Description
`name`	`String`
`node_functions`	`HashMap<String, Py<PyAny>>`
`node_map`	`HashMap<String, PyNodeDecl>`
`execution_order`	`Vec<String>`
`react_mode`	`String`
`accumulators`	`Vec<String>`
`has_reactor`	`bool`	`true` when the graph is bound to a reactor. Trigger-less graphs (`false`) cannot be turned into a `ComputationGraphDeclaration`.
`reactor_name`	`Option<String>`	Reactor name from the `@cloaca.reactor`-decorated class. Threaded into the resulting `ComputationGraphDeclaration.reactor_name` so two packages declaring the same reactor share a single runtime instance (T-0544 M5 cross-package fan-out).

Methods

`execute_sync`

pub

fn execute_sync(
    &self,
    py: Python<'_>,
    inputs: &HashMap<String, Py<PyAny>>,
) -> PyResult<Py<PyAny>>

Execute the graph synchronously from Python with dict inputs.

Used by ComputationGraphBuilder.execute() for tutorials.

Source

    pub fn execute_sync(
        &self,
        py: Python<'_>,
        inputs: &HashMap<String, Py<PyAny>>,
    ) -> PyResult<Py<PyAny>> {
        // Convert PyObject inputs to serde_json::Value for the executor
        let mut cache_values: HashMap<String, serde_json::Value> = HashMap::new();
        for (name, obj) in inputs {
            let val = pythonize::depythonize::<serde_json::Value>(obj.bind(py))?;
            cache_values.insert(name.clone(), val);
        }

        let result = execute_graph_sync(
            py,
            &self.node_functions,
            &self.execution_order,
            &self.node_map,
            &cache_values,
        );

        match result {
            Ok(outputs) => {
                // Terminal results are Box<dyn Any + Send> containing serde_json::Value
                if let Some(last) = outputs.last() {
                    if let Some(json_val) = last.downcast_ref::<serde_json::Value>() {
                        let py_obj = pythonize::pythonize(py, json_val).map_err(|e| {
                            PyValueError::new_err(format!("result conversion failed: {}", e))
                        })?;
                        Ok(py_obj.unbind())
                    } else {
                        Ok(py.None())
                    }
                } else {
                    Ok(py.None())
                }
            }
            Err(e) => Err(PyValueError::new_err(format!(
                "graph execution failed: {}",
                e
            ))),
        }
    }

`execute`

pub async

async fn execute(
    &self,
    cache: &cloacina::computation_graph::types::InputCache,
) -> GraphResult

Execute the graph with the given input cache.

Source

    pub async fn execute(
        &self,
        cache: &cloacina::computation_graph::types::InputCache,
    ) -> GraphResult {
        let executor = self.clone();

        // Deserialize cache inputs. Boundary frames are bincode(Vec<u8>) of
        // raw JSON (passthrough events AND — since CLOACI-T-0842 — state
        // windows). The old `cache.get::<serde_json::Value>` could NEVER
        // decode either shape (`Value` can't deserialize from bincode —
        // deserialize_any), so node inputs were silently empty; decode the
        // real wire shape.
        let mut cache_values: HashMap<String, serde_json::Value> = HashMap::new();
        for acc_name in &executor.accumulators {
            if let Some(Ok(json_bytes)) = cache.get::<Vec<u8>>(acc_name) {
                if let Ok(val) = serde_json::from_slice::<serde_json::Value>(&json_bytes) {
                    cache_values.insert(acc_name.clone(), val);
                }
            }
        }

        let result = tokio::task::spawn_blocking(move || {
            Python::attach(|py| {
                execute_graph_sync(
                    py,
                    &executor.node_functions,
                    &executor.execution_order,
                    &executor.node_map,
                    &cache_values,
                )
            })
        })
        .await;

        match result {
            Ok(Ok(outputs)) => GraphResult::completed(outputs),
            Ok(Err(e)) => GraphResult::error(e),
            Err(join_err) => GraphResult::error(GraphError::NodeExecution(format!(
                "graph execution panicked: {}",
                join_err
            ))),
        }
    }

`terminal_names`

pub

fn terminal_names(&self) -> Vec<String>

Terminal node names in execution_order, in the order their outputs will be appended to the result vector. Used by @cloaca.task(invokes=...) to route outputs back into the task context under their terminal name.

Source

    pub fn terminal_names(&self) -> Vec<String> {
        self.execution_order
            .iter()
            .filter(|n| {
                matches!(
                    self.node_map.get(*n).map(|d| &d.edge),
                    Some(PyEdgeDecl::Terminal)
                )
            })
            .cloned()
            .collect()
    }

`execute_trigger_less`

pub async

async fn execute_trigger_less(
    &self,
    ctx: Py<PyAny>,
) -> Result<Vec<(String, serde_json::Value)>, GraphError>

Execute a trigger-less graph with a Python Context object.

Each “root” node — one with no upstream edges and no cache inputs — receives the context as its single argument. Downstream nodes get their predecessor’s output as before. Returns (terminal_name, value) pairs in the order each terminal actually fired — so routing branches that were skipped do not appear in the output. The task wrapper routes each pair into the task context under its terminal node name.

Source

    pub async fn execute_trigger_less(
        &self,
        ctx: Py<PyAny>,
    ) -> Result<Vec<(String, serde_json::Value)>, GraphError> {
        let executor = self.clone();
        match tokio::task::spawn_blocking(move || {
            Python::attach(|py| {
                execute_graph_sync_named_terminals(
                    py,
                    &executor.node_functions,
                    &executor.execution_order,
                    &executor.node_map,
                    &ctx,
                )
            })
        })
        .await
        {
            Ok(Ok(pairs)) => Ok(pairs),
            Ok(Err(e)) => Err(e),
            Err(join_err) => Err(GraphError::NodeExecution(format!(
                "trigger-less graph execution panicked: {}",
                join_err
            ))),
        }
    }

Functions

`fn get_registered_accumulators`

pub

fn get_registered_accumulators() -> Vec<PyAccumulatorRegistration>

Get the accumulators registered in the thread’s current scope (plus the unscoped ones, for embedded/test use). CLOACI-T-0921: a tenant-scoped caller no longer sees another tenant’s accumulator declarations.

Source

pub fn get_registered_accumulators() -> Vec<PyAccumulatorRegistration> {
    let scope = current_registration_scope();
    ACCUMULATOR_REGISTRY
        .lock()
        .unwrap()
        .iter()
        .filter(|(key, _)| {
            if scope.is_unscoped() {
                true
            } else {
                key.matches_scope(&scope) || (key.tenant_id.is_none() && key.package.is_none())
            }
        })
        .map(|(_, (_, reg))| reg.clone())
        .collect()
}

`fn get_all_registered_accumulators`

pub

fn get_all_registered_accumulators() -> Vec<(GraphKey, PyAccumulatorRegistration)>

Every registered accumulator, across all scopes — diagnostics only.

Source

pub fn get_all_registered_accumulators() -> Vec<(GraphKey, PyAccumulatorRegistration)> {
    ACCUMULATOR_REGISTRY
        .lock()
        .unwrap()
        .iter()
        .map(|(key, (_, reg))| (key.clone(), reg.clone()))
        .collect()
}

`fn drain_accumulators`

pub

fn drain_accumulators() -> HashMap<GraphKey, (Py<PyAny>, PyAccumulatorRegistration)>

Drain all registered accumulators (used by builder on exit / test reset).

Source

pub fn drain_accumulators() -> HashMap<GraphKey, (Py<PyAny>, PyAccumulatorRegistration)> {
    let mut registry = ACCUMULATOR_REGISTRY.lock().unwrap();
    std::mem::take(&mut *registry)
}

`fn passthrough_accumulator_decorator`

Binding

Python API: cloaca.passthrough_accumulator_decorator

fn passthrough_accumulator_decorator(
    py: Python<'_>,
    func: Py<PyAny>,
) -> PyResult<Py<PyAny>>

The @cloaca.passthrough_accumulator decorator. Registers a function as a passthrough accumulator (Event → Output, no buffering).

Source

pub fn passthrough_accumulator_decorator(py: Python<'_>, func: Py<PyAny>) -> PyResult<Py<PyAny>> {
    let func_name: String = func.getattr(py, "__name__")?.extract(py)?;
    let reg = PyAccumulatorRegistration {
        name: func_name.clone(),
        accumulator_type: "passthrough".to_string(),
        config: HashMap::new(),
        capacity: 0,
    };
    register_accumulator(func_name, func.clone_ref(py), reg);
    Ok(func)
}

`fn stream_accumulator_decorator`

Binding

Python API: cloaca.stream_accumulator_decorator

fn stream_accumulator_decorator(
    py: Python<'_>,
    r#type: String,
    topic: String,
    group: Option<String>,
) -> PyResult<Py<PyAny>>

Factory for @cloaca.stream_accumulator(type=..., topic=...). Returns a decorator function that registers the target with stream config.

Source

pub fn stream_accumulator_decorator(
    py: Python<'_>,
    r#type: String,
    topic: String,
    group: Option<String>,
) -> PyResult<Py<PyAny>> {
    let config_type = r#type;
    let config_topic = topic;
    let config_group = group;

    // Return a decorator function
    let decorator = PyCFunction::new_closure(
        py,
        None,
        None,
        move |args: &Bound<'_, PyTuple>,
              _kwargs: Option<&Bound<'_, PyDict>>|
              -> PyResult<Py<PyAny>> {
            let _py = args.py();
            let func = args.get_item(0)?;
            let func_name: String = func.getattr("__name__")?.extract()?;

            let mut config = HashMap::new();
            config.insert("type".to_string(), config_type.clone());
            config.insert("topic".to_string(), config_topic.clone());
            if let Some(ref g) = config_group {
                config.insert("group".to_string(), g.clone());
            }

            let reg = PyAccumulatorRegistration {
                name: func_name.clone(),
                accumulator_type: "stream".to_string(),
                config,
                capacity: 0,
            };
            register_accumulator(func_name, func.clone().unbind(), reg);
            Ok(func.clone().unbind())
        },
    )?;

    Ok(decorator.into())
}

`fn polling_accumulator_decorator`

Binding

Python API: cloaca.polling_accumulator_decorator

fn polling_accumulator_decorator(
    py: Python<'_>,
    interval: String,
) -> PyResult<Py<PyAny>>

Factory for @cloaca.polling_accumulator(interval=...).

Source

pub fn polling_accumulator_decorator(py: Python<'_>, interval: String) -> PyResult<Py<PyAny>> {
    let config_interval = interval;

    let decorator = PyCFunction::new_closure(
        py,
        None,
        None,
        move |args: &Bound<'_, PyTuple>,
              _kwargs: Option<&Bound<'_, PyDict>>|
              -> PyResult<Py<PyAny>> {
            let _py = args.py();
            let func = args.get_item(0)?;
            let func_name: String = func.getattr("__name__")?.extract()?;

            let mut config = HashMap::new();
            config.insert("interval".to_string(), config_interval.clone());

            let reg = PyAccumulatorRegistration {
                name: func_name.clone(),
                accumulator_type: "polling".to_string(),
                config,
                capacity: 0,
            };
            register_accumulator(func_name, func.clone().unbind(), reg);
            Ok(func.clone().unbind())
        },
    )?;

    Ok(decorator.into())
}

`fn batch_accumulator_decorator`

Binding

Python API: cloaca.batch_accumulator_decorator

fn batch_accumulator_decorator(
    py: Python<'_>,
    flush_interval: String,
    max_buffer_size: Option<usize>,
) -> PyResult<Py<PyAny>>

Factory for @cloaca.batch_accumulator(flush_interval=..., max_buffer_size=...).

Source

pub fn batch_accumulator_decorator(
    py: Python<'_>,
    flush_interval: String,
    max_buffer_size: Option<usize>,
) -> PyResult<Py<PyAny>> {
    let config_interval = flush_interval;
    let config_max = max_buffer_size;

    let decorator = PyCFunction::new_closure(
        py,
        None,
        None,
        move |args: &Bound<'_, PyTuple>,
              _kwargs: Option<&Bound<'_, PyDict>>|
              -> PyResult<Py<PyAny>> {
            let _py = args.py();
            let func = args.get_item(0)?;
            let func_name: String = func.getattr("__name__")?.extract()?;

            let mut config = HashMap::new();
            config.insert("flush_interval".to_string(), config_interval.clone());
            if let Some(max) = config_max {
                config.insert("max_buffer_size".to_string(), max.to_string());
            }

            let reg = PyAccumulatorRegistration {
                name: func_name.clone(),
                accumulator_type: "batch".to_string(),
                config,
                capacity: 0,
            };
            register_accumulator(func_name, func.clone().unbind(), reg);
            Ok(func.clone().unbind())
        },
    )?;

    Ok(decorator.into())
}

`fn state_accumulator_decorator`

Binding

Python API: cloaca.state_accumulator_decorator

fn state_accumulator_decorator(
    py: Python<'_>,
    capacity: i32,
) -> PyResult<Py<PyAny>>

Factory for @cloaca.state_accumulator(capacity=N).

Registers a function as a state accumulator: a bounded VecDeque that receives values, persists to the DAL on every write, and emits the full list back as the boundary (enabling cyclic feedback patterns). Mirrors Rust’s #[state_accumulator(capacity=…)].

Capacity semantics (see StateAccumulator in cloacina core):

> 0: bounded — evicts oldest when at capacity
< 0: unbounded — grows without limit
0: write-only sink — no history emitted back

Source

pub fn state_accumulator_decorator(py: Python<'_>, capacity: i32) -> PyResult<Py<PyAny>> {
    let config_capacity = capacity;

    let decorator = PyCFunction::new_closure(
        py,
        None,
        None,
        move |args: &Bound<'_, PyTuple>,
              _kwargs: Option<&Bound<'_, PyDict>>|
              -> PyResult<Py<PyAny>> {
            let _py = args.py();
            let func = args.get_item(0)?;
            let func_name: String = func.getattr("__name__")?.extract()?;

            // Mirror capacity into config too, so it survives the
            // String-keyed config map used at the FFI/override boundary.
            let mut config = HashMap::new();
            config.insert("capacity".to_string(), config_capacity.to_string());

            let reg = PyAccumulatorRegistration {
                name: func_name.clone(),
                accumulator_type: "state".to_string(),
                config,
                capacity: config_capacity,
            };
            register_accumulator(func_name, func.clone().unbind(), reg);
            Ok(func.clone().unbind())
        },
    )?;

    Ok(decorator.into())
}

`fn node`

Binding

Python API: cloaca.node

fn node(py: Python<'_>, func: Py<PyAny>) -> PyResult<Py<PyAny>>

The @cloaca.node decorator. Registers a function as a node in the current ComputationGraphBuilder context.

Source

pub fn node(py: Python<'_>, func: Py<PyAny>) -> PyResult<Py<PyAny>> {
    let _ctx = current_graph_context().ok_or_else(|| {
        PyValueError::new_err(
            "@cloaca.node must be used inside a ComputationGraphBuilder context manager",
        )
    })?;

    let func_name: String = func.getattr(py, "__name__")?.extract(py)?;
    register_node(func_name, func.clone_ref(py));

    // Return the function unchanged (transparent decorator)
    Ok(func)
}

`fn get_graph_executor`

pub

fn get_graph_executor(name: &str) -> Option<PythonGraphExecutor>

Get a registered graph executor by name, resolved within the thread’s current RegistrationScope (for testing / reactor / fleet-agent use).

Source

pub fn get_graph_executor(name: &str) -> Option<PythonGraphExecutor> {
    let registry = GRAPH_EXECUTORS.lock().unwrap();
    let key = resolve_graph_key(&registry, name)?;
    registry.get(&key).cloned()
}

`fn get_graph_executor_by_key`

pub

fn get_graph_executor_by_key(
    key: &GraphKey,
) -> Option<PythonGraphExecutor>

Get a registered graph executor by its full (tenant, package, name) key, ignoring the thread’s ambient scope.

Source

pub fn get_graph_executor_by_key(key: &GraphKey) -> Option<PythonGraphExecutor> {
    GRAPH_EXECUTORS.lock().unwrap().get(key).cloned()
}

`fn registered_graph_keys`

pub

fn registered_graph_keys() -> Vec<GraphKey>

Every registered graph executor key — diagnostics / tests.

Source

pub fn registered_graph_keys() -> Vec<GraphKey> {
    GRAPH_EXECUTORS.lock().unwrap().keys().cloned().collect()
}

`fn get_graph_executors_for_reactor`

pub

fn get_graph_executors_for_reactor(
    reactor: &str,
) -> Vec<PythonGraphExecutor>

All registered graph executors subscribed to reactor (CLOACI-T-0841). A reactor firing fans out to its subscriber graphs; the fleet agent receives the REACTOR name in the packet and uses this to find the graph(s) to execute — the agent-side analog of the scheduler’s subscriber dispatcher.

Source

pub fn get_graph_executors_for_reactor(reactor: &str) -> Vec<PythonGraphExecutor> {
    // CLOACI-T-0921: when a scope is installed, only that tenant's subscriber
    // graphs (plus unscoped ones) fan out. The fleet agent dispatches with no
    // scope installed and still sees everything it hosts.
    let scope = current_registration_scope();
    GRAPH_EXECUTORS
        .lock()
        .unwrap()
        .iter()
        .filter(|(key, _)| {
            scope.tenant_id.is_none() || key.same_tenant(&scope) || key.tenant_id.is_none()
        })
        .map(|(_, ex)| ex)
        .filter(|ex| ex.has_reactor && ex.reactor_name.as_deref() == Some(reactor))
        .cloned()
        .collect()
}

`fn build_python_graph_declaration`

pub

fn build_python_graph_declaration(
    graph_name: &str,
    tenant_id: Option<String>,
    accumulator_overrides: &[cloacina_workflow_plugin::types::AccumulatorConfig],
) -> Option<cloacina::computation_graph::scheduler::ComputationGraphDeclaration>

Build a ComputationGraphDeclaration from a registered Python graph executor.

This bridges the Python CG world (decorators + ComputationGraphBuilder) into the Rust ComputationGraphScheduler by wrapping the Python executor in a CompiledGraphFn.

Source

pub fn build_python_graph_declaration(
    graph_name: &str,
    tenant_id: Option<String>,
    accumulator_overrides: &[cloacina_workflow_plugin::types::AccumulatorConfig],
) -> Option<cloacina::computation_graph::scheduler::ComputationGraphDeclaration> {
    use cloacina::computation_graph::packaging_bridge::{
        PassthroughAccumulatorFactory, ProviderStreamAccumulatorFactory, StateAccumulatorFactory,
    };
    use cloacina::computation_graph::reactor::{InputStrategy, ReactionCriteria};
    use cloacina::computation_graph::scheduler::{
        AccumulatorDeclaration, ComputationGraphDeclaration, ReactorDeclaration,
    };
    use cloacina_computation_graph::InputCache;
    use std::sync::Arc;

    let executor = get_graph_executor(graph_name)?;

    // Trigger-less graphs aren't reactor-driven and therefore have no
    // ComputationGraphDeclaration to publish. Callers that need the graph
    // (e.g. a `@cloaca.task(invokes=...)` body) look it up through
    // `get_graph_executor` directly.
    if !executor.has_reactor {
        return None;
    }

    let criteria = match executor.react_mode.as_str() {
        "when_all" => ReactionCriteria::WhenAll,
        _ => ReactionCriteria::WhenAny,
    };

    let accumulator_names = executor.accumulators.clone();
    let reactor_name = executor.reactor_name.clone();

    // Capture the node/edge topology before `executor` is moved into the
    // graph_fn closure, so the health API can render this Python CG's DAG
    // the same way Rust CGs do. (CLOACI-T-0673)
    let topology = Some(python_graph_topology_json(&executor));

    // Build CompiledGraphFn from the Python executor
    let graph_fn: cloacina_computation_graph::CompiledGraphFn =
        Arc::new(move |cache: InputCache| {
            let exec = executor.clone();
            Box::pin(async move { exec.execute(&cache).await })
        });

    // Build accumulator declarations — use package.toml overrides when available
    let accumulators = accumulator_names
        .iter()
        .map(|name| {
            let factory: Arc<dyn cloacina::computation_graph::scheduler::AccumulatorFactory> =
                if let Some(override_cfg) = accumulator_overrides.iter().find(|a| a.name == *name) {
                    match override_cfg.accumulator_type.as_str() {
                        // CLOACI-T-0898: stream sources come from a bundled
                        // constructor provider (`provider`/`constructor` config
                        // keys); a declaration without one fails loudly at spawn.
                        "stream" => Arc::new(ProviderStreamAccumulatorFactory::new(
                            override_cfg.config.clone(),
                        )),
                        "state" => Arc::new(StateAccumulatorFactory::new(
                            override_cfg
                                .config
                                .get("capacity")
                                .and_then(|c| c.parse::<i32>().ok())
                                .unwrap_or(0),
                        )),
                        _ => Arc::new(PassthroughAccumulatorFactory),
                    }
                } else {
                    Arc::new(PassthroughAccumulatorFactory)
                };
            AccumulatorDeclaration {
                name: name.clone(),
                factory,
            }
        })
        .collect();

    Some(ComputationGraphDeclaration {
        name: graph_name.to_string(),
        accumulators,
        reactor: ReactorDeclaration {
            criteria,
            strategy: InputStrategy::Latest,
            graph_fn,
            // CLOACI-T-0830: the Python CG path doesn't author WASM reactor
            // constructors — native dirty-flag firing only.
            constructor: None,
        },
        tenant_id,
        // T-0544 M5: thread the reactor name from the `@cloaca.reactor`
        // class through to the scheduler so Python packages naming the
        // same reactor as a Rust (or Python) package share a runtime
        // instance via M2's idempotent registration path.
        reactor_name,
        topology,
    })
}

cloacina-python::computation_graph

Structs

struct PyAccumulatorRegistration

Fields

struct GraphKey

Fields

Methods

new

current

describe

class ComputationGraphBuilder

Fields

Methods

new

NAME

_​_​enter_​_​

_​_​exit_​_​

_​_​repr_​_​

execute

class PythonGraphExecutor

Fields

Methods

execute_​sync

execute

terminal_​names

execute_​trigger_​less

Functions

fn get_​registered_​accumulators

fn get_​all_​registered_​accumulators

fn drain_​accumulators

fn passthrough_​accumulator_​decorator

fn stream_​accumulator_​decorator

fn polling_​accumulator_​decorator

fn batch_​accumulator_​decorator

fn state_​accumulator_​decorator

fn node

fn get_​graph_​executor

fn get_​graph_​executor_​by_​key

fn registered_​graph_​keys

fn get_​graph_​executors_​for_​reactor

fn build_​python_​graph_​declaration

`struct PyAccumulatorRegistration`

`struct GraphKey`

`new`

`current`

`describe`

`class ComputationGraphBuilder`

`new`

`NAME`

`enter`

`exit`

`repr`

`execute`

`class PythonGraphExecutor`

`execute_sync`

`execute`

`terminal_names`

`execute_trigger_less`

`fn get_registered_accumulators`

`fn get_all_registered_accumulators`

`fn drain_accumulators`

`fn passthrough_accumulator_decorator`

`fn stream_accumulator_decorator`

`fn polling_accumulator_decorator`

`fn batch_accumulator_decorator`

`fn state_accumulator_decorator`

`fn node`

`fn get_graph_executor`

`fn get_graph_executor_by_key`

`fn registered_graph_keys`

`fn get_graph_executors_for_reactor`

`fn build_python_graph_declaration`