Design Patterns — Athomic

Reference for which patterns are in use, where they live, and when to apply each one. Before implementing a feature, identify which pattern(s) apply. Patterns are not optional style choices — they are the correct solution to recurring structural problems.

---

Pattern vs Anti-Pattern: The Core Rule

A design pattern solves a recurring problem by establishing a proven structure. An anti-pattern is what you get when you solve the problem ad-hoc:

Anti-pattern	Correct Pattern
if isinstance(x, A): ... elif isinstance(x, B): ...	Strategy + Registry
Constructor with 10+ parameters	Builder or Parameter Object
One giant class with 50 methods	Facade + Extract Class
Direct new ConcreteClass() everywhere	Factory Method
Copy-paste code with slight variations	Template Method
Module importing from every other module	Dependency Inversion + Protocol

---

Patterns In Use — With Their Location

1. Factory Method

What it solves: Decouples object creation from usage. Callers ask for an object by name, the factory decides which concrete class to instantiate.

Where it is:

ai/features/engines/factory.py          ← EngineFactory: "numpy" → AsyncNumpyEngine
ai/features/transformations/registry.py ← TransformationRegistry: "minmax" → MinMaxTransformation
ai/features/encoders/factory.py         ← FeatureEncoderFactory (5-step assembly)
database/factory.py                     ← connection_manager_factory
database/kvstore/factory.py             ← KVStoreFactory
integration/messaging/producers/factory.py
resilience/locking/factory.py           ← LockingFactory

The pattern in this codebase:

# ✅ CORRECT — factory + registry
class EngineFactory:
    def create(self, settings: EngineSettings) -> FeatureExecutionEngineProtocol:
        engine_cls = EngineRegistry.get(settings.engine)  # lookup by string key
        return engine_cls(settings)

# ✅ CORRECT — new engine registered without touching factory
EngineRegistry.register("ray", RayEngine)

When to apply: Any time you need to create objects whose concrete type is determined at runtime (from config, settings, or registry key).

Anti-pattern to avoid:

# ❌ WRONG — direct instantiation scattered everywhere
if settings.engine == "numpy":
    engine = AsyncNumpyEngine(settings)
elif settings.engine == "torch":
    engine = TorchEngine(settings)
# Adding "ray" requires modifying THIS code AND finding all other places

---

2. Registry

What it solves: Maps string keys to implementations. Combined with Factory, enables extension without modification (OCP).

Where it is:

ai/features/engines/registry.py                  ← EngineRegistry
ai/features/transformations/registry.py          ← TransformationRegistry
ai/features/resilience/adaptive_concurrency/discipline/registry.py
ai/features/resilience/adaptive_concurrency/policies/registry.py

Pattern:

class TransformationRegistry:
    _registry: dict[str, type[BaseFeatureTransformation]] = {}

    @classmethod
    def register(cls, key: str, transformation_cls: type) -> None:
        cls._registry[key] = transformation_cls

    @classmethod
    def get(cls, key: str) -> type[BaseFeatureTransformation]:
        if key not in cls._registry:
            raise KeyError(f"No transformation registered for key '{key}'")
        return cls._registry[key]

    @classmethod
    def register_defaults(cls) -> None:
        cls.register("minmax", MinMaxTransformation)
        cls.register("zscore", ZScoreTransformation)
        ...

When to apply: When the set of implementations is open-ended and should be extensible by adding new registrations rather than modifying existing code.

---

3. Strategy

What it solves: Encapsulates a family of interchangeable algorithms. The context uses the algorithm through a Protocol, not a concrete class.

Where it is:

resilience/adaptive_concurrency/algorithms/      ← AIMD, Gradient2 strategies
resilience/adaptive_concurrency/discipline/      ← FIFO, LIFO queue strategies
resilience/adaptive_concurrency/policies/        ← reject, probabilistic overload strategies
integration/messaging/consumers/strategies/      ← consumption, orchestration strategies
integration/messaging/producers/strategies/      ← direct, delayed strategies

