haive.core.contracts.orchestrator

Orchestrator for contract enforcement.

This module provides the central orchestrator that ensures all contracts are respected during execution, providing the control layer over the dynamic runtime system.

Classes

Orchestrator

Orchestrates execution with contract enforcement.

Module Contents

class haive.core.contracts.orchestrator.Orchestrator

Orchestrates execution with contract enforcement.

This is the central coordinator that ensures all contracts are respected during execution, providing the control layer over the dynamic runtime system.

components

Registered components (engines and nodes)

contracts

Contracts for each component

access_rules

Derived access permissions

execution_log

Log of all executions

_dependency_graph

Component dependency graph

Examples

>>> # Create orchestrator
>>> orchestrator = Orchestrator()
>>>
>>> # Register components
>>> orchestrator.register_engine("llm", llm_engine)
>>> orchestrator.register_node(validation_node)
>>>
>>> # Execute with contract enforcement
>>> state = BoundedState(initial_data)
>>> result = orchestrator.execute("llm", state)

Initialize orchestrator.

execute(component_name, state, validate_only=False)

Execute component with full contract enforcement.

Parameters:

• component_name (str) – Component to execute

• state (haive.core.contracts.boundaries.BoundedState) – Bounded state container

• validate_only (bool) – Only validate, don’t execute

Returns:

Execution result

Raises:

• ContractViolation – If any contract violated

• ValueError – If component not found

Return type:

Any

Examples

>>> # Execute with enforcement
>>> result = orchestrator.execute("llm", state)
>>>
>>> # Validate only
>>> orchestrator.execute("llm", state, validate_only=True)

execute_chain(component_names, state, stop_on_error=True)

Execute chain of components in sequence.

Parameters:

• component_names (List[str]) – Components to execute in order

• state (haive.core.contracts.boundaries.BoundedState) – Bounded state container

• stop_on_error (bool) – Whether to stop on first error

Returns:

Combined results from all components

Return type:

Dict[str, Any]

Examples

>>> # Execute pipeline
>>> results = orchestrator.execute_chain(
...     ["retriever", "llm", "validator"],
...     state
... )

get_execution_plan(components)

Get execution plan for components.

Parameters:

components (List[str]) – Component names

Returns:

Execution plan with details

Return type:

List[Dict[str, Any]]

get_execution_summary()

Get summary of all executions.

Returns:

Execution statistics

Return type:

Dict[str, Any]

register_engine(name, engine, auto_register_state=True)

Register engine with its contract.

Parameters:

• name (str) – Engine identifier

• engine (haive.core.contracts.engine_contracts.EngineInterface) – Engine implementing EngineInterface

• auto_register_state (bool) – Whether to auto-register with state

Return type:

None

Examples

>>> orchestrator.register_engine("llm", ContractualLLMEngine())
>>> orchestrator.register_engine("retriever", retriever_engine)

register_node(node, auto_register_state=True)

Register node with its contract.

Parameters:

• node (haive.core.contracts.node_contracts.ContractualNode) – Node with contract

• auto_register_state (bool) – Whether to auto-register with state

Return type:

None

Examples

>>> validation_node = ContractualNode("validator", contract, validate_fn)
>>> orchestrator.register_node(validation_node)

validate_composition(components)

Validate that components can be composed.

Parameters:

components (List[str]) – List of component names in execution order

Returns:

List of compatibility issues

Return type:

List[str]

Examples

>>> issues = orchestrator.validate_composition(["llm", "validator"])
>>> if issues:
...     print(f"Issues found: {issues}")