haive.core.contracts.engine_contracts¶

Engine contracts for explicit behavior specification.

This module defines contracts that engines must implement to make their behavior explicit and verifiable at runtime.

Classes¶

ContractAdapter

Base adapter for adding contracts to existing engines.

EngineContract

Complete contract for an engine.

EngineInterface

Interface all engines must implement for contracts.

FieldContract

Contract for a single field.

Module Contents¶

class haive.core.contracts.engine_contracts.ContractAdapter(engine)[source]¶

Bases: EngineInterface

Base adapter for adding contracts to existing engines.

This provides a base implementation for adapting existing engines to support contracts without modifying the original implementation.

engine¶

The wrapped engine

contract¶

The engine’s contract

Examples

>>> # Adapt existing engine
>>> class MyEngineAdapter(ContractAdapter):
...     def build_contract(self):
...         return EngineContract(...)
...
...     def execute(self, state):
...         # Adapt state and call engine
...         return self.engine.invoke(state)

Initialize adapter with existing engine.

Parameters:

engine (Any) – Engine to wrap with contracts

abstractmethod build_contract()[source]¶

Build contract for the wrapped engine.

Returns:

Engine contract specification

Return type:

EngineContract

get_contract()[source]¶

Get engine’s contract.

Returns:

Engine contract

Return type:

EngineContract

log_violation(phase, details)[source]¶

Log a contract violation.

Parameters:
  • phase (str) – Phase where violation occurred

  • details (str) – Violation details

Return type:

None

validate_input(state)[source]¶

Validate input against contract.

Parameters:

state (Dict[str, Any]) – Input state

Returns:

True if valid

Return type:

bool

validate_output(result)[source]¶

Validate output against contract.

Parameters:

result (Any) – Execution result

Returns:

True if valid

Return type:

bool

class haive.core.contracts.engine_contracts.EngineContract(/, **data)[source]¶

Bases: pydantic.BaseModel

Complete contract for an engine.

Defines the full contract for an engine including inputs, outputs, side effects, and conditions that must hold before and after execution.

Parameters:

data (Any)

inputs¶

Input field contracts

outputs¶

Output field contracts

side_effects¶

Fields modified as side effects

preconditions¶

Conditions that must be true before execution

postconditions¶

Conditions guaranteed after execution

error_handling¶

How errors are handled

performance¶

Performance characteristics

Examples

>>> # LLM engine contract
>>> llm_contract = EngineContract(
...     inputs=[
...         FieldContract(name="messages", field_type=list, required=True),
...         FieldContract(name="temperature", field_type=float, required=False)
...     ],
...     outputs=[
...         FieldContract(name="response", field_type=str, required=True)
...     ],
...     side_effects=["conversation_history"],
...     preconditions=["len(messages) > 0"],
...     postconditions=["response is not empty"]
... )

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_guaranteed_outputs()[source]¶

Get list of guaranteed output field names.

Returns:

List of guaranteed output field names

Return type:

List[str]

get_optional_inputs()[source]¶

Get optional inputs with defaults.

Returns:

Dictionary of optional field names to defaults

Return type:

Dict[str, Any]

get_required_inputs()[source]¶

Get list of required input field names.

Returns:

List of required field names

Return type:

List[str]

validate_inputs(state)[source]¶

Validate state against input contracts.

Parameters:

state (Dict[str, Any]) – State to validate

Returns:

(is_valid, list_of_issues)

Return type:

tuple[bool, List[str]]

validate_outputs(result)[source]¶

Validate result against output contracts.

Parameters:

result (Dict[str, Any]) – Result to validate

Returns:

(is_valid, list_of_issues)

Return type:

tuple[bool, List[str]]

class haive.core.contracts.engine_contracts.EngineInterface[source]¶

Bases: abc.ABC

Interface all engines must implement for contracts.

This ensures every engine explicitly declares its contract, making dependencies and effects clear at runtime.

get_contract()[source]¶

Return the engine’s contract

Return type:

EngineContract

validate_input()[source]¶

Check if state is valid for execution

Parameters:

state (Dict[str, Any])

Return type:

bool

validate_output()[source]¶

Check if output meets contract

Parameters:

result (Any)

Return type:

bool

execute()[source]¶

Execute the engine with contract enforcement

Parameters:

state (Dict[str, Any])

Return type:

Dict[str, Any]

check_postconditions(state)[source]¶

Check which postconditions are not met.

Parameters:

state (Dict[str, Any]) – State after execution

Returns:

List of unmet postconditions

Return type:

List[str]

check_preconditions(state)[source]¶

Check which preconditions are not met.

Parameters:

state (Dict[str, Any]) – Current state

Returns:

List of unmet preconditions

Return type:

List[str]

abstractmethod execute(state)[source]¶

Execute engine with state.

Parameters:

state (Dict[str, Any]) – Input state

Returns:

Execution result

Raises:

ContractViolation – If contract is violated

Return type:

Dict[str, Any]

abstractmethod get_contract()[source]¶

Get engine’s contract.

Returns:

Complete contract specification

Return type:

EngineContract

Examples

>>> contract = engine.get_contract()
>>> print(f"Required inputs: {contract.get_required_inputs()}")
>>> print(f"Guaranteed outputs: {contract.get_guaranteed_outputs()}")
get_contract_summary()[source]¶

Get human-readable contract summary.

Returns:

Contract summary with key information

Return type:

Dict[str, Any]

abstractmethod validate_input(state)[source]¶

Validate state meets input requirements.

Parameters:

state (Dict[str, Any]) – Current state

Returns:

True if state is valid for execution

Return type:

bool

abstractmethod validate_output(result)[source]¶

Validate output meets contract.

Parameters:

result (Any) – Execution result

Returns:

True if output is valid

Return type:

bool

class haive.core.contracts.engine_contracts.FieldContract(/, **data)[source]¶

Bases: pydantic.BaseModel

Contract for a single field.

Defines the contract for an individual field including its type, requirements, validation, and documentation.

Parameters:

data (Any)

name¶

Field identifier

field_type¶

Expected type for the field

required¶

Whether field must be present

default¶

Default value if not required

description¶

Human-readable field purpose

validator¶

Optional validation function

examples¶

Example values for documentation

Examples

>>> # Define a messages field contract
>>> messages_field = FieldContract(
...     name="messages",
...     field_type=list,
...     required=True,
...     description="List of conversation messages",
...     validator=lambda x: len(x) > 0
... )
>>>
>>> # Define optional temperature field
>>> temp_field = FieldContract(
...     name="temperature",
...     field_type=float,
...     required=False,
...     default=0.7,
...     validator=lambda x: 0 <= x <= 2
... )

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.

validate_value(value)[source]¶

Validate a value against this field contract.

Parameters:

value (Any) – Value to validate

Returns:

(is_valid, error_message)

Return type:

tuple[bool, Optional[str]]