cleveragents-core/.opencode/skills/programming-patterns/references/architectural/hexagonal.md

Hexagonal Architecture (Ports & Adapters)

Problem

Application logic becomes tightly coupled to infrastructure concerns — databases, HTTP frameworks, message queues, external APIs. Testing requires spinning up real infrastructure. Swapping a database or switching from REST to gRPC requires rewriting business logic. The application cannot be driven by different actors (tests, CLI, web) without duplication.

Solution

Hexagonal Architecture (also known as Ports & Adapters) places the application core at the center, surrounded by ports and adapters:

• Core / Domain — Pure business logic with no knowledge of the outside world.
• Ports — Interfaces that define how the core interacts with the outside (inbound ports for driving, outbound ports for driven).
• Adapters — Concrete implementations that connect ports to infrastructure (HTTP adapter, database adapter, test adapter).

The dependency rule: adapters depend on ports, ports are defined by the core. The core depends on nothing external.

When to Use

• Applications with complex domain logic that must be infrastructure-agnostic.
• Systems that need to be testable without real infrastructure.
• When you anticipate swapping infrastructure (e.g., database migration).
• Multi-interface applications (web, CLI, message queue all drive the same core).

When to Avoid

• Simple CRUD apps where the indirection is not justified.
• Prototypes or throwaway code.
• When the team is small and the infrastructure is unlikely to change.

Pseudocode

// Inbound port (driving)
INTERFACE OrderUseCase:
    METHOD place_order(customer_id, items) -> order_id

// Outbound port (driven)
INTERFACE OrderStore:
    METHOD save(order)
    METHOD find_by_id(id) -> order

// Core implementation
CLASS OrderService IMPLEMENTS OrderUseCase:
    CONSTRUCTOR(store: OrderStore)
    METHOD place_order(customer_id, items):
        order = Order(customer_id, items)
        store.save(order)
        RETURN order.id

// Adapter: HTTP inbound
CLASS HTTPAdapter:
    METHOD handle_post(request):
        result = use_case.place_order(request.customer_id, request.items)
        RETURN response(result)

// Adapter: Database outbound
CLASS PostgresOrderStore IMPLEMENTS OrderStore:
    METHOD save(order):
        INSERT INTO orders ...

Python

from abc import ABC, abstractmethod
from dataclasses import dataclass, field


# === DOMAIN (Core) ===

@dataclass
class Notification:
    recipient: str
    message: str
    channel: str = ""


# === PORTS ===

class NotificationSender(ABC):
    """Outbound port: how notifications are sent."""
    @abstractmethod
    def send(self, notification: Notification) -> bool: ...


class NotificationLogger(ABC):
    """Outbound port: how notifications are logged."""
    @abstractmethod
    def log(self, notification: Notification, success: bool) -> None: ...


class NotificationService(ABC):
    """Inbound port: use cases for the notification system."""
    @abstractmethod
    def notify(self, recipient: str, message: str) -> bool: ...

    @abstractmethod
    def get_log(self) -> list[str]: ...


# === CORE IMPLEMENTATION ===

class NotificationServiceImpl(NotificationService):
    def __init__(self, sender: NotificationSender, logger: NotificationLogger) -> None:
        self._sender = sender
        self._logger = logger

    def notify(self, recipient: str, message: str) -> bool:
        notification = Notification(recipient, message)
        success = self._sender.send(notification)
        self._logger.log(notification, success)
        return success

    def get_log(self) -> list[str]:
        if isinstance(self._logger, InMemoryLogger):
            return self._logger.entries
        return []


# === ADAPTERS ===

class ConsoleNotificationSender(NotificationSender):
    """Adapter: sends notification to console."""
    def send(self, notification: Notification) -> bool:
        print(f"    [ConsoleSender] Sending to {notification.recipient}: {notification.message}")
        return True


class FailingNotificationSender(NotificationSender):
    """Adapter: simulates failure."""
    def send(self, notification: Notification) -> bool:
        print(f"    [FailingSender] Failed to send to {notification.recipient}")
        return False


class InMemoryLogger(NotificationLogger):
    """Adapter: logs to in-memory list."""
    def __init__(self) -> None:
        self.entries: list[str] = []

    def log(self, notification: Notification, success: bool) -> None:
        status = "OK" if success else "FAILED"
        entry = f"[{status}] -> {notification.recipient}: {notification.message}"
        self.entries.append(entry)
        print(f"    [Logger] {entry}")


# === DRIVING ADAPTERS ===

class CLIAdapter:
    """Inbound adapter: simulates CLI interaction."""
    def __init__(self, service: NotificationService) -> None:
        self._service = service

    def run_command(self, recipient: str, message: str) -> None:
        print(f"  [CLI] Sending notification...")
        result = self._service.notify(recipient, message)
        print(f"  [CLI] Result: {'sent' if result else 'failed'}")


def main() -> None:
    # Wiring: compose adapters with core
    logger = InMemoryLogger()

    print("--- With console sender (success) ---")
    sender = ConsoleNotificationSender()
    service = NotificationServiceImpl(sender, logger)
    cli = CLIAdapter(service)

    cli.run_command("alice@example.com", "Your order shipped!")
    cli.run_command("bob@example.com", "Payment received.")

    print("\n--- With failing sender ---")
    fail_sender = FailingNotificationSender()
    fail_service = NotificationServiceImpl(fail_sender, logger)
    fail_cli = CLIAdapter(fail_service)

    fail_cli.run_command("charlie@example.com", "This will fail")

    print("\n--- Full log ---")
    for entry in logger.entries:
        print(f"  {entry}")


