cleveragents-core/docs/api/core.md

cleveragents.core — Core Utilities

The core package provides the exception hierarchy, error classification, retry patterns, circuit breaker, and async resource cleanup used throughout the entire CleverAgents platform.

---

Domain Base Model

Module: cleveragents.domain.models.base

DomainBaseModel

from cleveragents.domain.models.base import DomainBaseModel

class DomainBaseModel(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,
        validate_assignment=True,
        arbitrary_types_allowed=False,
        populate_by_name=True,
        use_enum_values=True,
    )

Shared Pydantic base class for all standard domain-layer models. Inherit from DomainBaseModel instead of pydantic.BaseModel directly to get the canonical domain-layer configuration in one place.

Configuration semantics:

Setting	Effect
str_strip_whitespace	Leading/trailing whitespace stripped from all str fields on assignment and validation
validate_assignment	Field assignments after construction are validated like constructor arguments
arbitrary_types_allowed=False	All field types must be Pydantic-compatible — keeps the domain layer clean
populate_by_name	Models can be constructed using either the Python field name or the JSON alias
use_enum_values	Enum fields are stored and serialised as their underlying primitive values

Usage:

from cleveragents.domain.models.base import DomainBaseModel
from pydantic import Field

class MyDomainModel(DomainBaseModel):
    name: str
    count: int = Field(ge=0)

Note: This class was introduced in v3.7.0 (PR #1941) to eliminate the duplicated model_config that previously appeared in 14 separate domain model files. It is a pure structural refactor with no behavioral changes.

---

Exception Hierarchy

All exceptions inherit from CleverAgentsError. Catch the most specific type you can handle; let everything else propagate.

CleverAgentsError
├── DomainError
│   ├── ValidationError
│   ├── BusinessRuleViolation
│   │   ├── LockConflictError
│   │   └── DecisionPhaseViolationError
│   ├── ResourceNotFoundError  (alias: NotFoundError)
│   ├── ResourceConflictError
│   ├── LockExpiredError
│   └── PlanError
├── InfrastructureError
│   ├── DatabaseError
│   │   └── MigrationNotApprovedError
│   ├── NetworkError
│   └── ExternalServiceError
├── ProviderError
│   ├── RateLimitError
│   ├── ModelNotAvailableError
│   └── TokenLimitExceededError
├── AuthenticationError
├── AuthorizationError
├── ConfigurationError
│   └── MissingConfigurationError
├── FileSystemError
├── ExecutionError
├── StreamRoutingError
└── UnsafeConfigurationError

CleverAgentsError

class CleverAgentsError(Exception):
    message: str
    details: dict[str, Any]

Base class for all platform exceptions. Always carries a human-readable message and an optional details dict for structured context.

---

ResourceNotFoundError

class ResourceNotFoundError(DomainError):
    resource_type: str | None
    resource_id: str | None

Raised when a requested entity does not exist. If resource_type and resource_id are provided the message is auto-generated.

from cleveragents.core.exceptions import ResourceNotFoundError

raise ResourceNotFoundError(resource_type="plan", resource_id="plan-42")
# → "Plan 'plan-42' not found"

---

LockConflictError

class LockConflictError(BusinessRuleViolation):
    resource_type: str
    resource_id: str
    owner_id: str

Raised when an advisory lock cannot be acquired because another owner holds it.

---

RateLimitError

class RateLimitError(ProviderError):
    retry_after: int | None   # seconds to wait before retrying

---

DecisionPhaseViolationError

class DecisionPhaseViolationError(BusinessRuleViolation):
    decision_type: str
    plan_phase: str
    allowed_types: frozenset[str]

Raised when a decision type is incompatible with the plan's current phase.

---

Error Handling Utilities

Module: cleveragents.core.error_handling

ErrorCode

IntEnum mapping exception categories to HTTP-like numeric codes used for consistent CLI and API output.

classify_error(exc) → ErrorCode

Returns the ErrorCode for any exception, falling back to ErrorCode.INTERNAL for unknown types.

redact_error_details(details) → dict

Strips sensitive keys (tokens, passwords, secrets) from an error details dict before logging.

wrap_unexpected(exc, *, context=None) → CleverAgentsError

Wraps an unexpected exception in a CleverAgentsError with a safe user-facing message that hides internal stack details.

from cleveragents.core.error_handling import wrap_unexpected

try:
    risky_operation()
except Exception as exc:
    safe = wrap_unexpected(exc, context={"plan_id": plan_id})
    raise safe from exc

---

Retry Patterns

Module: cleveragents.core.retry_patterns

Provides decorators and helpers for exponential-backoff retry with jitter, configurable per exception type.

---

Async Cleanup

Module: cleveragents.core.async_cleanup

AsyncResourceTracker — unified async resource lifecycle with timeout-bounded cleanup, leak detection via finalizer, and async context manager support.

async with AsyncResourceTracker() as tracker:
    conn = await tracker.register(open_connection())
    # conn is automatically closed on exit, even on error

---

Circuit Breaker

Module: cleveragents.core.circuit_breaker

Implements the circuit-breaker pattern to prevent cascading failures when calling external services.

---

Invariant Service

Module: cleveragents.application.services.invariant_service

The InvariantService manages invariant constraints across scopes and is automatically invoked at every plan phase transition by the InvariantReconciliationActor.

InvariantService

from cleveragents.application.services.invariant_service import InvariantService
from cleveragents.domain.models.core.invariant import InvariantScope

service = InvariantService(event_bus=event_bus)
inv = service.add_invariant(
    text="All output files must be UTF-8 encoded",
    scope=InvariantScope.PROJECT,
    source_name="my-project",
)

Constructor parameters:

Parameter	Type	Default	Description
event_bus	EventBus | None	None	Optional event bus for emitting INVARIANT_VIOLATED / INVARIANT_ENFORCED / INVARIANT_RECONCILED events

Methods:

Method	Returns	Description
add_invariant(text, scope, source_name)	Invariant	Add a new invariant; validates and sanitizes text
list_invariants(scope, source_name, effective)	list[Invariant]	Filter invariants; pass effective=True to get the merged precedence chain
remove_invariant(invariant_id)	Invariant	Soft-delete an invariant (sets active=False)
get_effective_invariants(plan_id, project_name)	list[Invariant]	Return merged invariants using plan > project > global precedence
enforce_invariants(plan_id, invariants, actor_response, violated_invariant_ids)	list[InvariantEnforcementRecord]	Create enforcement records; emits INVARIANT_VIOLATED for each violation

Merge precedence: plan > project > global. Duplicate invariant texts are de-duplicated by merge_invariants().

Events emitted:

Event	When
INVARIANT_VIOLATED	An invariant in violated_invariant_ids was not satisfied
INVARIANT_ENFORCED	Each invariant enforcement record is created
INVARIANT_RECONCILED	Once per enforce_invariants() call (batch summary)

Note: The InvariantService is registered as a Singleton provider in the DI container. Obtain it via container.invariant_service() rather than constructing it directly.