merge: integrate messaging R2 polling producer

# Conflicts:
#	docs/superpowers/specs/2026-07-28-messaging-production-capability-design.md
#	src/app-bootstrap/gradle.lockfile
#	src/app-bootstrap/src/test/java/dev/caskeleton/bootstrap/architecture/ArchitectureViolationFixtureTest.java
#	src/build.gradle

src/application-core/CLAUDE.md

@@ -21,6 +21,8 @@ Package root: `dev.caskeleton.application`.
 - Own application transaction boundaries through the `TransactionPort` abstraction.
 - Expose framework-free invocation context through ports such as `CorrelationIdPort`; adapters own
   MDC or other concrete storage.
+- Own the framework-free semantic integration-event draft, validated-event value contract and
+  exact typed payload contribution SPI. This is the messaging semantic contract R1 boundary only.
 
 ## Allowed
 
@@ -46,6 +48,8 @@ Package root: `dev.caskeleton.application`.
 - Persistence-layer transaction annotations of any kind inside this module.
 - Diagnostic frameworks (`org.slf4j`, `java.util.logging`, Logback, Log4j, Micrometer). Express
   diagnostic intent through a specific outbound `*Port`; adapters own rendering.
+- Messaging provider/runtime types: physical topic, Kafka record or metadata, JSON tree/raw JSON
+  payload, serializer/schema-validator implementation, security topology and publication epoch.
 
 ## Contract types
 
@@ -63,6 +67,10 @@ Package root: `dev.caskeleton.application`.
 | `capability.UseCaseCapability` | Mandatory annotation on every concrete use case: declares `transactionMode`, `idempotency`, `repositoryAccess`, `externalOutboundAllowed`. |
 | `capability.Idempotency` | `IDEMPOTENT` / `KEYED` / `NOT_IDEMPOTENT`. |
 | `capability.RepositoryAccess` | `NONE` / `READ_REPOSITORY` / `WRITE_REPOSITORY`. |
+| `messaging.contract.IntegrationEventContractContribution<P>` | Closed exact-record payload type, canonical component order, local schema identity/hash and provider-neutral descriptor contribution. |
+| `messaging.event.IntegrationEventDraft<P>` | Typed semantic event before local encoding; never JSON, Kafka or persistence state. |
+| `messaging.event.ValidatedIntegrationEvent` | Stable semantic identities plus immutable exact encoded bytes and hashes, ready for a later durable append boundary. |
+| `messaging.event.IntegrationEventEncoderPort` | Framework-free local draft-to-validated-event boundary implemented by an outbound adapter. |
 
 ## Notification R1 application boundary
 