Pattern:

# Protocol defines the algorithm interface
class QueueDisciplineProtocol(Protocol):
    def push(self, item: RequestSlot) -> None: ...
    def pop(self) -> RequestSlot: ...
    def peek(self) -> RequestSlot | None: ...

# Strategies implement the Protocol
class FIFODiscipline:
    def push(self, item: RequestSlot) -> None:
        self._queue.append(item)
    def pop(self) -> RequestSlot:
        return self._queue.popleft()

class LIFODiscipline:
    def push(self, item: RequestSlot) -> None:
        self._stack.append(item)
    def pop(self) -> RequestSlot:
        return self._stack.pop()

# Context uses Protocol — doesn't know which concrete strategy it holds
class AdmissionQueue:
    def __init__(self, discipline: QueueDisciplineProtocol) -> None:
        self._discipline = discipline

When to apply: When a class needs to support multiple variants of the same behavior, and those variants should be interchangeable at runtime.

Anti-pattern to avoid:

# ❌ WRONG — behavior switch via flag or isinstance
class AdmissionQueue:
    def pop(self):
        if self._mode == "fifo":
            return self._queue.popleft()
        elif self._mode == "lifo":
            return self._stack.pop()
        # OCP violated: adding "priority" requires modifying this class

---

4. Decorator (Structural)

What it solves: Adds behavior to an object by wrapping it, without subclassing. Each decorator adds one concern. The core is clean.

Where it is:

ai/features/encoders/decorator_stack.py     ← EncoderDecoratorStack
ai/features/encoders/telemetry.py           ← TelemetryInstrumentedEncoder
ai/features/encoders/lineage.py             ← LineageTrackingEncoder
ai/features/encoders/admission.py           ← AdmissionGuardedEncoder
ai/features/caching/proxy.py               ← CachedTransformation (wraps any transformer)

The stack in this codebase:

TelemetryInstrumentedEncoder     ← outermost: OTel spans + Prometheus
  └── LineageTrackingEncoder     ← OpenLineage governance events
       └── AdmissionGuardedEncoder ← pre-checks memory before allocation
            └── CachedEncoder    ← batch-level cache (optional)
                 └── BackpressureAwareEncoder ← throttle (optional)
                      └── FeatureEncoderService  ← core: no cross-cutting concerns

Pattern:

class TelemetryInstrumentedEncoder:
    def __init__(self, inner: FeatureEncoderProtocol) -> None:
        self._inner = inner  # the next decorator or the core

    async def encode_tensor_batch(self, raw_states) -> FeatureResponse:
        with self._tracer.start_span("feature.encode"):
            result = await self._inner.encode_tensor_batch(raw_states)
            self._histogram.observe(time.perf_counter() - start)
            return result

When to apply: Cross-cutting concerns (logging, metrics, caching, security checks) that should be composable and independent of the core logic.

---

5. Protocol (Structural Subtyping)

What it solves: Defines a contract without coupling to an implementation. Enables Dependency Inversion across layers.

Where it is: Every layer boundary in this codebase.

ai/features/engines/protocol.py             ← FeatureExecutionEngineProtocol
ai/features/transformations/protocol.py     ← TransformerProtocol
ai/features/imputation/protocol.py          ← FeatureImputationProtocol
database/kvstore/protocol.py                ← KVStoreProtocol
resilience/locking/protocol.py              ← LockingProtocol
integration/messaging/consumers/protocol.py ← ConsumerProtocol

Critical rule: The Protocol is defined at the lower layer. The higher layer implements it structurally (no inheritance needed).

# ✅ CORRECT — Protocol at the consumer's layer (lower level)
# ai/features/quality/protocol.py
class DriftBaselineProtocol(Protocol):
    def get_feature_stats(
        self, vector_id: str, feature_key: str
    ) -> tuple[float | None, float | None]: ...

