FOREWORD: WHY ARCHITECTURE STILL MATTERS — FROM MICROCONTROLLERS TO THE CLOUD
Every decade or so, the software industry rediscovers a fundamental truth: the way you organize code determines everything. It determines how fast you can change things, how confidently you can deploy, how easily a new team member can find their footing, and how gracefully the system behaves when something goes wrong at three in the morning. This truth holds whether the system runs on a 64-core cloud server processing millions of transactions per second, or on a 32-kilobyte microcontroller managing the fuel injection timing of an engine.
Layered architectures gave us separation of concerns. Microservices gave us independent deployability. Hexagonal architecture gave us clean boundaries between core logic and the outside world. Event-driven architecture gave us loose coupling at the cost of traceability. AUTOSAR gave embedded systems a component model. OSEK/VDX gave real-time operating systems a standardized API. Each of these patterns solved real problems and introduced new ones.
Capability Centric Architecture, or CCA, takes a step back and asks a deceptively simple question: what can this system actually do? Not how is it structured, not which team owns which service, not which RTOS task owns which peripheral, but what are its fundamental units of capability? CCA 0.1 introduced the idea. CCA 0.2 formalized it with a rich conceptual model built around three concentric rings, evolution envelopes, contracts, policies, composition, and observability. CCA 0.3, which this article describes in full, takes the pattern to a new level of maturity.
Critically, CCA 0.3 is designed from the ground up to span the full spectrum of computing environments. It applies equally to a cloud-native microservice processing payment transactions, to a Cortex-M4 microcontroller reading temperature sensors on a factory floor, to a Raspberry Pi acting as an industrial gateway, and to the hybrid systems that combine all three. The same conceptual model — three rings, evolution envelope, lifecycle, contract, composition — applies at every scale. The implementation adapts to the constraints of the environment, but the architecture remains coherent.
CCA 0.3 preserves and deepens every concept from CCA 0.2 while adding capability lifecycle management, a distributed capability mesh, capability channels for event-driven communication, a formal security model based on capability tokens, a built-in testing framework, context-scoped execution, enhanced composition patterns including sagas and fan-out/fan-in, a formal dependency graph with cycle detection, SLO-aware observability, embedded systems support with hardware abstraction capabilities, and the first concrete steps toward AI-assisted capability discovery.
This is not a minor revision. CCA 0.3 is a substantial deepening of a pattern that was already mature, motivated by real-world experience with CCA 0.2 deployments across enterprise software, industrial control systems, and IoT edge devices, and by the growing demand for systems that are not just correct but also governable, auditable, and composable at scale — regardless of whether that scale is measured in cloud instances or in bytes of RAM.
CHAPTER 1: FOUNDATIONS — THE CCA 0.2 MODEL REVISITED
Before we can understand what is new in CCA 0.3, we must be crystal clear about what CCA 0.2 established. CCA 0.2 is not a simple pattern. It is a rich conceptual model with several interlocking ideas, and CCA 0.3 builds directly on all of them.
THE THREE-RING MODEL
The most distinctive structural idea in CCA 0.2 is the three-ring model. Every capability in CCA 0.2 is conceptually organized into three concentric rings: the Essence ring at the center, the Realization ring in the middle, and the Adaptation ring on the outside.
The Essence ring is the identity of the capability. It contains the capability's name, its version, its declared dependencies on other capabilities, and its contract. The Essence ring answers the question "what is this capability?" in the most abstract and stable terms possible. The Essence ring changes rarely, because it represents the fundamental identity of the capability. If the Essence changes significantly, it is usually a sign that a new capability should be created rather than the existing one modified.
The Realization ring is the implementation of the capability. It contains the actual code that performs the work the capability promises to do. The Realization ring is where business logic lives — or where firmware logic lives. It is hidden from callers behind the contract defined in the Essence ring. The Realization ring can change frequently as the implementation is optimized, corrected, or extended, as long as those changes do not violate the contract.
The Adaptation ring is the outermost layer. It contains the policies that govern how the capability behaves in its operational environment: retry policies, circuit breakers, rate limiters, timeout policies, and observability hooks. The Adaptation ring is what makes the capability production-ready. It is also the most environment-specific ring: the same Essence and Realization might be wrapped in different Adaptation rings in development, staging, production, or on a resource-constrained embedded device.
This three-ring model is not just a conceptual diagram. It has direct implications for how capabilities are structured in code, how they are tested, and how they evolve over time. A change to the Adaptation ring (for example, adjusting a retry count or a watchdog timeout) requires no testing of the Realization ring. A change to the Realization ring (for example, optimizing a sensor reading algorithm) requires testing the Realization but not the Essence. A change to the Essence ring (for example, adding a new field to the contract) requires careful versioning and potentially affects every caller.
The three-ring model in embedded systems: In a firmware context, the Essence ring of a temperature-reading capability declares its name ("temperature-sensor"), version, and contract (returns a temperature in degrees Celsius within a declared accuracy). The Realization ring contains the I2C register read sequence for the specific sensor IC. The Adaptation ring contains the retry logic for I2C bus errors, the watchdog kick, and the power management policy. When the sensor IC is replaced with a different model, only the Realization ring changes. The callers — a PID controller capability, a data logging capability, a safety monitor capability — are completely unaffected.
EVOLUTION ENVELOPES
The second major conceptual contribution of CCA 0.2 is the evolution envelope. An evolution envelope is the set of changes that a capability can undergo without breaking its callers. It is defined by the capability's contract and versioning policy.
CCA 0.2 establishes that every capability has an evolution envelope with two dimensions. The first dimension is the contract envelope: the set of changes to the contract that are backward-compatible. Adding an optional output field is inside the envelope; removing a required input field is outside it. The second dimension is the behavioral envelope: the set of changes to the implementation that preserve the observable behavior of the capability as seen by its callers.
CCA 0.3 extends the evolution envelope concept in two important ways. First, it makes the evolution envelope machine-checkable: the enhanced contract system, with its behavioral assertions and SLA declarations, provides a formal specification against which proposed changes can be automatically tested. Second, it adds a third dimension to the evolution envelope: the operational envelope, which is the set of changes to the Adaptation ring that are safe to make without redeploying the capability.
Evolution envelopes in embedded systems: In an automotive ECU, the evolution envelope of a fuel-injection capability is defined not just by its software contract but also by hardware timing constraints, safety integrity levels (SIL/ASIL), and certification boundaries. CCA 0.3's machine-checkable evolution envelope is directly applicable to embedded systems where the cost of a regression is not a failed test in CI but a field recall or a safety incident.
THE CAPABILITY AS A FIRST-CLASS CITIZEN
CCA 0.2 established the principle that a capability is a first-class citizen of the architecture, not a second-class implementation detail hidden inside a service or a module. This means that capabilities are named, versioned, discoverable, composable, and observable as a matter of architectural policy, not as an afterthought.
CCA 0.3 deepens this principle by adding lifecycle management, security governance, and formal dependency analysis as additional dimensions of first-class citizenship. In CCA 0.3, a capability is not just something the system can do; it is something the system can do, knows how to do, knows when it is ready to do, knows who is allowed to ask it to do, and knows how well it is doing it.
This principle applies with equal force in embedded systems. A GPIO driver, a UART transmitter, a CAN bus frame parser, a PID controller, a motor speed regulator — all of these are capabilities in the CCA sense. Making them explicit, versioned, contracted, and observable is exactly what distinguishes a maintainable embedded system from a tangled web of global variables and interrupt service routines.
WHAT IS A CAPABILITY?
With this background in place, we can state the CCA 0.3 definition of a capability with precision. A capability is a named, versioned, self-describing unit of functionality organized into three concentric rings (Essence, Realization, Adaptation), governed by an evolution envelope, managed through a formal lifecycle, secured by capability tokens, observable through structured metrics and SLO tracking, and composable with other capabilities through a set of well-defined composition patterns.
Think of a capability the way you think of a professional contractor. A plumber does not just show up and start working. The plumber has a license (the contract in the Essence ring), a set of skills and tools (the Realization ring), a set of professional standards and insurance policies (the Adaptation ring), a clear scope of work that defines what changes are within the agreed engagement (the evolution envelope), and a professional status that can be active, temporarily suspended, or retired (the lifecycle). CCA 0.3 formalizes exactly this model for software components — whether those components run in a Kubernetes pod or in the interrupt vector table of a Cortex-M microcontroller.
CHAPTER 2: THE CAPABILITY LIFECYCLE
One of the most important new concepts in CCA 0.3 is the formal capability lifecycle. In CCA 0.2, a capability was either registered or not. There was no model for what happens between the moment a capability is created and the moment it is ready to serve requests, nor for what happens when it starts to degrade or when it needs to be retired gracefully.
CCA 0.3 defines six lifecycle states. A capability begins in the CREATED state, which means it has been instantiated but has not yet been initialized or registered. From CREATED it transitions to INITIALIZING, during which it sets up its internal resources, validates its own configuration, and resolves its declared dependencies. If initialization succeeds, the capability moves to READY, which is the normal operating state. From READY it can transition to DEGRADED if its health checks begin to fail or if its SLO thresholds are being violated, and it can transition to SUSPENDED if an operator or the system itself decides to temporarily stop serving requests. From any of these active states it can transition to RETIRED, which is the terminal state indicating that the capability has been decommissioned and should no longer be used.
Lifecycle in embedded systems: The lifecycle maps naturally to embedded power and hardware states. CREATED corresponds to a peripheral's clock being enabled. INITIALIZING corresponds to the hardware initialization sequence (setting baud rates, configuring DMA, running self-test). READY corresponds to normal operation. DEGRADED corresponds to a sensor reporting out-of-range values or a communication bus experiencing errors. SUSPENDED corresponds to a peripheral being powered down in a low-power sleep mode. RETIRED corresponds to a hardware fault that requires a system reset or field service. The watchdog timer is a natural lifecycle observer: if a capability stays in INITIALIZING too long, the watchdog fires and the system resets.
All exceptions used by the lifecycle system are defined in a central exceptions.py module to avoid circular imports between the lifecycle, policy, and capability modules.
# cca/exceptions.py
# CCA 0.3 — Centralized exception hierarchy.
# Defined here to avoid circular imports between lifecycle, policies,
# capability, and dependency_graph modules.
from __future__ import annotations
from typing import List
class CCAError(Exception):
"""Base class for all CCA 0.3 framework exceptions."""
pass
class LifecycleError(CCAError):
"""Raised when an invalid lifecycle transition is attempted."""
pass
class CapabilitySecurityError(CCAError):
"""Raised when a capability invocation fails security verification."""
pass
class DeadlineExceededError(CCAError):
"""Raised when the invocation deadline has already passed."""
pass
class CapabilityNotReadyError(CCAError):
"""Raised when a capability is invoked while not in the READY state."""
pass
class DependencyCycleError(CCAError):
"""
Raised when a circular dependency is detected in the capability graph.
cycle_path contains the sequence of capability names forming the cycle,
making it straightforward to diagnose and fix the problem.
"""
def __init__(self, cycle_path: List[str]) -> None:
self.cycle_path = cycle_path
cycle_str = " -> ".join(cycle_path)
super().__init__(f"Circular dependency detected: {cycle_str}")
# cca/lifecycle.py
# CCA 0.3 — Capability Lifecycle State Machine
# Belongs to the Essence ring of every capability.
from __future__ import annotations
from enum import Enum, auto
from typing import Callable, Dict, List, Set
import logging
from .exceptions import LifecycleError
logger = logging.getLogger(__name__)
class LifecycleState(Enum):
"""
The six states of a CCA 0.3 capability lifecycle.
Transitions are strictly controlled and fire lifecycle events.
RETIRED is the only terminal state: no further transitions are allowed.
Embedded mapping:
CREATED -> peripheral clock enabled
INITIALIZING -> hardware init sequence running
READY -> normal operation
DEGRADED -> sensor out-of-range / bus errors
SUSPENDED -> low-power sleep mode
RETIRED -> hardware fault / field service required
"""
CREATED = auto()
INITIALIZING = auto()
READY = auto()
DEGRADED = auto()
SUSPENDED = auto()
RETIRED = auto()
# Valid transitions map: each state maps to the set of states it may move to.
VALID_TRANSITIONS: Dict[LifecycleState, Set[LifecycleState]] = {
LifecycleState.CREATED: {LifecycleState.INITIALIZING},
LifecycleState.INITIALIZING: {LifecycleState.READY,
LifecycleState.RETIRED},
LifecycleState.READY: {LifecycleState.DEGRADED,
LifecycleState.SUSPENDED,
LifecycleState.RETIRED},
LifecycleState.DEGRADED: {LifecycleState.READY,
LifecycleState.SUSPENDED,
LifecycleState.RETIRED},
LifecycleState.SUSPENDED: {LifecycleState.READY,
LifecycleState.RETIRED},
LifecycleState.RETIRED: set(), # Terminal: no further transitions.
}
# Type alias for observer callables: (capability_name, old_state, new_state) -> None
LifecycleObserver = Callable[[str, LifecycleState, LifecycleState], None]
class LifecycleMachine:
"""
Manages the lifecycle state of a single capability.
Belongs to the Essence ring.
Observers (plain callables) are notified on every valid transition.
Observer errors are caught and logged so they cannot crash the machine.
In embedded systems, observers can kick the watchdog, update a status LED,
or write a fault code to non-volatile memory.
"""
def __init__(self, capability_name: str) -> None:
self._name: str = capability_name
self._state: LifecycleState = LifecycleState.CREATED
self._observers: List[LifecycleObserver] = []
@property
def state(self) -> LifecycleState:
"""Return the current lifecycle state."""
return self._state
def add_observer(self, observer: LifecycleObserver) -> None:
"""Register a callable that will be invoked on every state transition."""
self._observers.append(observer)
def transition_to(self, new_state: LifecycleState) -> None:
"""
Attempt to move to new_state.
Raises LifecycleError if the transition is not permitted.
Notifies all registered observers after a successful transition.
"""
allowed: Set[LifecycleState] = VALID_TRANSITIONS.get(self._state, set())
if new_state not in allowed:
raise LifecycleError(
f"Capability '{self._name}': cannot transition from "
f"{self._state.name} to {new_state.name}. "
f"Allowed targets: {[s.name for s in allowed]}."
)
old_state = self._state
self._state = new_state
logger.info(
"Capability '%s' lifecycle: %s -> %s",
self._name, old_state.name, new_state.name,
)
self._notify_observers(old_state, new_state)
def _notify_observers(
self,
old_state: LifecycleState,
new_state: LifecycleState,
) -> None:
"""Notify all registered observers. Observer errors are swallowed."""
for observer in self._observers:
try:
observer(self._name, old_state, new_state)
except Exception as exc:
logger.warning(
"Lifecycle observer raised an exception for "
"capability '%s': %s",
self._name, exc,
)
The LifecycleMachine class is deliberately kept small and focused. It knows nothing about what a capability does; it only knows the rules governing state transitions. The VALID_TRANSITIONS dictionary is defined at module level rather than inside the class, which makes it easy to inspect, test, and even modify the transition rules without touching the class itself.
CHAPTER 3: THE CAPABILITY CONTEXT
The second major new concept in CCA 0.3 is the capability execution context. In CCA 0.2, when you called capability.execute(input_data), the capability had no way to know anything about the environment in which it was being called. It did not know which tenant was making the request, which trace was active, which security principal was involved, or what the deadline for the overall operation was.
CCA 0.3 introduces the CapabilityContext, which is a structured, immutable carrier of cross-cutting information that travels with every capability invocation. The context threads through all three rings: the Essence ring uses it for security verification and lifecycle decisions, the Realization ring uses it for deadline checking and tenant-specific logic, and the Adaptation ring uses it for trace propagation and structured logging.
Context in embedded systems: In an embedded system, the context carries different but equally important cross-cutting information. Instead of a tenant ID, it carries a device ID. Instead of a distributed trace, it carries a local sequence number or a hardware timestamp from a free-running timer. Instead of a security token from an OAuth server, it carries a hardware-attested identity from a TPM or a secure element. The deadline is particularly important in real-time systems: a temperature reading that arrives after the PID controller's deadline is worse than no reading at all, because it can cause the controller to make decisions based on stale data.
All datetime values in CCA 0.3 are timezone-aware UTC datetimes, created with datetime.now(timezone.utc). This avoids the subtle bugs that arise when mixing naive and aware datetimes, and it is consistent with Python 3.12's deprecation of datetime.utcnow(). In embedded systems without a real-time clock, the deadline can be expressed as a monotonic tick count instead of a wall-clock time.
# cca/context.py
# CCA 0.3 — Capability Execution Context
# Threads through all three rings (Essence, Realization, Adaptation).
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Dict, Optional
@dataclass(frozen=True)
class TraceContext:
"""
Carries distributed tracing identifiers compatible with W3C TraceContext.
Immutable so it can be safely shared across threads.
In embedded systems, trace_id may be a hardware timer value or a
monotonic sequence number rather than a UUID string.
"""
trace_id: str
span_id: str
parent_span_id: Optional[str] = None
@dataclass(frozen=True)
class SecurityContext:
"""
Carries the identity and authorization tokens of the calling principal.
The tokens field is a tuple (not a list) to preserve immutability.
CapabilityToken objects are defined in security.py.
In embedded systems, principal_id may be a hardware device serial number
or a certificate thumbprint from a secure element.
"""
principal_id: str
tokens: tuple = field(default_factory=tuple)
@dataclass(frozen=True)
class CapabilityContext:
"""
The unified execution context passed to every CCA 0.3 capability invocation.
All fields are immutable by reference (frozen=True).
Note: the metadata dict value is mutable by content; treat it as read-only.
Child contexts are created via derive(), which correctly propagates the
parent span identifier for distributed tracing.
All deadline values must be timezone-aware UTC datetimes.
In embedded systems without a real-time clock, store a monotonic tick
deadline in metadata and check it with time.monotonic() in is_expired().
"""
trace: TraceContext
security: SecurityContext
tenant_id: str
deadline: Optional[datetime] = None
metadata: Dict[str, Any] = field(default_factory=dict)
def derive(self, new_span_id: str, **overrides: Any) -> CapabilityContext:
"""
Create a child context for a nested capability invocation.
The child inherits all fields from the parent but receives a new span_id,
and the parent's span_id becomes the child's parent_span_id.
This correctly models the parent-child relationship in distributed tracing.
Additional field overrides can be supplied as keyword arguments.
"""
child_trace = TraceContext(
trace_id = self.trace.trace_id,
span_id = new_span_id,
parent_span_id = self.trace.span_id,
)
merged_metadata = {**self.metadata, **overrides.get("metadata", {})}
return CapabilityContext(
trace = child_trace,
security = overrides.get("security", self.security),
tenant_id = overrides.get("tenant_id", self.tenant_id),
deadline = overrides.get("deadline", self.deadline),
metadata = merged_metadata,
)
def is_expired(self) -> bool:
"""
Returns True if the deadline has passed.
Capabilities should call this before starting any expensive work
to implement fail-fast behavior under deadline pressure.
Both this method and any deadline stored in the context must use
timezone-aware UTC datetimes for correct comparison.
"""
if self.deadline is None:
return False
return datetime.now(timezone.utc) > self.deadline
The derive method deserves special attention. When a capability composes other capabilities, it does not pass its own context directly to the child invocations. Instead, it derives a new context with a fresh span identifier, so that the distributed trace correctly reflects the parent-child relationship between the invocations. This is exactly how distributed tracing systems like OpenTelemetry model the propagation of trace context, and CCA 0.3 makes this propagation automatic and correct by design.
The is_expired method gives every capability a simple way to implement deadline-aware behavior. Before starting any expensive computation, a well-behaved capability checks whether the deadline has already passed and, if so, raises a DeadlineExceededError immediately rather than doing work that will ultimately be discarded.
CHAPTER 4: ENHANCED CONTRACTS AND THE EVOLUTION ENVELOPE
CCA 0.2 contracts validated input and output schemas, which was a good start. But schema validation alone is not enough to fully specify the behavior of a capability. CCA 0.3 extends contracts with three new elements: behavioral assertions, SLA declarations, and error catalogs. These additions also directly strengthen the evolution envelope concept from CCA 0.2.
Contracts in embedded systems: In an embedded system, the contract of a sensor capability declares not just the data types of its outputs but also the physical units (degrees Celsius, not Fahrenheit), the measurement range (−40°C to +125°C), the accuracy (±0.5°C), and the maximum measurement latency (50 ms). The behavioral assertions enforce these physical constraints: an assertion that −40 <= output["temperature_c"] <= 125 catches a sensor that is returning garbage values due to a hardware fault. The SLA declaration specifies the maximum acceptable latency, which in a real-time control loop may be as tight as 1 millisecond. These machine-checkable specifications are directly analogous to the requirements in a hardware datasheet, and they serve the same purpose: they define the evolution envelope within which the implementation can change without breaking the system.
The validate_input and validate_output methods accept both int and float values for numeric fields, since Python's isinstance(1, float) returns False even though an integer is numerically valid in most contexts.
# cca/contract.py
# CCA 0.3 — Enhanced Capability Contract
# Belongs to the Essence ring. Defines the evolution envelope.
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union
# Numeric types accepted by contract validation when the schema specifies float.
# isinstance(1, float) is False in Python, so we accept both int and float.
_NUMERIC_TYPES: Tuple[Type, ...] = (int, float)
@dataclass
class SLADeclaration:
"""
Declares the performance guarantees of a capability.
Forms the operational dimension of the capability's evolution envelope.
These values are used by the observability subsystem to compute
SLO compliance and to trigger DEGRADED lifecycle transitions.
In embedded systems, max_latency_p99_seconds may be as tight as 0.001
(1 ms) for real-time control loops. min_throughput_rps reflects the
required sampling rate of a sensor or the required update rate of an
actuator control loop.
"""
# Maximum acceptable latency in seconds at the 99th percentile.
max_latency_p99_seconds: float
# Maximum acceptable error rate as a fraction between 0.0 and 1.0.
max_error_rate: float
# Minimum acceptable throughput in requests per second.
min_throughput_rps: float
@dataclass
class ErrorDescriptor:
"""
Describes a single error that a capability may raise.
Part of the error catalog in the capability's contract.
error_code is machine-readable; description is human-readable.
In embedded systems, error codes often map to fault codes stored in
non-volatile memory for field diagnostics (e.g., OBD-II DTCs in
automotive systems).
"""
error_code: str
description: str
severity: str # One of: "INFO", "WARNING", "ERROR", "CRITICAL"
retryable: bool
@dataclass
class Contract:
"""
The CCA 0.3 enhanced capability contract. Lives in the Essence ring.
Specifies input/output schemas, behavioral assertions, SLA declarations,
and an error catalog. Together these elements define the evolution envelope:
any change to the capability that violates these specifications falls
outside the envelope and requires a new capability version.
Schema values may be a single type or a tuple of types (like (int, float))
to accommodate Python's numeric type hierarchy.
In embedded systems, behavioral assertions enforce physical constraints
(measurement ranges, unit correctness) that are equivalent to the
specifications in a hardware datasheet.
"""
input_schema: Dict[str, Union[Type, Tuple[Type, ...]]]
output_schema: Dict[str, Union[Type, Tuple[Type, ...]]]
sla: SLADeclaration
error_catalog: List[ErrorDescriptor] = field(default_factory=list)
# Assertions: (input_data, output_data) -> bool.
assertions: List[Callable[[Any, Any], bool]] = field(default_factory=list)
def validate_input(self, data: Any) -> None:
"""
Validate input data against the declared input schema.
Raises ValueError if a required field is missing or has the wrong type.
Schema entries may specify a tuple of types, e.g. (int, float).
"""
if not isinstance(data, dict):
raise ValueError(
f"Contract violation: input must be a dict, "
f"got {type(data).__name__}."
)
for field_name, field_type in self.input_schema.items():
if field_name not in data:
raise ValueError(
f"Contract violation: required input field "
f"'{field_name}' is missing."
)
if not isinstance(data[field_name], field_type):
type_names = (
" or ".join(t.__name__ for t in field_type)
if isinstance(field_type, tuple)
else field_type.__name__
)
raise ValueError(
f"Contract violation: field '{field_name}' must be "
f"{type_names}, got {type(data[field_name]).__name__}."
)
def validate_output(self, data: Any) -> None:
"""
Validate output data against the declared output schema.
Raises ValueError if a required field is missing or has the wrong type.
"""
if not isinstance(data, dict):
raise ValueError(
f"Contract violation: output must be a dict, "
f"got {type(data).__name__}."
)
for field_name, field_type in self.output_schema.items():
if field_name not in data:
raise ValueError(
f"Contract violation: required output field "
f"'{field_name}' is missing."
)
if not isinstance(data[field_name], field_type):
type_names = (
" or ".join(t.__name__ for t in field_type)
if isinstance(field_type, tuple)
else field_type.__name__
)
raise ValueError(
f"Contract violation: output field '{field_name}' must be "
f"{type_names}, got "
f"{type(data[field_name]).__name__}."
)
def check_assertions(self, input_data: Any, output_data: Any) -> None:
"""
Evaluate all behavioral assertions against the input/output pair.
Raises AssertionError if any assertion returns False.
Assertion failures indicate that the Realization ring has violated
the behavioral dimension of the evolution envelope.
"""
for i, assertion in enumerate(self.assertions):
if not assertion(input_data, output_data):
raise AssertionError(
f"Contract behavioral assertion #{i} failed. "
f"input={input_data!r}, output={output_data!r}. "
f"This change may be outside the evolution envelope."
)
def lookup_error(self, error_code: str) -> Optional[ErrorDescriptor]:
"""Return the ErrorDescriptor for the given error_code, or None."""
return next(
(e for e in self.error_catalog if e.error_code == error_code),
None,
)
CHAPTER 5: CAPABILITY POLICIES
Policies are the Adaptation ring's mechanism for governing how a capability behaves in its operational environment without touching the Realization ring. CCA 0.2 introduced four policy types: retry, circuit breaker, rate limiting, and timeout. CCA 0.3 preserves all four and integrates them with the nine-step execution protocol in the Capability base class through before_execute and after_execute hooks.
Policies in embedded systems: In an embedded system, the Adaptation ring policies take on hardware-specific meanings. The retry policy governs how many times to retry an I2C read before declaring the sensor faulty. The circuit breaker policy governs when to stop attempting to communicate with a peripheral that is not responding and to switch to a safe fallback value. The rate limiting policy governs the maximum sampling rate of an ADC to avoid overloading the CPU. The timeout policy enforces the real-time deadline of a control loop iteration. These are not abstract software patterns; they are the concrete operational policies that embedded systems engineers have always needed but rarely expressed in a structured, reusable, testable form.
All exceptions are imported from cca/exceptions.py, eliminating any circular import risk.
# cca/policies.py
# CCA 0.3 — Capability Policies (Adaptation ring)
# RetryPolicy, CircuitBreakerPolicy, RateLimitingPolicy, TimeoutPolicy.
from __future__ import annotations
import threading
import time
import logging
from enum import Enum, auto
from typing import Any, Optional
from .exceptions import DeadlineExceededError
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Retry Policy
# ---------------------------------------------------------------------------
class RetryPolicy:
"""
Adaptation ring policy: automatically retry a failing capability invocation.
Applies exponential back-off with an optional jitter between attempts.
In embedded systems, back-off delays must be chosen carefully to avoid
violating real-time deadlines. For hard real-time systems, set
delay_seconds=0 and max_retries to a small number (1 or 2).
"""
def __init__(
self,
max_retries: int = 3,
delay_seconds: float = 0.5,
backoff_factor: float = 2.0,
retryable_errors: tuple = (Exception,),
) -> None:
self.max_retries = max_retries
self.delay_seconds = delay_seconds
self.backoff_factor = backoff_factor
self.retryable_errors = retryable_errors
def execute_with_retry(self, fn: Any, *args: Any, **kwargs: Any) -> Any:
"""
Call fn(*args, **kwargs) up to max_retries+1 times.
Sleeps between attempts using exponential back-off.
Re-raises the last exception if all attempts fail.
"""
delay = self.delay_seconds
last_exc: Optional[Exception] = None
for attempt in range(self.max_retries + 1):
try:
return fn(*args, **kwargs)
except self.retryable_errors as exc:
last_exc = exc
if attempt < self.max_retries:
logger.warning(
"RetryPolicy: attempt %d/%d failed (%s). "
"Retrying in %.2fs.",
attempt + 1, self.max_retries + 1, exc, delay,
)
time.sleep(delay)
delay *= self.backoff_factor
raise last_exc # type: ignore[misc]
# ---------------------------------------------------------------------------
# Circuit Breaker Policy
# ---------------------------------------------------------------------------
class CircuitState(Enum):
CLOSED = auto() # Normal operation; requests pass through.
OPEN = auto() # Failing; requests are rejected immediately.
HALF_OPEN = auto() # Probing; one request is allowed through.
class CircuitBreakerOpenError(Exception):
"""Raised by CircuitBreakerPolicy when the circuit is OPEN."""
pass
class CircuitBreakerPolicy:
"""
Adaptation ring policy: stop calling a failing capability.
Transitions: CLOSED -> OPEN after failure_threshold consecutive failures.
OPEN -> HALF_OPEN after recovery_timeout_seconds.
HALF_OPEN -> CLOSED on success; -> OPEN on failure.
Thread-safe.
In embedded systems, the circuit breaker can switch to a safe fallback
value (e.g., last known good sensor reading) when the circuit is OPEN,
rather than propagating an error to the control loop.
"""
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout_seconds: float = 30.0,
) -> None:
self._threshold = failure_threshold
self._timeout = recovery_timeout_seconds
self._state = CircuitState.CLOSED
self._failures = 0
self._opened_at: Optional[float] = None
self._lock = threading.RLock()
@property
def state(self) -> CircuitState:
with self._lock:
return self._state
def before_execute(self, capability: Any, input_data: Any, context: Any) -> None:
"""
Called before capability execution.
Raises CircuitBreakerOpenError if the circuit is OPEN.
Transitions OPEN -> HALF_OPEN if the recovery timeout has elapsed.
"""
with self._lock:
if self._state == CircuitState.OPEN:
elapsed = time.monotonic() - (self._opened_at or 0.0)
if elapsed >= self._timeout:
logger.info(
"CircuitBreaker for '%s': OPEN -> HALF_OPEN "
"(recovery timeout elapsed).", capability.name,
)
self._state = CircuitState.HALF_OPEN
else:
raise CircuitBreakerOpenError(
f"Circuit breaker for '{capability.name}' is OPEN. "
f"Retry after {self._timeout - elapsed:.1f}s."
)
def after_execute(
self,
capability: Any,
input_data: Any,
result: Any,
context: Any,
) -> None:
"""Called after a successful execution. Resets failure count."""
with self._lock:
if self._state in (CircuitState.HALF_OPEN, CircuitState.CLOSED):
self._failures = 0
self._state = CircuitState.CLOSED
logger.debug(
"CircuitBreaker for '%s': reset to CLOSED.", capability.name,
)
def record_failure(self, capability_name: str) -> None:
"""
Record a failure. Called by the capability's execute() on exception.
Transitions CLOSED/HALF_OPEN -> OPEN when threshold is reached.
"""
with self._lock:
self._failures += 1
if self._failures >= self._threshold:
self._state = CircuitState.OPEN
self._opened_at = time.monotonic()
logger.warning(
"CircuitBreaker for '%s': CLOSED -> OPEN "
"after %d failures.", capability_name, self._failures,
)
# ---------------------------------------------------------------------------
# Rate Limiting Policy
# ---------------------------------------------------------------------------
class RateLimitExceededError(Exception):
"""Raised by RateLimitingPolicy when the rate limit is exceeded."""
pass
class RateLimitingPolicy:
"""
Adaptation ring policy: limit the number of invocations per second.
Uses a token bucket algorithm. Thread-safe.
In embedded systems, rate limiting enforces the maximum sampling rate
of a sensor or the maximum update rate of an actuator to protect the
hardware and to stay within CPU budget.
"""
def __init__(self, max_calls_per_second: float) -> None:
self._rate = max_calls_per_second
self._tokens = max_calls_per_second
self._last_check = time.monotonic()
self._lock = threading.RLock()
def before_execute(self, capability: Any, input_data: Any, context: Any) -> None:
"""
Consume one token from the bucket.
Raises RateLimitExceededError if no tokens are available.
"""
with self._lock:
now = time.monotonic()
elapsed = now - self._last_check
self._tokens = min(
self._rate,
self._tokens + elapsed * self._rate,
)
self._last_check = now
if self._tokens < 1.0:
raise RateLimitExceededError(
f"Rate limit exceeded for capability '{capability.name}'. "
f"Max {self._rate} calls/second."
)
self._tokens -= 1.0
# ---------------------------------------------------------------------------
# Timeout Policy
# ---------------------------------------------------------------------------
class TimeoutPolicy:
"""
Adaptation ring policy: enforce a maximum execution time.
Checks the context deadline before execution and warns if the remaining
time is less than the policy's configured maximum.
In embedded systems, this policy enforces the real-time deadline of a
control loop iteration. A missed deadline in a hard real-time system
is a safety violation, not just a performance problem.
"""
def __init__(self, max_seconds: float) -> None:
self.max_seconds = max_seconds
def before_execute(self, capability: Any, input_data: Any, context: Any) -> None:
"""
Verify that the context deadline allows at least max_seconds of work.
Raises DeadlineExceededError if the deadline has already passed.
Logs a warning if the remaining time is less than max_seconds.
"""
from datetime import datetime, timezone
if context.deadline is not None:
remaining = (
context.deadline - datetime.now(timezone.utc)
).total_seconds()
if remaining <= 0:
raise DeadlineExceededError(
f"Capability '{capability.name}': deadline already passed."
)
if remaining < self.max_seconds:
logger.warning(
"TimeoutPolicy for '%s': only %.2fs remaining, "
"policy max is %.2fs.",
capability.name, remaining, self.max_seconds,
)
The circuit breaker's record_failure method is called from the Capability base class execute() method whenever the Realization ring raises an exception. This closes the feedback loop between execution failures and the circuit breaker's state machine without requiring the Realization ring to know anything about the circuit breaker.
CHAPTER 6: THE CAPABILITY MESH
The most architecturally significant structural change in CCA 0.3 is the replacement of the simple central registry with the Capability Mesh. The registry in CCA 0.2 was a single in-process dictionary. It worked well for small systems but did not scale to distributed deployments, did not support federation across organizational boundaries, and was a single point of failure.
The Capability Mesh is a distributed, federated registry that allows capabilities to be discovered and invoked across process boundaries, across data centers, and even across organizational boundaries. The mesh is inspired by service mesh technologies like Istio and Linkerd, but it operates at the capability level rather than at the network level.
The mesh in embedded and hybrid systems: In an embedded system, the mesh operates within the constraints of the hardware. On a single microcontroller, the mesh is an in-memory lookup table with no networking overhead. On an edge device like a Raspberry Pi or an industrial PC, the mesh can span multiple processes communicating over local IPC. In a hybrid embedded-enterprise system, the mesh federates across the embedded device and the cloud, using MQTT, CoAP, or OPC-UA as the transport. A cloud service can discover a temperature sensor capability running on a factory floor device through the same mesh API it uses to discover a currency conversion capability running in a Kubernetes pod.
# cca/mesh.py
# CCA 0.3 — Capability Mesh
# Replaces the CCA 0.2 central registry with a distributed, routable mesh.
from __future__ import annotations
import random
import threading
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, Dict, List, Optional, Set
from .lifecycle import LifecycleState
from .contract import Contract
@dataclass
class CapabilityRegistration:
"""
A record in the capability mesh describing one registered capability instance.
Each instance has a unique instance_id even if multiple instances of the
same capability name and version are running simultaneously.
The endpoint is an opaque string interpreted by the transport layer
(a URL, a queue name, a CAN node ID, an MQTT topic, or an in-process
callable reference).
lifecycle_state is mutable so the mesh can update it without replacing
the entire registration record.
"""
instance_id: str
name: str
version: str
tags: Set[str]
contract: Contract
lifecycle_state: LifecycleState
endpoint: str
weight: int = 50 # Used for weighted routing (range 1–100).
connections: int = 0 # Active connections; used by LEAST_CONNECTIONS.
class RoutingStrategy(Enum):
"""Determines how the mesh selects among multiple matching instances."""
ROUND_ROBIN = "round_robin"
WEIGHTED = "weighted"
LEAST_CONNECTIONS = "least_connections"
RANDOM = "random"
class CapabilityMesh:
"""
The CCA 0.3 distributed capability mesh.
Manages registration, discovery, and routing of capability instances.
Thread-safe for concurrent registration and discovery operations.
In a single-process or embedded deployment this operates entirely in memory.
In a distributed deployment the _registry would be backed by a
distributed store such as etcd, Consul, Redis, or an MQTT broker.
In a hybrid embedded-enterprise deployment, the mesh federates across
device and cloud using MQTT, CoAP, or OPC-UA as the transport layer.
"""
def __init__(
self,
routing_strategy: RoutingStrategy = RoutingStrategy.ROUND_ROBIN,
) -> None:
self._registry: Dict[tuple, List[CapabilityRegistration]] = {}
self._lock = threading.RLock()
self._strategy = routing_strategy
self._rr_counters: Dict[tuple, int] = {}
self._lifecycle_hooks: List[Callable[[CapabilityRegistration], None]] = []
def register(self, registration: CapabilityRegistration) -> None:
"""
Add a new capability instance to the mesh.
If an instance with the same instance_id already exists, it is replaced.
"""
key = (registration.name, registration.version)
with self._lock:
instances = self._registry.setdefault(key, [])
self._registry[key] = [
r for r in instances
if r.instance_id != registration.instance_id
]
self._registry[key].append(registration)
self._notify_lifecycle_hooks(registration)
def deregister(self, instance_id: str) -> None:
"""Remove a capability instance from the mesh by its instance_id."""
with self._lock:
for key in list(self._registry.keys()):
self._registry[key] = [
r for r in self._registry[key]
if r.instance_id != instance_id
]
if not self._registry[key]:
del self._registry[key]
def discover(
self,
name: str,
version: Optional[str] = None,
tags: Optional[Set[str]] = None,
require_state: LifecycleState = LifecycleState.READY,
) -> Optional[CapabilityRegistration]:
"""
Find a single capability instance matching the given criteria.
Returns None if no matching instance is found.
Applies the configured routing strategy when multiple matches exist.
"""
with self._lock:
candidates = self._find_candidates(name, version, tags, require_state)
if not candidates:
return None
return self._apply_routing(name, version, candidates)
def discover_all(
self,
name: str,
version: Optional[str] = None,
tags: Optional[Set[str]] = None,
require_state: LifecycleState = LifecycleState.READY,
) -> List[CapabilityRegistration]:
"""
Return all matching capability instances.
Used by the fan-out composition pattern to broadcast to every instance.
"""
with self._lock:
return self._find_candidates(name, version, tags, require_state)
def update_lifecycle_state(
self,
instance_id: str,
new_state: LifecycleState,
) -> None:
"""Update the lifecycle state of a registered instance in the mesh."""
with self._lock:
for instances in self._registry.values():
for reg in instances:
if reg.instance_id == instance_id:
reg.lifecycle_state = new_state
self._notify_lifecycle_hooks(reg)
return
def add_lifecycle_hook(
self,
hook: Callable[[CapabilityRegistration], None],
) -> None:
"""
Register a hook called whenever a registration changes lifecycle state.
Hooks can trigger alerts, spin up replacements, or update dashboards.
In embedded systems, hooks can update a status LED or write a fault
code to non-volatile memory.
"""
self._lifecycle_hooks.append(hook)
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _find_candidates(
self,
name: str,
version: Optional[str],
tags: Optional[Set[str]],
require_state: LifecycleState,
) -> List[CapabilityRegistration]:
"""Filter the registry to find all instances matching the criteria."""
results: List[CapabilityRegistration] = []
for (reg_name, reg_version), instances in self._registry.items():
if reg_name != name:
continue
if version is not None and reg_version != version:
continue
for inst in instances:
if inst.lifecycle_state != require_state:
continue
if tags and not tags.issubset(inst.tags):
continue
results.append(inst)
return results
def _apply_routing(
self,
name: str,
version: Optional[str],
candidates: List[CapabilityRegistration],
) -> CapabilityRegistration:
"""Select one candidate according to the configured routing strategy."""
if self._strategy == RoutingStrategy.ROUND_ROBIN:
key = (name, version)
idx = self._rr_counters.get(key, 0) % len(candidates)
self._rr_counters[key] = idx + 1
return candidates[idx]
if self._strategy == RoutingStrategy.WEIGHTED:
total_weight = sum(c.weight for c in candidates)
r = random.uniform(0, total_weight)
cumulative = 0.0
for candidate in candidates:
cumulative += candidate.weight
if r <= cumulative:
return candidate
return candidates[-1]
if self._strategy == RoutingStrategy.LEAST_CONNECTIONS:
return min(candidates, key=lambda c: c.connections)
if self._strategy == RoutingStrategy.RANDOM:
return random.choice(candidates)
return candidates[0]
def _notify_lifecycle_hooks(
self,
registration: CapabilityRegistration,
) -> None:
"""Notify all registered lifecycle hooks. Hook errors are swallowed."""
for hook in self._lifecycle_hooks:
try:
hook(registration)
except Exception:
pass
---
## CHAPTER 7: THE CAPABILITY SECURITY MODEL
Security was listed as a future direction in CCA 0.2, and CCA 0.3 delivers on that promise with a formal capability-based security model. The model is inspired by the object-capability security model pioneered by researchers like Mark Miller and the E programming language, adapted for the practical needs of distributed software systems.
**Security in embedded systems**: Embedded systems have historically had poor security, partly because security was seen as a luxury that resource-constrained devices could not afford. CCA 0.3's security model is designed to be applicable even in resource-constrained environments. On devices with a hardware security module (HSM) or a Trusted Platform Module (TPM), the token authority can use hardware-backed key storage and hardware-accelerated cryptography. On devices with ARM TrustZone, the token authority runs in the secure world and the capability execution runs in the normal world, with the token verification crossing the security boundary. On the most constrained devices, a simplified version of the token model uses pre-shared symmetric keys and a simple counter-based nonce instead of full HMAC-SHA256. The CCA 0.3 security model is a spectrum, not a binary choice.
In industrial control systems governed by IEC 62443, capability tokens map directly to the concept of security zones and conduits: a token grants access to capabilities within a specific security zone, and the token authority acts as the conduit controller.
```python
# cca/security.py
# CCA 0.3 — Capability-Based Security Model
# Token enforcement belongs to the Adaptation ring.
# Required permission declaration belongs to the Essence ring.
from __future__ import annotations
import hashlib
import hmac
import json
import time
import uuid
from dataclasses import dataclass, field
from typing import FrozenSet, Optional
@dataclass(frozen=True)
class CapabilityToken:
"""
A signed credential granting the holder the right to invoke a specific
capability with specific permissions. Tokens are immutable once issued.
Signatures use HMAC-SHA256 over a canonical JSON serialization of the payload.
In embedded systems with an HSM or TPM, the signature field may contain
a hardware-backed ECDSA signature instead of an HMAC-SHA256 digest.
"""
token_id: str
capability_name: str
capability_version: str
principal_id: str
permissions: FrozenSet[str]
issued_at: float
expires_at: float
signature: str
def is_valid(self, secret_key: str) -> bool:
"""
Verify that the token has not been tampered with and has not expired.
Uses hmac.compare_digest to prevent timing-based side-channel attacks.
Returns True only if both the signature and the expiry check pass.
"""
if time.time() > self.expires_at:
return False
expected_sig = self._compute_signature(secret_key)
return hmac.compare_digest(self.signature, expected_sig)
def has_permission(self, permission: str) -> bool:
"""Return True if this token grants the specified permission."""
return permission in self.permissions
def _compute_signature(self, secret_key: str) -> str:
"""
Compute the HMAC-SHA256 signature over a canonical JSON payload.
The payload keys are sorted to ensure deterministic serialization.
Uses hmac.new(key, msg, digestmod) — the standard Python 3 HMAC API.
"""
payload = json.dumps(
{
"token_id": self.token_id,
"capability_name": self.capability_name,
"capability_version": self.capability_version,
"principal_id": self.principal_id,
"permissions": sorted(self.permissions),
"issued_at": self.issued_at,
"expires_at": self.expires_at,
},
sort_keys=True,
)
return hmac.new(
secret_key.encode("utf-8"),
payload.encode("utf-8"),
hashlib.sha256,
).hexdigest()
class TokenAuthority:
"""
Issues and verifies capability tokens.
In production this would integrate with an identity provider (OAuth2/OIDC).
In embedded systems, this integrates with an HSM, TPM, or secure element.
The secret_key must be kept confidential and rotated periodically.
"""
def __init__(
self,
secret_key: str,
default_ttl_seconds: float = 3600.0,
) -> None:
self._secret_key = secret_key
self._default_ttl = default_ttl_seconds
def issue(
self,
capability_name: str,
capability_version: str,
principal_id: str,
permissions: FrozenSet[str],
ttl_seconds: Optional[float] = None,
) -> CapabilityToken:
"""
Issue a new capability token for the given principal and capability.
The token is signed with the authority's secret key.
"""
now = time.time()
ttl = ttl_seconds if ttl_seconds is not None else self._default_ttl
token_id = str(uuid.uuid4())
temp = CapabilityToken(
token_id = token_id,
capability_name = capability_name,
capability_version = capability_version,
principal_id = principal_id,
permissions = permissions,
issued_at = now,
expires_at = now + ttl,
signature = "",
)
real_sig = temp._compute_signature(self._secret_key)
return CapabilityToken(
token_id = token_id,
capability_name = capability_name,
capability_version = capability_version,
principal_id = principal_id,
permissions = permissions,
issued_at = now,
expires_at = now + ttl,
signature = real_sig,
)
def verify(self, token: CapabilityToken) -> bool:
"""Verify a token. Returns True if valid and unexpired, False otherwise."""
return token.is_valid(self._secret_key)
CHAPTER 8: COMPOSITION PATTERNS
CCA 0.2 supported three composition patterns: sequential, parallel, and conditional. CCA 0.3 preserves all three and adds four new patterns: pipeline with backpressure, saga, fan-out/fan-in, and scatter-gather. Together, these seven patterns cover virtually every composition scenario that arises in real-world systems.
Composition in embedded systems: In an embedded system, composition is the mechanism by which simple hardware-facing capabilities are assembled into complex behaviors. A sensor fusion capability composes three individual sensor capabilities (accelerometer, gyroscope, magnetometer) using parallel composition, then feeds their outputs into a Kalman filter capability using sequential composition. A motor control capability composes a position sensor capability, a PID controller capability, and a PWM output capability using sequential composition in a tight control loop. The saga pattern is used in embedded systems for multi-step hardware initialization sequences where each step has a well-defined undo operation: if step 3 of a 5-step initialization fails, the saga compensates by undoing steps 2 and 1 in reverse order, leaving the hardware in a clean state.
# cca/composition/sequential.py
# CCA 0.3 — Sequential composition (originally from CCA 0.2).
# Belongs to the Realization ring of composed capabilities.
from __future__ import annotations
from typing import Any, List
class SequentialComposition:
"""
CCA sequential composition: the output of each capability is passed as
the input to the next. The final capability's output is returned.
Capabilities must expose an execute(input_data, context) interface.
Embedded use case: sensor reading -> unit conversion -> filtering ->
control law computation -> actuator command.
"""
def __init__(self, capabilities: List[Any]) -> None:
if not capabilities:
raise ValueError(
"SequentialComposition requires at least one capability."
)
self._capabilities = capabilities
def execute(self, input_data: Any, context: Any) -> Any:
"""Execute capabilities in order, chaining outputs to inputs."""
result = input_data
for cap in self._capabilities:
result = cap.execute(result, context)
return result
# cca/composition/parallel.py
# CCA 0.3 — Parallel composition (originally from CCA 0.2).
# Belongs to the Realization ring of composed capabilities.
from __future__ import annotations
import concurrent.futures
from typing import Any, List
class ParallelComposition:
"""
CCA parallel composition: all capabilities receive the same input and
execute concurrently. Returns a list of results in submission order.
Exceptions from individual capabilities are re-raised during collection.
Embedded use case: reading multiple independent sensors simultaneously
to minimize total sampling latency (sensor fusion).
Note: In embedded systems without OS threading, parallel composition
degrades gracefully to sequential execution — the interface is identical,
only the execution model changes.
"""
def __init__(
self,
capabilities: List[Any],
max_workers: int = 10,
timeout: float = 30.0,
) -> None:
if not capabilities:
raise ValueError(
"ParallelComposition requires at least one capability."
)
self._capabilities = capabilities
self._max_workers = max_workers
self._timeout = timeout
def execute(self, input_data: Any, context: Any) -> List[Any]:
"""
Execute all capabilities concurrently and return their results
in the same order as the capabilities list (submission order).
"""
with concurrent.futures.ThreadPoolExecutor(
max_workers=self._max_workers,
) as executor:
# Submit all futures and keep them in submission order.
futures = [
executor.submit(cap.execute, input_data, context)
for cap in self._capabilities
]
# Collect results in submission order (not completion order).
return [f.result(timeout=self._timeout) for f in futures]
# cca/composition/conditional.py
# CCA 0.3 — Conditional composition (originally from CCA 0.2).
# Belongs to the Realization ring of composed capabilities.
from __future__ import annotations
from typing import Any, Callable, List, Optional, Tuple
class ConditionalComposition:
"""
CCA conditional composition: selects which capability to invoke based on
a predicate applied to the input data.
branches is a list of (predicate, capability) pairs evaluated in order.
The first branch whose predicate returns True is executed.
If no predicate matches and default_capability is set, it is used.
Raises ValueError if no branch matches and no default is configured.
Embedded use case: selecting between a primary sensor and a redundant
backup sensor based on the primary sensor's health status.
"""
def __init__(
self,
branches: List[Tuple[Callable[[Any], bool], Any]],
default_capability: Optional[Any] = None,
) -> None:
if not branches:
raise ValueError(
"ConditionalComposition requires at least one branch."
)
self._branches = branches
self._default = default_capability
def execute(self, input_data: Any, context: Any) -> Any:
"""Evaluate predicates in order and execute the first matching branch."""
for predicate, capability in self._branches:
if predicate(input_data):
return capability.execute(input_data, context)
if self._default is not None:
return self._default.execute(input_data, context)
raise ValueError(
"ConditionalComposition: no branch matched the input "
f"and no default capability is configured. input={input_data!r}"
)
THE PIPELINE WITH BACKPRESSURE
The sequential pattern passes the output of one capability directly to the next. This works well when all capabilities in the chain process data at roughly the same rate. But when one capability is significantly slower than the others, the fast capabilities produce data faster than the slow capability can consume it, and the system runs out of memory. CCA 0.3's pipeline pattern addresses this with backpressure.
Pipeline in embedded systems: In an embedded data acquisition system, the pipeline pattern maps naturally to a DMA-driven processing chain. The first stage reads raw ADC samples via DMA. The second stage applies a digital filter. The third stage computes derived quantities (RMS, FFT). The fourth stage formats and transmits the results over UART or CAN. The bounded queues between stages enforce backpressure: if the transmission stage cannot keep up, the computation stage slows down, which in turn slows the ADC sampling rate. This prevents buffer overflows and data loss without any explicit flow control logic in the individual stages.
# cca/composition/pipeline.py
# CCA 0.3 — Pipeline composition with backpressure.
# Belongs to the Realization ring of composed capabilities.
from __future__ import annotations
import queue
import threading
from typing import Any, Callable, List
class PipelineStage:
"""
Wraps a capability's execute callable as a named pipeline stage.
Reads items from an input queue, processes them, and writes to an output queue.
Errors are wrapped in a sentinel dict and forwarded rather than silently dropped.
Embedded use case: one stage per processing step in a DSP pipeline
(acquire -> filter -> compute -> transmit).
"""
def __init__(
self,
name: str,
execute: Callable[[Any], Any],
workers: int = 1,
) -> None:
self.name = name
self.execute = execute
self.workers = workers
def run(
self,
input_q: queue.Queue,
output_q: queue.Queue,
stop_event: threading.Event,
) -> None:
"""
Process items from input_q and put results into output_q.
Stops when stop_event is set and the input queue is drained.
Errors are wrapped in a sentinel dict and forwarded downstream.
"""
while not (stop_event.is_set() and input_q.empty()):
try:
item = input_q.get(timeout=0.1)
except queue.Empty:
continue
try:
result = self.execute(item)
output_q.put(result)
except Exception as exc:
output_q.put({"__error__": str(exc), "__input__": item})
finally:
input_q.task_done()
class BackpressurePipeline:
"""
A CCA 0.3 pipeline that connects multiple capability stages with
bounded queues to enforce backpressure between stages.
The buffer_size parameter controls the maximum number of items
allowed to be in flight between any two adjacent stages.
"""
def __init__(
self,
stages: List[PipelineStage],
buffer_size: int = 10,
) -> None:
if not stages:
raise ValueError("BackpressurePipeline requires at least one stage.")
self._stages = stages
self._buffer_size = buffer_size
def process(self, items: List[Any]) -> List[Any]:
"""
Push all items through the pipeline and return the collected results.
Blocks until all items have been processed by all stages.
Each intermediate queue is joined in order to guarantee that all items
have propagated fully through the pipeline before results are collected.
"""
queues: List[queue.Queue] = [
queue.Queue(maxsize=self._buffer_size)
for _ in range(len(self._stages) + 1)
]
stop_event = threading.Event()
threads: List[threading.Thread] = []
for i, stage in enumerate(self._stages):
t = threading.Thread(
target = stage.run,
args = (queues[i], queues[i + 1], stop_event),
daemon = True,
name = f"pipeline-{stage.name}",
)
t.start()
threads.append(t)
for item in items:
queues[0].put(item)
# Wait for all intermediate queues to be fully drained in order.
for q in queues[:-1]:
q.join()
stop_event.set()
for t in threads:
t.join(timeout=5.0)
results: List[Any] = []
while not queues[-1].empty():
results.append(queues[-1].get_nowait())
return results
THE SAGA PATTERN
The saga pattern is used for distributed transactions that span multiple capabilities. In a traditional database transaction, you can roll back all changes if something goes wrong. In a distributed system, there is no global transaction manager, so you need a different approach. A saga is a sequence of capability invocations where each step has a corresponding compensating action.
# cca/composition/saga.py
# CCA 0.3 — Saga composition pattern for distributed transactions.
# Belongs to the Realization ring of composed capabilities.
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Callable, List, Tuple
import logging
logger = logging.getLogger(__name__)
@dataclass
class SagaStep:
"""
One step in a saga.
action: the forward operation, receives the current accumulated result.
compensate: the undo operation, receives the result produced by action.
Embedded use case: multi-step hardware initialization where each step
has a well-defined undo (e.g., enable clock -> configure registers ->
run self-test; compensate: disable clock <- reset registers).
"""
name: str
action: Callable[[Any], Any]
compensate: Callable[[Any], None]
class SagaExecutionError(Exception):
"""
Raised when a saga step fails and compensation has been attempted.
Carries the name of the failed step and the original exception.
"""
def __init__(self, failed_step: str, original_error: Exception) -> None:
self.failed_step = failed_step
self.original_error = original_error
super().__init__(
f"Saga failed at step '{failed_step}': {original_error}. "
f"Compensation has been applied to all previously completed steps."
)
class Saga:
"""
Executes a sequence of SagaSteps with automatic compensation on failure.
Implements the CCA 0.3 saga composition pattern for distributed transactions
across multiple capabilities without a global transaction manager.
"""
def __init__(self, name: str, steps: List[SagaStep]) -> None:
if not steps:
raise ValueError(f"Saga '{name}' requires at least one step.")
self._name = name
self._steps = steps
def execute(self, initial_input: Any) -> Any:
"""
Execute all steps in order. If any step raises an exception,
compensate all previously completed steps in reverse order,
then raise SagaExecutionError.
Returns the result of the final step on success.
"""
completed: List[Tuple[SagaStep, Any]] = []
current_input = initial_input
for step in self._steps:
logger.info(
"Saga '%s': executing step '%s'.", self._name, step.name,
)
try:
result = step.action(current_input)
completed.append((step, result))
current_input = result
except Exception as exc:
logger.error(
"Saga '%s': step '%s' failed: %s. Starting compensation.",
self._name, step.name, exc,
)
self._compensate(completed)
raise SagaExecutionError(step.name, exc) from exc
logger.info(
"Saga '%s': all %d steps completed successfully.",
self._name, len(self._steps),
)
return current_input
def _compensate(self, completed: List[Tuple[SagaStep, Any]]) -> None:
"""
Execute compensating actions in reverse order.
Compensation errors are logged but do not prevent other compensations
from running, because partial compensation is better than none.
"""
for step, result in reversed(completed):
logger.info(
"Saga '%s': compensating step '%s'.", self._name, step.name,
)
try:
step.compensate(result)
except Exception as comp_exc:
logger.error(
"Saga '%s': compensation for step '%s' failed: %s.",
self._name, step.name, comp_exc,
)
THE FAN-OUT/FAN-IN PATTERN
The fan-out/fan-in pattern sends the same input to multiple capability instances simultaneously and then aggregates their results.
# cca/composition/fanout.py
# CCA 0.3 — Fan-out/Fan-in composition pattern.
# Belongs to the Realization ring of composed capabilities.
from __future__ import annotations
import concurrent.futures
import logging
from typing import Any, Callable, List
logger = logging.getLogger(__name__)
class FanOutFanIn:
"""
CCA 0.3 fan-out/fan-in composition.
Sends the same input_data to all provided capability callables concurrently,
collects their results (and errors), then passes the result list to the
aggregator callable which returns a single combined value.
Partial results are still aggregated even if some capabilities fail.
Embedded use case: querying redundant sensors and aggregating their
readings (e.g., voting among three temperature sensors for fault tolerance).
"""
def __init__(
self,
capabilities: List[Callable[[Any], Any]],
aggregator: Callable[[List[Any]], Any],
max_workers: int = 10,
timeout: float = 30.0,
) -> None:
if not capabilities:
raise ValueError("FanOutFanIn requires at least one capability.")
self._capabilities = capabilities
self._aggregator = aggregator
self._max_workers = max_workers
self._timeout = timeout
def execute(self, input_data: Any) -> Any:
"""
Fan out input_data to all capabilities, collect results, and aggregate.
Partial results are still aggregated even if some capabilities fail.
"""
results: List[Any] = []
errors: List[Exception] = []
with concurrent.futures.ThreadPoolExecutor(
max_workers=self._max_workers,
) as executor:
future_to_cap = {
executor.submit(cap, input_data): cap
for cap in self._capabilities
}
for future in concurrent.futures.as_completed(
future_to_cap, timeout=self._timeout,
):
try:
results.append(future.result())
except Exception as exc:
errors.append(exc)
if errors:
logger.warning(
"FanOutFanIn: %d of %d capabilities raised errors.",
len(errors), len(self._capabilities),
)
return self._aggregator(results)
# cca/composition/__init__.py
# CCA 0.3 — Composition package exports.
from .sequential import SequentialComposition
from .parallel import ParallelComposition
from .conditional import ConditionalComposition
from .pipeline import BackpressurePipeline, PipelineStage
from .saga import Saga, SagaStep, SagaExecutionError
from .fanout import FanOutFanIn
__all__ = [
"SequentialComposition",
"ParallelComposition",
"ConditionalComposition",
"BackpressurePipeline",
"PipelineStage",
"Saga",
"SagaStep",
"SagaExecutionError",
"FanOutFanIn",
]
CHAPTER 9: CAPABILITY CHANNELS
CCA 0.2 capabilities communicated synchronously. CCA 0.3 introduces capability channels, which provide an asynchronous, event-driven communication model between capabilities. A channel is a named, typed conduit through which capabilities can publish events and subscribe to events.
Channels in embedded systems: In an embedded system, capability channels map naturally to the message queues of an RTOS (FreeRTOS, Zephyr, ThreadX), to CAN bus message identifiers, to MQTT topics on an IoT gateway, or to shared memory regions in a multi-core embedded system. The channel abstraction hides the underlying transport mechanism from the capabilities that use it. A temperature sensor capability publishes to a "sensor.temperature" channel without knowing whether the subscriber is a local PID controller task, a remote SCADA system connected via Modbus, or a cloud analytics platform connected via MQTT. The channel registry acts as the broker, and the transport layer is swappable without changing the capability code.
# cca/channels.py
# CCA 0.3 — Capability Channels for asynchronous event-driven communication.
# Belongs to the Adaptation ring.
from __future__ import annotations
import threading
import time
import uuid
from dataclasses import dataclass
from typing import Any, Callable, Dict, Optional
@dataclass
class ChannelEvent:
"""
An event published to a capability channel.
The payload must conform to the channel's declared schema.
partition_key is used to preserve ordering within a logical partition.
In embedded systems, partition_key may be a sensor node ID or a
CAN bus arbitration ID.
"""
event_id: str
channel_name: str
publisher_id: str
partition_key: str
payload: Any
timestamp: float
schema_version: str = "1.0"
class ChannelSubscription:
"""
Represents an active subscription to a channel.
Calling cancel() removes the subscription from the channel.
"""
def __init__(
self,
subscription_id: str,
cancel_fn: Callable[[], None],
) -> None:
self._id = subscription_id
self._cancel_fn = cancel_fn
@property
def subscription_id(self) -> str:
return self._id
def cancel(self) -> None:
"""Unsubscribe from the channel."""
self._cancel_fn()
class CapabilityChannel:
"""
A CCA 0.3 capability channel.
Provides publish/subscribe communication between capabilities
without direct coupling between publisher and subscriber.
This in-process implementation is suitable for single-process and
embedded deployments. In distributed deployments, replace with a
broker-backed implementation using Apache Kafka, RabbitMQ, AWS
EventBridge, MQTT, or a FreeRTOS message queue.
"""
def __init__(self, name: str, schema: Dict[str, Any]) -> None:
self._name = name
self._schema = schema
self._lock = threading.RLock()
self._subscribers: Dict[str, Callable[[ChannelEvent], None]] = {}
@property
def name(self) -> str:
return self._name
def publish(
self,
publisher_id: str,
payload: Any,
partition_key: str = "",
) -> str:
"""
Publish an event to the channel. Returns the event_id.
Subscriber errors are swallowed so one bad subscriber cannot
prevent other subscribers from receiving the event.
"""
event = ChannelEvent(
event_id = str(uuid.uuid4()),
channel_name = self._name,
publisher_id = publisher_id,
partition_key = partition_key,
payload = payload,
timestamp = time.time(),
)
with self._lock:
handlers = list(self._subscribers.values())
for handler in handlers:
try:
handler(event)
except Exception:
pass
return event.event_id
def subscribe(
self,
handler: Callable[[ChannelEvent], None],
) -> ChannelSubscription:
"""
Subscribe to events on this channel.
Returns a ChannelSubscription whose cancel() method removes the subscription.
"""
sub_id = str(uuid.uuid4())
with self._lock:
self._subscribers[sub_id] = handler
def cancel() -> None:
with self._lock:
self._subscribers.pop(sub_id, None)
return ChannelSubscription(sub_id, cancel)
class ChannelRegistry:
"""
A registry of named capability channels.
Capabilities use this to find or create channels by name.
Thread-safe for concurrent access.
"""
def __init__(self) -> None:
self._channels: Dict[str, CapabilityChannel] = {}
self._lock = threading.RLock()
def get_or_create(
self,
channel_name: str,
schema: Dict[str, Any],
) -> CapabilityChannel:
"""Return an existing channel or create a new one with the given schema."""
with self._lock:
if channel_name not in self._channels:
self._channels[channel_name] = CapabilityChannel(
channel_name, schema,
)
return self._channels[channel_name]
def get(self, channel_name: str) -> Optional[CapabilityChannel]:
"""Return an existing channel, or None if it does not exist."""
with self._lock:
return self._channels.get(channel_name)
CHAPTER 10: THE DEPENDENCY GRAPH AND CYCLE DETECTION
As systems grow and capabilities accumulate, the web of dependencies between them becomes increasingly complex. CCA 0.3 introduces a formal dependency graph that is built automatically from the declarations of all registered capabilities, and a cycle detection algorithm that runs at registration time to prevent circular dependencies.
Dependency graphs in embedded systems: In an embedded system, the dependency graph governs the hardware initialization sequence. The UART capability depends on the clock capability. The Modbus capability depends on the UART capability. The temperature sensor capability depends on the I2C capability. The PID controller capability depends on the temperature sensor capability and the PWM output capability. The dependency graph makes this initialization order explicit and automatically verifiable. If a firmware engineer accidentally creates a circular dependency (for example, by making the clock capability depend on a timer capability that itself depends on the clock capability), the cycle is detected at startup before any hardware is touched.
# cca/dependency_graph.py
# CCA 0.3 — Formal Capability Dependency Graph with cycle detection.
# Operates on the Essence ring's dependency declarations.
# Uses iterative DFS to avoid Python's default recursion limit.
from __future__ import annotations
from typing import Dict, List, Optional, Set
from .exceptions import DependencyCycleError
class CapabilityDependencyGraph:
"""
Maintains a directed graph of capability dependencies.
Nodes are capability names; an edge A -> B means A depends on B
(B must be initialized before A).
Provides cycle detection and topological ordering for initialization.
Cycle detection uses iterative DFS to avoid Python's recursion limit.
In embedded systems, topological_order() determines the hardware
peripheral initialization sequence at firmware startup.
"""
def __init__(self) -> None:
self._adjacency: Dict[str, Set[str]] = {}
def add_capability(self, name: str, dependencies: List[str]) -> None:
"""
Add a capability and its dependencies to the graph.
Raises DependencyCycleError if adding this capability creates a cycle.
Rolls back the addition before raising so the graph remains consistent.
"""
self._adjacency[name] = set(dependencies)
for dep in dependencies:
if dep not in self._adjacency:
self._adjacency[dep] = set()
cycle = self._find_cycle()
if cycle:
del self._adjacency[name]
raise DependencyCycleError(cycle)
def topological_order(self) -> List[str]:
"""
Return a topological ordering of capabilities for initialization.
Capabilities with no dependencies come first.
An edge A -> B (A depends on B) means B must appear before A.
Raises DependencyCycleError if the graph contains a cycle.
Uses Kahn's algorithm (BFS-based), which is iterative and safe
for graphs of any size.
"""
precedes: Dict[str, Set[str]] = {node: set() for node in self._adjacency}
in_degree: Dict[str, int] = {node: 0 for node in self._adjacency}
for node in self._adjacency:
for dep in self._adjacency[node]:
precedes[dep].add(node)
in_degree[node] += 1
queue: List[str] = [n for n, d in in_degree.items() if d == 0]
order: List[str] = []
while queue:
node = queue.pop(0)
order.append(node)
for successor in precedes.get(node, set()):
in_degree[successor] -= 1
if in_degree[successor] == 0:
queue.append(successor)
if len(order) != len(self._adjacency):
cycle = self._find_cycle()
raise DependencyCycleError(cycle or ["unknown cycle"])
return order
def _find_cycle(self) -> Optional[List[str]]:
"""
Iterative DFS-based cycle detection.
Returns the cycle as a list of node names (start node repeated at end),
or None if the graph is acyclic.
Uses an explicit stack to avoid Python's recursion limit.
"""
visited: Set[str] = set()
rec_stack: Set[str] = set()
for start in self._adjacency:
if start in visited:
continue
stack = [(start, iter(self._adjacency.get(start, set())), [start])]
rec_stack.add(start)
while stack:
node, neighbors, path = stack[-1]
try:
neighbor = next(neighbors)
if neighbor not in visited:
if neighbor in rec_stack:
cycle_start = path.index(neighbor)
return path[cycle_start:] + [neighbor]
rec_stack.add(neighbor)
stack.append((
neighbor,
iter(self._adjacency.get(neighbor, set())),
path + [neighbor],
))
elif neighbor in rec_stack:
cycle_start = path.index(neighbor)
return path[cycle_start:] + [neighbor]
except StopIteration:
visited.add(node)
rec_stack.discard(node)
stack.pop()
return None
CHAPTER 11: ENHANCED OBSERVABILITY WITH SLO TRACKING
CCA 0.2 had basic observability in the Adaptation ring. CCA 0.3 takes observability to a new level by adding SLO tracking, health probes, and structured event emission. The observability subsystem is aware of the SLA declarations in capability contracts, and it automatically computes whether each capability is meeting its declared objectives.
Observability in embedded systems: In an embedded system, observability takes different forms than in a cloud system. There is no Prometheus, no Grafana, no distributed tracing backend. Instead, observability is implemented through hardware debug interfaces (JTAG, SWD), through non-volatile fault logs written to EEPROM or flash, through CAN bus diagnostic frames, through UART debug output, and through hardware performance counters. CCA 0.3's SLOTracker and ObservableCapabilityWrapper are designed to be backend-agnostic: the on_slo_violation callback can write to any output — a cloud metrics endpoint, a local UART, a CAN bus diagnostic frame, or a non-volatile fault log. This makes the observability model equally applicable to a cloud microservice and to a microcontroller with 64 KB of RAM.
# cca/observability.py
# CCA 0.3 — Enhanced Observability with SLO Tracking.
# Belongs to the Adaptation ring. Reads SLA declarations from the Essence ring.
from __future__ import annotations
import collections
import logging
import time
import threading
from dataclasses import dataclass
from typing import Any, Callable, Deque, Dict, Optional
logger = logging.getLogger(__name__)
@dataclass
class InvocationRecord:
"""Records the outcome of a single capability invocation."""
capability_name: str
started_at: float
duration_seconds: float
succeeded: bool
error_code: Optional[str] = None
class SLOTracker:
"""
Tracks SLO compliance for a single capability using a sliding window
of recent invocation records. Thread-safe.
In embedded systems, the window_size should be set to a small value
(e.g., 10–50) to limit RAM usage. The SLO thresholds should match
the real-time requirements of the control loop.
"""
def __init__(
self,
capability_name: str,
max_latency_p99_seconds: float,
max_error_rate: float,
window_size: int = 1000,
) -> None:
self._name = capability_name
self._max_latency = max_latency_p99_seconds
self._max_error = max_error_rate
self._window: Deque[InvocationRecord] = collections.deque(maxlen=window_size)
self._lock = threading.RLock()
def record(self, record: InvocationRecord) -> None:
"""Add an invocation record to the sliding window."""
with self._lock:
self._window.append(record)
def current_error_rate(self) -> float:
"""Return the fraction of recent invocations that failed."""
with self._lock:
if not self._window:
return 0.0
failures = sum(1 for r in self._window if not r.succeeded)
return failures / len(self._window)
def current_p99_latency(self) -> float:
"""Return the 99th percentile latency of recent invocations in seconds."""
with self._lock:
if not self._window:
return 0.0
latencies = sorted(r.duration_seconds for r in self._window)
idx = int(len(latencies) * 0.99)
return latencies[min(idx, len(latencies) - 1)]
def is_slo_compliant(self) -> bool:
"""
Return True if the capability is currently meeting its SLO.
Both the error rate and the p99 latency must be within the
bounds declared in the Essence ring's SLA declaration.
"""
return (
self.current_error_rate() <= self._max_error
and self.current_p99_latency() <= self._max_latency
)
def compliance_report(self) -> Dict[str, Any]:
"""Return a dictionary summarizing current SLO compliance."""
return {
"capability": self._name,
"sample_size": len(self._window),
"error_rate": round(self.current_error_rate(), 4),
"p99_latency_s": round(self.current_p99_latency(), 4),
"slo_compliant": self.is_slo_compliant(),
"max_error_rate": self._max_error,
"max_latency_p99": self._max_latency,
}
class ObservableCapabilityWrapper:
"""
Wraps any CCA 0.3 capability to add structured logging, SLO tracking,
and automatic notification when SLOs are violated.
The on_slo_violation callback can trigger a lifecycle DEGRADED transition,
write to a fault log, send a CAN diagnostic frame, or publish to a
cloud metrics endpoint — the callback is backend-agnostic.
"""
def __init__(
self,
capability: Any,
slo_tracker: SLOTracker,
on_slo_violation: Optional[Callable[[str, Dict[str, Any]], None]] = None,
) -> None:
self._cap = capability
self._slo_tracker = slo_tracker
self._on_slo_violation = on_slo_violation
self.name = capability.name
self.version = capability.version
def execute(self, input_data: Any, context: Any = None) -> Any:
"""
Execute the wrapped capability, recording timing and outcome,
and checking SLO compliance after each invocation.
"""
started_at = time.monotonic()
succeeded = False
error_code: Optional[str] = None
logger.info(
"Capability '%s' v%s: invocation started.", self.name, self.version,
)
try:
result = self._cap.execute(input_data, context)
succeeded = True
logger.info(
"Capability '%s' v%s: invocation succeeded.",
self.name, self.version,
)
return result
except Exception as exc:
error_code = getattr(exc, "error_code", type(exc).__name__)
logger.error(
"Capability '%s' v%s: invocation failed with %s: %s.",
self.name, self.version, error_code, exc,
)
raise
finally:
duration = time.monotonic() - started_at
rec = InvocationRecord(
capability_name = self.name,
started_at = started_at,
duration_seconds = duration,
succeeded = succeeded,
error_code = error_code,
)
self._slo_tracker.record(rec)
if not self._slo_tracker.is_slo_compliant():
report = self._slo_tracker.compliance_report()
logger.warning(
"Capability '%s': SLO violation detected: %s",
self.name, report,
)
if self._on_slo_violation:
self._on_slo_violation(self.name, report)
CHAPTER 12: THE FULL CCA 0.3 CAPABILITY BASE CLASS
Now that all the new components of CCA 0.3 have been introduced, we can look at the full Capability base class that ties everything together. This class embodies the three-ring model directly: the __init__ parameters and the lifecycle machine represent the Essence ring, the _execute method represents the Realization ring, and the policies, observability hooks, and security enforcement in execute represent the Adaptation ring.
# cca/capability.py
# CCA 0.3 — The unified Capability base class.
# Embodies the three-ring model: Essence, Realization, and Adaptation.
from __future__ import annotations
import uuid
from typing import Any, List, Optional
from .lifecycle import LifecycleMachine, LifecycleState
from .context import CapabilityContext
from .contract import Contract
from .security import CapabilityToken, TokenAuthority
from .exceptions import (
CapabilitySecurityError,
DeadlineExceededError,
CapabilityNotReadyError,
)
class Capability:
"""
The CCA 0.3 base capability class. Embodies the three-ring model.
Essence ring : name, version, contract, dependencies, required_permission,
lifecycle machine.
Realization : _execute() — overridden by concrete subclasses.
Adaptation : policies, token_authority, security enforcement,
contract validation, behavioral assertion checking.
The execute() method enforces a nine-step protocol covering all
cross-cutting concerns so that _execute() contains only business logic
(or firmware logic in embedded systems).
Circuit breaker policies must implement record_failure(name: str) to be
notified when the Realization ring raises an exception.
Embedded use: subclass Capability for each hardware peripheral or
firmware function. The three-ring model ensures that hardware-specific
code (Realization) is isolated from operational policies (Adaptation)
and from the interface contract (Essence), making it straightforward
to swap sensor ICs, change communication protocols, or adjust
real-time policies without touching the rest of the firmware.
"""
def __init__(
self,
name: str,
version: str,
contract: Contract,
dependencies: List[str],
policies: List[Any],
required_permission: str = "invoke",
token_authority: Optional[TokenAuthority] = None,
) -> None:
# --- Essence ring ---
self.name = name
self.version = version
self.contract = contract
self.dependencies = dependencies
self._required_permission = required_permission
self._lifecycle = LifecycleMachine(name)
self.instance_id = str(uuid.uuid4())
# --- Adaptation ring ---
self.policies = policies
self._token_authority = token_authority
@property
def lifecycle_state(self) -> LifecycleState:
"""Expose the current lifecycle state from the Essence ring."""
return self._lifecycle.state
def add_lifecycle_observer(self, observer: Any) -> None:
"""Register a lifecycle observer on the Essence ring's lifecycle machine."""
self._lifecycle.add_observer(observer)
def initialize(self) -> None:
"""
Initialize the capability.
Transitions: CREATED -> INITIALIZING -> READY (on success)
CREATED -> INITIALIZING -> RETIRED (on failure)
Calls _on_initialize() for subclass resource setup.
In embedded systems, _on_initialize() runs the hardware peripheral
initialization sequence (clock enable, register config, self-test).
"""
self._lifecycle.transition_to(LifecycleState.INITIALIZING)
try:
self._on_initialize()
self._lifecycle.transition_to(LifecycleState.READY)
except Exception as exc:
self._lifecycle.transition_to(LifecycleState.RETIRED)
raise RuntimeError(
f"Capability '{self.name}' failed to initialize: {exc}"
) from exc
def retire(self) -> None:
"""Gracefully retire the capability, releasing resources.
In embedded systems, _on_retire() disables the peripheral and
releases any DMA channels or interrupt vectors it was using.
"""
self._on_retire()
self._lifecycle.transition_to(LifecycleState.RETIRED)
def execute(self, input_data: Any, context: CapabilityContext) -> Any:
"""
The public invocation entry point. Enforces a nine-step protocol:
1. Lifecycle check — only READY capabilities accept invocations.
2. Deadline check — fail fast if the deadline has already passed.
3. Security check — verify the context carries a valid token.
4. Input validation — contract schema check (Essence ring).
5. Pre-policies — Adaptation ring before_execute hooks.
6. Execution — delegate to _execute() (Realization ring).
7. Output validation — contract schema check (Essence ring).
8. Assertions — behavioral assertion check (Essence ring).
9. Post-policies — Adaptation ring after_execute hooks.
On exception in step 6, circuit breaker policies are notified via
record_failure() before the exception is re-raised.
"""
# Step 1: Lifecycle check.
if self._lifecycle.state != LifecycleState.READY:
raise CapabilityNotReadyError(
f"Capability '{self.name}' is in state "
f"{self._lifecycle.state.name} and cannot accept invocations."
)
# Step 2: Deadline check.
if context.is_expired():
raise DeadlineExceededError(
f"Capability '{self.name}': invocation deadline has passed."
)
# Step 3: Security check.
if self._token_authority is not None:
self._verify_security(context)
# Step 4: Contract input validation (Essence ring).
self.contract.validate_input(input_data)
# Step 5: Pre-execution policies (Adaptation ring).
for policy in self.policies:
if hasattr(policy, "before_execute"):
policy.before_execute(self, input_data, context)
# Step 6: Execute the Realization ring.
try:
result = self._execute(input_data, context)
except Exception:
for policy in self.policies:
if hasattr(policy, "record_failure"):
policy.record_failure(self.name)
raise
# Step 7: Contract output validation (Essence ring).
self.contract.validate_output(result)
# Step 8: Behavioral assertion checking (Essence ring).
self.contract.check_assertions(input_data, result)
# Step 9: Post-execution policies (Adaptation ring).
for policy in self.policies:
if hasattr(policy, "after_execute"):
policy.after_execute(self, input_data, result, context)
return result
def _execute(self, input_data: Any, context: CapabilityContext) -> Any:
"""
Realization ring: subclasses override this to implement business logic
or firmware logic. The context is available for deadline checking,
tracing, and tenant/device-specific logic.
"""
raise NotImplementedError(
f"Capability '{self.name}' must implement _execute()."
)
def _on_initialize(self) -> None:
"""
Called during initialization (Realization ring setup).
Subclasses override this to acquire resources such as DB connections,
hardware peripheral handles, DMA channels, or interrupt vectors.
"""
pass
def _on_retire(self) -> None:
"""
Called before retirement (Realization ring teardown).
Subclasses override this to release resources gracefully.
In embedded systems: disable peripheral, release DMA, deregister ISR.
"""
pass
def _verify_security(self, context: CapabilityContext) -> None:
"""
Adaptation ring: verify that the context contains a valid token
for this capability with the required permission.
Raises CapabilitySecurityError if verification fails.
"""
matching_token: Optional[CapabilityToken] = None
for token in context.security.tokens:
if (
token.capability_name == self.name
and token.capability_version == self.version
):
matching_token = token
break
if matching_token is None:
raise CapabilitySecurityError(
f"No token found for capability '{self.name}' v{self.version}."
)
if not self._token_authority.verify(matching_token):
raise CapabilitySecurityError(
f"Token for capability '{self.name}' is invalid or expired."
)
if not matching_token.has_permission(self._required_permission):
raise CapabilitySecurityError(
f"Token for capability '{self.name}' does not grant "
f"permission '{self._required_permission}'."
)
# cca/__init__.py
# CCA 0.3 — Package root. Exports the primary public API.
from .exceptions import (
CCAError,
LifecycleError,
CapabilitySecurityError,
DeadlineExceededError,
CapabilityNotReadyError,
DependencyCycleError,
)
from .capability import Capability
from .lifecycle import LifecycleMachine, LifecycleState
from .context import CapabilityContext, TraceContext, SecurityContext
from .contract import Contract, SLADeclaration, ErrorDescriptor
from .security import CapabilityToken, TokenAuthority
from .mesh import CapabilityMesh, CapabilityRegistration, RoutingStrategy
from .channels import CapabilityChannel, ChannelRegistry, ChannelEvent
from .dependency_graph import CapabilityDependencyGraph
from .observability import SLOTracker, ObservableCapabilityWrapper, InvocationRecord
from .policies import (
RetryPolicy,
CircuitBreakerPolicy,
CircuitBreakerOpenError,
RateLimitingPolicy,
RateLimitExceededError,
TimeoutPolicy,
)
__all__ = [
# Exceptions
"CCAError",
"LifecycleError",
"CapabilitySecurityError",
"DeadlineExceededError",
"CapabilityNotReadyError",
"DependencyCycleError",
# Capability base
"Capability",
# Lifecycle
"LifecycleMachine",
"LifecycleState",
# Context
"CapabilityContext",
"TraceContext",
"SecurityContext",
# Contract
"Contract",
"SLADeclaration",
"ErrorDescriptor",
# Security
"CapabilityToken",
"TokenAuthority",
# Mesh
"CapabilityMesh",
"CapabilityRegistration",
"RoutingStrategy",
# Channels
"CapabilityChannel",
"ChannelRegistry",
"ChannelEvent",
# Dependency graph
"CapabilityDependencyGraph",
# Observability
"SLOTracker",
"ObservableCapabilityWrapper",
"InvocationRecord",
# Policies
"RetryPolicy",
"CircuitBreakerPolicy",
"CircuitBreakerOpenError",
"RateLimitingPolicy",
"RateLimitExceededError",
"TimeoutPolicy",
]
CHAPTER 13: THE CAPABILITY TESTING FRAMEWORK
Testing was not a first-class concern in CCA 0.2. CCA 0.3 changes this by introducing a built-in testing framework with three distinct testing modes: unit testing of individual capabilities, contract testing to verify that a capability honors its declared contract, and chaos testing to verify that a capability behaves correctly under adverse conditions.
Testing in embedded systems: Embedded systems have historically been difficult to test because the code is tightly coupled to hardware. CCA 0.3's testing framework addresses this directly. Because every capability's Realization ring is hidden behind a contract, it can be tested against a mock hardware abstraction layer (HAL) without any physical hardware present. This is the software-in-the-loop (SIL) testing model. The MockCapability class can simulate a temperature sensor returning a sequence of values, including out-of-range values and communication errors, allowing the PID controller capability and the safety monitor capability to be tested exhaustively without a physical sensor. For hardware-in-the-loop (HIL) testing, the mock is replaced with the real hardware, and the same test cases run unchanged — because the interface (the contract) is identical.
# cca/testing.py
# CCA 0.3 — Built-in Capability Testing Framework.
# Tests the Realization ring against the Essence ring's contract.
# Supports unit, contract, and chaos testing modes.
# Applicable to both enterprise software and embedded firmware (SIL/HIL).
from __future__ import annotations
import time
import unittest
from typing import Any, List, Optional
class MockCapability:
"""
A test double for a CCA 0.3 capability.
Supports configuring return values and exceptions in sequence.
Records all calls for later assertion in tests.
In embedded testing (SIL): use MockCapability to simulate sensors,
actuators, and communication peripherals without physical hardware.
In HIL testing: replace MockCapability with the real capability backed
by physical hardware — the test cases are identical.
"""
def __init__(self, name: str, version: str = "1.0.0") -> None:
self.name = name
self.version = version
self._responses: List[Any] = []
self._call_log: List[dict] = []
self._response_index = 0
def will_return(self, *values: Any) -> MockCapability:
"""Configure the mock to return the given values in sequence."""
self._responses.extend(values)
return self
def will_raise(self, exception: Exception) -> MockCapability:
"""Configure the mock to raise the given exception on the next call."""
self._responses.append(exception)
return self
def execute(self, input_data: Any, context: Any = None) -> Any:
"""
Return the next configured response, or raise it if it is an exception.
Records every call for later inspection.
Raises RuntimeError if no more responses have been configured.
"""
self._call_log.append({"input": input_data, "context": context})
if self._response_index >= len(self._responses):
raise RuntimeError(
f"MockCapability '{self.name}': no more configured responses. "
f"Call will_return() or will_raise() to add more."
)
response = self._responses[self._response_index]
self._response_index += 1
if isinstance(response, Exception):
raise response
return response
@property
def call_count(self) -> int:
"""Return the total number of times execute() has been called."""
return len(self._call_log)
@property
def last_call_input(self) -> Optional[Any]:
"""Return the input from the most recent execute() call, or None."""
return self._call_log[-1]["input"] if self._call_log else None
class ContractTestCase(unittest.TestCase):
"""
Base class for CCA 0.3 contract tests.
Subclasses must set capability_under_test, valid_test_inputs, and
invalid_test_inputs as class-level attributes before the tests run.
The framework then automatically generates and runs contract verification
tests, checking both the Essence ring's schema validation and its
behavioral assertions.
"""
capability_under_test: Any = None
valid_test_inputs: List[Any] = []
invalid_test_inputs: List[Any] = []
def setUp(self) -> None:
if self.capability_under_test is None:
self.skipTest(
"ContractTestCase: capability_under_test is not set. "
"Set it as a class attribute in the subclass."
)
def test_valid_inputs_succeed(self) -> None:
"""All valid inputs should produce outputs that pass contract validation."""
cap = self.capability_under_test
for valid_input in self.valid_test_inputs:
with self.subTest(input=valid_input):
result = cap.execute(valid_input)
cap.contract.validate_output(result)
def test_invalid_inputs_raise_contract_error(self) -> None:
"""All invalid inputs should raise ValueError from contract validation."""
cap = self.capability_under_test
for invalid_input in self.invalid_test_inputs:
with self.subTest(input=invalid_input):
with self.assertRaises((ValueError, TypeError)):
cap.execute(invalid_input)
def test_behavioral_assertions_hold(self) -> None:
"""Behavioral assertions from the Essence ring must hold for all valid inputs."""
cap = self.capability_under_test
for valid_input in self.valid_test_inputs:
with self.subTest(input=valid_input):
result = cap.execute(valid_input)
cap.contract.check_assertions(valid_input, result)
class ChaosTestCase(unittest.TestCase):
"""
Base class for CCA 0.3 chaos tests.
Injects failures into capability dependencies and verifies that the
Adaptation ring's policies (retry, circuit breaker, timeout) respond correctly.
In embedded testing, chaos tests inject I2C bus errors, ADC timeouts,
and power glitches to verify the robustness of the Adaptation ring.
"""
def assert_retried_n_times(
self,
mock_dep: MockCapability,
expected_calls: int,
) -> None:
"""Assert that a mock dependency was called exactly expected_calls times."""
self.assertEqual(
mock_dep.call_count,
expected_calls,
f"Expected {expected_calls} calls to '{mock_dep.name}', "
f"got {mock_dep.call_count}.",
)
def assert_completes_within(
self,
capability: Any,
input_data: Any,
max_seconds: float,
context: Any = None,
) -> Any:
"""Assert that a capability completes within the given time budget."""
start = time.monotonic()
result = capability.execute(input_data, context)
elapsed = time.monotonic() - start
self.assertLessEqual(
elapsed,
max_seconds,
f"Capability '{capability.name}' took {elapsed:.3f}s, "
f"exceeding the {max_seconds}s SLA budget.",
)
return result
CHAPTER 14: CCA 0.3 IN EMBEDDED SYSTEMS
This chapter provides concrete, runnable embedded systems examples that demonstrate how CCA 0.3 applies to firmware and edge computing environments. These examples run on CPython (for development and SIL testing) and are structured so that the Realization ring can be replaced with MicroPython or C implementations for deployment on actual hardware.
THE EMBEDDED ARCHITECTURE
Consider a smart industrial temperature monitoring node. The node consists of a microcontroller (or a Raspberry Pi for demonstration purposes) connected to a temperature sensor via I2C, a status LED via GPIO, and a cloud backend via MQTT. The node has four capabilities:
TemperatureSensorCapability— reads temperature from an I2C sensor (Realization: hardware I2C read)LEDStatusCapability— controls a status LED via GPIO (Realization: GPIO write)TemperatureAlertCapability— applies threshold logic and generates alerts (Realization: pure logic)MQTTGatewayCapability— publishes sensor data to a cloud MQTT broker (Realization: network I/O)
These four capabilities are composed using sequential composition (sensor → alert → gateway) and the LED is driven by a lifecycle observer on the sensor capability. The dependency graph ensures correct initialization order: I2C bus → sensor → alert logic → MQTT gateway.
CapabilityDependencyGraph (embedded node)
|
+-- i2c-bus (no dependencies)
+-- gpio-controller (no dependencies)
+-- temperature-sensor (depends on: i2c-bus)
+-- led-status (depends on: gpio-controller)
+-- temperature-alert (depends on: temperature-sensor)
+-- mqtt-gateway (depends on: temperature-alert)
Sequential composition:
temperature-sensor -> temperature-alert -> mqtt-gateway
Lifecycle observer on temperature-sensor:
DEGRADED -> led-status blinks red
READY -> led-status shows green
RETIRED -> led-status off
Channel: sensor.temperature.readings
publisher: temperature-sensor
subscriber: temperature-alert (primary)
subscriber: cloud-dashboard (secondary, via MQTT gateway)
# capabilities/embedded/__init__.py
# CCA 0.3 — Embedded capabilities package.
from .temperature_sensor import TemperatureSensorCapability, TEMPERATURE_SENSOR_CONTRACT
from .led_status import LEDStatusCapability, LED_STATUS_CONTRACT
from .temperature_alert import TemperatureAlertCapability, TEMPERATURE_ALERT_CONTRACT
from .mqtt_gateway import MQTTGatewayCapability, MQTT_GATEWAY_CONTRACT
__all__ = [
"TemperatureSensorCapability",
"TEMPERATURE_SENSOR_CONTRACT",
"LEDStatusCapability",
"LED_STATUS_CONTRACT",
"TemperatureAlertCapability",
"TEMPERATURE_ALERT_CONTRACT",
"MQTTGatewayCapability",
"MQTT_GATEWAY_CONTRACT",
]
# capabilities/embedded/temperature_sensor.py
# CCA 0.3 — Temperature sensor capability.
# Essence ring: contract declares physical units, range, accuracy, and SLA.
# Realization ring: reads from an I2C temperature sensor (stubbed for SIL).
# Adaptation ring: retry on I2C bus errors, circuit breaker on repeated failures.
#
# Hardware target: any I2C temperature sensor (TMP102, MCP9808, SHT31, etc.)
# SIL target: StubI2CTemperatureSensor (included below for testing)
# MicroPython note: replace threading.RLock with a no-op lock and
# replace uuid.uuid4() with a monotonic counter.
from __future__ import annotations
from typing import Any
from cca.capability import Capability, DeadlineExceededError
from cca.contract import Contract, SLADeclaration, ErrorDescriptor
# ---------------------------------------------------------------------------
# Contract (Essence ring) — defines the evolution envelope.
# Physical constraints are expressed as behavioral assertions.
# ---------------------------------------------------------------------------
TEMPERATURE_SENSOR_CONTRACT = Contract(
input_schema={
# oversample_count: number of ADC samples to average (1–16).
"oversample_count": int,
},
output_schema={
# temperature_c: measured temperature in degrees Celsius.
"temperature_c": (int, float),
# raw_adc_counts: raw ADC value before conversion (for diagnostics).
"raw_adc_counts": int,
# sensor_id: hardware identifier of the sensor that was read.
"sensor_id": str,
},
sla=SLADeclaration(
# Real-time requirement: sensor reading must complete within 50 ms.
max_latency_p99_seconds = 0.050,
# Fault tolerance: at most 1% of readings may fail.
max_error_rate = 0.01,
# Minimum sampling rate for the control loop: 10 Hz.
min_throughput_rps = 10.0,
),
error_catalog=[
ErrorDescriptor(
error_code = "I2C_BUS_ERROR",
description = "I2C communication error; sensor did not ACK.",
severity = "ERROR",
retryable = True,
),
ErrorDescriptor(
error_code = "SENSOR_OUT_OF_RANGE",
description = "Sensor returned a value outside the physical range.",
severity = "WARNING",
retryable = False,
),
ErrorDescriptor(
error_code = "SENSOR_NOT_READY",
description = "Sensor conversion not complete within timeout.",
severity = "WARNING",
retryable = True,
),
],
assertions=[
# Physical range: TMP102 operating range is -40°C to +125°C.
lambda inp, out: -40.0 <= out["temperature_c"] <= 125.0,
# Oversample count must produce a valid raw value (non-negative).
lambda inp, out: out["raw_adc_counts"] >= 0,
# Sensor ID must be a non-empty string.
lambda inp, out: len(out["sensor_id"]) > 0,
],
)
class TemperatureSensorCapability(Capability):
"""
CCA 0.3 temperature sensor capability.
Essence ring : name="temperature-sensor", version="1.0.0",
TEMPERATURE_SENSOR_CONTRACT,
dependencies=["i2c-bus"].
Realization : _execute() reads from the I2C sensor via the HAL.
Adaptation : policies (retry, circuit breaker) supplied at construction.
The i2c_hal parameter is a hardware abstraction layer object that must
implement read_temperature(oversample_count: int) -> (float, int, str).
This abstraction makes the capability testable without physical hardware
(SIL testing) and swappable between different sensor ICs without
changing the contract or the Adaptation ring.
"""
def __init__(self, i2c_hal: Any, policies: list = None) -> None:
super().__init__(
name = "temperature-sensor",
version = "1.0.0",
contract = TEMPERATURE_SENSOR_CONTRACT,
dependencies = ["i2c-bus"],
policies = policies or [],
)
self._i2c_hal = i2c_hal
def _on_initialize(self) -> None:
"""
Verify that the I2C HAL is reachable and the sensor responds.
In production: send a configuration register write and read it back.
"""
# Ping the sensor to verify it is present on the bus.
self._i2c_hal.ping()
def _on_retire(self) -> None:
"""
Place the sensor in low-power shutdown mode before retiring.
In production: write the shutdown bit to the sensor's config register.
"""
self._i2c_hal.shutdown()
def _execute(self, input_data: Any, context: Any) -> Any:
"""
Realization ring: read temperature from the I2C sensor.
Checks the deadline before initiating the (potentially slow) I2C read.
"""
if context.is_expired():
raise DeadlineExceededError(
"Deadline expired before I2C temperature read could be performed."
)
oversample_count = input_data["oversample_count"]
try:
temperature_c, raw_adc_counts, sensor_id = (
self._i2c_hal.read_temperature(oversample_count)
)
except OSError as exc:
# I2C bus error (NACK, bus collision, timeout).
err = RuntimeError("I2C bus error during temperature read.")
err.error_code = "I2C_BUS_ERROR" # type: ignore[attr-defined]
raise err from exc
# Validate physical range before returning (defense in depth).
if not (-40.0 <= temperature_c <= 125.0):
err = RuntimeError(
f"Sensor returned out-of-range value: {temperature_c}°C"
)
err.error_code = "SENSOR_OUT_OF_RANGE" # type: ignore[attr-defined]
raise err
return {
"temperature_c": temperature_c,
"raw_adc_counts": raw_adc_counts,
"sensor_id": sensor_id,
}
# capabilities/embedded/led_status.py
# CCA 0.3 — LED status indicator capability.
# Essence ring: contract declares valid LED states and blink patterns.
# Realization ring: writes to a GPIO pin via the HAL.
# Adaptation ring: no retry (GPIO writes are instantaneous and idempotent).
from __future__ import annotations
from typing import Any
from cca.capability import Capability
from cca.contract import Contract, SLADeclaration, ErrorDescriptor
LED_STATUS_CONTRACT = Contract(
input_schema={
# state: one of "off", "on", "blink_slow", "blink_fast", "blink_sos"
"state": str,
},
output_schema={
# previous_state: the LED state before this command was applied.
"previous_state": str,
# current_state: the LED state after this command was applied.
"current_state": str,
},
sla=SLADeclaration(
# GPIO writes are near-instantaneous; 5 ms is very conservative.
max_latency_p99_seconds = 0.005,
max_error_rate = 0.0, # GPIO writes must never fail.
min_throughput_rps = 100.0,
),
error_catalog=[
ErrorDescriptor(
error_code = "INVALID_LED_STATE",
description = "The requested LED state is not supported.",
severity = "ERROR",
retryable = False,
),
ErrorDescriptor(
error_code = "GPIO_WRITE_ERROR",
description = "GPIO write failed (hardware fault).",
severity = "CRITICAL",
retryable = True,
),
],
assertions=[
# The current_state in the output must match the requested state.
lambda inp, out: out["current_state"] == inp["state"],
# previous_state must be a valid state string.
lambda inp, out: out["previous_state"] in {
"off", "on", "blink_slow", "blink_fast", "blink_sos",
},
],
)
_VALID_STATES = {"off", "on", "blink_slow", "blink_fast", "blink_sos"}
class LEDStatusCapability(Capability):
"""
CCA 0.3 LED status indicator capability.
Essence ring : name="led-status", version="1.0.0",
LED_STATUS_CONTRACT, dependencies=["gpio-controller"].
Realization : _execute() writes to a GPIO pin via the HAL.
Adaptation : no retry policy (GPIO writes are idempotent).
The gpio_hal parameter must implement:
set_state(state: str) -> None
get_state() -> str
"""
def __init__(self, gpio_hal: Any, policies: list = None) -> None:
super().__init__(
name = "led-status",
version = "1.0.0",
contract = LED_STATUS_CONTRACT,
dependencies = ["gpio-controller"],
policies = policies or [],
)
self._gpio_hal = gpio_hal
def _on_initialize(self) -> None:
"""Initialize the LED to the 'off' state."""
self._gpio_hal.set_state("off")
def _execute(self, input_data: Any, context: Any) -> Any:
"""Realization ring: apply the requested LED state via the GPIO HAL."""
requested_state = input_data["state"]
if requested_state not in _VALID_STATES:
err = ValueError(
f"Invalid LED state: '{requested_state}'. "
f"Valid states: {sorted(_VALID_STATES)}"
)
err.error_code = "INVALID_LED_STATE" # type: ignore[attr-defined]
raise err
previous_state = self._gpio_hal.get_state()
self._gpio_hal.set_state(requested_state)
return {
"previous_state": previous_state,
"current_state": requested_state,
}
# capabilities/embedded/temperature_alert.py
# CCA 0.3 — Temperature threshold alert capability.
# Essence ring: contract declares alert thresholds and output structure.
# Realization ring: pure logic — no hardware dependency.
# Adaptation ring: no retry (pure computation, no I/O).
#
# This capability is identical whether running on a Cortex-M4 or in the cloud.
# It demonstrates that CCA 0.3 capabilities can be hardware-independent
# even within an embedded system.
from __future__ import annotations
from typing import Any
from cca.capability import Capability
from cca.contract import Contract, SLADeclaration
TEMPERATURE_ALERT_CONTRACT = Contract(
input_schema={
"temperature_c": (int, float),
"high_threshold_c": (int, float),
"low_threshold_c": (int, float),
},
output_schema={
"alert_level": str, # "NORMAL", "HIGH", "LOW", "CRITICAL_HIGH", "CRITICAL_LOW"
"alert_message": str,
"temperature_c": (int, float),
},
sla=SLADeclaration(
# Pure computation: should complete in under 1 ms.
max_latency_p99_seconds = 0.001,
max_error_rate = 0.0,
min_throughput_rps = 1000.0,
),
assertions=[
# Alert level must be one of the defined values.
lambda inp, out: out["alert_level"] in {
"NORMAL", "HIGH", "LOW", "CRITICAL_HIGH", "CRITICAL_LOW"
},
# Temperature in output must match temperature in input.
lambda inp, out: abs(out["temperature_c"] - inp["temperature_c"]) < 0.001,
# If temperature is within thresholds, alert level must be NORMAL.
lambda inp, out: not (
inp["low_threshold_c"] <= inp["temperature_c"] <= inp["high_threshold_c"]
) or out["alert_level"] == "NORMAL",
],
)
# Critical thresholds are 10°C beyond the user-configured thresholds.
_CRITICAL_MARGIN_C = 10.0
class TemperatureAlertCapability(Capability):
"""
CCA 0.3 temperature threshold alert capability.
Pure logic: no hardware dependency. Identical on embedded and cloud.
Essence ring : name="temperature-alert", version="1.0.0",
TEMPERATURE_ALERT_CONTRACT,
dependencies=["temperature-sensor"].
Realization : _execute() applies threshold logic.
Adaptation : no policies needed (pure computation).
"""
def __init__(self, policies: list = None) -> None:
super().__init__(
name = "temperature-alert",
version = "1.0.0",
contract = TEMPERATURE_ALERT_CONTRACT,
dependencies = ["temperature-sensor"],
policies = policies or [],
)
def _execute(self, input_data: Any, context: Any) -> Any:
"""Realization ring: apply threshold logic to determine alert level."""
temp = input_data["temperature_c"]
high_thr = input_data["high_threshold_c"]
low_thr = input_data["low_threshold_c"]
if temp >= high_thr + _CRITICAL_MARGIN_C:
level = "CRITICAL_HIGH"
message = (
f"CRITICAL: Temperature {temp:.1f}°C exceeds critical "
f"high threshold {high_thr + _CRITICAL_MARGIN_C:.1f}°C."
)
elif temp >= high_thr:
level = "HIGH"
message = (
f"WARNING: Temperature {temp:.1f}°C exceeds high "
f"threshold {high_thr:.1f}°C."
)
elif temp <= low_thr - _CRITICAL_MARGIN_C:
level = "CRITICAL_LOW"
message = (
f"CRITICAL: Temperature {temp:.1f}°C is below critical "
f"low threshold {low_thr - _CRITICAL_MARGIN_C:.1f}°C."
)
elif temp <= low_thr:
level = "LOW"
message = (
f"WARNING: Temperature {temp:.1f}°C is below low "
f"threshold {low_thr:.1f}°C."
)
else:
level = "NORMAL"
message = f"Temperature {temp:.1f}°C is within normal range."
return {
"alert_level": level,
"alert_message": message,
"temperature_c": temp,
}
# capabilities/embedded/mqtt_gateway.py
# CCA 0.3 — MQTT gateway capability: the embedded-to-enterprise bridge.
# Essence ring: contract declares the message schema published to the broker.
# Realization ring: publishes to an MQTT broker via the HAL.
# Adaptation ring: retry on network errors, circuit breaker on broker outage.
#
# This capability is the bridge between the embedded world (sensor data)
# and the enterprise world (cloud analytics, SCADA, dashboards).
# The same CCA 0.3 contract governs both sides of the bridge.
from __future__ import annotations
import json
from typing import Any
from cca.capability import Capability, DeadlineExceededError
from cca.contract import Contract, SLADeclaration, ErrorDescriptor
MQTT_GATEWAY_CONTRACT = Contract(
input_schema={
# topic: MQTT topic string (e.g., "factory/line1/node3/temperature")
"topic": str,
# payload: dict to be JSON-serialized and published.
"payload": dict,
# qos: MQTT QoS level (0, 1, or 2).
"qos": int,
},
output_schema={
# message_id: MQTT message identifier assigned by the broker.
"message_id": int,
# topic: the topic the message was published to.
"topic": str,
# published: True if the broker acknowledged the message.
"published": bool,
},
sla=SLADeclaration(
# Network I/O: 500 ms is acceptable for a non-real-time gateway.
max_latency_p99_seconds = 0.500,
max_error_rate = 0.005, # 0.5% — network is unreliable.
min_throughput_rps = 5.0,
),
error_catalog=[
ErrorDescriptor(
error_code = "MQTT_CONNECTION_LOST",
description = "Lost connection to the MQTT broker.",
severity = "ERROR",
retryable = True,
),
ErrorDescriptor(
error_code = "MQTT_PUBLISH_TIMEOUT",
description = "Broker did not acknowledge the message within timeout.",
severity = "WARNING",
retryable = True,
),
ErrorDescriptor(
error_code = "INVALID_QOS",
description = "QoS level must be 0, 1, or 2.",
severity = "ERROR",
retryable = False,
),
],
assertions=[
# QoS must be 0, 1, or 2.
lambda inp, out: inp["qos"] in {0, 1, 2},
# Published flag must be True for QoS > 0.
lambda inp, out: inp["qos"] == 0 or out["published"],
# Output topic must match input topic.
lambda inp, out: out["topic"] == inp["topic"],
],
)
class MQTTGatewayCapability(Capability):
"""
CCA 0.3 MQTT gateway capability: the embedded-to-enterprise bridge.
Essence ring : name="mqtt-gateway", version="1.0.0",
MQTT_GATEWAY_CONTRACT,
dependencies=["temperature-alert"].
Realization : _execute() publishes to an MQTT broker via the HAL.
Adaptation : retry on network errors, circuit breaker on broker outage.
The mqtt_hal parameter must implement:
publish(topic: str, payload: str, qos: int) -> (int, bool)
is_connected() -> bool
connect() -> None
In production: use paho-mqtt, asyncio-mqtt, or a MicroPython MQTT client.
In SIL testing: use StubMQTTHAL (included in tests/).
"""
def __init__(self, mqtt_hal: Any, policies: list = None) -> None:
super().__init__(
name = "mqtt-gateway",
version = "1.0.0",
contract = MQTT_GATEWAY_CONTRACT,
dependencies = ["temperature-alert"],
policies = policies or [],
)
self._mqtt_hal = mqtt_hal
def _on_initialize(self) -> None:
"""Establish the MQTT broker connection during initialization."""
self._mqtt_hal.connect()
def _on_retire(self) -> None:
"""Disconnect from the MQTT broker gracefully."""
if hasattr(self._mqtt_hal, "disconnect"):
self._mqtt_hal.disconnect()
def _execute(self, input_data: Any, context: Any) -> Any:
"""Realization ring: publish the payload to the MQTT broker."""
if context.is_expired():
raise DeadlineExceededError(
"Deadline expired before MQTT publish could be performed."
)
topic = input_data["topic"]
payload = input_data["payload"]
qos = input_data["qos"]
if qos not in {0, 1, 2}:
err = ValueError(f"Invalid QoS level: {qos}. Must be 0, 1, or 2.")
err.error_code = "INVALID_QOS" # type: ignore[attr-defined]
raise err
if not self._mqtt_hal.is_connected():
try:
self._mqtt_hal.connect()
except Exception as exc:
err = RuntimeError("Lost connection to MQTT broker.")
err.error_code = "MQTT_CONNECTION_LOST" # type: ignore[attr-defined]
raise err from exc
try:
message_id, published = self._mqtt_hal.publish(
topic,
json.dumps(payload),
qos,
)
except Exception as exc:
err = RuntimeError("MQTT publish timed out.")
err.error_code = "MQTT_PUBLISH_TIMEOUT" # type: ignore[attr-defined]
raise err from exc
return {
"message_id": message_id,
"topic": topic,
"published": published,
}
STUB HARDWARE ABSTRACTION LAYERS FOR SIL TESTING
The following stub HAL implementations allow all four embedded capabilities to be tested without physical hardware. They are used in the embedded demo (main_embedded.py) and in the embedded test suite.
# capabilities/embedded/stubs.py
# CCA 0.3 — Stub hardware abstraction layers for SIL (Software-in-the-Loop) testing.
# Replace these with real HAL implementations for HIL (Hardware-in-the-Loop) testing
# or for production deployment on physical hardware.
from __future__ import annotations
import time
from typing import Tuple
class StubI2CTemperatureSensor:
"""
Stub HAL for a TMP102-compatible I2C temperature sensor.
Returns a configurable sequence of temperature readings.
Simulates I2C bus errors when configured to do so.
"""
def __init__(
self,
sensor_id: str = "TMP102-0x48",
base_temp_c: float = 22.5,
noise_scale: float = 0.1,
) -> None:
self._sensor_id = sensor_id
self._base_temp = base_temp_c
self._noise_scale = noise_scale
self._call_count = 0
self._fail_on: set = set() # Set of call indices that should fail.
self._is_shutdown = False
def configure_failure(self, call_index: int) -> None:
"""Configure the stub to raise an OSError on the given call index."""
self._fail_on.add(call_index)
def ping(self) -> None:
"""Simulate a sensor presence check."""
if self._is_shutdown:
raise OSError("Sensor is in shutdown mode.")
def shutdown(self) -> None:
"""Simulate placing the sensor in low-power shutdown mode."""
self._is_shutdown = True
def read_temperature(self, oversample_count: int) -> Tuple[float, int, str]:
"""
Return (temperature_c, raw_adc_counts, sensor_id).
Simulates ADC noise proportional to noise_scale.
Raises OSError on configured failure call indices.
"""
idx = self._call_count
self._call_count += 1
if idx in self._fail_on:
raise OSError(f"I2C NACK on call {idx} (simulated bus error).")
# Simulate a small amount of ADC noise.
import random
noise = random.gauss(0, self._noise_scale)
temp_c = round(self._base_temp + noise, 3)
# TMP102 has 12-bit resolution: 0.0625°C per LSB.
raw_counts = int((temp_c + 40.0) / 0.0625)
return temp_c, raw_counts, self._sensor_id
class StubGPIOController:
"""
Stub HAL for a GPIO-controlled LED.
Records all state transitions for test assertion.
"""
def __init__(self) -> None:
self._state = "off"
self._history = ["off"]
def set_state(self, state: str) -> None:
"""Set the LED state and record the transition."""
self._state = state
self._history.append(state)
def get_state(self) -> str:
"""Return the current LED state."""
return self._state
@property
def history(self):
"""Return the full state transition history."""
return list(self._history)
class StubMQTTHAL:
"""
Stub HAL for an MQTT client.
Records all published messages for test assertion.
Simulates connection failures when configured to do so.
"""
def __init__(self, broker_url: str = "mqtt://localhost:1883") -> None:
self._broker_url = broker_url
self._connected = False
self._published: list = []
self._message_id = 0
self._fail_connect = False
self._fail_publish = False
def configure_connect_failure(self, fail: bool = True) -> None:
self._fail_connect = fail
def configure_publish_failure(self, fail: bool = True) -> None:
self._fail_publish = fail
def connect(self) -> None:
if self._fail_connect:
raise ConnectionError(
f"Cannot connect to MQTT broker at {self._broker_url}."
)
self._connected = True
def disconnect(self) -> None:
self._connected = False
def is_connected(self) -> bool:
return self._connected
def publish(
self,
topic: str,
payload: str,
qos: int,
) -> Tuple[int, bool]:
if self._fail_publish:
raise TimeoutError("MQTT broker did not acknowledge (simulated).")
self._message_id += 1
self._published.append({
"message_id": self._message_id,
"topic": topic,
"payload": payload,
"qos": qos,
})
return self._message_id, True
@property
def published_messages(self) -> list:
"""Return all published messages for test assertion."""
return list(self._published)
THE EMBEDDED DEMONSTRATION
# main_embedded.py
# CCA 0.3 — Embedded systems demonstration.
# Simulates a smart industrial temperature monitoring node using stub HALs.
# Replace stub HALs with real hardware HALs for production deployment.
from __future__ import annotations
import logging
import uuid
from datetime import datetime, timedelta, timezone
logging.basicConfig(
level = logging.INFO,
format = "%(asctime)s %(levelname)-8s %(name)s: %(message)s",
)
from cca import (
CapabilityMesh,
CapabilityRegistration,
RoutingStrategy,
CapabilityContext,
TraceContext,
SecurityContext,
LifecycleState,
SLOTracker,
ObservableCapabilityWrapper,
RetryPolicy,
CircuitBreakerPolicy,
)
from cca.channels import ChannelRegistry
from cca.dependency_graph import CapabilityDependencyGraph
from cca.composition import SequentialComposition
from capabilities.embedded.temperature_sensor import TemperatureSensorCapability
from capabilities.embedded.led_status import LEDStatusCapability
from capabilities.embedded.temperature_alert import TemperatureAlertCapability
from capabilities.embedded.mqtt_gateway import MQTTGatewayCapability
from capabilities.embedded.stubs import (
StubI2CTemperatureSensor,
StubGPIOController,
StubMQTTHAL,
)
def make_embedded_context(device_id: str, deadline_seconds: float = 5.0) -> CapabilityContext:
"""
Create an execution context for an embedded capability invocation.
In a real embedded system, trace_id would be a hardware timer value
or a monotonic sequence number rather than a UUID.
"""
return CapabilityContext(
trace = TraceContext(
trace_id = str(uuid.uuid4()),
span_id = str(uuid.uuid4()),
),
security = SecurityContext(
principal_id = device_id,
tokens = tuple(),
),
tenant_id = device_id,
deadline = datetime.now(timezone.utc) + timedelta(seconds=deadline_seconds),
metadata = {"device_type": "temperature-node", "firmware_version": "1.0.0"},
)
def main() -> None:
log = logging.getLogger("cca.embedded.demo")
log.info("=" * 60)
log.info("CCA 0.3 Embedded Systems Demonstration")
log.info("Smart Industrial Temperature Monitoring Node")
log.info("=" * 60)
# ------------------------------------------------------------------ #
# 1. Build the embedded dependency graph. #
# ------------------------------------------------------------------ #
dep_graph = CapabilityDependencyGraph()
dep_graph.add_capability("i2c-bus", [])
dep_graph.add_capability("gpio-controller", [])
dep_graph.add_capability("temperature-sensor",["i2c-bus"])
dep_graph.add_capability("led-status", ["gpio-controller"])
dep_graph.add_capability("temperature-alert", ["temperature-sensor"])
dep_graph.add_capability("mqtt-gateway", ["temperature-alert"])
init_order = dep_graph.topological_order()
log.info("Embedded capability initialization order: %s", init_order)
# ------------------------------------------------------------------ #
# 2. Instantiate stub HALs (replace with real HALs for production). #
# ------------------------------------------------------------------ #
i2c_hal = StubI2CTemperatureSensor(
sensor_id = "TMP102-0x48",
base_temp_c = 24.3,
)
gpio_hal = StubGPIOController()
mqtt_hal = StubMQTTHAL(broker_url="mqtt://iot.factory.example.com:1883")
# ------------------------------------------------------------------ #
# 3. Create capabilities with Adaptation ring policies. #
# ------------------------------------------------------------------ #
retry_i2c = RetryPolicy(max_retries=2, delay_seconds=0.01)
cb_i2c = CircuitBreakerPolicy(failure_threshold=3,
recovery_timeout_seconds=5.0)
retry_mqtt = RetryPolicy(max_retries=3, delay_seconds=0.1)
cb_mqtt = CircuitBreakerPolicy(failure_threshold=5,
recovery_timeout_seconds=30.0)
temp_sensor_cap = TemperatureSensorCapability(
i2c_hal = i2c_hal,
policies = [cb_i2c],
)
led_cap = LEDStatusCapability(gpio_hal=gpio_hal)
alert_cap = TemperatureAlertCapability()
mqtt_gateway_cap = MQTTGatewayCapability(
mqtt_hal = mqtt_hal,
policies = [retry_mqtt, cb_mqtt],
)
# ------------------------------------------------------------------ #
# 4. Register a lifecycle observer on the sensor to drive the LED. #
# ------------------------------------------------------------------ #
def sensor_lifecycle_observer(
name: str,
old_state: LifecycleState,
new_state: LifecycleState,
) -> None:
"""Drive the status LED based on the sensor's lifecycle state."""
if new_state == LifecycleState.READY:
ctx = make_embedded_context("device-001")
led_cap.execute({"state": "on"}, ctx)
log.info("LED -> ON (sensor READY)")
elif new_state == LifecycleState.DEGRADED:
ctx = make_embedded_context("device-001")
led_cap.execute({"state": "blink_fast"}, ctx)
log.info("LED -> BLINK_FAST (sensor DEGRADED)")
elif new_state == LifecycleState.RETIRED:
ctx = make_embedded_context("device-001")
led_cap.execute({"state": "off"}, ctx)
log.info("LED -> OFF (sensor RETIRED)")
temp_sensor_cap.add_lifecycle_observer(sensor_lifecycle_observer)
# ------------------------------------------------------------------ #
# 5. Initialize all capabilities in dependency order. #
# ------------------------------------------------------------------ #
log.info("Initializing capabilities in dependency order...")
led_cap.initialize() # gpio-controller dependency satisfied by stub
temp_sensor_cap.initialize() # i2c-bus dependency satisfied by stub
alert_cap.initialize()
mqtt_gateway_cap.initialize()
log.info("All capabilities initialized.")
# ------------------------------------------------------------------ #
# 6. Wrap the sensor with observability (Adaptation ring). #
# ------------------------------------------------------------------ #
slo_tracker = SLOTracker(
capability_name = temp_sensor_cap.name,
max_latency_p99_seconds = 0.050,
max_error_rate = 0.01,
window_size = 50, # Small window for embedded RAM budget.
)
def on_sensor_slo_violation(name: str, report: dict) -> None:
"""
SLO violation handler: in production, write to fault log and
transition the sensor to DEGRADED state.
"""
log.warning("EMBEDDED SLO VIOLATION for '%s': %s", name, report)
observable_sensor = ObservableCapabilityWrapper(
capability = temp_sensor_cap,
slo_tracker = slo_tracker,
on_slo_violation = on_sensor_slo_violation,
)
# ------------------------------------------------------------------ #
# 7. Register capabilities in the embedded mesh. #
# ------------------------------------------------------------------ #
mesh = CapabilityMesh(routing_strategy=RoutingStrategy.ROUND_ROBIN)
for cap in [temp_sensor_cap, led_cap, alert_cap, mqtt_gateway_cap]:
mesh.register(CapabilityRegistration(
instance_id = cap.instance_id,
name = cap.name,
version = cap.version,
tags = {"embedded", "factory-floor"},
contract = cap.contract,
lifecycle_state = LifecycleState.READY,
endpoint = "in-process",
))
# ------------------------------------------------------------------ #
# 8. Set up the sensor.temperature.readings channel. #
# ------------------------------------------------------------------ #
channel_registry = ChannelRegistry()
sensor_channel = channel_registry.get_or_create(
"sensor.temperature.readings",
schema={"temperature_c": float, "alert_level": str},
)
readings_log = []
def on_sensor_reading(event) -> None:
readings_log.append(event.payload)
log.info(
"Channel 'sensor.temperature.readings': %s", event.payload,
)
sub = sensor_channel.subscribe(on_sensor_reading)
# ------------------------------------------------------------------ #
# 9. Run the monitoring loop: sensor -> alert -> MQTT gateway. #
# ------------------------------------------------------------------ #
log.info("Starting embedded monitoring loop (5 iterations)...")
# Build the sequential composition: sensor reading -> alert -> MQTT.
# The sensor stage is wrapped in a lambda to fix the input format.
device_id = "device-001"
for iteration in range(5):
ctx = make_embedded_context(device_id, deadline_seconds=1.0)
log.info("--- Iteration %d ---", iteration + 1)
# Step A: Read temperature from sensor.
sensor_result = observable_sensor.execute(
{"oversample_count": 4},
ctx,
)
log.info(
"Sensor reading: %.3f°C (raw ADC: %d)",
sensor_result["temperature_c"],
sensor_result["raw_adc_counts"],
)
# Step B: Apply threshold logic.
child_ctx = ctx.derive(new_span_id=str(uuid.uuid4()))
alert_result = alert_cap.execute(
{
"temperature_c": sensor_result["temperature_c"],
"high_threshold_c": 30.0,
"low_threshold_c": 10.0,
},
child_ctx,
)
log.info(
"Alert: [%s] %s",
alert_result["alert_level"],
alert_result["alert_message"],
)
# Step C: Publish to MQTT gateway (embedded-to-enterprise bridge).
mqtt_ctx = ctx.derive(new_span_id=str(uuid.uuid4()))
mqtt_result = mqtt_gateway_cap.execute(
{
"topic": f"factory/line1/{device_id}/temperature",
"payload": {
"temperature_c": sensor_result["temperature_c"],
"alert_level": alert_result["alert_level"],
"iteration": iteration + 1,
},
"qos": 1,
},
mqtt_ctx,
)
log.info(
"MQTT published: message_id=%d, topic=%s",
mqtt_result["message_id"],
mqtt_result["topic"],
)
# Step D: Publish to the local sensor channel for other subscribers.
sensor_channel.publish(
publisher_id = temp_sensor_cap.name,
payload = {
"temperature_c": sensor_result["temperature_c"],
"alert_level": alert_result["alert_level"],
},
partition_key = device_id,
)
# ------------------------------------------------------------------ #
# 10. Print SLO compliance report. #
# ------------------------------------------------------------------ #
log.info("Sensor SLO compliance report: %s", slo_tracker.compliance_report())
log.info(
"MQTT messages published: %d",
len(mqtt_hal.published_messages),
)
log.info(
"Channel readings received: %d",
len(readings_log),
)
log.info(
"LED state history: %s",
gpio_hal.history,
)
# ------------------------------------------------------------------ #
# 11. Retire all capabilities in reverse dependency order. #
# ------------------------------------------------------------------ #
log.info("Retiring capabilities in reverse dependency order...")
sub.cancel()
mqtt_gateway_cap.retire()
alert_cap.retire()
temp_sensor_cap.retire() # Triggers LED -> OFF via lifecycle observer.
led_cap.retire()
log.info("All embedded capabilities retired.")
if __name__ == "__main__":
main()
CHAPTER 15: PUTTING IT ALL TOGETHER — ENTERPRISE EXAMPLE
To see how all the pieces of CCA 0.3 fit together in an enterprise context, we walk through the construction of an order processing service for an e-commerce platform. The service has four capabilities: ValidateOrder, ReserveInventory, ChargePayment, and DispatchShipment. These are composed using the saga pattern to implement a distributed transaction. A currency conversion capability is also included to demonstrate the full contract and SLO machinery.
CapabilityMesh
|
+-- CapabilityDependencyGraph (validates init order)
|
+-- CurrencyConversionCapability (Essence + Realization + Adaptation)
| |
| +-- Adaptation: RetryPolicy, CircuitBreakerPolicy, SLOTracker
|
+-- ValidateOrder <-- Essence: contract, lifecycle
+-- ReserveInventory <-- depends on InventoryService stub
+-- ChargePayment <-- depends on PaymentGateway stub
+-- DispatchShipment <-- depends on ShippingService stub
|
+-- publishes to Channel: order.events
|
+-- NotificationCapability (subscriber)
+-- AnalyticsCapability (subscriber)
Saga: ProcessOrder
orchestrates ValidateOrder -> ReserveInventory
-> ChargePayment -> DispatchShipment
with automatic compensation on any failure
# capabilities/currency_conversion.py
# CCA 0.3 — Currency conversion capability: a complete three-ring example.
from __future__ import annotations
from typing import Any
from cca.capability import Capability, DeadlineExceededError
from cca.contract import Contract, SLADeclaration, ErrorDescriptor
CURRENCY_CONVERSION_CONTRACT = Contract(
input_schema={
"amount": (int, float),
"from_currency": str,
"to_currency": str,
},
output_schema={
"converted_amount": (int, float),
"exchange_rate": (int, float),
},
sla=SLADeclaration(
max_latency_p99_seconds = 0.1,
max_error_rate = 0.001,
min_throughput_rps = 200.0,
),
error_catalog=[
ErrorDescriptor(
error_code = "UNSUPPORTED_CURRENCY",
description = "One or both currency codes are not supported.",
severity = "ERROR",
retryable = False,
),
ErrorDescriptor(
error_code = "RATE_SOURCE_UNAVAILABLE",
description = "The exchange rate data source is temporarily unavailable.",
severity = "WARNING",
retryable = True,
),
],
assertions=[
lambda inp, out: out["converted_amount"] > 0,
lambda inp, out: out["exchange_rate"] > 0,
lambda inp, out: (
inp["from_currency"] != inp["to_currency"]
or abs(out["converted_amount"] - inp["amount"]) < 0.0001
),
],
)
class CurrencyConversionCapability(Capability):
"""
A concrete CCA 0.3 capability that converts an amount from one currency
to another using a configurable exchange rate source.
Essence ring : name="currency-conversion", version="1.0.0",
CURRENCY_CONVERSION_CONTRACT,
dependencies=["exchange-rate-source"].
Realization : _execute() performs the currency conversion.
Adaptation : policies are supplied by the caller at construction time.
"""
def __init__(self, rate_source: Any, policies: list = None) -> None:
super().__init__(
name = "currency-conversion",
version = "1.0.0",
contract = CURRENCY_CONVERSION_CONTRACT,
dependencies = ["exchange-rate-source"],
policies = policies or [],
)
self._rate_source = rate_source
def _on_initialize(self) -> None:
"""Verify that the rate source is reachable during initialization."""
pass
def _execute(self, input_data: Any, context: Any) -> Any:
"""Realization ring: perform the currency conversion."""
if context.is_expired():
raise DeadlineExceededError(
"Deadline expired before exchange rate lookup could be performed."
)
from_currency = input_data["from_currency"]
to_currency = input_data["to_currency"]
amount = input_data["amount"]
if from_currency == to_currency:
return {"converted_amount": float(amount), "exchange_rate": 1.0}
try:
rate = self._rate_source.get_rate(from_currency, to_currency)
except Exception as exc:
err = RuntimeError("Exchange rate source unavailable.")
err.error_code = "RATE_SOURCE_UNAVAILABLE" # type: ignore[attr-defined]
raise err from exc
return {
"converted_amount": round(float(amount) * rate, 6),
"exchange_rate": float(rate),
}
# capabilities/order_processing.py
# CCA 0.3 — Order processing saga: demonstrates Saga composition,
# CapabilityContext propagation, and CapabilityChannel publishing.
from __future__ import annotations
import uuid
from datetime import datetime, timedelta, timezone
from typing import Any
from cca.context import CapabilityContext, TraceContext, SecurityContext
from cca.channels import CapabilityChannel
from cca.composition.saga import Saga, SagaStep
def build_order_processing_saga(
validate_cap: Any,
reserve_cap: Any,
charge_cap: Any,
dispatch_cap: Any,
order_channel: CapabilityChannel,
) -> Saga:
"""
Construct the order processing saga from four capabilities and a channel.
Each SagaStep has a forward action and a compensating action.
"""
def validate_action(order_data: dict) -> dict:
ctx = order_data["context"]
child_ctx = ctx.derive(new_span_id=str(uuid.uuid4()))
validated = validate_cap.execute(order_data["order"], child_ctx)
return {**order_data, "validated_order": validated}
def validate_compensate(result: dict) -> None:
pass
def reserve_action(order_data: dict) -> dict:
ctx = order_data["context"]
child_ctx = ctx.derive(new_span_id=str(uuid.uuid4()))
reservation = reserve_cap.execute(order_data["validated_order"], child_ctx)
return {**order_data, "reservation_id": reservation["reservation_id"]}
def reserve_compensate(result: dict) -> None:
reservation_id = result.get("reservation_id")
if reservation_id:
reserve_cap.release(reservation_id)
def charge_action(order_data: dict) -> dict:
ctx = order_data["context"]
child_ctx = ctx.derive(new_span_id=str(uuid.uuid4()))
charge = charge_cap.execute(order_data["validated_order"], child_ctx)
return {**order_data, "charge_id": charge["charge_id"]}
def charge_compensate(result: dict) -> None:
charge_id = result.get("charge_id")
if charge_id:
charge_cap.refund(charge_id)
def dispatch_action(order_data: dict) -> dict:
ctx = order_data["context"]
child_ctx = ctx.derive(new_span_id=str(uuid.uuid4()))
shipment = dispatch_cap.execute(order_data["validated_order"], child_ctx)
order_channel.publish(
publisher_id = "order-processing-saga",
payload = {
"order_id": order_data["validated_order"]["order_id"],
"shipment_id": shipment["shipment_id"],
},
partition_key = order_data["validated_order"]["customer_id"],
)
return {**order_data, "shipment_id": shipment["shipment_id"]}
def dispatch_compensate(result: dict) -> None:
shipment_id = result.get("shipment_id")
if shipment_id:
dispatch_cap.cancel_shipment(shipment_id)
return Saga(
name = "ProcessOrder",
steps = [
SagaStep("ValidateOrder", validate_action, validate_compensate),
SagaStep("ReserveInventory", reserve_action, reserve_compensate),
SagaStep("ChargePayment", charge_action, charge_compensate),
SagaStep("DispatchShipment", dispatch_action, dispatch_compensate),
],
)
def process_order(order_data: dict, saga: Saga) -> dict:
"""
Entry point for order processing.
Creates the execution context with a 10-second deadline and runs the saga.
"""
context = CapabilityContext(
trace = TraceContext(
trace_id = str(uuid.uuid4()),
span_id = str(uuid.uuid4()),
),
security = SecurityContext(
principal_id = order_data.get("customer_id", "anonymous"),
tokens = tuple(),
),
tenant_id = order_data.get("tenant_id", "default"),
deadline = datetime.now(timezone.utc) + timedelta(seconds=10),
)
initial_input = {"order": order_data, "context": context}
return saga.execute(initial_input)
# capabilities/__init__.py
# CCA 0.3 — Capabilities package.
from .currency_conversion import CurrencyConversionCapability
from .order_processing import build_order_processing_saga, process_order
__all__ = [
"CurrencyConversionCapability",
"build_order_processing_saga",
"process_order",
]
CHAPTER 16: PROJECT STRUCTURE, INSTALLATION, AND DEPLOYMENT
A CCA 0.3 project has a well-defined directory structure that reflects the three-ring model and the separation of concerns between the framework components, the enterprise capabilities, and the embedded capabilities.
cca_project/
cca/
__init__.py
exceptions.py
capability.py
lifecycle.py
context.py
contract.py
security.py
policies.py
mesh.py
channels.py
dependency_graph.py
observability.py
testing.py
composition/
__init__.py
sequential.py
parallel.py
conditional.py
pipeline.py
saga.py
fanout.py
capabilities/
__init__.py
currency_conversion.py
order_processing.py
embedded/
__init__.py
temperature_sensor.py
led_status.py
temperature_alert.py
mqtt_gateway.py
stubs.py
tests/
__init__.py
test_lifecycle.py
test_contract.py
test_currency_conversion.py
test_order_processing.py
test_embedded.py
requirements.txt
setup.py
main.py
main_embedded.py
The requirements.txt file lists the external dependencies. The CCA 0.3 core framework uses only the Python standard library. All external packages are optional enhancements for production deployments.
# requirements.txt
# CCA 0.3 framework dependencies.
# The core framework uses only the Python standard library (Python >= 3.9).
# The following packages are optional enhancements for production deployments.
# JSON schema validation (optional — replaces the simple dict-based validation)
jsonschema>=4.0.0
# Pydantic for richer schema validation and data modelling (optional)
pydantic>=2.0.0
# OpenTelemetry for production-grade distributed tracing (optional)
opentelemetry-api>=1.20.0
opentelemetry-sdk>=1.20.0
# MQTT client for embedded-to-enterprise gateway capabilities (optional)
paho-mqtt>=1.6.0
# For running the test suite
pytest>=7.0.0
pytest-cov>=4.0.0
# For MicroPython-based embedded deployments:
# Use the MicroPython package manager (mip) instead of pip.
# The cca/ package can be cross-compiled with mpy-cross for deployment
# to MicroPython-capable microcontrollers (ESP32, RP2040, STM32).
# Example MicroPython installation:
# import mip
# mip.install("github:your-org/cca-micropython")
The setup.py file makes the framework installable as a Python package.
# setup.py
# CCA 0.3 — Package installation configuration.
from setuptools import setup, find_packages
setup(
name = "cca",
version = "0.3.0",
description = (
"Capability Centric Architecture 0.3 Framework — "
"for enterprise, embedded, and hybrid systems"
),
author = "CCA Contributors",
python_requires = ">=3.9",
packages = find_packages(),
install_requires = [], # Core framework has no mandatory external dependencies.
extras_require = {
"validation": ["jsonschema>=4.0.0", "pydantic>=2.0.0"],
"tracing": ["opentelemetry-api>=1.20.0", "opentelemetry-sdk>=1.20.0"],
"mqtt": ["paho-mqtt>=1.6.0"],
"dev": ["pytest>=7.0.0", "pytest-cov>=4.0.0"],
},
classifiers=[
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Topic :: Software Development :: Embedded Systems",
"Topic :: Software Development :: Libraries :: Application Frameworks",
],
)
The main.py file provides a complete, runnable enterprise demonstration.
# main.py
# CCA 0.3 — Complete runnable enterprise demonstration.
# Wires together the mesh, dependency graph, channels,
# observability, policies, and a concrete capability.
from __future__ import annotations
import logging
import uuid
from datetime import datetime, timedelta, timezone
logging.basicConfig(
level = logging.INFO,
format = "%(asctime)s %(levelname)-8s %(name)s: %(message)s",
)
from cca import (
CapabilityMesh, CapabilityRegistration, RoutingStrategy,
CapabilityContext, TraceContext, SecurityContext,
LifecycleState,
SLOTracker, ObservableCapabilityWrapper,
RetryPolicy, CircuitBreakerPolicy,
)
from cca.channels import ChannelRegistry
from cca.dependency_graph import CapabilityDependencyGraph
from capabilities.currency_conversion import CurrencyConversionCapability
class StubRateSource:
"""A stub exchange rate source for demonstration purposes."""
_rates = {
("USD", "EUR"): 0.92,
("EUR", "USD"): 1.087,
("USD", "GBP"): 0.79,
("GBP", "USD"): 1.266,
}
def get_rate(self, from_currency: str, to_currency: str) -> float:
key = (from_currency, to_currency)
if key not in self._rates:
raise ValueError(
f"Unsupported currency pair: {from_currency}/{to_currency}"
)
return self._rates[key]
def main() -> None:
log = logging.getLogger("cca.demo")
# ------------------------------------------------------------------ #
# 1. Build the dependency graph and verify initialization order. #
# ------------------------------------------------------------------ #
dep_graph = CapabilityDependencyGraph()
dep_graph.add_capability("exchange-rate-source", [])
dep_graph.add_capability("currency-conversion", ["exchange-rate-source"])
init_order = dep_graph.topological_order()
log.info("Capability initialization order: %s", init_order)
# ------------------------------------------------------------------ #
# 2. Create the capability with Adaptation ring policies. #
# ------------------------------------------------------------------ #
rate_source = StubRateSource()
circuit_policy = CircuitBreakerPolicy(
failure_threshold=5, recovery_timeout_seconds=30.0,
)
cap = CurrencyConversionCapability(
rate_source = rate_source,
policies = [circuit_policy],
)
cap.initialize()
log.info(
"Capability '%s' v%s initialized. State: %s",
cap.name, cap.version, cap.lifecycle_state.name,
)
# ------------------------------------------------------------------ #
# 3. Wrap with observability (Adaptation ring). #
# ------------------------------------------------------------------ #
slo_tracker = SLOTracker(
capability_name = cap.name,
max_latency_p99_seconds = 0.1,
max_error_rate = 0.001,
window_size = 100,
)
def on_slo_violation(name: str, report: dict) -> None:
log.warning("SLO VIOLATION for '%s': %s", name, report)
observable_cap = ObservableCapabilityWrapper(
capability = cap,
slo_tracker = slo_tracker,
on_slo_violation = on_slo_violation,
)
# ------------------------------------------------------------------ #
# 4. Register in the capability mesh. #
# ------------------------------------------------------------------ #
mesh = CapabilityMesh(routing_strategy=RoutingStrategy.ROUND_ROBIN)
mesh.register(CapabilityRegistration(
instance_id = cap.instance_id,
name = cap.name,
version = cap.version,
tags = {"finance", "currency"},
contract = cap.contract,
lifecycle_state = LifecycleState.READY,
endpoint = "in-process",
))
def on_lifecycle_change(reg: CapabilityRegistration) -> None:
log.info(
"Mesh lifecycle hook: '%s' state is now %s.",
reg.name, reg.lifecycle_state.name,
)
mesh.add_lifecycle_hook(on_lifecycle_change)
# ------------------------------------------------------------------ #
# 5. Set up a capability channel for downstream events. #
# ------------------------------------------------------------------ #
channel_registry = ChannelRegistry()
finance_channel = channel_registry.get_or_create(
"finance.events",
schema={"event_type": str, "amount": float},
)
sub = finance_channel.subscribe(
lambda event: log.info(
"Channel 'finance.events' received event: %s", event.payload,
)
)
# ------------------------------------------------------------------ #
# 6. Build the execution context (timezone-aware UTC deadline). #
# ------------------------------------------------------------------ #
context = CapabilityContext(
trace = TraceContext(
trace_id = str(uuid.uuid4()),
span_id = str(uuid.uuid4()),
),
security = SecurityContext(
principal_id = "demo-user",
tokens = tuple(),
),
tenant_id = "demo-tenant",
deadline = datetime.now(timezone.utc) + timedelta(seconds=5),
)
# ------------------------------------------------------------------ #
# 7. Invoke the capability (via the observable wrapper). #
# ------------------------------------------------------------------ #
result = observable_cap.execute(
{"amount": 100, "from_currency": "USD", "to_currency": "EUR"},
context,
)
log.info("Conversion result: %s", result)
# ------------------------------------------------------------------ #
# 8. Publish a downstream event on the finance channel. #
# ------------------------------------------------------------------ #
finance_channel.publish(
publisher_id = cap.name,
payload = {
"event_type": "conversion_completed",
"amount": result["converted_amount"],
},
partition_key = "demo-tenant",
)
# ------------------------------------------------------------------ #
# 9. Demonstrate context derivation for a child invocation. #
# ------------------------------------------------------------------ #
child_context = context.derive(new_span_id=str(uuid.uuid4()))
log.info(
"Child context: trace_id=%s, span_id=%s, parent_span_id=%s",
child_context.trace.trace_id,
child_context.trace.span_id,
child_context.trace.parent_span_id,
)
child_result = observable_cap.execute(
{"amount": 50, "from_currency": "EUR", "to_currency": "USD"},
child_context,
)
log.info("Child invocation result: %s", child_result)
# ------------------------------------------------------------------ #
# 10. Print SLO compliance report. #
# ------------------------------------------------------------------ #
log.info("SLO compliance report: %s", slo_tracker.compliance_report())
# ------------------------------------------------------------------ #
# 11. Discover the capability from the mesh and verify routing. #
# ------------------------------------------------------------------ #
discovered = mesh.discover("currency-conversion", version="1.0.0")
log.info(
"Mesh discovery: found '%s' at endpoint '%s'.",
discovered.name if discovered else "none",
discovered.endpoint if discovered else "n/a",
)
# ------------------------------------------------------------------ #
# 12. Retire the capability gracefully and clean up. #
# ------------------------------------------------------------------ #
sub.cancel()
cap.retire()
mesh.deregister(cap.instance_id)
log.info("Capability '%s' retired and deregistered.", cap.name)
if __name__ == "__main__":
main()
# tests/__init__.py
# CCA 0.3 — Test package marker (empty).
# tests/test_lifecycle.py
# CCA 0.3 — Unit tests for the LifecycleMachine and DependencyGraph.
import unittest
from cca.lifecycle import LifecycleMachine, LifecycleState
from cca.exceptions import LifecycleError, DependencyCycleError
from cca.dependency_graph import CapabilityDependencyGraph
class TestLifecycleMachine(unittest.TestCase):
def test_initial_state_is_created(self):
m = LifecycleMachine("test-cap")
self.assertEqual(m.state, LifecycleState.CREATED)
def test_valid_transition_created_to_initializing(self):
m = LifecycleMachine("test-cap")
m.transition_to(LifecycleState.INITIALIZING)
self.assertEqual(m.state, LifecycleState.INITIALIZING)
def test_invalid_transition_raises_lifecycle_error(self):
m = LifecycleMachine("test-cap")
with self.assertRaises(LifecycleError):
m.transition_to(LifecycleState.READY)
def test_retired_is_terminal(self):
m = LifecycleMachine("test-cap")
m.transition_to(LifecycleState.INITIALIZING)
m.transition_to(LifecycleState.RETIRED)
with self.assertRaises(LifecycleError):
m.transition_to(LifecycleState.READY)
def test_observer_is_notified_on_transition(self):
events = []
m = LifecycleMachine("test-cap")
m.add_observer(lambda name, old, new: events.append((name, old, new)))
m.transition_to(LifecycleState.INITIALIZING)
self.assertEqual(len(events), 1)
self.assertEqual(events[0], (
"test-cap",
LifecycleState.CREATED,
LifecycleState.INITIALIZING,
))
def test_observer_error_does_not_crash_machine(self):
"""
An observer that raises must not crash the lifecycle machine.
Uses a proper raising function (not a generator trick) to ensure
a real RuntimeError is raised and swallowed by _notify_observers.
"""
def bad_observer(name, old, new):
raise RuntimeError("Simulated observer failure.")
m = LifecycleMachine("test-cap")
m.add_observer(bad_observer)
# Must not raise — the lifecycle machine swallows observer errors.
m.transition_to(LifecycleState.INITIALIZING)
self.assertEqual(m.state, LifecycleState.INITIALIZING)
class TestDependencyGraph(unittest.TestCase):
def test_topological_order_respects_dependencies(self):
g = CapabilityDependencyGraph()
g.add_capability("A", [])
g.add_capability("B", ["A"])
g.add_capability("C", ["B"])
order = g.topological_order()
self.assertLess(order.index("A"), order.index("B"))
self.assertLess(order.index("B"), order.index("C"))
def test_cycle_detection_raises(self):
g = CapabilityDependencyGraph()
g.add_capability("A", [])
g.add_capability("B", ["A"])
with self.assertRaises(DependencyCycleError):
# Creates B -> A -> B cycle.
g.add_capability("A", ["B"])
def test_embedded_init_order(self):
"""Verify that embedded hardware init order is correctly computed."""
g = CapabilityDependencyGraph()
g.add_capability("i2c-bus", [])
g.add_capability("gpio-controller", [])
g.add_capability("temperature-sensor", ["i2c-bus"])
g.add_capability("led-status", ["gpio-controller"])
g.add_capability("temperature-alert", ["temperature-sensor"])
g.add_capability("mqtt-gateway", ["temperature-alert"])
order = g.topological_order()
self.assertLess(order.index("i2c-bus"),
order.index("temperature-sensor"))
self.assertLess(order.index("temperature-sensor"),
order.index("temperature-alert"))
self.assertLess(order.index("temperature-alert"),
order.index("mqtt-gateway"))
if __name__ == "__main__":
unittest.main()
# tests/test_currency_conversion.py
# CCA 0.3 — Contract and unit tests for CurrencyConversionCapability.
import unittest
from datetime import datetime, timedelta, timezone
import uuid
from cca.context import CapabilityContext, TraceContext, SecurityContext
from cca.exceptions import DeadlineExceededError
from capabilities.currency_conversion import CurrencyConversionCapability
def _make_context(seconds: float = 10.0) -> CapabilityContext:
return CapabilityContext(
trace = TraceContext(
trace_id = str(uuid.uuid4()),
span_id = str(uuid.uuid4()),
),
security = SecurityContext(principal_id="test", tokens=tuple()),
tenant_id = "test-tenant",
deadline = datetime.now(timezone.utc) + timedelta(seconds=seconds),
)
class StubRateSource:
_rates = {("USD", "EUR"): 0.92, ("EUR", "USD"): 1.087}
def get_rate(self, from_currency, to_currency):
key = (from_currency, to_currency)
if key not in self._rates:
raise ValueError(f"Unsupported: {from_currency}/{to_currency}")
return self._rates[key]
class TestCurrencyConversion(unittest.TestCase):
def setUp(self):
self.cap = CurrencyConversionCapability(rate_source=StubRateSource())
self.cap.initialize()
self.ctx = _make_context()
def tearDown(self):
self.cap.retire()
def test_usd_to_eur_conversion(self):
result = self.cap.execute(
{"amount": 100, "from_currency": "USD", "to_currency": "EUR"},
self.ctx,
)
self.assertAlmostEqual(result["converted_amount"], 92.0, places=4)
self.assertAlmostEqual(result["exchange_rate"], 0.92, places=4)
def test_same_currency_returns_original_amount(self):
result = self.cap.execute(
{"amount": 50, "from_currency": "USD", "to_currency": "USD"},
self.ctx,
)
self.assertAlmostEqual(result["converted_amount"], 50.0, places=4)
self.assertEqual(result["exchange_rate"], 1.0)
def test_integer_amount_is_accepted(self):
result = self.cap.execute(
{"amount": 200, "from_currency": "USD", "to_currency": "EUR"},
self.ctx,
)
self.assertGreater(result["converted_amount"], 0)
def test_contract_assertions_hold(self):
inp = {"amount": 100, "from_currency": "USD", "to_currency": "EUR"}
out = self.cap.execute(inp, self.ctx)
self.cap.contract.check_assertions(inp, out)
def test_missing_field_raises_value_error(self):
with self.assertRaises(ValueError):
self.cap.execute(
{"amount": 100, "from_currency": "USD"},
self.ctx,
)
def test_expired_deadline_raises(self):
expired_ctx = _make_context(seconds=-1)
with self.assertRaises(DeadlineExceededError):
self.cap.execute(
{"amount": 100, "from_currency": "USD", "to_currency": "EUR"},
expired_ctx,
)
if __name__ == "__main__":
unittest.main()
# tests/test_embedded.py
# CCA 0.3 — SIL (Software-in-the-Loop) tests for embedded capabilities.
# These tests run entirely without physical hardware using stub HALs.
# For HIL (Hardware-in-the-Loop) testing, replace stub HALs with real
# hardware HAL implementations — the test cases are identical.
import unittest
from datetime import datetime, timedelta, timezone
import uuid
from cca.context import CapabilityContext, TraceContext, SecurityContext
from cca.exceptions import DeadlineExceededError
from capabilities.embedded.temperature_sensor import TemperatureSensorCapability
from capabilities.embedded.led_status import LEDStatusCapability
from capabilities.embedded.temperature_alert import TemperatureAlertCapability
from capabilities.embedded.mqtt_gateway import MQTTGatewayCapability
from capabilities.embedded.stubs import (
StubI2CTemperatureSensor,
StubGPIOController,
StubMQTTHAL,
)
def _make_embedded_context(seconds: float = 5.0) -> CapabilityContext:
return CapabilityContext(
trace = TraceContext(
trace_id = str(uuid.uuid4()),
span_id = str(uuid.uuid4()),
),
security = SecurityContext(principal_id="device-001", tokens=tuple()),
tenant_id = "device-001",
deadline = datetime.now(timezone.utc) + timedelta(seconds=seconds),
)
class TestTemperatureSensorCapability(unittest.TestCase):
def setUp(self):
self.hal = StubI2CTemperatureSensor(
sensor_id = "TMP102-0x48",
base_temp_c = 22.5,
noise_scale = 0.0, # Zero noise for deterministic tests.
)
self.cap = TemperatureSensorCapability(i2c_hal=self.hal)
self.cap.initialize()
self.ctx = _make_embedded_context()
def tearDown(self):
self.cap.retire()
def test_normal_reading_within_range(self):
result = self.cap.execute({"oversample_count": 1}, self.ctx)
self.assertAlmostEqual(result["temperature_c"], 22.5, places=1)
self.assertGreater(result["raw_adc_counts"], 0)
self.assertEqual(result["sensor_id"], "TMP102-0x48")
def test_contract_assertions_hold(self):
inp = {"oversample_count": 4}
out = self.cap.execute(inp, self.ctx)
self.cap.contract.check_assertions(inp, out)
def test_i2c_error_raises_capability_error(self):
"""Simulates an I2C bus error and verifies the capability raises."""
self.hal.configure_failure(call_index=0)
with self.assertRaises(RuntimeError) as cm:
self.cap.execute({"oversample_count": 1}, self.ctx)
self.assertIn("I2C bus error", str(cm.exception))
def test_expired_deadline_raises(self):
expired_ctx = _make_embedded_context(seconds=-1)
with self.assertRaises(DeadlineExceededError):
self.cap.execute({"oversample_count": 1}, expired_ctx)
def test_missing_oversample_count_raises_contract_error(self):
with self.assertRaises(ValueError):
self.cap.execute({}, self.ctx)
class TestLEDStatusCapability(unittest.TestCase):
def setUp(self):
self.hal = StubGPIOController()
self.cap = LEDStatusCapability(gpio_hal=self.hal)
self.cap.initialize()
self.ctx = _make_embedded_context()
def tearDown(self):
self.cap.retire()
def test_set_led_on(self):
result = self.cap.execute({"state": "on"}, self.ctx)
self.assertEqual(result["current_state"], "on")
self.assertEqual(self.hal.get_state(), "on")
def test_set_led_blink_fast(self):
result = self.cap.execute({"state": "blink_fast"}, self.ctx)
self.assertEqual(result["current_state"], "blink_fast")
def test_invalid_state_raises_value_error(self):
with self.assertRaises(ValueError):
self.cap.execute({"state": "rainbow"}, self.ctx)
def test_initialize_sets_led_off(self):
"""LED must be in 'off' state immediately after initialization."""
self.assertIn("off", self.hal.history)
def test_contract_assertion_current_state_matches_input(self):
inp = {"state": "blink_slow"}
out = self.cap.execute(inp, self.ctx)
self.cap.contract.check_assertions(inp, out)
class TestTemperatureAlertCapability(unittest.TestCase):
def setUp(self):
self.cap = TemperatureAlertCapability()
self.cap.initialize()
self.ctx = _make_embedded_context()
def tearDown(self):
self.cap.retire()
def test_normal_temperature(self):
result = self.cap.execute(
{"temperature_c": 22.5, "high_threshold_c": 30.0, "low_threshold_c": 10.0},
self.ctx,
)
self.assertEqual(result["alert_level"], "NORMAL")
def test_high_temperature_alert(self):
result = self.cap.execute(
{"temperature_c": 35.0, "high_threshold_c": 30.0, "low_threshold_c": 10.0},
self.ctx,
)
self.assertEqual(result["alert_level"], "HIGH")
def test_critical_high_temperature_alert(self):
result = self.cap.execute(
{"temperature_c": 45.0, "high_threshold_c": 30.0, "low_threshold_c": 10.0},
self.ctx,
)
self.assertEqual(result["alert_level"], "CRITICAL_HIGH")
def test_low_temperature_alert(self):
result = self.cap.execute(
{"temperature_c": 5.0, "high_threshold_c": 30.0, "low_threshold_c": 10.0},
self.ctx,
)
self.assertEqual(result["alert_level"], "LOW")
def test_contract_assertions_hold_for_all_levels(self):
test_cases = [
{"temperature_c": 22.5, "high_threshold_c": 30.0, "low_threshold_c": 10.0},
{"temperature_c": 35.0, "high_threshold_c": 30.0, "low_threshold_c": 10.0},
{"temperature_c": 45.0, "high_threshold_c": 30.0, "low_threshold_c": 10.0},
{"temperature_c": 5.0, "high_threshold_c": 30.0, "low_threshold_c": 10.0},
{"temperature_c": -5.0, "high_threshold_c": 30.0, "low_threshold_c": 10.0},
]
for inp in test_cases:
with self.subTest(inp=inp):
out = self.cap.execute(inp, self.ctx)
self.cap.contract.check_assertions(inp, out)
class TestMQTTGatewayCapability(unittest.TestCase):
def setUp(self):
self.hal = StubMQTTHAL(broker_url="mqtt://test.broker:1883")
self.cap = MQTTGatewayCapability(mqtt_hal=self.hal)
self.cap.initialize()
self.ctx = _make_embedded_context()
def tearDown(self):
self.cap.retire()
def test_publish_succeeds(self):
result = self.cap.execute(
{
"topic": "factory/line1/device-001/temperature",
"payload": {"temperature_c": 22.5, "alert_level": "NORMAL"},
"qos": 1,
},
self.ctx,
)
self.assertTrue(result["published"])
self.assertEqual(result["topic"], "factory/line1/device-001/temperature")
self.assertEqual(len(self.hal.published_messages), 1)
def test_invalid_qos_raises_value_error(self):
with self.assertRaises(ValueError):
self.cap.execute(
{"topic": "test/topic", "payload": {}, "qos": 5},
self.ctx,
)
def test_connection_failure_raises_runtime_error(self):
self.hal.configure_connect_failure(True)
self.hal.disconnect() # Force disconnected state.
with self.assertRaises(RuntimeError):
self.cap.execute(
{"topic": "test/topic", "payload": {}, "qos": 0},
self.ctx,
)
def test_expired_deadline_raises(self):
expired_ctx = _make_embedded_context(seconds=-1)
with self.assertRaises(DeadlineExceededError):
self.cap.execute(
{"topic": "test/topic", "payload": {}, "qos": 0},
expired_ctx,
)
if __name__ == "__main__":
unittest.main()
To install the framework for development, clone the repository and run the following commands in a terminal.
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
On Windows, activate the virtual environment with:
.venv\Scripts\activate
To run the full test suite with coverage reporting:
pytest tests/ --cov=cca --cov-report=term-missing -v
To run the enterprise demonstration:
python main.py
To run the embedded systems demonstration (SIL mode — no physical hardware required):
python main_embedded.py
Expected output from main_embedded.py (timestamps and UUIDs will differ):
2025-01-15 10:00:00 INFO cca.embedded.demo: ============================================================
2025-01-15 10:00:00 INFO cca.embedded.demo: CCA 0.3 Embedded Systems Demonstration
2025-01-15 10:00:00 INFO cca.embedded.demo: Smart Industrial Temperature Monitoring Node
2025-01-15 10:00:00 INFO cca.embedded.demo: ============================================================
2025-01-15 10:00:00 INFO cca.embedded.demo: Embedded capability initialization order: ['i2c-bus', 'gpio-controller', 'temperature-sensor', 'led-status', 'temperature-alert', 'mqtt-gateway']
2025-01-15 10:00:00 INFO cca.lifecycle: Capability 'led-status' lifecycle: CREATED -> INITIALIZING
2025-01-15 10:00:00 INFO cca.lifecycle: Capability 'led-status' lifecycle: INITIALIZING -> READY
2025-01-15 10:00:00 INFO cca.lifecycle: Capability 'temperature-sensor' lifecycle: CREATED -> INITIALIZING
2025-01-15 10:00:00 INFO cca.lifecycle: Capability 'temperature-sensor' lifecycle: INITIALIZING -> READY
2025-01-15 10:00:00 INFO cca.embedded.demo: LED -> ON (sensor READY)
2025-01-15 10:00:00 INFO cca.embedded.demo: --- Iteration 1 ---
2025-01-15 10:00:00 INFO cca.embedded.demo: Sensor reading: 24.300°C (raw ADC: 1012)
2025-01-15 10:00:00 INFO cca.embedded.demo: Alert: [NORMAL] Temperature 24.3°C is within normal range.
2025-01-15 10:00:00 INFO cca.embedded.demo: MQTT published: message_id=1, topic=factory/line1/device-001/temperature
2025-01-15 10:00:00 INFO cca.embedded.demo: Sensor SLO compliance report: {'capability': 'temperature-sensor', ...}
2025-01-15 10:00:00 INFO cca.embedded.demo: All embedded capabilities retired.
CHAPTER 17: COMPARISON WITH OTHER ARCHITECTURES
CCA 0.3 occupies a unique position in the landscape of software architecture patterns. Understanding how it relates to the patterns that practitioners are already familiar with — in both enterprise and embedded domains — helps clarify both what CCA 0.3 offers and where it fits in the broader architectural toolkit.
Compared to layered architecture, CCA 0.3 is organized around what the system can do rather than how it is technically structured. Adding a new business capability does not require touching all three layers; each capability is self-contained in its three rings.
Compared to microservices, CCA 0.3 is more fine-grained and more formally specified. CCA 0.3 capabilities can be deployed within a single process or across multiple processes, making the architecture scale-independent. More importantly, CCA 0.3 adds formal contracts, lifecycle management, security tokens, evolution envelopes, and SLO tracking as first-class citizens.
Compared to hexagonal architecture (ports and adapters), CCA 0.3 shares the emphasis on clean boundaries between core logic and external concerns. The Essence ring corresponds to the hexagonal core, and the Adaptation ring corresponds to the adapters. But hexagonal architecture is primarily a structural pattern for a single application, while CCA 0.3 is a system-level pattern that governs how multiple capabilities interact across an entire system.
Compared to event-driven architecture, CCA 0.3 is a superset. CCA 0.3 supports both synchronous request-response interactions through direct capability invocation and asynchronous event-driven interactions through capability channels.
Compared to service mesh architectures like Istio, CCA 0.3 operates at a higher level of abstraction. A service mesh manages network-level concerns. CCA 0.3 manages capability-level concerns. The two are complementary.
Compared to AUTOSAR (AUTomotive Open System ARchitecture): AUTOSAR is the dominant component architecture for automotive embedded software. It defines software components (SWCs), ports, interfaces, and a runtime environment (RTE). CCA 0.3 is conceptually similar to AUTOSAR's component model but is more general, more lightweight, and directly applicable to non-automotive embedded systems. The CCA 0.3 Essence ring corresponds to AUTOSAR's software component interface description. The Realization ring corresponds to the AUTOSAR runnable. The Adaptation ring corresponds to the AUTOSAR RTE configuration (scheduling, inter-runnable variables, mode management). The key difference is that CCA 0.3 is not tied to the AUTOSAR toolchain and can be implemented in Python, C, C++, or Rust. For automotive systems that must comply with ISO 26262, CCA 0.3's formal contracts and behavioral assertions provide a natural foundation for the required software unit testing and integration testing evidence.
Compared to OSEK/VDX and AUTOSAR OS: OSEK/VDX and AUTOSAR OS define the task scheduling and interrupt management model for embedded real-time systems. CCA 0.3 sits above the OS layer: it organizes the application software into capabilities, but it does not replace the RTOS. In a Zephyr RTOS deployment, each CCA 0.3 capability runs as one or more Zephyr threads, and the capability's Adaptation ring policies (rate limiting, timeout) interact with the Zephyr scheduler through standard RTOS APIs.
Compared to IEC 61508 and IEC 62443 functional safety and security frameworks: IEC 61508 requires that safety-related software be structured into independently testable units with well-defined interfaces and behavioral specifications. CCA 0.3's three-ring model, evolution envelopes, and behavioral assertions directly satisfy these requirements. IEC 62443 requires that industrial control systems implement security zones and conduits. CCA 0.3's capability token model maps directly to this requirement: each capability defines its security zone (the set of principals authorized to invoke it), and the token authority acts as the conduit controller.
The key differentiator of CCA 0.3 compared to all of these patterns is the combination of the three-ring model, the evolution envelope, and the formal lifecycle — applied uniformly across enterprise software, embedded firmware, and hybrid systems. No other mainstream architecture pattern provides a principled answer to all three of these questions simultaneously across the full spectrum of computing environments.
CHAPTER 18: MIGRATION FROM CCA 0.2 TO CCA 0.3
For teams that have already adopted CCA 0.2, migration to CCA 0.3 is designed to be incremental. The core concepts are the same; CCA 0.3 adds new capabilities without breaking the existing model.
The migration can be done in five steps. The first step is to add lifecycle management to existing capabilities. The second step is to replace the central registry with the CapabilityMesh. The third step is to add CapabilityContext to all execute calls. The fourth step is to enhance contracts with SLA declarations and behavioral assertions. The fifth step is to adopt the new composition patterns, security model, channels, dependency graph, and testing framework as needed.
Migration in embedded systems: For embedded systems teams migrating from an ad-hoc firmware architecture to CCA 0.3, the migration path is slightly different. The first step is to identify the existing firmware functions that correspond to capabilities: hardware drivers, communication stacks, control algorithms, and data processing functions. The second step is to wrap each identified function in a Capability subclass with a contract that captures its existing interface (input parameters, output values, error codes). The third step is to add lifecycle management, replacing the existing initialization and shutdown sequences with initialize() and retire(). The fourth step is to add policies, replacing the existing retry loops and error handling with RetryPolicy and CircuitBreakerPolicy. The fifth step is to add observability, replacing the existing debug output with SLOTracker and ObservableCapabilityWrapper.
The following table summarizes the mapping between CCA 0.2 and CCA 0.3 concepts, with embedded system equivalents.
CCA 0.2 Concept CCA 0.3 Equivalent Embedded Equivalent
-----------------------------------------------------------------------
Capability Capability (3-ring model) Hardware driver / firmware function
Registry CapabilityMesh In-memory peripheral table
Contract (schema) Contract (schema+SLA+assert) Hardware datasheet spec
Policies (4 types) Policies (same 4, integrated) RTOS task policies
Composition (3 patterns) Composition (7 patterns) ISR chains, DMA pipelines
Observability (basic) Observability (SLO) JTAG counters, fault logs
No lifecycle LifecycleMachine (6 states) Power/hardware state machine
No context CapabilityContext RTOS task parameters
No security CapabilityToken + Authority HSM / TrustZone tokens
No channels CapabilityChannel RTOS message queues / CAN IDs
No dependency graph CapabilityDependencyGraph Hardware init sequence
No testing framework ContractTestCase + Chaos SIL / HIL test framework
Evolution envelope Machine-checkable envelope Datasheet revision control
CHAPTER 19: WHAT COMES AFTER CCA 0.3?
CCA 0.3 is a substantial step forward, but the evolution of the pattern is far from complete. Several exciting directions are already being explored for future versions.
The capability marketplace concept becomes much more concrete in the context of CCA 0.3. Because every capability has a formal contract, a version, tags, an SLA declaration, and an evolution envelope, it is possible to build a searchable catalog of capabilities that can be shared across teams and organizations. This applies equally to enterprise software capabilities (currency conversion, payment processing, address validation) and to embedded firmware capabilities (I2C temperature sensor drivers, CAN bus parsers, PID controller implementations). A firmware engineer looking for a certified, tested implementation of a TMP102 temperature sensor driver could find it in the marketplace, complete with its contract, its SLA, its behavioral assertions, and its SIL test suite.
AI-assisted capability discovery is another promising direction. Given a natural-language description of what a developer or firmware engineer wants to do, an AI system could search the capability marketplace, find matching capabilities, and even suggest how to compose them to achieve the desired result. The formal contracts, tags, and SLA declarations in CCA 0.3 provide exactly the structured metadata that an AI system needs to make intelligent recommendations — whether the target is a cloud microservice or a Cortex-M4 firmware image.
Formal verification of capability contracts is a longer-term goal. The behavioral assertions in CCA 0.3 contracts are a first step: they are machine-checkable specifications that can be evaluated at runtime. In safety-critical embedded systems governed by IEC 61508 SIL 3 or ISO 26262 ASIL D, formal verification of these assertions would provide the mathematical proof of correctness required by the certification standards.
The dependency graph introduced in CCA 0.3 opens the door to automated impact analysis. When a capability is about to be changed — whether it is a cloud microservice or a firmware driver — the system could automatically identify all capabilities that depend on it, run their contract tests against the new version, and report any compatibility issues before the change is deployed or flashed to a device.
Finally, the combination of lifecycle management, SLO tracking, and the capability mesh creates the foundation for self-healing systems at every scale: cloud systems that automatically detect when a microservice is degrading and spin up replacement instances, and embedded systems that automatically detect when a sensor is failing and switch to a redundant backup sensor — all governed by the same CCA 0.3 lifecycle and observability model.
CONCLUSION
CCA 0.3 is not just an incremental improvement over CCA 0.2. It is a mature, production-ready architectural pattern that takes the foundational ideas of CCA 0.2 — the three-ring model of Essence, Realization, and Adaptation; the evolution envelope; the capability as a first-class citizen; the four policy types; the three original composition patterns — and extends them with the tools needed to build large-scale, distributed, observable, and governable software systems at every scale of computing.
By adding formal lifecycle management, execution contexts, enhanced contracts with SLA declarations and behavioral assertions that make the evolution envelope machine-checkable, a distributed capability mesh with four routing strategies, a capability-based security model using HMAC-SHA256 tokens, seven composition patterns including sagas and fan-out/fan-in, capability channels for event-driven communication, a formal dependency graph with iterative cycle detection, a built-in testing framework with contract and chaos testing modes, SLO-aware observability with sliding-window compliance tracking, and a complete embedded systems capability model with hardware abstraction layers and SIL/HIL testing support, CCA 0.3 provides a comprehensive answer to the question of how to organize complex software systems — from a 32-kilobyte microcontroller to a 10,000-node cloud cluster.
The pattern is not prescriptive about technology choices. The examples in this article use Python, but the concepts apply equally to Java, Go, Rust, TypeScript, C, or C++. The in-process implementations of the mesh, channels, and other components can be replaced with distributed implementations backed by Kafka, Redis, etcd, MQTT, or any other appropriate technology. The stub hardware abstraction layers can be replaced with real HAL implementations for any target hardware platform.
What CCA 0.3 is prescriptive about is the way of thinking. Systems are collections of capabilities. Capabilities have three rings: Essence, Realization, and Adaptation. Capabilities have evolution envelopes that define how they can change without breaking their callers. Capabilities have lifecycles that are managed explicitly. Security is built in. Observability is automatic. Composition is explicit and structured. Dependencies are declared and verified. These principles are not optional extras; they are the foundation on which reliable, maintainable, and evolvable systems are built — whether those systems run in a Kubernetes pod, on a factory floor gateway, or in the interrupt vector table of a safety-critical microcontroller.
The question is not whether your system has capabilities. Every system does — from the cloud service that processes a million transactions per minute to the microcontroller that reads a temperature sensor ten times per second. The question is whether those capabilities are explicit, versioned, contracted, governed, observed, and composable. CCA 0.3 gives you the tools to make them all of those things, across the full spectrum of computing environments. The rest is up to you.
No comments:
Post a Comment