@@ -218,3 +226,6 @@ cd src
 ./gradlew :application-core:test
 ./gradlew :app-bootstrap:test --tests '*CleanArchitectureTest'
 ```
+
+The messaging types above establish semantic contract R1 only. They do not claim JSON Schema
+qualification, Kafka publication, durable outbox persistence or any messaging R2 capability.

src/application-core/README.md

@@ -22,6 +22,34 @@ Micrometer meter/tag 렌더링은 Redis adapter가 소유한다. 관측 실패
 
 ---
 
+## 메시징 semantic contract R1
+
+`messaging.contract`와 `messaging.event`는 feature가 integration event를 동적 JSON이나 provider
+타입으로 넘기지 않게 만드는 application 경계다.
+
+- `IntegrationPayload` 구현은 feature가 소유한 불변 typed record다.
+- `IntegrationEventContractContribution`은 contract ID와 payload version을 분리하고, exact final
+  record type token, 실제 record component 순서, repository-local schema resource/hash와
+  provider-neutral `ContractDescriptor`만 기여한다. assignable-type 탐색, `Class.forName`, Java
+  class-name routing, `Map`, raw JSON string/tree는 이 SPI에 들어오지 않는다.
+- `IntegrationEventDraft`는 canonical event/aggregate/order/correlation identity와 typed payload를
+  보유한다. tenant가 없는 모드도 null 대신 canonical system tenant scope를
+  `AggregateIdentity`에 넣어 dedupe/order identity가 PostgreSQL nullable uniqueness에 기대지
+  않게 한다.
+- `IntegrationEventEncoderPort` 뒤의 adapter가 deterministic encoding과 schema validation을
+  수행하고 `ValidatedIntegrationEvent`를 돌려준다. 결과는 logical destination, exact US-ASCII
+  partition key, exact encoded envelope bytes, schema/envelope hash와 catalog/binding revision을
+  defensive copy로 보존한다.
+- `ContractDescriptor`는 owner module, logical destination, serializer ID, ordering requirement,
+  payload/envelope byte limit, sensitivity classification, same-event requeue horizon만 표현한다.
+  physical topic, Kafka cluster/security topology는 deployment binding의 책임이다.
+
+이 단계의 완성 범위는 **framework-free semantic contract R1**이다. JSON Schema validator와
+deterministic writer, Kafka ACK producer, PostgreSQL outbox append/relay는 후속 R2 작업이며 여기서
+구현되었거나 검증됐다고 주장하지 않는다.
+
+---
+
 ## 유스케이스 계약 (usecase / command / query / capability)
 
 ### UseCase / CommandUseCase / QueryUseCase

src/application-core/src/main/java/dev/caskeleton/application/messaging/contract/ContractDescriptor.java

@@ -0,0 +1,57 @@
+package dev.caskeleton.application.messaging.contract;
+
+import java.time.Duration;
+
+/** Provider-neutral semantic metadata for one integration-event contract. */
+public record ContractDescriptor(
+    String ownerModule,
+    LogicalDestinationId logicalDestination,
+    String serializerId,
+    boolean orderingRequired,
+    int maximumPayloadBytes,
+    int maximumEnvelopeBytes,
+    SensitivityClassification sensitivityClassification,
+    Duration sameEventRequeueHorizon) {
+
+  private static final String SEMANTIC_ID = "[a-z][a-z0-9]*(?:[.-][a-z0-9]+)*";
+  private static final Duration MAXIMUM_REQUEUE_HORIZON = Duration.ofDays(365);
+
+  public ContractDescriptor {
+    if (ownerModule == null
+        || ownerModule.length() > 96
+        || !ownerModule.matches("[a-z][a-z0-9]*(?:-[a-z0-9]+)*")) {
+      throw new IllegalArgumentException("ownerModule must be a canonical module identifier");
+    }
+    if (logicalDestination == null) {
+      throw new IllegalArgumentException("logicalDestination must not be null");
+    }
+    if (serializerId == null || serializerId.length() > 96 || !serializerId.matches(SEMANTIC_ID)) {
+      throw new IllegalArgumentException("serializerId must be a canonical semantic identifier");
+    }
+    if (maximumPayloadBytes <= 0) {
+      throw new IllegalArgumentException("maximumPayloadBytes must be positive");
+    }
+    if (maximumEnvelopeBytes <= 0 || maximumEnvelopeBytes < maximumPayloadBytes) {
+      throw new IllegalArgumentException(
+          "maximumEnvelopeBytes must be positive and at least maximumPayloadBytes");
+    }
+    if (sensitivityClassification == null) {
+      throw new IllegalArgumentException("sensitivityClassification must not be null");
+    }
+    if (sameEventRequeueHorizon == null
+        || sameEventRequeueHorizon.isZero()
+        || sameEventRequeueHorizon.isNegative()
+        || sameEventRequeueHorizon.compareTo(MAXIMUM_REQUEUE_HORIZON) > 0) {
+      throw new IllegalArgumentException(
+          "sameEventRequeueHorizon must be positive and at most 365 days");
+    }
+  }
+
+  /** Closed vocabulary for payload handling policy. */
+  public enum SensitivityClassification {
+    PUBLIC,
+    INTERNAL,
+    CONFIDENTIAL,
+    RESTRICTED
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/messaging/contract/ContractId.java

@@ -0,0 +1,19 @@
+package dev.caskeleton.application.messaging.contract;
+
+/** Stable semantic contract identity. Payload versions are represented separately. */
+public record ContractId(String value) {
+
+  private static final int MAXIMUM_LENGTH = 160;
+  private static final String SEGMENT = "[a-z][a-z0-9]*(?:-[a-z0-9]+)*";
+  private static final String GRAMMAR = SEGMENT + "(?:\\." + SEGMENT + ")+";
+
+  public ContractId {
+    if (value == null
+        || value.length() > MAXIMUM_LENGTH
+        || !value.matches(GRAMMAR)
+        || value.matches(".*(?:\\.|-)v[0-9]+$")) {
+      throw new IllegalArgumentException(
+          "contractId must be a version-free canonical lower-case semantic identifier");
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/messaging/contract/IntegrationEventContractContribution.java

@@ -0,0 +1,25 @@
+package dev.caskeleton.application.messaging.contract;
+
+import java.util.List;
+
+/**
+ * Framework-free contribution to the closed integration-event contract catalog.
+ *
+ * @param <P> exact feature-owned payload record type
+ */
+public interface IntegrationEventContractContribution<P extends IntegrationPayload> {
+
+  ContractId contractId();
+
+  int payloadVersion();
+
+  Class<P> exactPayloadRecordType();
+
+  List<String> canonicalRecordComponentOrder();
+
+  SchemaResourceId payloadSchemaResource();
+
+  Sha256 payloadSchemaHash();
+
+  ContractDescriptor descriptor();
+}

src/application-core/src/main/java/dev/caskeleton/application/messaging/contract/IntegrationPayload.java

@@ -0,0 +1,4 @@
+package dev.caskeleton.application.messaging.contract;
+
+/** Marker for a feature-owned, immutable integration-event payload record. */
+public interface IntegrationPayload {}

src/application-core/src/main/java/dev/caskeleton/application/messaging/contract/LogicalDestinationId.java

@@ -0,0 +1,17 @@
+package dev.caskeleton.application.messaging.contract;
+
+/** Logical delivery-class identity; never a physical topic, cluster or Java class name. */
+public record LogicalDestinationId(String value) {
+
+  private static final String GRAMMAR = "[a-z][a-z0-9]*(?:-[a-z0-9]+)*";
+
+  public LogicalDestinationId {
+    if (value == null
+        || value.length() > 96
+        || !value.matches(GRAMMAR)
+        || value.matches(".*-v[0-9]+$")) {
+      throw new IllegalArgumentException(
+          "logicalDestinationId must be a version-free canonical lower-case semantic identifier");
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/messaging/contract/SchemaResourceId.java

@@ -0,0 +1,15 @@
+package dev.caskeleton.application.messaging.contract;
+
+/** Canonical repository-local JSON Schema resource identity. */
+public record SchemaResourceId(String value) {
+
+  private static final String GRAMMAR =
+      "contracts/messaging/[a-z][a-z0-9]*(?:[-.][a-z0-9]+)*/v[1-9][0-9]*\\.schema\\.json";
+
+  public SchemaResourceId {
+    if (value == null || value.length() > 256 || !value.matches(GRAMMAR)) {
+      throw new IllegalArgumentException(
+          "schemaResourceId must identify a versioned local contracts/messaging JSON Schema");
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/messaging/contract/Sha256.java

@@ -0,0 +1,38 @@
+package dev.caskeleton.application.messaging.contract;
+
+import java.util.Arrays;
+import java.util.HexFormat;
+
+/** Immutable SHA-256 digest value. */
+@SuppressWarnings("ArrayRecordComponent")
+public record Sha256(byte[] bytes) {
+
+  public static final int BYTE_LENGTH = 32;
+
+  public Sha256 {
+    if (bytes == null || bytes.length != BYTE_LENGTH) {
+      throw new IllegalArgumentException("SHA-256 digest must contain exactly 32 bytes");
+    }
+    bytes = bytes.clone();
+  }
+
+  @Override
+  public byte[] bytes() {
+    return bytes.clone();
+  }
+
+  @Override
+  public boolean equals(Object other) {
+    return this == other || (other instanceof Sha256 sha256 && Arrays.equals(bytes, sha256.bytes));
+  }
+
+  @Override
+  public int hashCode() {
+    return Arrays.hashCode(bytes);
+  }
+
+  @Override
+  public String toString() {
+    return HexFormat.of().formatHex(bytes);
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/messaging/event/AggregateIdentity.java

@@ -0,0 +1,24 @@
+package dev.caskeleton.application.messaging.event;
+
+/** Canonical non-null tenant and aggregate ordering identity. */
+public record AggregateIdentity(String tenantScope, String aggregateType, String aggregateId) {
+
+  public AggregateIdentity {
+    if (tenantScope == null
+        || tenantScope.length() > 96
+        || !tenantScope.matches("[a-z0-9]+(?:[._:-][a-z0-9]+)*")) {
+      throw new IllegalArgumentException(
+          "tenantScope must be a canonical non-null lower-case identifier");
+    }
+    if (aggregateType == null
+        || aggregateType.length() > 64
+        || !aggregateType.matches("[a-z][a-z0-9]*(?:[.-][a-z0-9]+)*")) {
+      throw new IllegalArgumentException("aggregateType must be a canonical lower-case identifier");
+    }
+    if (aggregateId == null
+        || aggregateId.length() > 160
+        || !aggregateId.matches("[A-Za-z0-9][A-Za-z0-9._:-]*")) {
+      throw new IllegalArgumentException("aggregateId must be a canonical bounded identifier");
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/messaging/event/AggregateOrder.java

@@ -0,0 +1,14 @@
+package dev.caskeleton.application.messaging.event;
+
+/** Total order within one aggregate identity. */
+public record AggregateOrder(long sequence, int eventIndex) {
+
+  public AggregateOrder {
+    if (sequence <= 0) {
+      throw new IllegalArgumentException("aggregate sequence must be positive");
+    }
+    if (eventIndex < 0) {
+      throw new IllegalArgumentException("eventIndex must be non-negative");
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/messaging/event/EventId.java

@@ -0,0 +1,14 @@
+package dev.caskeleton.application.messaging.event;
+
+/** Canonical integration-event identity. */
+public record EventId(String value) {
+
+  private static final String GRAMMAR = "[A-Za-z0-9][A-Za-z0-9._:-]*";
+
+  public EventId {
+    if (value == null || value.length() > 96 || !value.matches(GRAMMAR)) {
+      throw new IllegalArgumentException(
+          "eventId must be 1-96 US-ASCII characters matching " + "[A-Za-z0-9][A-Za-z0-9._:-]*");
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/messaging/event/IntegrationEventDraft.java

@@ -0,0 +1,53 @@
+package dev.caskeleton.application.messaging.event;
+
+import dev.caskeleton.application.messaging.contract.ContractId;
+import dev.caskeleton.application.messaging.contract.IntegrationPayload;
+import dev.caskeleton.application.messaging.contract.LogicalDestinationId;
+import java.time.Instant;
+import java.util.Optional;
+
+/** Feature-owned semantic event before local encoding and schema validation. */
+public record IntegrationEventDraft<P extends IntegrationPayload>(
+    EventId eventId,
+    ContractId contractId,
+    int payloadVersion,
+    LogicalDestinationId destinationId,
+    AggregateIdentity aggregate,
+    AggregateOrder order,
+    Instant occurredAt,
+    String correlationId,
+    Optional<String> causationId,
+    P featurePayload) {
+
+  private static final String CORRELATION_GRAMMAR = "[A-Za-z0-9][A-Za-z0-9._:-]*";
+
+  public IntegrationEventDraft {
+    if (eventId == null
+        || contractId == null
+        || destinationId == null
+        || aggregate == null
+        || order == null) {
+      throw new IllegalArgumentException("event identities and order must not be null");
+    }
+    if (payloadVersion <= 0) {
+      throw new IllegalArgumentException("payloadVersion must be positive");
+    }
+    if (occurredAt == null) {
+      throw new IllegalArgumentException("occurredAt must not be null");
+    }
+    requireCanonicalCorrelationIdentity("correlationId", correlationId);
+    if (causationId == null) {
+      throw new IllegalArgumentException("causationId Optional must not be null");
+    }
+    causationId.ifPresent(value -> requireCanonicalCorrelationIdentity("causationId", value));
+    if (featurePayload == null) {
+      throw new IllegalArgumentException("featurePayload must not be null");
+    }
+  }
+
+  private static void requireCanonicalCorrelationIdentity(String field, String value) {
+    if (value == null || value.length() > 96 || !value.matches(CORRELATION_GRAMMAR)) {
+      throw new IllegalArgumentException(field + " must be a canonical bounded US-ASCII identity");
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/messaging/event/IntegrationEventEncoderPort.java

@@ -0,0 +1,7 @@
+package dev.caskeleton.application.messaging.event;
+
+/** Local, framework-free boundary for deterministic encoding and contract validation. */
+public interface IntegrationEventEncoderPort {
+
+  ValidatedIntegrationEvent encode(IntegrationEventDraft<?> draft);
+}

src/application-core/src/main/java/dev/caskeleton/application/messaging/event/ValidatedIntegrationEvent.java

@@ -0,0 +1,162 @@
+package dev.caskeleton.application.messaging.event;
+
+import dev.caskeleton.application.messaging.contract.ContractId;
+import dev.caskeleton.application.messaging.contract.LogicalDestinationId;
+import dev.caskeleton.application.messaging.contract.Sha256;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.Objects;
+import java.util.Optional;
+
+/** Immutable, locally encoded integration event ready for durable append. */
+@SuppressWarnings("ArrayRecordComponent")
+public record ValidatedIntegrationEvent(
+    EventId eventId,
+    ContractId contractId,
+    int envelopeVersion,
+    int payloadVersion,
+    LogicalDestinationId logicalDestinationId,
+    AggregateIdentity aggregate,
+    AggregateOrder order,
+    Instant occurredAt,
+    String correlationId,
+    Optional<String> causationId,
+    String partitionKeyText,
+    byte[] partitionKeyBytes,
+    byte[] envelopeBytes,
+    String contentType,
+    Sha256 schemaSetHash,
+    Sha256 envelopeSha256,
+    Sha256 envelopeSchemaHash,
+    Sha256 payloadSchemaHash,
+    String contractCatalogRevision,
+    String destinationBindingRevision) {
+
+  private static final String CORRELATION_GRAMMAR = "[A-Za-z0-9][A-Za-z0-9._:-]*";
+  private static final String REVISION_GRAMMAR = "[a-z0-9][a-z0-9._:-]{0,95}";
+  private static final String CONTENT_TYPE_GRAMMAR = "[a-z0-9!#$&^_.+-]+/[a-z0-9!#$&^_.+-]+";
+
+  public ValidatedIntegrationEvent {
+    if (eventId == null
+        || contractId == null
+        || logicalDestinationId == null
+        || aggregate == null
+        || order == null) {
+      throw new IllegalArgumentException("stable event identities and order must not be null");
+    }
+    if (envelopeVersion <= 0 || payloadVersion <= 0) {
+      throw new IllegalArgumentException("envelopeVersion and payloadVersion must be positive");
+    }
+    if (occurredAt == null) {
+      throw new IllegalArgumentException("occurredAt must not be null");
+    }
+    requireCanonicalIdentity("correlationId", correlationId, CORRELATION_GRAMMAR);
+    if (causationId == null) {
+      throw new IllegalArgumentException("causationId Optional must not be null");
+    }
+    causationId.ifPresent(
+        value -> requireCanonicalIdentity("causationId", value, CORRELATION_GRAMMAR));
+    if (partitionKeyText == null || !partitionKeyText.matches("[0-9a-f]{64}")) {
+      throw new IllegalArgumentException(
+          "partitionKeyText must contain exactly 64 lower-case hexadecimal characters");
+    }
+    if (partitionKeyBytes == null
+        || !Arrays.equals(
+            partitionKeyBytes, partitionKeyText.getBytes(StandardCharsets.US_ASCII))) {
+      throw new IllegalArgumentException(
+          "partitionKeyBytes must exactly equal the US-ASCII partitionKeyText bytes");
+    }
+    if (envelopeBytes == null || envelopeBytes.length == 0) {
+      throw new IllegalArgumentException("envelopeBytes must not be null or empty");
+    }
+    if (contentType == null
+        || contentType.length() > 96
+        || !contentType.matches(CONTENT_TYPE_GRAMMAR)) {
+      throw new IllegalArgumentException("contentType must be a canonical bounded media type");
+    }
+    if (schemaSetHash == null
+        || envelopeSha256 == null
+        || envelopeSchemaHash == null
+        || payloadSchemaHash == null) {
+      throw new IllegalArgumentException("validated schema and envelope hashes must not be null");
+    }
+    requireCanonicalIdentity("contractCatalogRevision", contractCatalogRevision, REVISION_GRAMMAR);
+    requireCanonicalIdentity(
+        "destinationBindingRevision", destinationBindingRevision, REVISION_GRAMMAR);
+    partitionKeyBytes = partitionKeyBytes.clone();
+    envelopeBytes = envelopeBytes.clone();
+  }
+
+  @Override
+  public byte[] partitionKeyBytes() {
+    return partitionKeyBytes.clone();
+  }
+
+  @Override
+  public byte[] envelopeBytes() {
+    return envelopeBytes.clone();
+  }
+
+  @Override
+  public boolean equals(Object other) {
+    if (this == other) {
+      return true;
+    }
+    if (!(other instanceof ValidatedIntegrationEvent that)) {
+      return false;
+    }
+    return envelopeVersion == that.envelopeVersion
+        && payloadVersion == that.payloadVersion
+        && eventId.equals(that.eventId)
+        && contractId.equals(that.contractId)
+        && logicalDestinationId.equals(that.logicalDestinationId)
+        && aggregate.equals(that.aggregate)
+        && order.equals(that.order)
+        && occurredAt.equals(that.occurredAt)
+        && correlationId.equals(that.correlationId)
+        && causationId.equals(that.causationId)
+        && partitionKeyText.equals(that.partitionKeyText)
+        && Arrays.equals(partitionKeyBytes, that.partitionKeyBytes)
+        && Arrays.equals(envelopeBytes, that.envelopeBytes)
+        && contentType.equals(that.contentType)
+        && schemaSetHash.equals(that.schemaSetHash)
+        && envelopeSha256.equals(that.envelopeSha256)
+        && envelopeSchemaHash.equals(that.envelopeSchemaHash)
+        && payloadSchemaHash.equals(that.payloadSchemaHash)
+        && contractCatalogRevision.equals(that.contractCatalogRevision)
+        && destinationBindingRevision.equals(that.destinationBindingRevision);
+  }
+
+  @Override
+  public int hashCode() {
+    int result =
+        Objects.hash(
+            eventId,
+            contractId,
+            envelopeVersion,
+            payloadVersion,
+            logicalDestinationId,
+            aggregate,
+            order,
+            occurredAt,
+            correlationId,
+            causationId,
+            partitionKeyText,
+            contentType,
+            schemaSetHash,
+            envelopeSha256,
+            envelopeSchemaHash,
+            payloadSchemaHash,
+            contractCatalogRevision,
+            destinationBindingRevision);
+    result = 31 * result + Arrays.hashCode(partitionKeyBytes);
+    return 31 * result + Arrays.hashCode(envelopeBytes);
+  }
+
+  private static void requireCanonicalIdentity(String field, String value, String grammar) {
+    if (value == null || !value.matches(grammar)) {
+      throw new IllegalArgumentException(field + " must be a canonical bounded US-ASCII identity");
+    }
+  }
+}

src/application-core/src/test/java/dev/caskeleton/application/messaging/contract/IntegrationEventContractContributionTest.java

@@ -0,0 +1,217 @@
+package dev.caskeleton.application.messaging.contract;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.lang.reflect.Modifier;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.Test;
+
+class IntegrationEventContractContributionTest {
+
+  @Test
+  void contributionUsesAnExactFinalRecordAndCanonicalComponentOrder() {
+    TestContribution contribution = new TestContribution();
+
+    assertThat(contribution.exactPayloadRecordType()).isEqualTo(TestPayload.class);
+    assertThat(contribution.exactPayloadRecordType().isRecord()).isTrue();
+    assertThat(Modifier.isFinal(contribution.exactPayloadRecordType().getModifiers())).isTrue();
+    assertThat(contribution.canonicalRecordComponentOrder())
+        .containsExactlyElementsOf(
+            Arrays.stream(TestPayload.class.getRecordComponents())
+                .map(component -> component.getName())
+                .toList());
+    assertThat(contribution.contractId()).isEqualTo(new ContractId("test.event.created"));
+    assertThat(contribution.payloadVersion()).isEqualTo(1);
+    assertThat(contribution.payloadSchemaResource())
+        .isEqualTo(new SchemaResourceId("contracts/messaging/test.event.created/v1.schema.json"));
+    assertThat(contribution.payloadSchemaHash()).isEqualTo(new Sha256(new byte[32]));
+  }
+
+  @Test
+  void contributionSpiIsClosedAndDoesNotExposeDynamicPayloadOrRoutingApis() {
+    Set<String> methods =
+        Arrays.stream(IntegrationEventContractContribution.class.getDeclaredMethods())
+            .map(method -> method.getName())
+            .collect(Collectors.toSet());
+
+    assertThat(methods)
+        .containsExactlyInAnyOrder(
+            "contractId",
+            "payloadVersion",
+            "exactPayloadRecordType",
+            "canonicalRecordComponentOrder",
+            "payloadSchemaResource",
+            "payloadSchemaHash",
+            "descriptor");
+    assertThat(
+            Arrays.stream(IntegrationEventContractContribution.class.getDeclaredMethods())
+                .filter(method -> method.isDefault()))
+        .isEmpty();
+    assertThat(
+            Arrays.stream(IntegrationEventContractContribution.class.getDeclaredMethods())
+                .flatMap(
+                    method ->
+                        java.util.stream.Stream.concat(
+                            java.util.stream.Stream.of(method.getReturnType()),
+                            Arrays.stream(method.getParameterTypes()))))
+        .doesNotContain(Map.class, String.class);
+    assertThat(methods)
+        .noneMatch(
+            name ->
+                name.contains("assignable")
+                    || name.contains("className")
+                    || name.contains("json")
+                    || name.contains("tree"));
+  }
+
+  @Test
+  void contractAndDestinationIdentifiersAreStableSemanticIdsWithSeparateVersions() {
+    assertThat(new ContractId("portfolio.worklog.reserved").value())
+        .isEqualTo("portfolio.worklog.reserved");
+    assertThat(new LogicalDestinationId("portfolio-domain-events").value())
+        .isEqualTo("portfolio-domain-events");
+
+    assertThatThrownBy(() -> new ContractId("portfolio.worklog.reserved.v1"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new ContractId("dev.caskeleton.WorkLogReservedPayload"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new ContractId("portfolio-.worklog.reserved"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new ContractId("kafka://portfolio.domain-events.v1"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new LogicalDestinationId("portfolio.domain-events.v1"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new LogicalDestinationId("WorkLogReservedPayload"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new LogicalDestinationId("portfolio-domain-events-v1"))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  void schemaResourceIsAClosedLocalContractResource() {
+    assertThat(
+            new SchemaResourceId("contracts/messaging/test.event.created/v1.schema.json").value())
+        .isEqualTo("contracts/messaging/test.event.created/v1.schema.json");
+
+    assertThatThrownBy(() -> new SchemaResourceId("https://schemas.example/test.schema.json"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new SchemaResourceId("../test.schema.json"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new SchemaResourceId("contracts/messaging/test..event/v1.schema.json"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new SchemaResourceId("dev.caskeleton.TestPayload"))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  void sha256HasFixedLengthContentEqualityAndDefensiveCopies() {
+    byte[] bytes = new byte[32];
+    bytes[0] = 42;
+    Sha256 hash = new Sha256(bytes);
+    Sha256 equalHash = new Sha256(bytes.clone());
+
+    bytes[0] = 0;
+    byte[] exposed = hash.bytes();
+    exposed[0] = 0;
+
+    assertThat(hash).isEqualTo(equalHash);
+    assertThat(hash.hashCode()).isEqualTo(equalHash.hashCode());
+    assertThat(hash.bytes()[0]).isEqualTo((byte) 42);
+    assertThatThrownBy(() -> new Sha256(new byte[31])).isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new Sha256(null)).isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  void descriptorContainsOnlyBoundedProviderNeutralContractMetadata() {
+    ContractDescriptor descriptor = descriptor();
+
+    assertThat(descriptor.ownerModule()).isEqualTo("sample-portfolio");
+    assertThat(descriptor.logicalDestination())
+        .isEqualTo(new LogicalDestinationId("portfolio-domain-events"));
+    assertThat(descriptor.maximumPayloadBytes()).isEqualTo(64 * 1024);
+    assertThat(descriptor.maximumEnvelopeBytes()).isEqualTo(128 * 1024);
+    assertThat(descriptor.sameEventRequeueHorizon()).isEqualTo(Duration.ofDays(7));
+    assertThat(
+            Arrays.stream(ContractDescriptor.class.getRecordComponents())
+                .map(component -> component.getName()))
+        .noneMatch(
+            name ->
+                name.toLowerCase(Locale.ROOT).contains("topic")
+                    || name.toLowerCase(Locale.ROOT).contains("kafka")
+                    || name.toLowerCase(Locale.ROOT).contains("bootstrap")
+                    || name.toLowerCase(Locale.ROOT).contains("security"));
+
+    assertThatThrownBy(
+            () ->
+                new ContractDescriptor(
+                    "sample-portfolio",
+                    new LogicalDestinationId("portfolio-domain-events"),
+                    "json-schema-envelope-v1",
+                    true,
+                    0,
+                    128 * 1024,
+                    ContractDescriptor.SensitivityClassification.INTERNAL,
+                    Duration.ofDays(7)))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  private static ContractDescriptor descriptor() {
+    return new ContractDescriptor(
+        "sample-portfolio",
+        new LogicalDestinationId("portfolio-domain-events"),
+        "json-schema-envelope-v1",
+        true,
+        64 * 1024,
+        128 * 1024,
+        ContractDescriptor.SensitivityClassification.INTERNAL,
+        Duration.ofDays(7));
+  }
+
+  private record TestPayload(String workLogId, long revision) implements IntegrationPayload {}
+
+  private static final class TestContribution
+      implements IntegrationEventContractContribution<TestPayload> {
+
+    @Override
+    public ContractId contractId() {
+      return new ContractId("test.event.created");
+    }
+
+    @Override
+    public int payloadVersion() {
+      return 1;
+    }
+
+    @Override
+    public Class<TestPayload> exactPayloadRecordType() {
+      return TestPayload.class;
+    }
+
+    @Override
+    public List<String> canonicalRecordComponentOrder() {
+      return List.of("workLogId", "revision");
+    }
+
+    @Override
+    public SchemaResourceId payloadSchemaResource() {
+      return new SchemaResourceId("contracts/messaging/test.event.created/v1.schema.json");
+    }
+
+    @Override
+    public Sha256 payloadSchemaHash() {
+      return new Sha256(new byte[32]);
+    }
+
+    @Override
+    public ContractDescriptor descriptor() {
+      return IntegrationEventContractContributionTest.descriptor();
+    }
+  }
+}

src/application-core/src/test/java/dev/caskeleton/application/messaging/event/IntegrationEventDraftTest.java

@@ -0,0 +1,119 @@
+package dev.caskeleton.application.messaging.event;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import dev.caskeleton.application.messaging.contract.ContractId;
+import dev.caskeleton.application.messaging.contract.IntegrationPayload;
+import dev.caskeleton.application.messaging.contract.LogicalDestinationId;
+import java.time.Instant;
+import java.util.Optional;
+import org.junit.jupiter.api.Test;
+
+class IntegrationEventDraftTest {
+
+  @Test
+  void eventIdAcceptsOnlyBoundedCanonicalUsAscii() {
+    assertThat(new EventId("A0:event.id-1").value()).isEqualTo("A0:event.id-1");
+
+    assertThatThrownBy(() -> new EventId("")).isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new EventId("-event")).isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new EventId("event 한글")).isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new EventId("e".repeat(97)))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  void aggregateIdentityRequiresCanonicalNonNullTenantAndAggregateValues() {
+    AggregateIdentity identity = new AggregateIdentity("tenant-a", "worklog", "worklog-42");
+
+    assertThat(identity.tenantScope()).isEqualTo("tenant-a");
+    assertThatThrownBy(() -> new AggregateIdentity(null, "worklog", "worklog-42"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new AggregateIdentity(" ", "worklog", "worklog-42"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new AggregateIdentity("Tenant A", "worklog", "worklog-42"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new AggregateIdentity("tenant-", "worklog", "worklog-42"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new AggregateIdentity("tenant-a", "WorkLog", "worklog-42"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new AggregateIdentity("tenant-a", "worklog", " worklog-42"))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  void aggregateOrderRequiresPositiveSequenceAndNonNegativeEventIndex() {
+    assertThat(new AggregateOrder(1, 0)).isEqualTo(new AggregateOrder(1, 0));
+    assertThatThrownBy(() -> new AggregateOrder(0, 0)).isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new AggregateOrder(1, -1))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  void draftRequiresPositiveVersionTimeCanonicalCorrelationAndTypedPayload() {
+    TestPayload payload = new TestPayload("worklog-42");
+    IntegrationEventDraft<TestPayload> draft = draft(payload, Optional.of("cause-1"));
+
+    assertThat(draft.featurePayload()).isSameAs(payload);
+    assertThat(draft.occurredAt()).isEqualTo(Instant.parse("2026-07-28T05:10:30.123Z"));
+    assertThatThrownBy(
+            () ->
+                new IntegrationEventDraft<>(
+                    new EventId("event-1"),
+                    new ContractId("portfolio.worklog.reserved"),
+                    0,
+                    new LogicalDestinationId("portfolio-domain-events"),
+                    new AggregateIdentity("tenant-a", "worklog", "worklog-42"),
+                    new AggregateOrder(17, 0),
+                    Instant.parse("2026-07-28T05:10:30.123Z"),
+                    "corr-1",
+                    Optional.empty(),
+                    payload))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(
+            () ->
+                new IntegrationEventDraft<>(
+                    new EventId("event-1"),
+                    new ContractId("portfolio.worklog.reserved"),
+                    1,
+                    new LogicalDestinationId("portfolio-domain-events"),
+                    new AggregateIdentity("tenant-a", "worklog", "worklog-42"),
+                    new AggregateOrder(17, 0),
+                    null,
+                    "corr-1",
+                    Optional.empty(),
+                    payload))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> draft(payload, null)).isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> draft(null, Optional.empty()))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  void encoderPortOwnsOnlyTheProviderNeutralDraftToValidatedBoundary() throws Exception {
+    assertThat(
+            IntegrationEventEncoderPort.class
+                .getDeclaredMethod("encode", IntegrationEventDraft.class)
+                .getReturnType())
+        .isEqualTo(ValidatedIntegrationEvent.class);
+    assertThat(IntegrationEventEncoderPort.class.getDeclaredMethods()).hasSize(1);
+  }
+
+  private static IntegrationEventDraft<TestPayload> draft(
+      TestPayload payload, Optional<String> causationId) {
+    return new IntegrationEventDraft<>(
+        new EventId("event-1"),
+        new ContractId("portfolio.worklog.reserved"),
+        1,
+        new LogicalDestinationId("portfolio-domain-events"),
+        new AggregateIdentity("tenant-a", "worklog", "worklog-42"),
+        new AggregateOrder(17, 0),
+        Instant.parse("2026-07-28T05:10:30.123Z"),
+        "corr-1",
+        causationId,
+        payload);
+  }
+
+  private record TestPayload(String workLogId) implements IntegrationPayload {}
+}

src/application-core/src/test/java/dev/caskeleton/application/messaging/event/ValidatedIntegrationEventTest.java

@@ -0,0 +1,158 @@
+package dev.caskeleton.application.messaging.event;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import dev.caskeleton.application.messaging.contract.ContractId;
+import dev.caskeleton.application.messaging.contract.LogicalDestinationId;
+import dev.caskeleton.application.messaging.contract.Sha256;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.Locale;
+import java.util.Optional;
+import org.junit.jupiter.api.Test;
+
+class ValidatedIntegrationEventTest {
+
+  @Test
+  void validatedEventOwnsExactEncodedBytesThroughDefensiveCopies() {
+    byte[] keyBytes = "a".repeat(64).getBytes(StandardCharsets.US_ASCII);
+    byte[] envelopeBytes = "{\"envelopeVersion\":1}".getBytes(StandardCharsets.UTF_8);
+    ValidatedIntegrationEvent event = event(keyBytes, envelopeBytes);
+
+    keyBytes[0] = 'b';
+    envelopeBytes[0] = 'x';
+    byte[] exposedKey = event.partitionKeyBytes();
+    byte[] exposedEnvelope = event.envelopeBytes();
+    exposedKey[0] = 'c';
+    exposedEnvelope[0] = 'y';
+
+    assertThat(event.partitionKeyBytes())
+        .containsExactly("a".repeat(64).getBytes(StandardCharsets.US_ASCII));
+    assertThat(event.envelopeBytes())
+        .containsExactly("{\"envelopeVersion\":1}".getBytes(StandardCharsets.UTF_8));
+  }
+
+  @Test
+  void validatedEventValueEqualityUsesEncodedByteContents() {
+    ValidatedIntegrationEvent first =
+        event(
+            "a".repeat(64).getBytes(StandardCharsets.US_ASCII),
+            "{}".getBytes(StandardCharsets.UTF_8));
+    ValidatedIntegrationEvent equal =
+        event(
+            "a".repeat(64).getBytes(StandardCharsets.US_ASCII),
+            "{}".getBytes(StandardCharsets.UTF_8));
+
+    assertThat(first).isEqualTo(equal);
+    assertThat(first.hashCode()).isEqualTo(equal.hashCode());
+  }
+
+  @Test
+  void validatedEventRequiresPositiveVersionsAndMatchingCanonicalAsciiPartitionKey() {
+    byte[] envelope = "{}".getBytes(StandardCharsets.UTF_8);
+
+    assertThatThrownBy(
+            () ->
+                event(
+                    "b".repeat(64).getBytes(StandardCharsets.US_ASCII),
+                    envelope,
+                    "a".repeat(64),
+                    0,
+                    1))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(
+            () ->
+                event(
+                    "b".repeat(64).getBytes(StandardCharsets.US_ASCII),
+                    envelope,
+                    "a".repeat(64),
+                    1,
+                    0))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(
+            () ->
+                event(
+                    "b".repeat(64).getBytes(StandardCharsets.US_ASCII),
+                    envelope,
+                    "a".repeat(64),
+                    1,
+                    1))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(
+            () ->
+                event(
+                    "é".repeat(64).getBytes(StandardCharsets.UTF_8),
+                    envelope,
+                    "é".repeat(64),
+                    1,
+                    1))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  void validatedEventCarriesStableSemanticIdentityAndProviderNeutralEvidenceOnly() {
+    ValidatedIntegrationEvent event =
+        event(
+            "a".repeat(64).getBytes(StandardCharsets.US_ASCII),
+            "{}".getBytes(StandardCharsets.UTF_8));
+
+    assertThat(event.eventId()).isEqualTo(new EventId("event-1"));
+    assertThat(event.contractId()).isEqualTo(new ContractId("portfolio.worklog.reserved"));
+    assertThat(event.logicalDestinationId())
+        .isEqualTo(new LogicalDestinationId("portfolio-domain-events"));
+    assertThat(event.aggregate())
+        .isEqualTo(new AggregateIdentity("tenant-a", "worklog", "worklog-42"));
+    assertThat(event.order()).isEqualTo(new AggregateOrder(17, 0));
+    assertThat(event.contentType()).isEqualTo("application/json");
+    assertThat(event.envelopeSha256()).isEqualTo(new Sha256(new byte[32]));
+    assertThat(event.contractCatalogRevision()).isEqualTo("catalog-r1");
+    assertThat(event.destinationBindingRevision()).isEqualTo("binding-r1");
+
+    assertThat(
+            Arrays.stream(ValidatedIntegrationEvent.class.getRecordComponents())
+                .map(component -> component.getName()))
+        .noneMatch(
+            name ->
+                name.toLowerCase(Locale.ROOT).contains("topic")
+                    || name.toLowerCase(Locale.ROOT).contains("kafka")
+                    || name.toLowerCase(Locale.ROOT).contains("metadata")
+                    || name.toLowerCase(Locale.ROOT).contains("publicationepoch")
+                    || name.toLowerCase(Locale.ROOT).contains("validator"));
+  }
+
+  private static ValidatedIntegrationEvent event(byte[] keyBytes, byte[] envelopeBytes) {
+    return event(keyBytes, envelopeBytes, "a".repeat(64), 1, 1);
+  }
+
+  private static ValidatedIntegrationEvent event(
+      byte[] keyBytes,
+      byte[] envelopeBytes,
+      String keyText,
+      int envelopeVersion,
+      int payloadVersion) {
+    Sha256 hash = new Sha256(new byte[32]);
+    return new ValidatedIntegrationEvent(
+        new EventId("event-1"),
+        new ContractId("portfolio.worklog.reserved"),
+        envelopeVersion,
+        payloadVersion,
+        new LogicalDestinationId("portfolio-domain-events"),
+        new AggregateIdentity("tenant-a", "worklog", "worklog-42"),
+        new AggregateOrder(17, 0),
+        Instant.parse("2026-07-28T05:10:30.123Z"),
+        "corr-1",
+        Optional.of("cause-1"),
+        keyText,
+        keyBytes,
+        envelopeBytes,
+        "application/json",
+        hash,
+        hash,
+        hash,
+        hash,
+        "catalog-r1",
+        "binding-r1");
+  }
+}

src/application-core/src/test/java/dev/caskeleton/application/outbox/PublishPendingOutboxEventsUseCaseTest.java

@@ -7,6 +7,7 @@ import dev.caskeleton.application.transaction.TransactionPort;
 import java.time.Clock;
 import java.time.Duration;
 import java.time.Instant;
+import java.time.ZoneId;
 import java.time.ZoneOffset;
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
@@ -61,7 +62,7 @@ class PublishPendingOutboxEventsUseCaseTest {
   // ---- success path ----
 
   @Test
-  void successfulPublishTransitionsEventToPublished() {
+  void legacyVoidPublisherNormalReturnTransitionsEventToPublishedCharacterization() {
     OutboxEvent event = makeEvent("evt-1", "UserCreated", "agg-1", NOW.minusSeconds(60), 1);
     store.addClaimable(event);
 
@@ -93,7 +94,7 @@ class PublishPendingOutboxEventsUseCaseTest {
   // ---- transient failure path ----
 
   @Test
-  void transientFailureTransitionsEventToFailedWithBackoff() {
+  void senderExceptionTransitionsEventToFailedWithBackoffCharacterization() {
     OutboxEvent event = makeEvent("evt-fail", "UserCreated", "agg-1", NOW.minusSeconds(60), 1);
     store.addClaimable(event);
     publishPort.failOn("evt-fail", new RuntimeException("broker down"));
@@ -122,7 +123,7 @@ class PublishPendingOutboxEventsUseCaseTest {
   }
 
   @Test
-  void failureOnThirdAttemptTransitionsEventToDead() {
+  void senderExceptionAtRetryLimitTransitionsEventToDeadCharacterization() {
     // attemptCount=3 means this is the 3rd attempt — next failure should DEAD
     OutboxEvent event = makeEvent("evt-dead", "UserCreated", "agg-1", NOW.minusSeconds(60), 3);
     store.addClaimable(event);
@@ -202,8 +203,8 @@ class PublishPendingOutboxEventsUseCaseTest {
   // ---- markPublished failure — spec §엣지·실패·의존 semantics ----
 
   /**
-   * When {@code store.markPublished} throws after a SUCCESSFUL publish, the exception must
-   * propagate out of {@code handle()} rather than being caught and misclassified as a publish
+   * When {@code store.markPublished} throws after a normal void publisher return, the exception
+   * must propagate out of {@code handle()} rather than being caught and misclassified as a publish
    * failure (which would trigger FAILED/DEAD state machine and potentially dead-letter a
    * successfully-delivered event).
    *
@@ -213,19 +214,21 @@ class PublishPendingOutboxEventsUseCaseTest {
    *   <li>The exception propagates — it is NOT swallowed inside {@code publishOne}.
    *   <li>{@code markFailed} is NOT called for the event (no misclassification).
    *   <li>{@code markDead} is NOT called for the event (no misclassification).
-   *   <li>{@code publishPort.publish} was called exactly once.
-   *   <li>The row remains IN_FLIGHT and is recovered via the orphan visibility-timeout reclaim path
-   *       on the next tick — re-published → duplicate absorbed by consumer dedupe (at-least-once).
+   *   <li>{@code publishPort.publish} is not called again before the in-flight timeout.
+   *   <li>The row remains IN_FLIGHT and can be recovered via the orphan visibility-timeout reclaim
+   *       path on a later tick. Because the prior void publisher return is not broker
+   *       acknowledgement evidence, a later publish can be a duplicate.
    * </ul>
    */
   @Test
-  void markPublishedFailurePropagatesAndDoesNotMisclassifyAsPublishFailure() {
+  void legacyMarkPublishedFailureLeavesInFlightAndDuplicatePossibleCharacterization() {
     OutboxEvent event =
         makeEvent("evt-store-fail", "UserCreated", "agg-1", NOW.minusSeconds(60), 1);
 
-    ThrowingOnMarkPublishedStorePort throwingStore =
-        new ThrowingOnMarkPublishedStorePort(new RuntimeException("DB down on markPublished"));
-    throwingStore.addClaimable(event);
+    MutableClock mutableClock = new MutableClock(NOW, ZoneOffset.UTC);
+    TimeoutAwareThrowingOnMarkPublishedStorePort throwingStore =
+        new TimeoutAwareThrowingOnMarkPublishedStorePort(
+            event, NOW, new RuntimeException("DB down on markPublished"));
 
     PublishPendingOutboxEventsUseCase useCaseWithThrowingStore =
         new PublishPendingOutboxEventsUseCase(
@@ -234,7 +237,7 @@ class PublishPendingOutboxEventsUseCaseTest {
             reporter,
             tx,
             backoffPolicy,
-            clock,
+            mutableClock,
             BATCH_SIZE,
             IN_FLIGHT_TIMEOUT);
 
@@ -244,17 +247,36 @@ class PublishPendingOutboxEventsUseCaseTest {
         .isInstanceOf(RuntimeException.class)
         .hasMessage("DB down on markPublished");
 
-    // publish was called exactly once — the broker call succeeded.
+    // A normal void return records only that the legacy publisher call completed.
     assertThat(publishPort.publishedEvents).containsExactly("evt-store-fail");
+    assertThat(throwingStore.currentStatus()).isEqualTo(OutboxEventStatus.IN_FLIGHT);
+    assertThat(throwingStore.attemptCount()).isEqualTo(1);
+    assertThat(throwingStore.nextAttemptAt()).isEqualTo(NOW.plus(IN_FLIGHT_TIMEOUT));
 
     // No misclassification: the event must NOT be marked FAILED or DEAD.
-    assertThat(throwingStore.failedEvents)
-        .as("markFailed must NOT be called when only markPublished fails")
-        .doesNotContainKey("evt-store-fail");
-    assertThat(throwingStore.deadEvents)
-        .as("markDead must NOT be called when only markPublished fails")
-        .doesNotContain("evt-store-fail");
+    assertThat(throwingStore.markFailedCalls()).isZero();
+    assertThat(throwingStore.markDeadCalls()).isZero();
     assertThat(reporter.reports).isEmpty();
+
+    mutableClock.advance(IN_FLIGHT_TIMEOUT.minusNanos(1));
+    OutboxRelayResult beforeTimeout =
+        useCaseWithThrowingStore.handle(PublishPendingOutboxEventsCommand.INSTANCE);
+    assertThat(beforeTimeout.claimedCount()).isZero();
+    assertThat(publishPort.publishedEvents).containsExactly("evt-store-fail");
+    assertThat(throwingStore.attemptCount()).isEqualTo(1);
+
+    // Move strictly past the timeout. The orphan becomes eligible and the same event is sent again
+    // because no PUBLISHED transition was persisted after the first normal void return.
+    mutableClock.advance(Duration.ofNanos(2));
+    assertThatThrownBy(
+            () -> useCaseWithThrowingStore.handle(PublishPendingOutboxEventsCommand.INSTANCE))
+        .isInstanceOf(RuntimeException.class)
+        .hasMessage("DB down on markPublished");
+    assertThat(throwingStore.currentStatus()).isEqualTo(OutboxEventStatus.IN_FLIGHT);
+    assertThat(throwingStore.attemptCount()).isEqualTo(2);
+    assertThat(publishPort.publishedEvents)
+        .as("mark failure leaves a duplicate-possible retry window")
+        .containsExactly("evt-store-fail", "evt-store-fail");
   }
 
   /**
@@ -487,6 +509,95 @@ class PublishPendingOutboxEventsUseCaseTest {
     }
   }
 
+  static final class TimeoutAwareThrowingOnMarkPublishedStorePort implements OutboxStorePort {
+    private final OutboxEvent event;
+    private final RuntimeException markPublishedException;
+    private OutboxEventStatus currentStatus = OutboxEventStatus.PENDING;
+    private int attemptCount;
+    private Instant nextAttemptAt;
+    private int markFailedCalls;
+    private int markDeadCalls;
+
+    TimeoutAwareThrowingOnMarkPublishedStorePort(
+        OutboxEvent event, Instant firstEligibleAt, RuntimeException markPublishedException) {
+      this.event = event;
+      this.nextAttemptAt = firstEligibleAt;
+      this.markPublishedException = markPublishedException;
+    }
+
+    @Override
+    public List<OutboxEvent> claimBatch(int batchSize, Instant now, Duration inFlightTimeout) {
+      if (batchSize == 0
+          || currentStatus == OutboxEventStatus.PUBLISHED
+          || currentStatus == OutboxEventStatus.DEAD
+          || nextAttemptAt.isAfter(now)) {
+        return List.of();
+      }
+      currentStatus = OutboxEventStatus.IN_FLIGHT;
+      attemptCount++;
+      nextAttemptAt = now.plus(inFlightTimeout);
+      return List.of(
+          new OutboxEvent(
+              event.eventId(),
+              event.eventType(),
+              event.aggregateId(),
+              event.payload(),
+              event.occurredAt(),
+              event.correlationId(),
+              event.idempotencyKey(),
+              currentStatus,
+              attemptCount));
+    }
+
+    @Override
+    public void markPublished(String eventId) {
+      throw markPublishedException;
+    }
+
+    @Override
+    public void markFailed(String eventId, Instant retryAt) {
+      markFailedCalls++;
+      currentStatus = OutboxEventStatus.FAILED;
+      nextAttemptAt = retryAt;
+    }
+
+    @Override
+    public void markDead(String eventId) {
+      markDeadCalls++;
+      currentStatus = OutboxEventStatus.DEAD;
+    }
+
+    @Override
+    public Map<OutboxEventStatus, Long> countByStatus() {
+      return Map.of(currentStatus, 1L);
+    }
+
+    @Override
+    public Map<String, Long> oldestUnpublishedAgeSecondsByEventType(Instant now) {
+      return Map.of();
+    }
+
+    OutboxEventStatus currentStatus() {
+      return currentStatus;
+    }
+
+    int attemptCount() {
+      return attemptCount;
+    }
+
+    Instant nextAttemptAt() {
+      return nextAttemptAt;
+    }
+
+    int markFailedCalls() {
+      return markFailedCalls;
+    }
+
+    int markDeadCalls() {
+      return markDeadCalls;
+    }
+  }
+
   static final class ThrowingTransitionStorePort extends FakeOutboxStorePort {
     private final RuntimeException markFailedException;
     private final RuntimeException markDeadException;
@@ -566,6 +677,35 @@ class PublishPendingOutboxEventsUseCaseTest {
     }
   }
 
+  static final class MutableClock extends Clock {
+    private Instant current;
+    private final ZoneId zone;
+
+    MutableClock(Instant current, ZoneId zone) {
+      this.current = current;
+      this.zone = zone;
+    }
+
+    void advance(Duration duration) {
+      current = current.plus(duration);
+    }
+
+    @Override
+    public ZoneId getZone() {
+      return zone;
+    }
+
+    @Override
+    public Clock withZone(ZoneId requestedZone) {
+      return new MutableClock(current, requestedZone);
+    }
+
+    @Override
+    public Instant instant() {
+      return current;
+    }
+  }
+
   /** Deterministic RandomGenerator that always returns 0 — produces zero jitter. */
   static final class ZeroRandom implements RandomGenerator {
     @Override