haive.core.contracts.boundaries

State boundaries with controlled access.

This module implements BoundedState and StateView to provide controlled access to state with explicit permissions, maintaining flexibility while adding runtime guarantees.

Classes

AccessPermissions	Define what fields a component can access.
BoundedState	State container with access boundaries.
StateView	Filtered view of state with access control.

Module Contents

class haive.core.contracts.boundaries.AccessPermissions(/, **data)

Bases: pydantic.BaseModel

Define what fields a component can access.

This provides fine-grained control over state access, allowing different components to have different levels of access to state fields.

Parameters:

data (Any)

readable

Fields component can read from state

writable

Fields component can write to state

append_only

Fields component can append to but not overwrite

compute_only

Fields component can derive from but not store

Examples

>>> # LLM engine permissions
>>> llm_permissions = AccessPermissions(
...     readable={"messages", "temperature"},
...     writable={"response"},
...     append_only={"conversation_history"}
... )
>>>
>>> # Tool node permissions
>>> tool_permissions = AccessPermissions(
...     readable={"tool_calls", "tools"},
...     writable={"tool_results"},
...     compute_only={"context"}
... )

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.

merge(other)

Merge with another set of permissions.

Parameters:

other (AccessPermissions) – Other permissions to merge with

Returns:

New AccessPermissions with combined permissions

Return type:

AccessPermissions

validate_consistency()

Check for permission conflicts.

Returns:

List of consistency issues found

Return type:

List[str]

class haive.core.contracts.boundaries.BoundedState(initial_data=None)

State container with access boundaries.

This maintains the actual state and provides controlled views to different components based on their access permissions. Supports versioning, checkpointing, and rollback.

_data

The actual state data

_access_rules

Permission rules for each component

_version

Current state version

_history

Checkpoint history for rollback

_global_access_log

Consolidated access log

Initialize bounded state.

Parameters:

initial_data (Optional[Dict[str, Any]]) – Initial state data

checkpoint(description='')

Create checkpoint in history.

Parameters:

description (str) – Optional checkpoint description

Returns:

Version number of checkpoint

Return type:

int

Examples

>>> version = state.checkpoint("Before processing")
>>> # Make changes...
>>> state.rollback(version)  # Restore to checkpoint

get_access_summary()

Get summary of access patterns.

Returns:

Summary statistics of field access

Return type:

Dict[str, Any]

get_history()

Get checkpoint history.

Returns:

List of checkpoint summaries (without full state)

Return type:

List[Dict[str, Any]]

get_view_for(component_name)

Get filtered state view for component.

Parameters:

component_name (str) – Name of component requesting view

Returns:

StateView with appropriate permissions

Raises:

ValueError – If component not registered

Return type:

StateView

Examples

>>> llm_view = state.get_view_for("llm")
>>> messages = llm_view.get("messages")
>>> llm_view.set("response", "Generated response")

merge_access_logs()

Merge access logs from all active views.

Returns:

Consolidated access log sorted by timestamp

Return type:

List[Dict[str, Any]]

register_component(name, permissions)

Register component with access permissions.

Parameters:

• name (str) – Component identifier

• permissions (AccessPermissions) – Access permissions for component

Raises:

ValueError – If permissions have conflicts

Return type:

None

Examples

>>> state = BoundedState()
>>> state.register_component("llm", llm_permissions)
>>> state.register_component("tools", tool_permissions)

rollback(version)

Rollback to previous version.

Parameters:

version (int) – Version number to rollback to

Raises:

ValueError – If version not found

Return type:

None

Examples

>>> state.rollback(3)  # Rollback to version 3
>>> state.rollback(state._version - 1)  # Rollback one version

snapshot()

Get immutable snapshot of current state.

Returns:

Deep copy of current state

Return type:

Dict[str, Any]

Examples

>>> snapshot = state.snapshot()
>>> # Modifications to snapshot don't affect state
>>> snapshot["temp"] = "value"
>>> assert "temp" not in state._data

update_permissions(component_name, permissions)

Update permissions for existing component.

Parameters:

• component_name (str) – Component to update

• permissions (AccessPermissions) – New permissions

Raises:

ValueError – If component not registered

Return type:

None

validate_state_consistency()

Check state consistency across components.

Returns:

List of consistency issues found

Return type:

List[str]

class haive.core.contracts.boundaries.StateView(state, permissions, component_name='unknown')

Filtered view of state with access control.

This provides a component with controlled access to state, enforcing permissions at runtime while maintaining flexibility. Each component gets its own view with specific permissions.

_state

Reference to actual state (not copied)

_permissions

Access permissions for this view

_access_log

Log of all access attempts

_component_name

Name of component using this view

Initialize state view with permissions.

Parameters:

• state (Dict[str, Any]) – Reference to actual state (not copied)

• permissions (AccessPermissions) – Access permissions for this view

• component_name (str) – Name of component using this view

append(field, item)

Append to list field with permission check.

Parameters:

• field (str) – Field name containing list

• item (Any) – Item to append

Raises:

• PermissionError – If field not appendable

• TypeError – If field is not a list

Return type:

None

Examples

>>> view.append("history", {"event": "processed"})
>>> view.append("immutable_list", "item")  # Raises PermissionError

compute_from(fields)

Get values for computation without storage permission.

Parameters:

fields (List[str]) – Fields to retrieve for computation

Returns:

Dictionary of field values (deep copied)

Raises:

PermissionError – If any field not compute_only

Return type:

Dict[str, Any]

Examples

>>> # Get fields for derived computation
>>> data = view.compute_from(["embeddings", "weights"])
>>> result = complex_computation(data)
>>> # Cannot store result back to those fields

get(field, default=None)

Get field value with permission check.

Parameters:

• field (str) – Field name to retrieve

• default (Any) – Default value if field not found or not accessible

Returns:

Field value or default (deep copied for safety)

Raises:

PermissionError – If field not readable

Return type:

Any

Examples

>>> view.get("messages")  # Returns deep copy
>>> view.get("private_field")  # Raises PermissionError
>>> view.get("optional", [])  # Returns default if not found

get_access_log()

Get access log for this view.

Returns:

List of access log entries

Return type:

List[Dict[str, Any]]

get_accessible_fields()

Get all accessible fields by operation type.

Returns:

Dictionary mapping operation to field sets

Return type:

Dict[str, Set[str]]

get_mutations()

Get list of mutations made through this view.

Returns:

List of mutation records

Return type:

List[Dict[str, Any]]

has_permission(operation, field)

Check if operation is permitted on field.

Parameters:

• operation (str) – Operation type (read, write, append, compute)

• field (str) – Field name

Returns:

True if operation is permitted

Return type:

bool

set(field, value)

Set field value with permission check.

Parameters:

• field (str) – Field name to set

• value (Any) – Value to set

Raises:

PermissionError – If field not writable

Return type:

None

Examples

>>> view.set("response", "Hello")  # If writable
>>> view.set("readonly", "value")  # Raises PermissionError