Consolidate functional observability decorators

AGENTS.md

@@ -198,6 +198,10 @@
 - Keep constructors to dependency and configuration assignment. Put startup
   parsing, object-graph construction, defaults, validation across components,
   and any other non-trivial creation logic in explicit composition factories.
+- Implement technical boundary decoration once over JDK functional interfaces
+  and adapt business-named ports through method references. Do not make domain
+  ports extend `Consumer` or `Function` merely to simplify a decorator, and do
+  not create one decorator class for every port shape.
 - Test the shared application through its public API with injected adapters and
   deterministic dependencies. Give adapter implementations focused contract
   tests and give composition factories a small wiring smoke test. A trivial

JOURNAL.md

@@ -201,3 +201,12 @@ maintenance updates. Direct-push CI now runs only for `main`; pull requests keep
 their own unit and integration run without a duplicate branch-push run. The
 license file matches GitHub's canonical Apache 2.0 template exactly, and
 hosting metadata identifies it as `Apache-2.0`.
+
+## 2026-08-10 — Consolidate functional boundary observation
+
+Replaced one observability class per port with a single package-private
+operation observer over JDK `Consumer`, `IntFunction`, `Supplier`, and
+`ToLongFunction`. The public observability factory adapts business port methods
+by method reference; the ports retain their business vocabulary and do not
+inherit generic JDK function names. Receiver successes now use the same debug
+logging path as submission, handling, and sending.

README.md

@@ -44,8 +44,9 @@ The modules have intentionally narrow responsibilities:
 - `adapters.messaging.postgresql` implements the transactional outbox.
 - `adapters.messaging.artemis` implements durable queues, topics,
   subscriptions, and polling with the Artemis Core client.
-- `observability` decorates API and messaging boundaries with JDK logging,
-  concurrent counters, latency histograms, immutable snapshots, and JMX.
+- `observability` adapts business ports through JDK functional interfaces and
+  decorates them with JDK logging, concurrent counters, latency histograms,
+  immutable snapshots, and JMX.
 - `app` is the only composition root. Nothing depends on it.
 
 See [ARCHITECTURE.md](ARCHITECTURE.md) for exact production module
@@ -187,8 +188,8 @@ The project’s long-term target is a compact, executable example showing that:
   APIs;
 - push handlers and pull receivers can share minimal messaging ports;
 - durable retries can be explicit without embedding workflow logic in tables;
-- logging and metrics can be composed around boundaries without annotations or
-  framework-managed proxies; and
+- logging and metrics can be composed once around JDK functions, then adapted
+  to business ports without annotations or framework-managed proxies; and
 - new HTTP, database, broker, LocalStack, or test adapters can be added as new
   modules rather than edits to core rules.
 

observability/src/main/java/dev/minimalaccounting/observability/Observability.java

@@ -1,34 +1,76 @@
 package dev.minimalaccounting.observability;
 
+import dev.minimalaccounting.ports.api.Transaction;
 import dev.minimalaccounting.ports.api.TransactionProcessor;
 import dev.minimalaccounting.ports.messaging.Handler;
+import dev.minimalaccounting.ports.messaging.Received;
 import dev.minimalaccounting.ports.messaging.Receiver;
 import dev.minimalaccounting.ports.messaging.Sender;
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.IntFunction;
+
 @RequiredArgsConstructor
 public final class Observability {
     @NonNull
     private final MetricsRegistry metrics;
 
-    public TransactionProcessor processor(String name, TransactionProcessor processor) {
-        return new ObservedTransactionProcessor(name, processor, metrics, logger(name));
+    public TransactionProcessor processor(
+            @NonNull String name,
+            @NonNull TransactionProcessor processor) {
+        Consumer<Transaction> observed = observer(name).consumer(name, processor::submit);
+        return observed::accept;
+    }
+
+    public <T> Handler<T> handler(@NonNull String name, @NonNull Handler<T> handler) {
+        Consumer<T> observed = observer(name).consumer(name, handler::handle);
+        return observed::accept;
     }
 
-    public <T> Handler<T> handler(String name, Handler<T> handler) {
-        return new ObservedHandler<>(name, handler, metrics, logger(name));
+    public <T> Sender<T> sender(@NonNull String name, @NonNull Sender<T> sender) {
+        Consumer<T> observed = observer(name).consumer(name, sender::send);
+        return observed::accept;
     }
 
-    public <T> Sender<T> sender(String name, Sender<T> sender) {
-        return new ObservedSender<>(name, sender, metrics, logger(name));
+    public <T> Receiver<T> receiver(@NonNull String name, @NonNull Receiver<T> receiver) {
+        OperationObserver observer = observer(name);
+        return new FunctionalReceiver<>(
+                observer.intFunction(
+                        name + ".receive",
+                        receiver::receive,
+                        List::size),
+                observer.consumer(name + ".ack", receiver::ack),
+                observer.consumer(name + ".nack", receiver::nack));
     }
 
-    public <T> Receiver<T> receiver(String name, Receiver<T> receiver) {
-        return new ObservedReceiver<>(name, receiver, metrics, logger(name));
+    private OperationObserver observer(String name) {
+        return new OperationObserver(metrics, logger(name));
     }
 
     private static System.Logger logger(String name) {
         return System.getLogger("dev.minimalaccounting." + name);
     }
+
+    private record FunctionalReceiver<T>(
+            @NonNull IntFunction<List<Received<T>>> poll,
+            @NonNull Consumer<String> acknowledge,
+            @NonNull Consumer<String> reject) implements Receiver<T> {
+        @Override
+        public List<Received<T>> receive(int limit) {
+            return poll.apply(limit);
+        }
+
+        @Override
+        public void ack(String messageId) {
+            acknowledge.accept(messageId);
+        }
+
+        @Override
+        public void nack(String messageId) {
+            reject.accept(messageId);
+        }
+    }
 }

observability/src/main/java/dev/minimalaccounting/observability/OperationObserver.java

@@ -0,0 +1,56 @@
+package dev.minimalaccounting.observability;
+
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+
+import java.util.function.Consumer;
+import java.util.function.IntFunction;
+import java.util.function.Supplier;
+import java.util.function.ToLongFunction;
+
+@RequiredArgsConstructor
+final class OperationObserver {
+    @NonNull
+    private final MetricsRegistry metrics;
+    @NonNull
+    private final System.Logger logger;
+
+    <T> Consumer<T> consumer(
+            @NonNull String operation,
+            @NonNull Consumer<? super T> delegate) {
+        return argument -> observe(operation, () -> {
+            delegate.accept(argument);
+            return null;
+        }, ignored -> 1);
+    }
+
+    <T> IntFunction<T> intFunction(
+            @NonNull String operation,
+            @NonNull IntFunction<? extends T> delegate,
+            @NonNull ToLongFunction<? super T> itemCount) {
+        return argument -> observe(
+                operation,
+                () -> delegate.apply(argument),
+                itemCount);
+    }
+
+    private <T> T observe(
+            String operation,
+            Supplier<? extends T> invocation,
+            ToLongFunction<? super T> itemCount) {
+        long started = System.nanoTime();
+        try {
+            T result = invocation.get();
+            metrics.success(
+                    operation,
+                    System.nanoTime() - started,
+                    itemCount.applyAsLong(result));
+            logger.log(System.Logger.Level.DEBUG, "{0} succeeded", operation);
+            return result;
+        } catch (RuntimeException failure) {
+            metrics.failure(operation, System.nanoTime() - started);
+            logger.log(System.Logger.Level.WARNING, "{0} failed", operation);
+            throw failure;
+        }
+    }
+}