init: 클린 아키텍처 백엔드

2026-07-24 14:29:36 +09:00
parent 9eed16d097
commit 821fe00c32
971 changed files with 74769 additions and 1 deletions
@@ -0,0 +1,29 @@
+# adapter:outbound:messaging — messaging adapter
+
+## Registered identity
+
+- Module ID: `adapter-outbound-messaging`
+- Gradle path: `:adapter:outbound:messaging`
+- Focused test: `./gradlew :adapter:outbound:messaging:test --console=plain`
+- Runtime baseline: Java 21; repository framework baseline: Spring Boot 4.0.0.
+- Registry SSOT: `.harness/project/modules.yaml`.
+
+Package root: `dev.caskeleton.adapter.outbound.messaging`.
+
+## Responsibility
+
+- Implement outbound message publication and broker integration behind application/domain ports.
+- Own broker settings, serialization envelope, disabled/fail-safe technical modes, and outbox
+  publication adaptation.
+- Reuse `adapter:outbound:support` for shared technical concerns.
+
+## Boundaries
+
+- Allowed dependency edges come only from `.harness/project/modules.yaml`.
+- No inbound DTO/controller, persistence repository/entity, bootstrap, or sample dependency.
+- Do not hide use-case sequencing or business routing policy in broker adapters.
+
+## Tests
+
+Use unit/contract tests with fake broker senders. No real network or broker is used in focused tests;
+settings records receive binding/validation tests when configuration changes.
@@ -0,0 +1,44 @@
+# adapter:outbound:messaging — 설계 결정 참조
+
+메시징(broker publish + outbox) 아웃바운드 어댑터 모듈. 패키지 루트:
+`dev.caskeleton.adapter.outbound.messaging`. `:adapter:outbound:support` 에 의존해 공유
+correlation / fail-open 의존성 로깅을 재사용한다.
+
+허용/금지 의존 정책은 `src/build.gradle` 의
+`allowedProjectDependencies['adapter:outbound:messaging']` 항목이 SSOT 다(이 모듈은 아직 별도
+CLAUDE.md 를 두지 않았다). 이 문서는 코드 주석에서 덜어낸 **설계 결정의 근거**를 모아둔 참조용
+기록이다.
+
+## 모듈 개요
+
+application-core 포트(`MessagePublisher` / `OutboxMessagePublishPort`) 뒤에 두는 **선택형**
+연동 어댑터다. `@ConditionalOnProperty` 로 게이팅되고 기본 비활성이며, 비활성 바인딩은
+`Disabled*` 구현으로 fail-fast 한다(Layer 3). 무거운 broker SDK 는 의도적으로 classpath 에
+최소화하고, 실제 broker client(`KafkaSender`)는 포킹 프로젝트가 채우는 seam 이다.
+
+## 두 포트를 하나의 활성 broker 에 조립
+
+`MessagingConfig` 는 두 messaging 포트를 단일 활성 `MessageBroker` 위에 조립한다 — broker
+추가는 새 broker 구현 파일 추가만으로 끝나고 이 config 는 바뀌지 않는다.
+
+## broker 선택 검증
+
+`app.messaging.broker` 가 설정됐는데 `MessageBroker` 빈이 없으면 startup 을 명시적 메시지로
+실패시킨다(조용한 no-op 아님). settings 와 활성 빈의 `brokerId()` 불일치도 startup 실패다.
+
+## 비활성 sentinel 두 개를 분리한 이유
+
+`DisabledMessagePublisher` 와 `DisabledOutboxMessagePublisher` 는 별도 클래스다 — 한 클래스가
+두 포트를 모두 구현하면 `getBean(MessagePublisher.class)` 가 모호해진다.
+
+## OutboxEnvelopeJson — 손수 짠 JSON
+
+이 모듈은 `jackson-databind` 를 classpath 에 두지 않아(스켈레톤을 가볍게 유지) outbox envelope
+직렬화는 의존성 없는 손수 짠 JSON 이다.
+
+## MessagePublisher vs OutboxMessagePublishPort
+
+`MessagePublisher` 는 fail-open 어댑터-로컬 발행기로, 발행 실패를 correlationId 와 함께
+로깅하고 삼켜(→ `:adapter:outbound:support` 의 `FailOpenDependencyLogger`) outbox/retry 로
+위임하므로 core 5xx 가 되지 않는다. 내구성 있는 전달이 필요하면 `OutboxMessagePublishPort` 를
+쓴다. 반환 타입을 void 로 둬 broker SDK 타입이 어댑터 밖으로 새지 않는다(B7).
@@ -0,0 +1,14 @@
+plugins { id 'groovy' }
+dependencies {
+    implementation project(':domain-core')
+    implementation project(':application-core')
+    implementation project(':shared-contract')
+    implementation project(':adapter:outbound:support')
+
+    implementation 'org.springframework.boot:spring-boot-autoconfigure'
+    implementation 'org.slf4j:slf4j-api'
+
+    testImplementation 'org.spockframework:spock-core:2.4-groovy-5.0'
+}
+tasks.withType(GroovyCompile).configureEach { groovyOptions.encoding = 'UTF-8'; options.encoding = 'UTF-8' }
+tasks.withType(JavaCompile).configureEach { options.encoding = 'UTF-8' }
@@ -0,0 +1,156 @@
+# This is a Gradle generated file for dependency locking.
+# Manual edits can break the build and are not advised.
+# This file is expected to be part of source control.
+biz.aQute.bnd:biz.aQute.bnd.annotation:7.1.0=testCompileClasspath
+ch.qos.logback:logback-classic:1.5.21=runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+ch.qos.logback:logback-core:1.5.21=runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+com.fasterxml.jackson.core:jackson-annotations:2.20=testCompileClasspath,testRuntimeClasspath
+com.github.ben-manes.caffeine:caffeine:3.2.3=annotationProcessor,testAnnotationProcessor
+com.github.kevinstern:software-and-algorithms:1.0=annotationProcessor,testAnnotationProcessor
+com.github.spotbugs:spotbugs-annotations:4.10.2=spotbugs
+com.github.spotbugs:spotbugs-annotations:4.8.6=testCompileClasspath
+com.github.spotbugs:spotbugs:4.10.2=spotbugs
+com.github.stephenc.jcip:jcip-annotations:1.0-1=spotbugs
+com.google.auto.service:auto-service-annotations:1.0.1=annotationProcessor,testAnnotationProcessor
+com.google.auto.value:auto-value-annotations:1.9=annotationProcessor,testAnnotationProcessor
+com.google.auto:auto-common:1.2.2=annotationProcessor,testAnnotationProcessor
+com.google.code.findbugs:jsr305:3.0.2=checkstyle,spotbugs,testCompileClasspath
+com.google.code.gson:gson:2.13.2=spotbugs
+com.google.errorprone:error_prone_annotation:2.49.0=annotationProcessor,testAnnotationProcessor
+com.google.errorprone:error_prone_annotations:2.38.0=testCompileClasspath
+com.google.errorprone:error_prone_annotations:2.41.0=spotbugs
+com.google.errorprone:error_prone_annotations:2.47.0=checkstyle
+com.google.errorprone:error_prone_annotations:2.49.0=annotationProcessor,testAnnotationProcessor
+com.google.errorprone:error_prone_check_api:2.49.0=annotationProcessor,testAnnotationProcessor
+com.google.errorprone:error_prone_core:2.49.0=annotationProcessor,testAnnotationProcessor
+com.google.googlejavaformat:google-java-format:1.35.0=annotationProcessor,testAnnotationProcessor
+com.google.guava:failureaccess:1.0.3=annotationProcessor,checkstyle,testAnnotationProcessor
+com.google.guava:guava:33.5.0-jre=annotationProcessor,testAnnotationProcessor
+com.google.guava:guava:33.6.0-jre=checkstyle
+com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava=annotationProcessor,checkstyle,testAnnotationProcessor
+com.google.j2objc:j2objc-annotations:3.1=annotationProcessor,checkstyle,testAnnotationProcessor
+com.google.protobuf:protobuf-java:4.33.2=annotationProcessor,testAnnotationProcessor
+com.h3xstream.findsecbugs:findsecbugs-plugin:1.14.0=spotbugsPlugins
+com.jayway.jsonpath:json-path:2.9.0=testCompileClasspath,testRuntimeClasspath
+com.puppycrawl.tools:checkstyle:13.5.0=checkstyle
+com.vaadin.external.google:android-json:0.0.20131108.vaadin1=testCompileClasspath,testRuntimeClasspath
+commons-beanutils:commons-beanutils:1.11.0=checkstyle
+commons-collections:commons-collections:3.2.2=checkstyle
+commons-io:commons-io:2.21.0=spotbugs
+commons-logging:commons-logging:1.3.5=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+info.picocli:picocli:4.7.7=checkstyle
+io.github.eisop:dataflow-errorprone:3.41.0-eisop1=annotationProcessor,testAnnotationProcessor
+io.github.java-diff-utils:java-diff-utils:4.12=annotationProcessor,testAnnotationProcessor
+io.leangen.geantyref:geantyref:1.3.16=testRuntimeClasspath
+io.micrometer:micrometer-commons:1.16.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+io.micrometer:micrometer-observation:1.16.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+jakarta.activation:jakarta.activation-api:2.1.4=testCompileClasspath,testRuntimeClasspath
+jakarta.annotation:jakarta.annotation-api:3.0.0=runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+jakarta.xml.bind:jakarta.xml.bind-api:4.0.4=testCompileClasspath,testRuntimeClasspath
+javax.inject:javax.inject:1=annotationProcessor,testAnnotationProcessor
+jaxen:jaxen:2.0.0=spotbugs
+net.bytebuddy:byte-buddy-agent:1.17.8=testCompileClasspath,testRuntimeClasspath
+net.bytebuddy:byte-buddy:1.17.8=testCompileClasspath,testRuntimeClasspath
+net.minidev:accessors-smart:2.6.0=testCompileClasspath,testRuntimeClasspath
+net.minidev:json-smart:2.6.0=testCompileClasspath,testRuntimeClasspath
+net.sf.saxon:Saxon-HE:12.9=checkstyle,spotbugs
+org.antlr:antlr4-runtime:4.13.2=checkstyle
+org.apache.bcel:bcel:6.12.0=spotbugs
+org.apache.commons:commons-lang3:3.20.0=checkstyle,spotbugs
+org.apache.commons:commons-text:1.15.0=spotbugs
+org.apache.commons:commons-text:1.3=checkstyle
+org.apache.groovy:groovy-bom:5.0.2=testCompileClasspath,testRuntimeClasspath
+org.apache.groovy:groovy:5.0.2=testCompileClasspath,testRuntimeClasspath
+org.apache.httpcomponents:httpclient:4.5.13=checkstyle
+org.apache.httpcomponents:httpcore:4.4.16=checkstyle
+org.apache.logging.log4j:log4j-api:2.25.2=runtimeClasspath,spotbugs,testCompileClasspath,testRuntimeClasspath
+org.apache.logging.log4j:log4j-core:2.25.2=spotbugs
+org.apache.logging.log4j:log4j-to-slf4j:2.25.2=runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+org.apache.maven.doxia:doxia-core:1.12.0=checkstyle
+org.apache.maven.doxia:doxia-logging-api:1.12.0=checkstyle
+org.apache.maven.doxia:doxia-module-xdoc:1.12.0=checkstyle
+org.apache.maven.doxia:doxia-sink-api:1.12.0=checkstyle
+org.apache.tomcat.embed:tomcat-embed-core:11.0.14=testCompileClasspath,testRuntimeClasspath
+org.apache.tomcat.embed:tomcat-embed-el:11.0.14=testCompileClasspath,testRuntimeClasspath
+org.apache.tomcat.embed:tomcat-embed-websocket:11.0.14=testCompileClasspath,testRuntimeClasspath
+org.apache.xbean:xbean-reflect:3.7=checkstyle
+org.apiguardian:apiguardian-api:1.1.2=testCompileClasspath
+org.assertj:assertj-core:3.27.6=testCompileClasspath,testRuntimeClasspath
+org.awaitility:awaitility:4.3.0=testCompileClasspath,testRuntimeClasspath
+org.codehaus.plexus:plexus-classworlds:2.6.0=checkstyle
+org.codehaus.plexus:plexus-component-annotations:2.1.0=checkstyle
+org.codehaus.plexus:plexus-container-default:2.1.0=checkstyle
+org.codehaus.plexus:plexus-utils:3.3.0=checkstyle
+org.dom4j:dom4j:2.2.0=spotbugs
+org.hamcrest:hamcrest:3.0=testCompileClasspath,testRuntimeClasspath
+org.javassist:javassist:3.28.0-GA=checkstyle
+org.jspecify:jspecify:1.0.0=annotationProcessor,checkstyle,compileClasspath,runtimeClasspath,testAnnotationProcessor,testCompileClasspath,testRuntimeClasspath
+org.junit.jupiter:junit-jupiter-api:6.0.1=testCompileClasspath,testRuntimeClasspath
+org.junit.jupiter:junit-jupiter-engine:6.0.1=testRuntimeClasspath
+org.junit.jupiter:junit-jupiter-params:6.0.1=testCompileClasspath,testRuntimeClasspath
+org.junit.jupiter:junit-jupiter:6.0.1=testCompileClasspath,testRuntimeClasspath
+org.junit.platform:junit-platform-commons:6.0.1=testCompileClasspath,testRuntimeClasspath
+org.junit.platform:junit-platform-engine:6.0.1=testCompileClasspath,testRuntimeClasspath
+org.junit.platform:junit-platform-launcher:6.0.1=testRuntimeClasspath
+org.junit:junit-bom:6.0.1=testCompileClasspath,testRuntimeClasspath
+org.junit:junit-bom:6.1.0=spotbugs
+org.mockito:mockito-core:5.20.0=testCompileClasspath,testRuntimeClasspath
+org.mockito:mockito-junit-jupiter:5.20.0=testCompileClasspath,testRuntimeClasspath
+org.objenesis:objenesis:3.3=testRuntimeClasspath
+org.opentest4j:opentest4j:1.3.0=testCompileClasspath,testRuntimeClasspath
+org.osgi:org.osgi.annotation.bundle:2.0.0=testCompileClasspath
+org.osgi:org.osgi.annotation.versioning:1.1.2=testCompileClasspath
+org.osgi:org.osgi.resource:1.0.0=testCompileClasspath
+org.osgi:org.osgi.service.serviceloader:1.0.0=testCompileClasspath
+org.ow2.asm:asm-analysis:9.10.1=spotbugs
+org.ow2.asm:asm-commons:9.10.1=spotbugs
+org.ow2.asm:asm-tree:9.10.1=spotbugs
+org.ow2.asm:asm-util:9.10.1=spotbugs
+org.ow2.asm:asm:9.10.1=spotbugs
+org.ow2.asm:asm:9.7.1=testCompileClasspath,testRuntimeClasspath
+org.pcollections:pcollections:4.0.1=annotationProcessor,testAnnotationProcessor
+org.reflections:reflections:0.10.2=checkstyle
+org.skyscreamer:jsonassert:1.5.3=testCompileClasspath,testRuntimeClasspath
+org.slf4j:jul-to-slf4j:2.0.17=runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+org.slf4j:slf4j-api:2.0.17=compileClasspath,runtimeClasspath,spotbugs,spotbugsSlf4j,testCompileClasspath,testRuntimeClasspath
+org.slf4j:slf4j-simple:2.0.17=checkstyle,spotbugsSlf4j
+org.spockframework:spock-bom:2.4-groovy-5.0=testCompileClasspath,testRuntimeClasspath
+org.spockframework:spock-core:2.4-groovy-5.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-autoconfigure:4.0.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-http-client:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-http-converter:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-jackson:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-restclient:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-resttestclient:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-servlet:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-starter-jackson-test:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-starter-jackson:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-starter-logging:4.0.0=runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-starter-test:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-starter-tomcat-runtime:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-starter-tomcat:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-starter-webmvc-test:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-starter-webmvc:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-starter:4.0.0=runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-test-autoconfigure:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-test:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-tomcat:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-web-server:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-webmvc-test:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot-webmvc:4.0.0=testCompileClasspath,testRuntimeClasspath
+org.springframework.boot:spring-boot:4.0.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+org.springframework:spring-aop:7.0.1=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+org.springframework:spring-beans:7.0.1=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+org.springframework:spring-context:7.0.1=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+org.springframework:spring-core:7.0.1=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+org.springframework:spring-expression:7.0.1=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+org.springframework:spring-test:7.0.1=testCompileClasspath,testRuntimeClasspath
+org.springframework:spring-web:7.0.1=testCompileClasspath,testRuntimeClasspath
+org.springframework:spring-webmvc:7.0.1=testCompileClasspath,testRuntimeClasspath
+org.xmlresolver:xmlresolver:5.3.3=checkstyle,spotbugs
+org.xmlunit:xmlunit-core:2.10.4=testCompileClasspath,testRuntimeClasspath
+org.yaml:snakeyaml:2.5=runtimeClasspath,testCompileClasspath,testRuntimeClasspath
+tools.jackson.core:jackson-core:3.0.2=testCompileClasspath,testRuntimeClasspath
+tools.jackson.core:jackson-databind:3.0.2=testCompileClasspath,testRuntimeClasspath
+tools.jackson:jackson-bom:3.0.2=testCompileClasspath,testRuntimeClasspath
+empty=
@@ -0,0 +1,72 @@
+package dev.caskeleton.adapter.outbound.messaging;
+
+import dev.caskeleton.adapter.outbound.messaging.core.DisabledMessagePublisher;
+import dev.caskeleton.adapter.outbound.messaging.core.MessageBroker;
+import dev.caskeleton.adapter.outbound.messaging.core.MessagePublisher;
+import dev.caskeleton.adapter.outbound.messaging.core.OutboundMessagePublisher;
+import dev.caskeleton.adapter.outbound.messaging.outbox.DisabledOutboxMessagePublisher;
+import dev.caskeleton.adapter.outbound.messaging.outbox.OutboxMessagePublishAdapter;
+import dev.caskeleton.adapter.outbound.support.FailOpenDependencyLogger;
+import dev.caskeleton.application.outbox.OutboxMessagePublishPort;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * Assembles both messaging ports onto the single active {@link MessageBroker} (the cache
+ * central-assembly pattern). A broker is contributed as a
+ * {@code @ConditionalOnProperty(app.messaging.broker=<id>)}-gated bean, at most one active — adding
+ * a broker is new files only and this config never changes. With no active broker it binds the
+ * disabled sentinels (fail-fast); if set but no contributing bean exists, startup fails.
+ */
+@Configuration
+@EnableConfigurationProperties(MessagingSettings.class)
+public class MessagingConfig {
+
+  @Bean
+  public MessagePublisher messagePublisher(
+      ObjectProvider<MessageBroker> brokerProvider,
+      MessagingSettings settings,
+      FailOpenDependencyLogger dependencyLogger) {
+    MessageBroker active = resolveBroker(brokerProvider, settings);
+    return (active == null)
+        ? new DisabledMessagePublisher()
+        : new OutboundMessagePublisher(active, dependencyLogger);
+  }
+
+  @Bean
+  public OutboxMessagePublishPort outboxMessagePublishPort(
+      ObjectProvider<MessageBroker> brokerProvider,
+      MessagingSettings settings,
+      FailOpenDependencyLogger dependencyLogger) {
+    MessageBroker active = resolveBroker(brokerProvider, settings);
+    return (active == null)
+        ? new DisabledOutboxMessagePublisher()
+        : new OutboxMessagePublishAdapter(active, dependencyLogger);
+  }
+
+  private static MessageBroker resolveBroker(
+      ObjectProvider<MessageBroker> brokerProvider, MessagingSettings settings) {
+    if (settings.broker().isBlank()) {
+      return null;
+    }
+    MessageBroker active = brokerProvider.getIfAvailable();
+    if (active == null) {
+      throw new IllegalStateException(
+          "app.messaging.broker="
+              + settings.broker()
+              + " but no MessageBroker bean contributes that id — enable the broker template"
+              + " (supply its client seam) or unset app.messaging.broker");
+    }
+    if (!settings.broker().equals(active.brokerId())) {
+      throw new IllegalStateException(
+          "app.messaging.broker="
+              + settings.broker()
+              + " but the active MessageBroker reports brokerId '"
+              + active.brokerId()
+              + "'");
+    }
+    return active;
+  }
+}
@@ -0,0 +1,20 @@
+package dev.caskeleton.adapter.outbound.messaging;
+
+import dev.caskeleton.adapter.outbound.messaging.core.MessageBroker;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Messaging template selection. {@code app.messaging.broker=<brokerId>} chooses the single active
+ * {@link MessageBroker} (e.g. {@code kafka}); unset/blank = no broker = fail-fast on use (the
+ * disabled sentinels {@code DisabledMessagePublisher} / {@code DisabledOutboxMessagePublisher}).
+ *
+ * @param broker the active broker id, matched against {@link MessageBroker#brokerId()}; blank means
+ *     the messaging template is disabled (the default)
+ */
+@ConfigurationProperties(prefix = "app.messaging")
+public record MessagingSettings(String broker) {
+
+  public MessagingSettings {
+    broker = (broker == null) ? "" : broker.trim();
+  }
+}
@@ -0,0 +1,17 @@
+package dev.caskeleton.adapter.outbound.messaging.core;
+
+import dev.caskeleton.shared.error.AdapterDisabledException;
+
+/**
+ * Fail-fast {@link MessagePublisher} binding when no broker is active ({@code app.messaging.broker}
+ * unset — the default). Any publish throws {@link AdapterDisabledException} — never a silent no-op
+ * (mirrors the cache router's unbound fail-fast). Broker-agnostic; the outbox counterpart is {@code
+ * DisabledOutboxMessagePublisher}.
+ */
+public class DisabledMessagePublisher implements MessagePublisher {
+
+  @Override
+  public void publish(OutboundMessage message) {
+    throw new AdapterDisabledException("messaging");
+  }
+}
@@ -0,0 +1,20 @@
+package dev.caskeleton.adapter.outbound.messaging.core;
+
+/**
+ * SPI a forking project contributes to bind the messaging template to a real broker. The active
+ * broker is selected by {@code app.messaging.broker=<brokerId>}. The skeleton carries no broker SDK
+ * — it is supplied by the project that selects the broker.
+ */
+public interface MessageBroker {
+
+  /** Stable broker identifier matched against {@code app.messaging.broker}. */
+  String brokerId();
+
+  /**
+   * Sends a message to the broker. May throw on any transport/broker failure; the binding decorator
+   * applies the fail-open (general) or fail-closed (outbox) policy.
+   *
+   * @throws Exception on any send failure (handled by the binding decorator)
+   */
+  void send(OutboundMessage message) throws Exception;
+}
@@ -0,0 +1,13 @@
+package dev.caskeleton.adapter.outbound.messaging.core;
+
+/**
+ * Adapter-local fire-and-forget (fail-open) publishing port for the optional adapter template. Not
+ * use-case-facing ({@code adapter-outbound} type, so application-core cannot hold it) — a use case
+ * that needs durable delivery uses the application-core {@code OutboxMessagePublishPort}. The
+ * active binding is selected by {@code app.messaging.broker}. The {@code void} return keeps broker
+ * SDK types from escaping the adapter (B7).
+ */
+public interface MessagePublisher {
+
+  void publish(OutboundMessage message);
+}
@@ -0,0 +1,25 @@
+package dev.caskeleton.adapter.outbound.messaging.core;
+
+import java.util.Objects;
+
+/**
+ * Transport-neutral message a {@link MessagePublisher} emits. Carries only the routing key family
+ * ({@code topic}, partition {@code key}) and an already-serialized {@code payload} string — it
+ * deliberately does NOT depend on any broker SDK type, so the messaging port stays
+ * template-portable across Kafka and any future broker.
+ *
+ * @param topic logical destination / topic
+ * @param key partition / ordering key (may be empty, never null)
+ * @param payload serialized message body (the producer is responsible for serialization)
+ */
+public record OutboundMessage(String topic, String key, String payload) {
+
+  public OutboundMessage {
+    Objects.requireNonNull(topic, "topic");
+    Objects.requireNonNull(key, "key");
+    Objects.requireNonNull(payload, "payload");
+    if (topic.isBlank()) {
+      throw new IllegalArgumentException("topic must not be blank");
+    }
+  }
+}
@@ -0,0 +1,37 @@
+package dev.caskeleton.adapter.outbound.messaging.core;
+
+import dev.caskeleton.adapter.outbound.support.FailOpenDependencyLogger;
+
+/**
+ * General {@link MessagePublisher} binding (fail-open). Delegates to the active {@link
+ * MessageBroker}; on failure it logs with the correlation id and swallows — a broker outage must
+ * never turn a core use case into a 5xx (durable delivery is delegated to the outbox/retry path).
+ * The fail-closed counterpart is {@code OutboxMessagePublishAdapter}.
+ *
+ * <p>Broker-agnostic: the same decorator serves any {@link MessageBroker}, so adding a broker never
+ * touches this class.
+ */
+public class OutboundMessagePublisher implements MessagePublisher {
+
+  private static final String DEPENDENCY_TYPE = "messaging";
+
+  private final MessageBroker broker;
+  private final FailOpenDependencyLogger dependencyLogger;
+
+  public OutboundMessagePublisher(MessageBroker broker, FailOpenDependencyLogger dependencyLogger) {
+    this.broker = broker;
+    this.dependencyLogger = dependencyLogger;
+  }
+
+  @Override
+  public void publish(OutboundMessage message) {
+    try {
+      broker.send(message);
+      dependencyLogger.logSuccess(broker.brokerId(), DEPENDENCY_TYPE, "publish");
+    } catch (Exception ex) {
+      // fail-open: observe with correlationId, delegate durability to outbox/retry,
+      // do NOT propagate — the core use case must still succeed.
+      dependencyLogger.logFailure(broker.brokerId(), DEPENDENCY_TYPE, "publish", ex);
+    }
+  }
+}
@@ -0,0 +1,32 @@
+package dev.caskeleton.adapter.outbound.messaging.kafka;
+
+import dev.caskeleton.adapter.outbound.messaging.core.MessageBroker;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * Layer 1 gating for the Kafka broker template: registers the Kafka {@link MessageBroker} only when
+ * {@code app.messaging.broker=kafka}. The central {@code MessagingConfig} binds both messaging
+ * ports onto it.
+ *
+ * <p>Needs a project-supplied {@link KafkaSender} bean (the integration seam) and a non-empty
+ * {@code app.messaging.kafka.brokers}. Adding another broker is a new config like this one — this
+ * file never changes.
+ */
+@Configuration
+@EnableConfigurationProperties(KafkaAdapterSettings.class)
+public class KafkaAdapterConfig {
+
+  @Bean
+  @ConditionalOnProperty(name = "app.messaging.broker", havingValue = "kafka")
+  public MessageBroker kafkaMessageBroker(KafkaSender sender, KafkaAdapterSettings settings) {
+    if (settings.brokers().isEmpty()) {
+      throw new IllegalStateException(
+          "app.messaging.broker=kafka requires a non-empty app.messaging.kafka.brokers "
+              + "(CSV of host:port)");
+    }
+    return new KafkaMessageBroker(sender);
+  }
+}
@@ -0,0 +1,28 @@
+package dev.caskeleton.adapter.outbound.messaging.kafka;
+
+import java.util.List;
+import java.util.regex.Pattern;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Kafka broker tuning bound from {@code app.messaging.kafka.*}. Validation is format-only ({@code
+ * host:port} per entry); the "Kafka selected ⇒ brokers required" cross-field rule is enforced in
+ * {@code KafkaAdapterConfig}, so an empty list is valid at bind time.
+ *
+ * @param brokers CSV of {@code host:port} broker endpoints (each entry format-validated)
+ */
+@ConfigurationProperties(prefix = "app.messaging.kafka")
+public record KafkaAdapterSettings(List<String> brokers) {
+
+  private static final Pattern HOST_PORT = Pattern.compile("^[^:\\s]+:\\d{1,5}$");
+
+  public KafkaAdapterSettings {
+    brokers = (brokers == null) ? List.of() : List.copyOf(brokers);
+    for (String broker : brokers) {
+      if (!HOST_PORT.matcher(broker.trim()).matches()) {
+        throw new IllegalArgumentException(
+            "APP_MESSAGING_KAFKA_BROKERS entry '" + broker + "' is not host:port");
+      }
+    }
+  }
+}
@@ -0,0 +1,31 @@
+package dev.caskeleton.adapter.outbound.messaging.kafka;
+
+import dev.caskeleton.adapter.outbound.messaging.core.MessageBroker;
+import dev.caskeleton.adapter.outbound.messaging.core.OutboundMessage;
+
+/**
+ * Kafka contribution of the {@link MessageBroker} SPI (brokerId {@code "kafka"}). Delegates the raw
+ * send to the project-supplied {@link KafkaSender} seam; the fail-open (general) and fail-closed
+ * (outbox) policies are applied by the binding decorators in the messaging package, not here — so
+ * this class carries no policy and no Kafka SDK.
+ */
+public class KafkaMessageBroker implements MessageBroker {
+
+  private static final String BROKER_ID = "kafka";
+
+  private final KafkaSender sender;
+
+  public KafkaMessageBroker(KafkaSender sender) {
+    this.sender = sender;
+  }
+
+  @Override
+  public String brokerId() {
+    return BROKER_ID;
+  }
+
+  @Override
+  public void send(OutboundMessage message) throws Exception {
+    sender.send(message);
+  }
+}
@@ -0,0 +1,19 @@
+package dev.caskeleton.adapter.outbound.messaging.kafka;
+
+import dev.caskeleton.adapter.outbound.messaging.core.OutboundMessage;
+
+/**
+ * Integration seam the forking project implements to bind the Kafka template to a real producer.
+ * The skeleton carries no Kafka SDK dependency — it is added by the project that enables Kafka. A
+ * send failure may throw; the binding decorator applies the failure policy.
+ */
+@FunctionalInterface
+public interface KafkaSender {
+
+  /**
+   * Sends a message to the broker. May throw on any transport/broker failure.
+   *
+   * @throws Exception on any send failure (caught and handled fail-open by the publisher)
+   */
+  void send(OutboundMessage message) throws Exception;
+}
@@ -0,0 +1,20 @@
+package dev.caskeleton.adapter.outbound.messaging.outbox;
+
+import dev.caskeleton.application.outbox.OutboxEvent;
+import dev.caskeleton.application.outbox.OutboxMessagePublishPort;
+import dev.caskeleton.shared.error.AdapterDisabledException;
+
+/**
+ * Fail-fast {@link OutboxMessagePublishPort} binding when no broker is active ({@code
+ * app.messaging.broker} unset — the default). Any publish throws {@link AdapterDisabledException} —
+ * never a silent no-op. Kept separate from {@code DisabledMessagePublisher} so each disabled bean
+ * implements exactly one port (a single class implementing both makes {@code
+ * getBean(MessagePublisher.class)} ambiguous).
+ */
+public class DisabledOutboxMessagePublisher implements OutboxMessagePublishPort {
+
+  @Override
+  public void publish(OutboxEvent event) {
+    throw new AdapterDisabledException("messaging");
+  }
+}
@@ -0,0 +1,70 @@
+package dev.caskeleton.adapter.outbound.messaging.outbox;
+
+import dev.caskeleton.application.outbox.OutboxEvent;
+
+/**
+ * Hand-rolled, dependency-free JSON serialiser for the outbox envelope (no Jackson — the module
+ * deliberately keeps {@code jackson-databind} off its classpath).
+ *
+ * <p>{@link OutboxEvent#payload()} MUST already be a valid serialised JSON value; it is inserted
+ * verbatim (no escaping) — serialisation policy is owned by the {@code schema-serialization}
+ * branch, this class only assembles the envelope. All other string fields go through {@link
+ * #escape(String)}.
+ */
+public final class OutboxEnvelopeJson {
+
+  private OutboxEnvelopeJson() {}
+
+  public static String toJson(OutboxEvent event) {
+    return "{"
+        + "\"eventId\":\""
+        + escape(event.eventId())
+        + "\","
+        + "\"eventType\":\""
+        + escape(event.eventType())
+        + "\","
+        + "\"aggregateId\":\""
+        + escape(event.aggregateId())
+        + "\","
+        + "\"occurredAt\":\""
+        + escape(event.occurredAt().toString())
+        + "\","
+        + "\"correlationId\":\""
+        + escape(event.correlationId())
+        + "\","
+        + "\"idempotencyKey\":\""
+        + escape(event.idempotencyKey())
+        + "\","
+        + "\"payload\":"
+        + event.payload()
+        + "}";
+  }
+
+  /** Escapes a string for a JSON string literal (RFC 8259 §7). */
+  static String escape(String value) {
+    StringBuilder sb = new StringBuilder(value.length() + 4);
+    for (int i = 0; i < value.length(); i++) {
+      char c = value.charAt(i);
+      if (c == '\\') {
+        sb.append("\\\\");
+      } else if (c == '"') {
+        sb.append("\\\"");
+      } else if (c == '\b') {
+        sb.append("\\b");
+      } else if (c == '\t') {
+        sb.append("\\t");
+      } else if (c == '\n') {
+        sb.append("\\n");
+      } else if (c == '\f') {
+        sb.append("\\f");
+      } else if (c == '\r') {
+        sb.append("\\r");
+      } else if (c < 0x20) {
+        sb.append(String.format("\\u%04x", (int) c));
+      } else {
+        sb.append(c);
+      }
+    }
+    return sb.toString();
+  }
+}
@@ -0,0 +1,49 @@
+package dev.caskeleton.adapter.outbound.messaging.outbox;
+
+import dev.caskeleton.adapter.outbound.messaging.core.MessageBroker;
+import dev.caskeleton.adapter.outbound.messaging.core.OutboundMessage;
+import dev.caskeleton.adapter.outbound.support.FailOpenDependencyLogger;
+import dev.caskeleton.application.outbox.OutboxEvent;
+import dev.caskeleton.application.outbox.OutboxMessagePublishPort;
+
+/**
+ * Outbox {@link OutboxMessagePublishPort} binding (fail-closed). Maps the claimed {@link
+ * OutboxEvent} to an {@link OutboundMessage} and delegates to the active {@link MessageBroker}; on
+ * failure it logs and re-throws so the relay can drive the FAILED/DEAD transition (the documented
+ * fail-closed contract — contrast the fail-open general {@code OutboundMessagePublisher}).
+ *
+ * <p>Broker-agnostic: the same decorator serves any {@link MessageBroker}, so adding a broker never
+ * touches this class.
+ */
+public class OutboxMessagePublishAdapter implements OutboxMessagePublishPort {
+
+  private static final String DEPENDENCY_TYPE = "messaging";
+
+  private final MessageBroker broker;
+  private final FailOpenDependencyLogger dependencyLogger;
+
+  public OutboxMessagePublishAdapter(
+      MessageBroker broker, FailOpenDependencyLogger dependencyLogger) {
+    this.broker = broker;
+    this.dependencyLogger = dependencyLogger;
+  }
+
+  @Override
+  public void publish(OutboxEvent event) {
+    String envelope = OutboxEnvelopeJson.toJson(event);
+    OutboundMessage message = new OutboundMessage(event.eventType(), event.aggregateId(), envelope);
+    try {
+      broker.send(message);
+      dependencyLogger.logSuccess(broker.brokerId(), DEPENDENCY_TYPE, "publish");
+    } catch (RuntimeException ex) {
+      // fail-closed: log then propagate — the relay must observe this to drive FAILED/DEAD.
+      dependencyLogger.logFailure(broker.brokerId(), DEPENDENCY_TYPE, "publish", ex);
+      throw ex;
+    } catch (Exception ex) {
+      // Wrap checked exceptions; preserve cause so the relay can inspect it.
+      dependencyLogger.logFailure(broker.brokerId(), DEPENDENCY_TYPE, "publish", ex);
+      throw new RuntimeException(
+          "outbox publish failed for broker '" + broker.brokerId() + "'", ex);
+    }
+  }
+}
@@ -0,0 +1,110 @@
+package dev.caskeleton.adapter.outbound.messaging.core;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import dev.caskeleton.adapter.outbound.support.FailOpenDependencyLogger;
+import dev.caskeleton.adapter.outbound.support.OutboundCorrelation;
+import java.util.ArrayList;
+import java.util.List;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+
+/**
+ * General {@link OutboundMessagePublisher} = fail-open: delegates to the active {@link
+ * MessageBroker}; a broker failure is logged with the correlation id and swallowed (never
+ * propagated). Broker-agnostic — a fake broker stands in for any real broker.
+ */
+class OutboundMessagePublisherTest {
+
+  private ch.qos.logback.classic.Logger logbackLogger;
+  private ListAppender<ILoggingEvent> appender;
+  private FailOpenDependencyLogger dependencyLogger;
+
+  @BeforeEach
+  void setUp() {
+    logbackLogger = (ch.qos.logback.classic.Logger) LoggerFactory.getLogger("test.messaging");
+    appender = new ListAppender<>();
+    appender.start();
+    logbackLogger.addAppender(appender);
+    logbackLogger.setLevel(Level.DEBUG);
+    dependencyLogger = new FailOpenDependencyLogger(logbackLogger);
+  }
+
+  @AfterEach
+  void tearDown() {
+    logbackLogger.detachAppender(appender);
+    MDC.clear();
+  }
+
+  /** Fake broker (brokerId "kafka") capturing sends, optionally failing. */
+  private static final class FakeBroker implements MessageBroker {
+    final List<OutboundMessage> sent = new ArrayList<>();
+    private final RuntimeException failure;
+
+    FakeBroker() {
+      this.failure = null;
+    }
+
+    FakeBroker(RuntimeException failure) {
+      this.failure = failure;
+    }
+
+    @Override
+    public String brokerId() {
+      return "kafka";
+    }
+
+    @Override
+    public void send(OutboundMessage message) {
+      if (failure != null) {
+        throw failure;
+      }
+      sent.add(message);
+    }
+  }
+
+  @Test
+  void publishDelegatesToTheActiveBrokerOnSuccess() {
+    FakeBroker broker = new FakeBroker();
+    OutboundMessagePublisher publisher = new OutboundMessagePublisher(broker, dependencyLogger);
+    OutboundMessage message = new OutboundMessage("worklog-events", "wl-1", "{}");
+
+    publisher.publish(message);
+
+    assertThat(broker.sent).containsExactly(message);
+  }
+
+  @Test
+  void publishFailureIsFailOpenAndNeverThrows() {
+    FakeBroker broker = new FakeBroker(new IllegalStateException("broker unavailable"));
+    OutboundMessagePublisher publisher = new OutboundMessagePublisher(broker, dependencyLogger);
+
+    // fail-open: a broker outage must not propagate to the core use case.
+    assertThatCode(() -> publisher.publish(new OutboundMessage("t", "k", "p")))
+        .doesNotThrowAnyException();
+  }
+
+  @Test
+  void publishFailureLogCarriesCorrelationIdAndBrokerId() {
+    MDC.put(OutboundCorrelation.MDC_KEY, "corr-msg-1");
+    FakeBroker broker = new FakeBroker(new IllegalStateException("broker unavailable"));
+    OutboundMessagePublisher publisher = new OutboundMessagePublisher(broker, dependencyLogger);
+
+    publisher.publish(new OutboundMessage("t", "k", "p"));
+
+    ILoggingEvent event =
+        appender.list.stream().filter(e -> e.getLevel() == Level.WARN).findFirst().orElseThrow();
+    assertThat(event.getFormattedMessage())
+        .contains("correlation_id=\"corr-msg-1\"")
+        .contains("dependency_name=\"kafka\"")
+        .contains("dependency_type=\"messaging\"")
+        .contains("operation=\"publish\"");
+  }
+}
@@ -0,0 +1,41 @@
+package dev.caskeleton.adapter.outbound.messaging.kafka;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+class KafkaAdapterSettingsTest {
+
+  @Test
+  void emptyOrNullBrokersIsTolerated() {
+    // bound globally via @ConfigurationPropertiesScan even when Kafka is not the active
+    // broker — an empty list must bind cleanly (the "required" guard lives in the config).
+    assertThat(new KafkaAdapterSettings((List<String>) null).brokers()).isEmpty();
+    assertThat(new KafkaAdapterSettings(List.of()).brokers()).isEmpty();
+  }
+
+  @Test
+  void rejectsABrokerThatIsNotHostPort() {
+    assertThatThrownBy(() -> new KafkaAdapterSettings(List.of("not-a-broker")))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("host:port");
+  }
+
+  @Test
+  void acceptsCsvOfHostPortBrokers() {
+    KafkaAdapterSettings settings =
+        new KafkaAdapterSettings(List.of("broker-1:9092", "broker-2:9092"));
+    assertThat(settings.brokers()).containsExactly("broker-1:9092", "broker-2:9092");
+  }
+
+  @Test
+  void brokersListIsDefensivelyCopied() {
+    List<String> mutable = new ArrayList<>(List.of("broker-1:9092"));
+    KafkaAdapterSettings settings = new KafkaAdapterSettings(mutable);
+    mutable.add("broker-2:9092");
+    assertThat(settings.brokers()).containsExactly("broker-1:9092");
+  }
+}
@@ -0,0 +1,266 @@
+package dev.caskeleton.adapter.outbound.messaging.outbox;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import dev.caskeleton.adapter.outbound.messaging.core.MessageBroker;
+import dev.caskeleton.adapter.outbound.messaging.core.OutboundMessage;
+import dev.caskeleton.adapter.outbound.support.FailOpenDependencyLogger;
+import dev.caskeleton.adapter.outbound.support.OutboundCorrelation;
+import dev.caskeleton.application.outbox.OutboxEvent;
+import dev.caskeleton.application.outbox.OutboxEventStatus;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+
+/**
+ * Broker-agnostic outbox publish adapter (fail-closed). Covers:
+ *
+ * <ul>
+ *   <li>Success — correct OutboundMessage (topic=eventType, key=aggregateId, payload=envelope JSON)
+ *       sent to the active broker.
+ *   <li>Fail-closed — broker failure is logged then propagated (never swallowed); checked
+ *       exceptions are wrapped.
+ *   <li>Envelope JSON fields and escaping.
+ * </ul>
+ */
+class OutboxMessagePublishAdapterTest {
+
+  private ch.qos.logback.classic.Logger logbackLogger;
+  private ListAppender<ILoggingEvent> appender;
+  private FailOpenDependencyLogger dependencyLogger;
+
+  @BeforeEach
+  void setUp() {
+    logbackLogger = (ch.qos.logback.classic.Logger) LoggerFactory.getLogger("test.outbox");
+    appender = new ListAppender<>();
+    appender.start();
+    logbackLogger.addAppender(appender);
+    logbackLogger.setLevel(Level.DEBUG);
+    dependencyLogger = new FailOpenDependencyLogger(logbackLogger);
+  }
+
+  @AfterEach
+  void tearDown() {
+    logbackLogger.detachAppender(appender);
+    MDC.clear();
+  }
+
+  /** Fake broker (brokerId "kafka") capturing sends, optionally failing with a given throwable. */
+  private static final class FakeBroker implements MessageBroker {
+    final List<OutboundMessage> sent = new ArrayList<>();
+    private final Exception failure;
+
+    FakeBroker() {
+      this.failure = null;
+    }
+
+    FakeBroker(Exception failure) {
+      this.failure = failure;
+    }
+
+    @Override
+    public String brokerId() {
+      return "kafka";
+    }
+
+    @Override
+    public void send(OutboundMessage message) throws Exception {
+      if (failure != null) {
+        throw failure;
+      }
+      sent.add(message);
+    }
+  }
+
+  private OutboxEvent sampleEvent() {
+    return new OutboxEvent(
+        "evt-01",
+        "WorkLogReserved",
+        "wl-agg-1",
+        "{\"workLogId\":\"wl-01\"}",
+        Instant.parse("2024-01-02T03:04:05Z"),
+        "corr-abc",
+        "idem-xyz",
+        OutboxEventStatus.IN_FLIGHT,
+        1);
+  }
+
+  @Nested
+  class SuccessPath {
+
+    @Test
+    void publishSendsMessageWithCorrectTopicKeyAndEnvelopePayload() {
+      FakeBroker broker = new FakeBroker();
+      OutboxMessagePublishAdapter adapter =
+          new OutboxMessagePublishAdapter(broker, dependencyLogger);
+
+      adapter.publish(sampleEvent());
+
+      assertThat(broker.sent).hasSize(1);
+      OutboundMessage sent = broker.sent.get(0);
+      assertThat(sent.topic()).isEqualTo("WorkLogReserved");
+      assertThat(sent.key()).isEqualTo("wl-agg-1");
+    }
+
+    @Test
+    void publishEnvelopeContainsAllFields() {
+      FakeBroker broker = new FakeBroker();
+      OutboxMessagePublishAdapter adapter =
+          new OutboxMessagePublishAdapter(broker, dependencyLogger);
+
+      adapter.publish(sampleEvent());
+
+      String payload = broker.sent.get(0).payload();
+      assertThat(payload).contains("\"eventId\"");
+      assertThat(payload).contains("\"eventType\"");
+      assertThat(payload).contains("\"aggregateId\"");
+      assertThat(payload).contains("\"occurredAt\"");
+      assertThat(payload).contains("\"correlationId\"");
+      assertThat(payload).contains("\"idempotencyKey\"");
+      assertThat(payload).contains("\"payload\"");
+    }
+
+    @Test
+    void publishEnvelopeFieldValuesMatchEvent() {
+      FakeBroker broker = new FakeBroker();
+      OutboxMessagePublishAdapter adapter =
+          new OutboxMessagePublishAdapter(broker, dependencyLogger);
+
+      adapter.publish(sampleEvent());
+
+      String payload = broker.sent.get(0).payload();
+      assertThat(payload).contains("\"evt-01\"");
+      assertThat(payload).contains("\"WorkLogReserved\"");
+      assertThat(payload).contains("\"wl-agg-1\"");
+      assertThat(payload).contains("2024-01-02T03:04:05Z");
+      assertThat(payload).contains("\"corr-abc\"");
+      assertThat(payload).contains("\"idem-xyz\"");
+      assertThat(payload).contains("{\"workLogId\":\"wl-01\"}");
+    }
+  }
+
+  @Nested
+  class FailClosedPath {
+
+    @Test
+    void publishFailurePropagatesAsRuntimeException() {
+      FakeBroker broker = new FakeBroker(new IllegalStateException("broker down"));
+      OutboxMessagePublishAdapter adapter =
+          new OutboxMessagePublishAdapter(broker, dependencyLogger);
+
+      assertThatThrownBy(() -> adapter.publish(sampleEvent())).isInstanceOf(RuntimeException.class);
+    }
+
+    @Test
+    void publishFailureIsLoggedBeforePropagation() {
+      MDC.put(OutboundCorrelation.MDC_KEY, "corr-fail-1");
+      FakeBroker broker = new FakeBroker(new IllegalStateException("broker unavailable"));
+      OutboxMessagePublishAdapter adapter =
+          new OutboxMessagePublishAdapter(broker, dependencyLogger);
+
+      try {
+        adapter.publish(sampleEvent());
+      } catch (RuntimeException ignored) {
+        // expected
+      }
+
+      boolean warnLogged =
+          appender.list.stream()
+              .anyMatch(
+                  e ->
+                      e.getLevel() == Level.WARN
+                          && e.getFormattedMessage().contains("corr-fail-1"));
+      assertThat(warnLogged).as("Expected a WARN log with the correlationId on failure").isTrue();
+    }
+
+    @Test
+    void publishFailureLogCarriesDependencyAndOperation() {
+      FakeBroker broker = new FakeBroker(new RuntimeException("connection refused"));
+      OutboxMessagePublishAdapter adapter =
+          new OutboxMessagePublishAdapter(broker, dependencyLogger);
+
+      try {
+        adapter.publish(sampleEvent());
+      } catch (RuntimeException ignored) {
+        // expected
+      }
+
+      String msg =
+          appender.list.stream()
+              .filter(e -> e.getLevel() == Level.WARN)
+              .findFirst()
+              .map(ILoggingEvent::getFormattedMessage)
+              .orElse("");
+      assertThat(msg)
+          .contains("dependency_name=\"kafka\"")
+          .contains("dependency_type=\"messaging\"")
+          .contains("operation=\"publish\"");
+    }
+
+    @Test
+    void publishWrapsCheckedExceptionInRuntimeException() {
+      FakeBroker broker = new FakeBroker(new IOException("network error"));
+      OutboxMessagePublishAdapter adapter =
+          new OutboxMessagePublishAdapter(broker, dependencyLogger);
+
+      assertThatThrownBy(() -> adapter.publish(sampleEvent()))
+          .isInstanceOf(RuntimeException.class)
+          .hasCauseInstanceOf(IOException.class);
+    }
+  }
+
+  @Nested
+  class JsonEscape {
+
+    @Test
+    void envelopeEscapesDoubleQuotesInStringFields() {
+      OutboxEvent eventWithQuote =
+          new OutboxEvent(
+              "evt-02",
+              "Has\"Quote",
+              "agg-1",
+              "{}",
+              Instant.parse("2024-01-01T00:00:00Z"),
+              "corr-1",
+              "idem-1",
+              OutboxEventStatus.IN_FLIGHT,
+              1);
+      FakeBroker broker = new FakeBroker();
+      new OutboxMessagePublishAdapter(broker, dependencyLogger).publish(eventWithQuote);
+
+      assertThat(broker.sent.get(0).payload()).contains("Has\\\"Quote");
+    }
+
+    @Test
+    void envelopeInsertsPayloadRawWithoutDoubleEncoding() {
+      OutboxEvent event =
+          new OutboxEvent(
+              "evt-05",
+              "SomeEvent",
+              "agg-2",
+              "{\"nested\":{\"a\":1}}",
+              Instant.parse("2024-01-01T00:00:00Z"),
+              "corr-2",
+              "idem-2",
+              OutboxEventStatus.IN_FLIGHT,
+              1);
+      FakeBroker broker = new FakeBroker();
+      new OutboxMessagePublishAdapter(broker, dependencyLogger).publish(event);
+
+      String payload = broker.sent.get(0).payload();
+      assertThat(payload).contains("{\"nested\":{\"a\":1}}");
+      assertThat(payload).doesNotContain("\"{\\\"nested\\\"");
+    }
+  }
+}