# Higher layer implements it without explicit inheritance
class DistributedQualityBaseline:
    # Structurally matches DriftBaselineProtocol — no 'implements' declaration
    def get_feature_stats(self, vector_id, feature_key):
        ...

Anti-pattern to avoid:

# ❌ WRONG — importing the concrete class from a HIGHER layer
# observability/metrics/feature_metrics.py
from nala.athomic.ai.features.quality.protocol import DriftBaselineProtocol
# observability/ depends on ai/ — inverted layer order — DIP violation

---

6. Template Method

What it solves: Defines the skeleton of an algorithm in a base class, letting subclasses override specific steps without changing the structure.

Where it is:

ai/features/transformations/base.py    ← BaseFeatureTransformation
ai/features/transformations/async_base.py ← AsyncBaseFeatureTransformation
athomic/services/base.py              ← BaseService (lifecycle)

Pattern:

class BaseFeatureTransformation(Generic[T_in]):
    # Template method — defines the structure, delegates to _execute_transform_batch
    async def transform_batch(self, values: Sequence[T_in]) -> TensorResult:
        if len(values) >= MIN_BATCH_THRESHOLD:
            return await self._run_in_process_pool(values)
        return await self._run_in_thread(values)

    # Abstract step — subclasses provide the actual logic
    @abstractmethod
    def _execute_transform_batch(self, values: Sequence[T_in]) -> TensorResult: ...

When to apply: When multiple classes share the same algorithm structure but differ in one or two steps.

---

7. Facade

What it solves: Provides a simplified interface to a complex subsystem. Clients use the Facade; they never reach into subsystem internals.

Where it is:

athomic/facade.py                         ← Athomic (the composition root)
ai/features/service.py                    ← FeaturePlatformService (ai layer facade)

Pattern:

# ✅ CORRECT — clients use FeaturePlatformService, not internal encoders
service = FeaturePlatformService(settings)
response = await service.encode("user_risk_v1", raw_state)

# ❌ WRONG — reaching into internals
registry = service._registry  # bypasses the facade contract
encoder = registry.get("user_risk_v1")  # internal — may change
result = await encoder._engine.execute(plan, context)  # internal

---

8. Observer (Event-Driven variant)

What it solves: Notifies multiple subscribers when a state change occurs, without the publisher knowing who's listening.

Where it is:

integration/messaging/producers/       ← Kafka producers (event publishers)
integration/messaging/consumers/       ← Kafka consumers (event subscribers)
database/outbox/                       ← Transactional outbox pattern

The outbox variant:

# Publish event atomically with the business transaction (outbox stores to Mongo)
await outbox.append(event)
# Background worker reads outbox → publishes to Kafka → consumers react
# Publisher and consumers are fully decoupled

---

When to Choose Which Pattern

Situation	Pattern
Need to create objects by string key	Factory Method + Registry
Multiple variants of an algorithm	Strategy
Adding cross-cutting concerns (logging, cache, auth)	Decorator
Decoupling layers (import direction)	Protocol (DIP)
Same algorithm structure, different steps	Template Method
Simplify complex subsystem API	Facade
React to events without tight coupling	Observer / Event Bus
Group related config params	Parameter Object (dataclass)
Build complex objects step by step	Builder

---

Anti-Patterns Banned in This Codebase

Anti-pattern	Why banned	Correct pattern
isinstance chain (3+)	OCP violation	Strategy + Registry
TYPE_CHECKING import	Hides broken architecture	Move to correct layer
Direct new ConcreteClass() in business logic	DIP violation	Factory + Protocol
10+ parameter constructors	Long Parameter List smell	Config dataclass
God class (>800 lines)	SRP violation	Extract Class
print() in production	Bypasses observability	logger.*
Bare except: pass	Swallows errors	Specific exception + log

---

Related Documents

• SOLID rules: docs/guides/architecture/solid-rules.md
• Code smells: docs/guides/architecture/code-smells.md
• Violations backlog: docs/backlog/todo/solid-violations.md