Error Handling

All ContextUnity services use a unified exception hierarchy rooted in ContextUnityError. This ensures consistent error codes, gRPC status mapping, and protocol-level error propagation across the service mesh.

Exception Hierarchy

Exception
└── ContextUnityError (INTERNAL_ERROR)
    ├── ConfigurationError (CONFIGURATION_ERROR)
    │   ├── RedisNotAvailable (REDIS_NOT_AVAILABLE)
    │   └── ServiceStartupError (SERVICE_STARTUP_ERROR)
    ├── RetrievalError (RETRIEVAL_ERROR)
    ├── IntentDetectionError (INTENT_ERROR)
    ├── ProviderError (PROVIDER_ERROR)
    │   └── StorageError (STORAGE_ERROR)
    │       ├── DatabaseConnectionError (DB_CONNECTION_ERROR)
    │       └── RedisConnectionError (REDIS_CONNECTION_ERROR)
    ├── SecurityError (SECURITY_ERROR)
    │   └── TamperDetectedError (TAMPER_DETECTED)
    ├── ConnectorError (CONNECTOR_ERROR)
    ├── ModelError (MODEL_ERROR)
    ├── IngestionError (INGESTION_ERROR)
    ├── GraphBuilderError (GRAPH_BUILDER_ERROR)
    ├── TransformerError (TRANSFORMER_ERROR)
    ├── LLMExecutionError (LLM_EXECUTION_ERROR)
    ├── FederatedToolTimeout (FEDERATED_TIMEOUT)
    └── PlatformServiceError (PLATFORM_SERVICE_ERROR)

Usage

from contextunity.core.exceptions import (
    ContextUnityError,
    ConfigurationError,
    SecurityError,
)

# Raise with default message
raise ConfigurationError()

# Raise with custom message + extra details
raise SecurityError(
    message="Token expired for tenant",
    tenant_id="my_app",
    expired_at="2026-01-01T00:00:00Z",
)

Every exception carries:

• code — stable error code string (e.g. "SECURITY_ERROR")
• message — human-readable description
• details — arbitrary keyword context dictionary

Error Codes Reference

Exception	Code	Description
ContextUnityError	INTERNAL_ERROR	Base — uncategorized platform error
ConfigurationError	CONFIGURATION_ERROR	Invalid or missing configuration
RetrievalError	RETRIEVAL_ERROR	Retrieval pipeline failure
IntentDetectionError	INTENT_ERROR	Intent classification failure
ProviderError	PROVIDER_ERROR	Storage/provider layer failure
SecurityError	SECURITY_ERROR	Token missing, invalid, or expired
TamperDetectedError	TAMPER_DETECTED	Prompt HMAC signature mismatch
ConnectorError	CONNECTOR_ERROR	Data connector failure
ModelError	MODEL_ERROR	LLM or embedding model failure
IngestionError	INGESTION_ERROR	Ingestion pipeline failure
GraphBuilderError	GRAPH_BUILDER_ERROR	Graph compilation failure
TransformerError	TRANSFORMER_ERROR	Data transformation failure
StorageError	STORAGE_ERROR	Database or storage operation failure
DatabaseConnectionError	DB_CONNECTION_ERROR	Database connection failure
LLMExecutionError	LLM_EXECUTION_ERROR	Model unavailable, rate-limited, or malformed response
FederatedToolTimeout	FEDERATED_TIMEOUT	Federated tool did not respond within deadline
PlatformServiceError	PLATFORM_SERVICE_ERROR	Cross-service call failure (Brain/Shield/Worker)
RedisNotAvailable	REDIS_NOT_AVAILABLE	Redis package not installed (gracefully degraded)
RedisConnectionError	REDIS_CONNECTION_ERROR	Redis server unreachable (includes TLS hint)
ServiceStartupError	SERVICE_STARTUP_ERROR	Service failed to start during platform bootstrap

gRPC Error Mapping

The @grpc_error_handler decorator automatically maps ContextUnityError to appropriate gRPC status codes:

from contextunity.core.grpc_errors import grpc_error_handler

class BrainService(brain_pb2_grpc.BrainServiceServicer):
    @grpc_error_handler
    async def Search(self, request, context):
        # If SecurityError is raised → grpc.StatusCode.PERMISSION_DENIED
        # If StorageError is raised → grpc.StatusCode.UNAVAILABLE
        ...

Error Code	gRPC Status
SECURITY_ERROR	PERMISSION_DENIED
TAMPER_DETECTED	ABORTED
CONFIGURATION_ERROR	FAILED_PRECONDITION
RETRIEVAL_ERROR	NOT_FOUND
PROVIDER_ERROR, STORAGE_ERROR, DB_CONNECTION_ERROR, CONNECTOR_ERROR	UNAVAILABLE
MODEL_ERROR, GRAPH_BUILDER_ERROR, INTENT_ERROR, LLM_EXECUTION_ERROR	INTERNAL
INGESTION_ERROR, TRANSFORMER_ERROR	INVALID_ARGUMENT
FEDERATED_TIMEOUT	DEADLINE_EXCEEDED
PLATFORM_SERVICE_ERROR	UNAVAILABLE

Two decorator variants exist:

Decorator	Use Case
@grpc_error_handler	Unary RPCs (async def method(request, context))
@grpc_stream_error_handler	Streaming RPCs (async def method(request, context) → AsyncIterator)

ErrorRegistry

The ErrorRegistry provides protocol-level error mapping — reconstruct exceptions from wire codes:

from contextunity.core.exceptions import error_registry

# Create exception from error code string (e.g., from gRPC trailing metadata)
exc = error_registry.from_code("SECURITY_ERROR", message="Token expired")
# → SecurityError("Token expired")

# List all registered codes
codes = error_registry.all_codes()

Service-Specific Subclasses

Services may define thin subclasses for domain-specific categorization:

# In services/brain/
class ContextbrainError(ContextUnityError):
    """Brain-specific error for domain context."""
    pass

Invariant

All exceptions in the platform must inherit from ContextUnityError. This is enforced by the @grpc_error_handler decorator — uncaught non-ContextUnityError exceptions are logged as INTERNAL errors with full tracebacks.

Module Layout

packages/core/src/contextunity/core/
├── exceptions.py    # Exception hierarchy + ErrorRegistry
└── grpc_errors.py   # @grpc_error_handler, @grpc_stream_error_handler, status mapping

Import rule: from contextunity.core.exceptions import ... for exception classes, from contextunity.core.grpc_errors import ... for gRPC decorators. Never import gRPC decorators from exceptions.py.