if __name__ == "__main__":
    main()

Output:

--- With console sender (success) ---
  [CLI] Sending notification...
    [ConsoleSender] Sending to alice@example.com: Your order shipped!
    [Logger] [OK] -> alice@example.com: Your order shipped!
  [CLI] Result: sent
  [CLI] Sending notification...
    [ConsoleSender] Sending to bob@example.com: Payment received.
    [Logger] [OK] -> bob@example.com: Payment received.
  [CLI] Result: sent

--- With failing sender ---
  [CLI] Sending notification...
    [FailingSender] Failed to send to charlie@example.com
    [Logger] [FAILED] -> charlie@example.com: This will fail
  [CLI] Result: failed

--- Full log ---
  [OK] -> alice@example.com: Your order shipped!
  [OK] -> bob@example.com: Payment received.
  [FAILED] -> charlie@example.com: This will fail

Go

package main

import "fmt"

// === DOMAIN ===

type Notification struct {
	Recipient string
	Message   string
}

// === PORTS ===

type NotificationSender interface {
	Send(n Notification) bool
}

type NotificationLogger interface {
	Log(n Notification, success bool)
}

type NotificationService interface {
	Notify(recipient, message string) bool
	GetLog() []string
}

// === CORE ===

type notificationServiceImpl struct {
	sender NotificationSender
	logger *InMemoryLogger
}

func NewNotificationService(sender NotificationSender, logger *InMemoryLogger) NotificationService {
	return &notificationServiceImpl{sender: sender, logger: logger}
}

func (s *notificationServiceImpl) Notify(recipient, message string) bool {
	n := Notification{recipient, message}
	success := s.sender.Send(n)
	s.logger.Log(n, success)
	return success
}

func (s *notificationServiceImpl) GetLog() []string {
	return s.logger.Entries
}

// === ADAPTERS ===

type ConsoleSender struct{}

func (c *ConsoleSender) Send(n Notification) bool {
	fmt.Printf("    [ConsoleSender] Sending to %s: %s\n", n.Recipient, n.Message)
	return true
}

type FailingSender struct{}

func (f *FailingSender) Send(n Notification) bool {
	fmt.Printf("    [FailingSender] Failed to send to %s\n", n.Recipient)
	return false
}

type InMemoryLogger struct {
	Entries []string
}

func (l *InMemoryLogger) Log(n Notification, success bool) {
	status := "OK"
	if !success {
		status = "FAILED"
	}
	entry := fmt.Sprintf("[%s] -> %s: %s", status, n.Recipient, n.Message)
	l.Entries = append(l.Entries, entry)
	fmt.Printf("    [Logger] %s\n", entry)
}

// === DRIVING ADAPTER ===

type CLIAdapter struct {
	service NotificationService
}

func (c *CLIAdapter) RunCommand(recipient, message string) {
	fmt.Println("  [CLI] Sending notification...")
	result := c.service.Notify(recipient, message)
	if result {
		fmt.Println("  [CLI] Result: sent")
	} else {
		fmt.Println("  [CLI] Result: failed")
	}
}

func main() {
	logger := &InMemoryLogger{}

	fmt.Println("--- With console sender (success) ---")
	sender := &ConsoleSender{}
	service := NewNotificationService(sender, logger)
	cli := &CLIAdapter{service: service}

	cli.RunCommand("alice@example.com", "Your order shipped!")
	cli.RunCommand("bob@example.com", "Payment received.")

	fmt.Println("\n--- With failing sender ---")
	failSender := &FailingSender{}
	failService := NewNotificationService(failSender, logger)
	failCli := &CLIAdapter{service: failService}

	failCli.RunCommand("charlie@example.com", "This will fail")

	fmt.Println("\n--- Full log ---")
	for _, entry := range logger.Entries {
		fmt.Printf("  %s\n", entry)
	}
}

Output:

--- With console sender (success) ---
  [CLI] Sending notification...
    [ConsoleSender] Sending to alice@example.com: Your order shipped!
    [Logger] [OK] -> alice@example.com: Your order shipped!
  [CLI] Result: sent
  [CLI] Sending notification...
    [ConsoleSender] Sending to bob@example.com: Payment received.
    [Logger] [OK] -> bob@example.com: Payment received.
  [CLI] Result: sent

--- With failing sender ---
  [CLI] Sending notification...
    [FailingSender] Failed to send to charlie@example.com
    [Logger] [FAILED] -> charlie@example.com: This will fail
  [CLI] Result: failed

--- Full log ---
  [OK] -> alice@example.com: Your order shipped!
  [OK] -> bob@example.com: Payment received.
  [FAILED] -> charlie@example.com: This will fail

Related Patterns

• Adapter — Each port in Hexagonal Architecture is implemented by an Adapter. The port is the interface; the Adapter is the concrete implementation.
• Dependency Injection — DI is the mechanism that plugs Adapters into ports at startup.
• Clean Architecture — Clean Architecture extends Hexagonal with explicit layer rules. Both enforce: domain core has no knowledge of external systems.
• Repository — Data storage is a port; Repository is the port interface; a concrete DB Adapter implements it.
• Facade — The application core can expose Facades as driving ports (inbound adapters call the Facade).