haive.core.contracts.node_contracts

Node contracts for graph execution.

This module defines contracts for graph nodes, ensuring explicit declaration of behavior and dependencies.

Exceptions

ContractViolation

Exception raised when contract is violated.

Classes

ContractualNode	Node that declares and enforces its contract.
NodeChain	Chain of contractual nodes with dependency validation.
NodeContract	Contract for a graph node.

Module Contents

exception haive.core.contracts.node_contracts.ContractViolation(violation)

Bases: Exception

Exception raised when contract is violated.

violation

Dictionary containing violation details

Initialize with violation details.

Parameters:

violation (Dict[str, Any]) – Violation information

class haive.core.contracts.node_contracts.ContractualNode(name, contract, execute_fn)

Node that declares and enforces its contract.

This ensures nodes explicitly declare their behavior, making graph composition predictable and debuggable.

name

Node identifier

contract

Node’s contract specification

execute_fn

Function to execute

_execution_count

Number of times executed

_contract_violations

List of contract violations

_execution_history

History of executions

Examples

>>> # Create validation node
>>> def validate(state_view):
...     response = state_view.get("response")
...     # Validation logic
...     return {"validated_response": response, "score": 0.95}
>>>
>>> node = ContractualNode(
...     name="validator",
...     contract=validation_contract,
...     execute_fn=validate
... )
>>>
>>> # Execute with state view
>>> result = node(state_view)

Initialize contractual node.

Parameters:

• name (str) – Node identifier

• contract (NodeContract) – Node’s contract

• execute_fn (Callable[[haive.core.contracts.boundaries.StateView], Dict[str, Any]]) – Function to execute with StateView

get_contract_summary()

Get human-readable contract summary.

Returns:

Contract details and execution statistics

Return type:

Dict[str, Any]

reset_statistics()

Reset execution statistics.

Return type:

None

class haive.core.contracts.node_contracts.NodeChain

Chain of contractual nodes with dependency validation.

Ensures nodes are executed in valid order with all dependencies met.

nodes

Dictionary of nodes by name

execution_order

Order to execute nodes

_executed

Set of executed node names

Examples

>>> # Create node chain
>>> chain = NodeChain()
>>> chain.add_node(llm_node)
>>> chain.add_node(validation_node)
>>> chain.add_node(output_node)
>>>
>>> # Validate and execute
>>> issues = chain.validate_chain()
>>> if not issues:
...     result = chain.execute(state_view)

Initialize empty node chain.

add_node(node)

Add node to chain.

Parameters:

node (ContractualNode) – Node to add

Return type:

None

execute(state_view)

Execute the node chain.

Parameters:

state_view (haive.core.contracts.boundaries.StateView) – State view for execution

Returns:

Combined results from all nodes

Raises:

ContractViolation – If any node violates contract

Return type:

Dict[str, Any]

get_execution_plan()

Get execution plan showing order and dependencies.

Returns:

List of execution steps

Return type:

List[Dict[str, Any]]

validate_chain()

Validate the node chain for consistency.

Returns:

List of validation issues

Return type:

List[str]

class haive.core.contracts.node_contracts.NodeContract(/, **data)

Bases: pydantic.BaseModel

Contract for a graph node.

Defines what a node requires, produces, and guarantees during execution in a graph workflow.

Parameters:

data (Any)

inputs

Required input fields from state

outputs

Produced output fields to state

transforms

How fields are transformed (source -> target)

dependencies

Other nodes this depends on

guarantees

What this node guarantees after execution

error_handling

How errors are handled

can_retry

Whether node can be retried on failure

Examples

>>> # Validation node contract
>>> validation_contract = NodeContract(
...     inputs=["response", "criteria"],
...     outputs=["validated_response", "validation_score"],
...     transforms={"response": "validated_response"},
...     guarantees=["validation_score between 0 and 1"]
... )
>>>
>>> # Tool node contract
>>> tool_contract = NodeContract(
...     inputs=["tool_calls", "tools"],
...     outputs=["tool_results"],
...     dependencies=["llm_node"],
...     can_retry=True
... )

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

get_all_fields()

Get all fields this node interacts with.

Returns:

Dictionary mapping field to interaction type

Return type:

Dict[str, str]

validate_dependencies(executed_nodes)

Check if dependencies have been executed.

Parameters:

executed_nodes (List[str]) – List of already executed node names

Returns:

List of missing dependencies

Return type:

List[str]